[exporter/awsemfexporter] add exponential histogram support

.chloggen/awsemf-exponential-histogram.yaml

@@ -0,0 +1,20 @@
+# Use this changelog template to create an entry for release notes.
+# If your change doesn't affect end users, such as a test fix or a tooling change,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: awsemfexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add exponential histogram support.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [22626]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:

exporter/awsemfexporter/datapoint.go

@@ -5,6 +5,7 @@ package awsemfexporter // import "github.com/open-telemetry/opentelemetry-collec
 
 import (
 	"fmt"
+	"math"
 	"strconv"
 	"time"
 
@@ -86,6 +87,13 @@ type histogramDataPointSlice struct {
 	pmetric.HistogramDataPointSlice
 }
 
+type exponentialHistogramDataPointSlice struct {
+	// TODO: Calculate delta value for count and sum value with exponential histogram
+	// https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/18245
+	deltaMetricMetadata
+	pmetric.ExponentialHistogramDataPointSlice
+}
+
 // summaryDataPointSlice is a wrapper for pmetric.SummaryDataPointSlice
 type summaryDataPointSlice struct {
 	deltaMetricMetadata
@@ -156,6 +164,88 @@ func (dps histogramDataPointSlice) CalculateDeltaDatapoints(i int, instrumentati
 	}}, true
 }
 
