[processor/metricstransform] Fix aggregation of exponential histograms

.chloggen/aggregateutil-exphist-panic.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: metricstransformprocessor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Fix aggregation of exponential histograms in metricstransform processor.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [39143]
+
+# (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: Fix a panic when the number of populated buckets varies, and fix summing of counts for the Zero bucket.
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

internal/coreinternal/aggregateutil/aggregate.go

@@ -296,17 +296,31 @@ func mergeExponentialHistogramDataPoints(dpsMap map[string]pmetric.ExponentialHi
 			}
 			dp.SetCount(dp.Count() + dps.At(i).Count())
 			dp.SetSum(dp.Sum() + dps.At(i).Sum())
+			dp.SetZeroCount(dp.ZeroCount() + dps.At(i).ZeroCount())
 			if dp.HasMin() && dp.Min() > dps.At(i).Min() {
 				dp.SetMin(dps.At(i).Min())
 			}
 			if dp.HasMax() && dp.Max() < dps.At(i).Max() {
 				dp.SetMax(dps.At(i).Max())
 			}
+			// Merge bucket counts.
+			// Note that groupExponentialHistogramDataPoints() has already ensured that we only try
+			// to merge exponential histograms with matching Scale and Positive/Negative Offsets,
+			// so the corresponding array items in BucketCounts have the same bucket boundaries.
+			// However, the number of buckets may differ depending on what values have been observed.
 			for b := 0; b < dps.At(i).Negative().BucketCounts().Len(); b++ {
-				negatives.SetAt(b, negatives.At(b)+dps.At(i).Negative().BucketCounts().At(b))
+				if b < negatives.Len() {
+					negatives.SetAt(b, negatives.At(b)+dps.At(i).Negative().BucketCounts().At(b))
+				} else {
+					negatives.Append(dps.At(i).Negative().BucketCounts().At(b))
+				}
 			}
 			for b := 0; b < dps.At(i).Positive().BucketCounts().Len(); b++ {
-				positives.SetAt(b, positives.At(b)+dps.At(i).Positive().BucketCounts().At(b))
+				if b < positives.Len() {
+					positives.SetAt(b, positives.At(b)+dps.At(i).Positive().BucketCounts().At(b))
+				} else {
+					positives.Append(dps.At(i).Positive().BucketCounts().At(b))
+				}
 			}
 			dps.At(i).Exemplars().MoveAndAppendTo(dp.Exemplars())
 			if dps.At(i).StartTimestamp() < dp.StartTimestamp() {

internal/coreinternal/aggregateutil/aggregate_test.go

@@ -365,7 +365,10 @@ func Test_GroupDataPoints(t *testing.T) {
 				d := s.DataPoints().AppendEmpty()
 				d.SetTimestamp(pcommon.NewTimestampFromTime(time.Time{}))
 				d.Attributes().PutStr("attr1", "val1")
-				d.SetCount(1)
+				d.SetCount(9)
+				d.SetZeroCount(2)
+				d.Positive().BucketCounts().Append(0, 1, 2, 3)
+				d.Negative().BucketCounts().Append(0, 1)
 				return m
 			},
 			want: AggGroups{
@@ -483,8 +486,11 @@ func Test_MergeDataPoints(t *testing.T) {
 				s := m.SetEmptyExponentialHistogram()
 				d := s.DataPoints().AppendEmpty()
 				d.Attributes().PutStr("attr1", "val1")
-				d.SetCount(3)
+				d.SetCount(16)
 				d.SetSum(0)
+				d.SetZeroCount(3)
+				d.Positive().BucketCounts().Append(0, 2, 4, 3)
+				d.Negative().BucketCounts().Append(0, 2, 2)
 				return m
 			},
 			in: func() pmetric.Metric {
@@ -547,18 +553,34 @@ func testDataExpHistogram() pmetric.ExponentialHistogramDataPointSlice {
 	data := pmetric.NewExponentialHistogramDataPointSlice()
 	d := data.AppendEmpty()
 	d.Attributes().PutStr("attr1", "val1")
-	d.SetCount(2)
+	d.SetCount(7)
+	d.SetZeroCount(1)
+	d.Positive().BucketCounts().Append(0, 1, 2)
+	d.Negative().BucketCounts().Append(0, 1, 2)
 	return data
 }
 
 func testDataExpHistogramDouble() pmetric.ExponentialHistogramDataPointSlice {
 	dataWant := pmetric.NewExponentialHistogramDataPointSlice()
+
 	dWant := dataWant.AppendEmpty()
 	dWant.Attributes().PutStr("attr1", "val1")
-	dWant.SetCount(2)
+	dWant.SetCount(7)
+	dWant.SetZeroCount(1)
+	dWant.Positive().BucketCounts().Append(0, 1, 2)
+	dWant.Negative().BucketCounts().Append(0, 1, 2)
+
 	dWant2 := dataWant.AppendEmpty()
 	dWant2.SetTimestamp(pcommon.NewTimestampFromTime(time.Time{}))
 	dWant2.Attributes().PutStr("attr1", "val1")
-	dWant2.SetCount(1)
+	dWant2.SetCount(9)
+	dWant2.SetZeroCount(2)
+	// Use a larger number of buckets than above to check that we expand the
+	// destination array as needed while merging.
+	dWant2.Positive().BucketCounts().Append(0, 1, 2, 3)
+	// Use a smaller number of buckets than above to check that we merge values
+	// into the correct, existing buckets.
+	dWant2.Negative().BucketCounts().Append(0, 1)
+
 	return dataWant
 }