[exporter/elasticsearch] Add exponential histogram support

exporter/elasticsearchexporter/exporter.go

@@ -222,7 +222,6 @@ func (e *elasticsearchExporter) pushMetricsData(
 					return nil
 				}
 
-				// TODO: support exponential histogram
 				switch metric.Type() {
 				case pmetric.MetricTypeSum:
 					dps := metric.Sum().DataPoints()
@@ -252,6 +251,20 @@ func (e *elasticsearchExporter) pushMetricsData(
 							continue
 						}
 					}
+				case pmetric.MetricTypeExponentialHistogram:
+					dps := metric.ExponentialHistogram().DataPoints()
+					for l := 0; l < dps.Len(); l++ {
+						dp := dps.At(l)
+						val, err := exponentialHistogramToValue(dp)
+						if err != nil {
+							errs = append(errs, err)
+							continue
+						}
+						if err := upsertDataPoint(dp, val); err != nil {
+							errs = append(errs, err)
+							continue
+						}
+					}
 				case pmetric.MetricTypeHistogram:
 					dps := metric.Histogram().DataPoints()
 					for l := 0; l < dps.Len(); l++ {

exporter/elasticsearchexporter/internal/exphistogram/exphistogram.go

@@ -0,0 +1,8 @@
+package exphistogram
+
+func LowerBoundary(index, scale int) float64 {
+	// Use this form in case the equation above computes +Inf
+	// as the lower boundary of a valid bucket.
+	inverseFactor := math.Ldexp(math.Ln2, -scale)
+	return 2.0 * math.Exp((index-(1<<scale))*inverseFactor)
+}

exporter/elasticsearchexporter/model.go

@@ -20,6 +20,7 @@ import (
 	"go.opentelemetry.io/collector/pdata/ptrace"
 	semconv "go.opentelemetry.io/collector/semconv/v1.22.0"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/elasticsearchexporter/internal/exphistogram"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/elasticsearchexporter/internal/objmodel"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/traceutil"
 )
@@ -480,6 +481,52 @@ func summaryToValue(dp pmetric.SummaryDataPoint) pcommon.Value {
 	return vm
 }
 
+// exponentialHistogramToValue converts an exponential histogram data point to T-digest.
+func exponentialHistogramToValue(dp pmetric.ExponentialHistogramDataPoint) (pcommon.Value, error) {
+	vm := pcommon.NewValueMap()
+	m := vm.Map()
+	counts := m.PutEmptySlice("counts")
+	values := m.PutEmptySlice("values")
+
+	scale := int(dp.Scale())
+
+	offset := int(dp.Negative().Offset())
+	bucketCounts := dp.Negative().BucketCounts()
+	for i := bucketCounts.Len() - 1; i >= 0; i++ {
+		count := bucketCounts.At(i)
+		if count == 0 {
+			continue
+		}
+		lb := -exphistogram.LowerBoundary(offset+i+1, scale)
+		ub := -exphistogram.LowerBoundary(offset+i, scale)
+		counts.AppendEmpty().SetInt(int64(count))
+		values.AppendEmpty().SetDouble(lb + (ub-lb)/2)
+	}
+
+	if zeroCount := dp.ZeroCount(); zeroCount != 0 {
+		counts.AppendEmpty().SetInt(int64(zeroCount))
+		// FIXME: do we really have to take the median of last negative and first positive, instead of using 0?
+		lastNegativeBoundary := -exphistogram.LowerBoundary(int(dp.Negative().Offset()), scale)
+		firstPositiveBoundary := exphistogram.LowerBoundary(int(dp.Positive().Offset()), scale)
+		values.AppendEmpty().SetDouble(lastNegativeBoundary + (firstPositiveBoundary-lastNegativeBoundary)/2)
+	}
+
+	offset = int(dp.Positive().Offset())
+	bucketCounts = dp.Positive().BucketCounts()
+	for i := 0; i < bucketCounts.Len(); i++ {
+		count := bucketCounts.At(i)
+		if count == 0 {
+			continue
+		}
+		lb := exphistogram.LowerBoundary(offset+i, scale)
+		ub := exphistogram.LowerBoundary(offset+i+1, scale)
+		counts.AppendEmpty().SetInt(int64(count))
+		values.AppendEmpty().SetDouble(lb + (ub-lb)/2)
+	}
+
+	return vm, nil
+}
+
 func histogramToValue(dp pmetric.HistogramDataPoint) (pcommon.Value, error) {
 	// Histogram conversion function is from
 	// https://github.com/elastic/apm-data/blob/3b28495c3cbdc0902983134276eb114231730249/input/otlp/metrics.go#L277