+// CalculateDeltaDatapoints retrieves the ExponentialHistogramDataPoint at the given index.
+func (dps exponentialHistogramDataPointSlice) CalculateDeltaDatapoints(idx int, instrumentationScopeName string, detailedMetrics bool) ([]dataPoint, bool) {
+	metric := dps.ExponentialHistogramDataPointSlice.At(idx)
+
+	scale := metric.Scale()
+	base := math.Pow(2, math.Pow(2, float64(-scale)))
+	arrayValues := []float64{}
+	arrayCounts := []float64{}
+	var bucketBegin float64
+	var bucketEnd float64
+
+	// Set mid-point of positive buckets in values/counts array.
+	positiveBuckets := metric.Positive()
+	positiveOffset := positiveBuckets.Offset()
+	positiveBucketCounts := positiveBuckets.BucketCounts()
+	bucketBegin = 0
+	bucketEnd = 0
+	for i := 0; i < positiveBucketCounts.Len(); i++ {
+		index := i + int(positiveOffset)
+		if bucketBegin == 0 {
+			bucketBegin = math.Pow(base, float64(index))
+		} else {
+			bucketBegin = bucketEnd
+		}
+		bucketEnd = math.Pow(base, float64(index+1))
+		metricVal := (bucketBegin + bucketEnd) / 2
+		count := positiveBucketCounts.At(i)
+		if count > 0 {
+			arrayValues = append(arrayValues, metricVal)
+			arrayCounts = append(arrayCounts, float64(count))
+		}
+	}
+
+	// Set count of zero bucket in values/counts array.
+	if metric.ZeroCount() > 0 {
+		arrayValues = append(arrayValues, 0)
+		arrayCounts = append(arrayCounts, float64(metric.ZeroCount()))
+	}
+
+	// Set mid-point of negative buckets in values/counts array.
+	// According to metrics spec, the value in histogram is expected to be non-negative.
+	// https://opentelemetry.io/docs/specs/otel/metrics/api/#histogram
+	// However, the negative support is defined in metrics data model.
+	// https://opentelemetry.io/docs/specs/otel/metrics/data-model/#exponentialhistogram
+	// The negative is also supported but only verified with unit test.
+
+	negativeBuckets := metric.Negative()
+	negativeOffset := negativeBuckets.Offset()
+	negativeBucketCounts := negativeBuckets.BucketCounts()
+	bucketBegin = 0
+	bucketEnd = 0
+	for i := 0; i < negativeBucketCounts.Len(); i++ {
+		index := i + int(negativeOffset)
+		if bucketEnd == 0 {
+			bucketEnd = -math.Pow(base, float64(index))
+		} else {
+			bucketEnd = bucketBegin
+		}
+		bucketBegin = -math.Pow(base, float64(index+1))
+		metricVal := (bucketBegin + bucketEnd) / 2
+		count := negativeBucketCounts.At(i)
+		if count > 0 {
+			arrayValues = append(arrayValues, metricVal)
+			arrayCounts = append(arrayCounts, float64(count))
+		}
+	}
+
+	return []dataPoint{{
+		name: dps.metricName,
+		value: &cWMetricHistogram{
+			Values: arrayValues,
+			Counts: arrayCounts,
+			Count:  metric.Count(),
+			Sum:    metric.Sum(),
+			Max:    metric.Max(),
+			Min:    metric.Min(),
+		},
+		labels:      createLabels(metric.Attributes(), instrumentationScopeName),
+		timestampMs: unixNanoToMilliseconds(metric.Timestamp()),
+	}}, true
+}
+
 // CalculateDeltaDatapoints retrieves the SummaryDataPoint at the given index and perform calculation with sum and count while retain the quantile value.
 func (dps summaryDataPointSlice) CalculateDeltaDatapoints(i int, instrumentationScopeName string, detailedMetrics bool) ([]dataPoint, bool) {
 	metric := dps.SummaryDataPointSlice.At(i)
@@ -263,6 +353,12 @@ func getDataPoints(pmd pmetric.Metric, metadata cWMetricMetadata, logger *zap.Lo
 			metricMetadata,
 			metric.DataPoints(),
 		}
+	case pmetric.MetricTypeExponentialHistogram:
+		metric := pmd.ExponentialHistogram()
+		dps = exponentialHistogramDataPointSlice{
+			metricMetadata,
+			metric.DataPoints(),
+		}
 	case pmetric.MetricTypeSummary:
 		metric := pmd.Summary()
 		// For summaries coming from the prometheus receiver, the sum and count are cumulative, whereas for summaries

exporter/awsemfexporter/datapoint_test.go

@@ -91,6 +91,34 @@ func generateTestHistogramMetric(name string) pmetric.Metrics {
 	return otelMetrics
 }
 
+func generateTestExponentialHistogramMetric(name string) pmetric.Metrics {
+	otelMetrics := pmetric.NewMetrics()
+	rs := otelMetrics.ResourceMetrics().AppendEmpty()
+	metrics := rs.ScopeMetrics().AppendEmpty().Metrics()
+	metric := metrics.AppendEmpty()
+	metric.SetName(name)
+	metric.SetUnit("Seconds")
+	exponentialHistogramMetric := metric.SetEmptyExponentialHistogram()
+
+	exponentialHistogramDatapoint := exponentialHistogramMetric.DataPoints().AppendEmpty()
+	exponentialHistogramDatapoint.SetCount(4)
+	exponentialHistogramDatapoint.SetSum(0)
+	exponentialHistogramDatapoint.SetMin(-4)
+	exponentialHistogramDatapoint.SetMax(4)
+	exponentialHistogramDatapoint.SetZeroCount(0)
+	exponentialHistogramDatapoint.SetScale(1)
+	exponentialHistogramDatapoint.Positive().SetOffset(1)
+	exponentialHistogramDatapoint.Positive().BucketCounts().FromRaw([]uint64{
+		1, 0, 1,
+	})
+	exponentialHistogramDatapoint.Negative().SetOffset(1)
+	exponentialHistogramDatapoint.Negative().BucketCounts().FromRaw([]uint64{
+		1, 0, 1,
+	})
+	exponentialHistogramDatapoint.Attributes().PutStr("label1", "value1")
+	return otelMetrics
+}
+
 func generateTestSummaryMetric(name string) pmetric.Metrics {
 	otelMetrics := pmetric.NewMetrics()
 	rs := otelMetrics.ResourceMetrics().AppendEmpty()
@@ -347,6 +375,106 @@ func TestCalculateDeltaDatapoints_HistogramDataPointSlice(t *testing.T) {
 
 }
 
+func TestCalculateDeltaDatapoints_ExponentialHistogramDataPointSlice(t *testing.T) {
+	deltaMetricMetadata := generateDeltaMetricMetadata(false, "foo", false)
+
+	testCases := []struct {
+		name              string
+		histogramDPS      pmetric.ExponentialHistogramDataPointSlice
+		expectedDatapoint dataPoint
+	}{
+		{
+			name: "Exponential histogram with min and max",
+			histogramDPS: func() pmetric.ExponentialHistogramDataPointSlice {
+				histogramDPS := pmetric.NewExponentialHistogramDataPointSlice()
+				histogramDP := histogramDPS.AppendEmpty()
+				histogramDP.SetCount(uint64(17))
+				histogramDP.SetSum(17.13)
+				histogramDP.SetMin(10)
+				histogramDP.SetMax(30)
+				histogramDP.Attributes().PutStr("label1", "value1")
+				return histogramDPS
+			}(),
+			expectedDatapoint: dataPoint{
+				name:   "foo",
+				value:  &cWMetricHistogram{Values: []float64{}, Counts: []float64{}, Sum: 17.13, Count: 17, Min: 10, Max: 30},
+				labels: map[string]string{oTellibDimensionKey: instrLibName, "label1": "value1"},
+			},
+		},
+		{
+			name: "Exponential histogram without min and max",
+			histogramDPS: func() pmetric.ExponentialHistogramDataPointSlice {
+				histogramDPS := pmetric.NewExponentialHistogramDataPointSlice()
+				histogramDP := histogramDPS.AppendEmpty()
+				histogramDP.SetCount(uint64(17))
+				histogramDP.SetSum(17.13)
+				histogramDP.Attributes().PutStr("label1", "value1")
+				return histogramDPS
+
+			}(),
+			expectedDatapoint: dataPoint{
+				name:   "foo",
+				value:  &cWMetricHistogram{Values: []float64{}, Counts: []float64{}, Sum: 17.13, Count: 17, Min: 0, Max: 0},
+				labels: map[string]string{oTellibDimensionKey: instrLibName, "label1": "value1"},
+			},
+		},
+		{
+			name: "Exponential histogram with buckets",
+			histogramDPS: func() pmetric.ExponentialHistogramDataPointSlice {
+				histogramDPS := pmetric.NewExponentialHistogramDataPointSlice()
+				histogramDP := histogramDPS.AppendEmpty()
+				histogramDP.Positive().BucketCounts().FromRaw([]uint64{1, 2, 3})
+				histogramDP.SetZeroCount(4)
+				histogramDP.Negative().BucketCounts().FromRaw([]uint64{1, 2, 3})
+				histogramDP.Attributes().PutStr("label1", "value1")
+				return histogramDPS
+			}(),
+			expectedDatapoint: dataPoint{
+				name:   "foo",
+				value:  &cWMetricHistogram{Values: []float64{1.5, 3, 6, 0, -1.5, -3, -6}, Counts: []float64{1, 2, 3, 4, 1, 2, 3}},
+				labels: map[string]string{oTellibDimensionKey: instrLibName, "label1": "value1"},
+			},
+		},
+		{
+			name: "Exponential histogram with different scale/offset/labels",
+			histogramDPS: func() pmetric.ExponentialHistogramDataPointSlice {
+				histogramDPS := pmetric.NewExponentialHistogramDataPointSlice()
+				histogramDP := histogramDPS.AppendEmpty()
+				histogramDP.SetScale(-1)
+				histogramDP.Positive().SetOffset(-1)
+				histogramDP.Positive().BucketCounts().FromRaw([]uint64{1, 2, 3})
+				histogramDP.SetZeroCount(4)
+				histogramDP.Negative().SetOffset(-1)
+				histogramDP.Negative().BucketCounts().FromRaw([]uint64{1, 2, 3})
+				histogramDP.Attributes().PutStr("label1", "value1")
+				histogramDP.Attributes().PutStr("label2", "value2")
+				return histogramDPS
+			}(),
+			expectedDatapoint: dataPoint{
+				name:   "foo",
+				value:  &cWMetricHistogram{Values: []float64{0.625, 2.5, 10, 0, -0.625, -2.5, -10}, Counts: []float64{1, 2, 3, 4, 1, 2, 3}},
+				labels: map[string]string{oTellibDimensionKey: instrLibName, "label1": "value1", "label2": "value2"},
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(_ *testing.T) {
+			// Given the histogram datapoints
+			exponentialHistogramDatapointSlice := exponentialHistogramDataPointSlice{deltaMetricMetadata, tc.histogramDPS}
+
+			// When calculate the delta datapoints for histograms
+			dps, retained := exponentialHistogramDatapointSlice.CalculateDeltaDatapoints(0, instrLibName, false)
+
+			// Then receiving the following datapoint with an expected length
+			assert.True(t, retained)
+			assert.Equal(t, 1, exponentialHistogramDatapointSlice.Len())
+			assert.Equal(t, tc.expectedDatapoint, dps[0])
+		})
+	}
+
+}
+
 func TestCalculateDeltaDatapoints_SummaryDataPointSlice(t *testing.T) {
 	for _, retainInitialValueOfDeltaMetric := range []bool{true, false} {
 		deltaMetricMetadata := generateDeltaMetricMetadata(true, "foo", retainInitialValueOfDeltaMetric)
@@ -486,6 +614,13 @@ func TestGetDataPoints(t *testing.T) {
 			expectedDatapointSlice: histogramDataPointSlice{cumulativeDeltaMetricMetadata, pmetric.HistogramDataPointSlice{}},
 			expectedAttributes:     map[string]interface{}{"label1": "value1"},
 		},
+		{
+			name:                   "ExponentialHistogram",
+			isPrometheusMetrics:    false,
+			metric:                 generateTestExponentialHistogramMetric("foo"),
+			expectedDatapointSlice: exponentialHistogramDataPointSlice{cumulativeDeltaMetricMetadata, pmetric.ExponentialHistogramDataPointSlice{}},
+			expectedAttributes:     map[string]interface{}{"label1": "value1"},
+		},
 		{
 			name:                   "Summary from SDK",
 			isPrometheusMetrics:    false,
@@ -540,6 +675,15 @@ func TestGetDataPoints(t *testing.T) {
 				assert.Equal(t, uint64(18), dp.Count())
 				assert.Equal(t, []float64{0, 10}, dp.ExplicitBounds().AsRaw())
 				assert.Equal(t, tc.expectedAttributes, dp.Attributes().AsRaw())
+			case exponentialHistogramDataPointSlice:
+				assert.Equal(t, 1, convertedDPS.Len())
+				dp := convertedDPS.ExponentialHistogramDataPointSlice.At(0)
+				assert.Equal(t, float64(0), dp.Sum())
+				assert.Equal(t, uint64(4), dp.Count())
+				assert.Equal(t, []uint64{1, 0, 1}, dp.Positive().BucketCounts().AsRaw())
+				assert.Equal(t, []uint64{1, 0, 1}, dp.Negative().BucketCounts().AsRaw())
+				assert.Equal(t, uint64(0), dp.ZeroCount())
+				assert.Equal(t, tc.expectedAttributes, dp.Attributes().AsRaw())
 			case summaryDataPointSlice:
 				expectedDPS := tc.expectedDatapointSlice.(summaryDataPointSlice)
 				assert.Equal(t, expectedDPS.deltaMetricMetadata, convertedDPS.deltaMetricMetadata)
@@ -587,6 +731,7 @@ func BenchmarkGetAndCalculateDeltaDataPoints(b *testing.B) {
 		generateTestGaugeMetric("int-gauge", intValueType),
 		generateTestGaugeMetric("int-gauge", doubleValueType),
 		generateTestHistogramMetric("histogram"),
+		generateTestExponentialHistogramMetric("exponential-histogram"),
 		generateTestSumMetric("int-sum", intValueType),
 		generateTestSumMetric("double-sum", doubleValueType),
 		generateTestSummaryMetric("summary"),

exporter/awsemfexporter/metric_translator.go

@@ -56,6 +56,17 @@ type cWMetricStats struct {
 	Sum   float64
 }
 
+// The SampleCount of CloudWatch metrics will be calculated by the sum of the 'Counts' array.
+// The 'Count' field should be same as the sum of the 'Counts' array and will be ignored in CloudWatch.
+type cWMetricHistogram struct {
+	Values []float64
+	Counts []float64
+	Max    float64
+	Min    float64
+	Count  uint64
+	Sum    float64
+}
+
 type groupedMetricMetadata struct {
 	namespace                  string
 	timestampMs                int64