[processor/metricstransform] Fix aggregation of exponential histograms (open-telemetry#39143)

#### Description

A panic would occur if the histograms being merged had different numbers
of populated entries in the BucketCounts array.

Also the counts for the Zero buckets were not being merged.

Example config:
```
processors:
  metricstransform:
    transforms:
    - action: combine
      aggregation_type: sum
      include: traces.span.metrics.duration
      new_name: traces.span.metrics.duration
```

Example panic:
```
panic: runtime error: index out of range [133] with length 133

goroutine 177 [running]:
go.opentelemetry.io/collector/pdata/pcommon.UInt64Slice.At(...)
	go.opentelemetry.io/collector/[email protected]/pcommon/generated_uint64slice.go:55
github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/aggregateutil.mergeExponentialHistogramDataPoints(0x40014b57e0?, {0x4007bc7220?, 0x400744fee8?})
	github.com/open-telemetry/opentelemetry-collector-contrib/internal/[email protected]/aggregateutil/aggregate.go:309 +0x888
github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/aggregateutil.MergeDataPoints({0x4001a435c0?, 0x400744fee8?}, {0x40011b4078?, 0x60?}, {0x0?, 0x0?, 0x0?, 0x4006a73410?})
	github.com/open-telemetry/opentelemetry-collector-contrib/internal/[email protected]/aggregateutil/aggregate.go:96 +0x208
github.com/open-telemetry/opentelemetry-collector-contrib/processor/metricstransformprocessor.groupMetrics({0x4001b048a0?, 0x400744fed8?}, {0x40011b4078, 0x3}, {0x4001a435c0?, 0x400744fee8?})
	github.com/open-telemetry/opentelemetry-collector-contrib/processor/[email protected]/metrics_transform_processor_otlp.go:451 +0xb8
github.com/open-telemetry/opentelemetry-collector-contrib/processor/metricstransformprocessor.combine({{0xadce910, 0x40018a0fd8}, {0x40011b4066, 0x7}, {0x4000777800, 0x1c}, 0x0, {0x40011b4078, 0x3}, {0x0, ...}, ...}, ...)
	github.com/open-telemetry/opentelemetry-collector-contrib/processor/[email protected]/metrics_transform_processor_otlp.go:438 +0x2c0
github.com/open-telemetry/opentelemetry-collector-contrib/processor/metricstransformprocessor.(*metricsTransformProcessor).processMetrics.func1.1({0x4000c0fd50?, 0x4007038b80?})
	github.com/open-telemetry/opentelemetry-collector-contrib/processor/[email protected]/metrics_transform_processor_otlp.go:258 +0x4e8
go.opentelemetry.io/collector/pdata/pmetric.ScopeMetricsSlice.RemoveIf({0x40041de338?, 0x4007038b80?}, 0x4000e3f5f8)
	go.opentelemetry.io/collector/[email protected]/pmetric/generated_scopemetricsslice.go:111 +0x80
github.com/open-telemetry/opentelemetry-collector-contrib/processor/metricstransformprocessor.(*metricsTransformProcessor).processMetrics.func1({0x40041de300?, 0x4007038b80?})
	github.com/open-telemetry/opentelemetry-collector-contrib/processor/[email protected]/metrics_transform_processor_otlp.go:235 +0x64
go.opentelemetry.io/collector/pdata/pmetric.ResourceMetricsSlice.RemoveIf({0x4000e3c378?, 0x4007038b80?}, 0x4000e3f6e8)
	go.opentelemetry.io/collector/[email protected]/pmetric/generated_resourcemetricsslice.go:111 +0x80
github.com/open-telemetry/opentelemetry-collector-contrib/processor/metricstransformprocessor.(*metricsTransformProcessor).processMetrics(0x4000e3f748?, {0x50ac40c?, 0x10?}, {0x4000e3c378?, 0x4007038b80?})
	github.com/open-telemetry/opentelemetry-collector-contrib/processor/[email protected]/metrics_transform_processor_otlp.go:234 +0x74
go.opentelemetry.io/collector/processor/processorhelper.NewMetrics.func1({0xad9c090, 0x11b475c0}, {0x4000e3c378?, 0x4007038b80?})
	go.opentelemetry.io/collector/[email protected]/processorhelper/metrics.go:55 +0xfc
go.opentelemetry.io/collector/consumer.ConsumeMetricsFunc.ConsumeMetrics(...)
	go.opentelemetry.io/collector/[email protected]/metrics.go:27
go.opentelemetry.io/collector/processor/processorhelper.NewMetrics.func1({0xad9c090, 0x11b475c0}, {0x4000e3c378?, 0x4007038b80?})
	go.opentelemetry.io/collector/[email protected]/processorhelper/metrics.go:66 +0x21c
go.opentelemetry.io/collector/consumer.ConsumeMetricsFunc.ConsumeMetrics(...)
	go.opentelemetry.io/collector/[email protected]/metrics.go:27
go.opentelemetry.io/collector/processor/processorhelper.NewMetrics.func1({0xad9c090, 0x11b475c0}, {0x4000e3c2e8?, 0x4007038390?})
	go.opentelemetry.io/collector/[email protected]/processorhelper/metrics.go:66 +0x21c
go.opentelemetry.io/collector/consumer.ConsumeMetricsFunc.ConsumeMetrics(...)
	go.opentelemetry.io/collector/[email protected]/metrics.go:27
go.opentelemetry.io/collector/processor/processorhelper.NewMetrics.func1({0xad9c090, 0x11b475c0}, {0x4000e3c2e8?, 0x4007038390?})
	go.opentelemetry.io/collector/[email protected]/processorhelper/metrics.go:66 +0x21c
go.opentelemetry.io/collector/consumer.ConsumeMetricsFunc.ConsumeMetrics(...)
	go.opentelemetry.io/collector/[email protected]/metrics.go:27
go.opentelemetry.io/collector/processor/processorhelper.NewMetrics.func1({0xad9c090, 0x11b475c0}, {0x4000e3c2e8?, 0x4007038390?})
	go.opentelemetry.io/collector/[email protected]/processorhelper/metrics.go:66 +0x21c
go.opentelemetry.io/collector/consumer.ConsumeMetricsFunc.ConsumeMetrics(...)
	go.opentelemetry.io/collector/[email protected]/metrics.go:27
go.opentelemetry.io/collector/consumer.ConsumeMetricsFunc.ConsumeMetrics(...)
	go.opentelemetry.io/collector/[email protected]/metrics.go:27
go.opentelemetry.io/collector/internal/fanoutconsumer.(*metricsConsumer).ConsumeMetrics(0x400197bd40, {0xad9c090, 0x11b475c0}, {0x4000e3c2e8?, 0x4007038390?})
	go.opentelemetry.io/collector/internal/[email protected]/metrics.go:60 +0x1ec
github.com/open-telemetry/opentelemetry-collector-contrib/connector/spanmetricsconnector.(*connectorImp).exportMetrics(0x4000203a40, {0xad9c090, 0x11b475c0})
	github.com/open-telemetry/opentelemetry-collector-contrib/connector/[email protected]/connector.go:258 +0x110
github.com/open-telemetry/opentelemetry-collector-contrib/connector/spanmetricsconnector.(*connectorImp).Start.func1()
	github.com/open-telemetry/opentelemetry-collector-contrib/connector/[email protected]/connector.go:213 +0x48
created by github.com/open-telemetry/opentelemetry-collector-contrib/connector/spanmetricsconnector.(*connectorImp).Start in goroutine 1
	github.com/open-telemetry/opentelemetry-collector-contrib/connector/[email protected]/connector.go:207 +0xa8
```

---------

Co-authored-by: Antoine Toulme <[email protected]>

Author: 2 people

.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
 }