[processor/cumulativetodelta] Add support for exponential histograms (#331)

* Remove unneeded min/max entries

* Add support for exponential histograms

* Update README

Author: dricross

processor/cumulativetodeltaprocessor/README.md

@@ -15,16 +15,16 @@
 
 ## Description
 
-The cumulative to delta processor (`cumulativetodeltaprocessor`) converts monotonic, cumulative sum and histogram metrics to monotonic, delta metrics. Non-monotonic sums and exponential histograms are excluded.
+The cumulative to delta processor (`cumulativetodeltaprocessor`) converts monotonic, cumulative sum, histogram, and exponential histogram metrics to monotonic, delta metrics. Non-monotonic sums are excluded.
 
 ## Configuration
 
 Configuration is specified through a list of metrics. The processor uses metric names to identify a set of cumulative metrics and converts them from cumulative to delta.
 
 The following settings can be optionally configured:
 
-- `include`: List of metrics names (case-insensitive), patterns or metric types to convert to delta. Valid values are: `sum`, `histogram`.
-- `exclude`: List of metrics names (case-insensitive), patterns or metric types to not convert to delta.  **If a metric name matches both include and exclude, exclude takes precedence.** Valid values are: `sum`, `histogram`.
+- `include`: List of metrics names (case-insensitive), patterns or metric types to convert to delta. Valid values are: `sum`, `histogram`, `exponentialhistogram`.
+- `exclude`: List of metrics names (case-insensitive), patterns or metric types to not convert to delta.  **If a metric name matches both include and exclude, exclude takes precedence.** Valid values are: `sum`, `histogram`, `exponentialhistogram`.
 - `max_staleness`: The total time a state entry will live past the time it was last seen. Set to 0 to retain state indefinitely. Default: 0
 - `initial_value`: Handling of the first observed point for a given metric identity.
   When the collector (re)starts, there's no record of how much of a given cumulative counter has already been converted to delta values.

processor/cumulativetodeltaprocessor/internal/tracking/identity.go

@@ -71,5 +71,5 @@ func (mi *MetricIdentity) IsFloatVal() bool {
 }
 
 func (mi *MetricIdentity) IsSupportedMetricType() bool {
-	return mi.MetricType == pmetric.MetricTypeSum || mi.MetricType == pmetric.MetricTypeHistogram
+	return mi.MetricType == pmetric.MetricTypeSum || mi.MetricType == pmetric.MetricTypeHistogram || mi.MetricType == pmetric.MetricTypeExponentialHistogram
 }

processor/cumulativetodeltaprocessor/internal/tracking/identity_test.go

@@ -167,23 +167,23 @@ func TestMetricIdentity_IsSupportedMetricType(t *testing.T) {
 			want: true,
 		},
 		{
-			name: "none",
+			name: "exponential_histogram",
 			fields: fields{
-				MetricType: pmetric.MetricTypeEmpty,
+				MetricType: pmetric.MetricTypeExponentialHistogram,
 			},
-			want: false,
+			want: true,
 		},
 		{
-			name: "gauge",
+			name: "none",
 			fields: fields{
-				MetricType: pmetric.MetricTypeGauge,
+				MetricType: pmetric.MetricTypeEmpty,
 			},
 			want: false,
 		},
 		{
-			name: "exponential_histogram",
+			name: "gauge",
 			fields: fields{
-				MetricType: pmetric.MetricTypeExponentialHistogram,
+				MetricType: pmetric.MetricTypeGauge,
 			},
 			want: false,
 		},

processor/cumulativetodeltaprocessor/internal/tracking/tracker.go

@@ -65,10 +65,11 @@ type State struct {
 }
 
 type DeltaValue struct {
-	StartTimestamp pcommon.Timestamp
-	FloatValue     float64
-	IntValue       int64
-	HistogramValue *HistogramPoint
+	StartTimestamp    pcommon.Timestamp
+	FloatValue        float64
+	IntValue          int64
+	HistogramValue    *HistogramPoint
+	ExpHistogramValue *ExpHistogramPoint
 }
 
 func NewMetricTracker(ctx context.Context, logger *zap.Logger, maxStaleness time.Duration, initalValue InitialValue) *MetricTracker {
@@ -123,7 +124,10 @@ func (t *MetricTracker) Convert(in MetricPoint) (out DeltaValue, valid bool) {
 		case pmetric.MetricTypeSum:
 			out.IntValue = metricPoint.IntValue
 			out.FloatValue = metricPoint.FloatValue
-		case pmetric.MetricTypeEmpty, pmetric.MetricTypeGauge, pmetric.MetricTypeExponentialHistogram, pmetric.MetricTypeSummary:
+		case pmetric.MetricTypeExponentialHistogram:
+			val := metricPoint.ExpHistogramValue.Clone()
+			out.ExpHistogramValue = &val
+		case pmetric.MetricTypeEmpty, pmetric.MetricTypeGauge, pmetric.MetricTypeSummary:
 		}
 		switch t.initialValue {
 		case InitialValueAuto:
@@ -171,6 +175,33 @@ func (t *MetricTracker) Convert(in MetricPoint) (out DeltaValue, valid bool) {
 		}
 
 		out.HistogramValue = &delta
+	case pmetric.MetricTypeExponentialHistogram:
+		value := metricPoint.ExpHistogramValue
+		prevValue := state.PrevPoint.ExpHistogramValue
+		if math.IsNaN(value.Sum) {
+			value.Sum = prevValue.Sum
+		}
+
+		if !value.isCompatible(prevValue) || value.isReset(prevValue) {
+			valid = false
+		}
+
+		delta := value.Clone()
+
+		if valid {
+			delta.Count -= prevValue.Count
+			delta.Sum -= prevValue.Sum
+			delta.ZeroCount -= prevValue.ZeroCount
+
+			for index := 0; index < len(delta.PosBuckets) && index < len(prevValue.PosBuckets); index++ {
+				delta.PosBuckets[index] -= prevValue.PosBuckets[index]
+			}
+			for index := 0; index < len(delta.NegBuckets) && index < len(prevValue.NegBuckets); index++ {
+				delta.NegBuckets[index] -= prevValue.NegBuckets[index]
+			}
+		}
+
+		out.ExpHistogramValue = &delta
 	case pmetric.MetricTypeSum:
 		if metricID.IsFloatVal() {
 			value := metricPoint.FloatValue
@@ -195,7 +226,7 @@ func (t *MetricTracker) Convert(in MetricPoint) (out DeltaValue, valid bool) {
 
 			out.IntValue = delta
 		}
-	case pmetric.MetricTypeEmpty, pmetric.MetricTypeGauge, pmetric.MetricTypeExponentialHistogram, pmetric.MetricTypeSummary:
+	case pmetric.MetricTypeEmpty, pmetric.MetricTypeGauge, pmetric.MetricTypeSummary:
 	}
 
 	state.PrevPoint = metricPoint

processor/cumulativetodeltaprocessor/internal/tracking/value.go

@@ -10,6 +10,7 @@ type ValuePoint struct {
 	FloatValue        float64
 	IntValue          int64
 	HistogramValue    *HistogramPoint
+	ExpHistogramValue *ExpHistogramPoint
 }
 
 type HistogramPoint struct {
@@ -28,3 +29,51 @@ func (point *HistogramPoint) Clone() HistogramPoint {
 		Buckets: bucketValues,
 	}
 }
+
+type ExpHistogramPoint struct {
+	Scale               int32
+	NegativeOffsetIndex int32
+	PositiveOffsetIndex int32
+	ZeroThreshold       float64
+	Count               uint64
+	Sum                 float64
+	ZeroCount           uint64
+	PosBuckets          []uint64
+	NegBuckets          []uint64
+}
+
+func (point *ExpHistogramPoint) Clone() ExpHistogramPoint {
+	posBucketValues := make([]uint64, len(point.PosBuckets))
+	copy(posBucketValues, point.PosBuckets)
+	negBucketValues := make([]uint64, len(point.NegBuckets))
+	copy(negBucketValues, point.NegBuckets)
+
+	return ExpHistogramPoint{
+		Count:               point.Count,
+		Sum:                 point.Sum,
+		Scale:               point.Scale,
+		ZeroCount:           point.ZeroCount,
+		ZeroThreshold:       point.ZeroThreshold,
+		PositiveOffsetIndex: point.PositiveOffsetIndex,
+		NegativeOffsetIndex: point.NegativeOffsetIndex,
+		PosBuckets:          posBucketValues,
+		NegBuckets:          negBucketValues,
+	}
+}
+
+func (point *ExpHistogramPoint) isCompatible(other *ExpHistogramPoint) bool {
+	// NOTE: this constraint could be relaxed but requires changes to tracker to handle perfect subsetting and handling
+	// index offsetting
+	return point.Scale == other.Scale &&
+		point.PositiveOffsetIndex == other.PositiveOffsetIndex &&
+		point.NegativeOffsetIndex == other.NegativeOffsetIndex &&
+		point.ZeroThreshold == other.ZeroThreshold
+}
+
+func (point *ExpHistogramPoint) isReset(prev *ExpHistogramPoint) bool {
+	// A drop in count or decrease in number of buckets indicates a reset
+	return point.Count < prev.Count ||
+		point.ZeroCount < prev.ZeroCount ||
+		len(point.PosBuckets) < len(prev.PosBuckets) ||
+		len(point.NegBuckets) < len(prev.NegBuckets)
+}

processor/cumulativetodeltaprocessor/processor.go

@@ -69,6 +69,8 @@ func getMetricTypeFilter(types []string) (map[pmetric.MetricType]bool, error) {
 			res[pmetric.MetricTypeSum] = true
 		case strings.ToLower(pmetric.MetricTypeHistogram.String()):
 			res[pmetric.MetricTypeHistogram] = true
+		case strings.ToLower(pmetric.MetricTypeExponentialHistogram.String()):
+			res[pmetric.MetricTypeExponentialHistogram] = true
 		default:
 			return nil, fmt.Errorf("unsupported metric type filter: %s", t)
 		}
@@ -131,7 +133,31 @@ func (ctdp *cumulativeToDeltaProcessor) processMetrics(_ context.Context, md pme
 
 					ms.SetAggregationTemporality(pmetric.AggregationTemporalityDelta)
 					return ms.DataPoints().Len() == 0
-				case pmetric.MetricTypeEmpty, pmetric.MetricTypeGauge, pmetric.MetricTypeExponentialHistogram, pmetric.MetricTypeSummary:
+				case pmetric.MetricTypeExponentialHistogram:
+					ms := m.ExponentialHistogram()
+					if ms.AggregationTemporality() != pmetric.AggregationTemporalityCumulative {
+						return false
+					}
+
+					if ms.DataPoints().Len() == 0 {
+						return false
+					}
+
+					baseIdentity := tracking.MetricIdentity{
+						Resource:               rm.Resource(),
+						InstrumentationLibrary: ilm.Scope(),
+						MetricType:             m.Type(),
+						MetricName:             m.Name(),
+						MetricUnit:             m.Unit(),
+						MetricIsMonotonic:      true,
+						MetricValueType:        pmetric.NumberDataPointValueTypeInt,
+					}
+
+					ctdp.convertExponentialHistogramDataPoints(ms.DataPoints(), baseIdentity)
+
+					ms.SetAggregationTemporality(pmetric.AggregationTemporalityDelta)
+					return ms.DataPoints().Len() == 0
+				case pmetric.MetricTypeEmpty, pmetric.MetricTypeGauge, pmetric.MetricTypeSummary:
 					fallthrough
 				default:
 					return false
@@ -240,3 +266,55 @@ func (ctdp *cumulativeToDeltaProcessor) convertHistogramDataPoints(in any, baseI
 		})
 	}
 }
+
+func (ctdp *cumulativeToDeltaProcessor) convertExponentialHistogramDataPoints(in any, baseIdentity tracking.MetricIdentity) {
+	if dps, ok := in.(pmetric.ExponentialHistogramDataPointSlice); ok {
+		dps.RemoveIf(func(dp pmetric.ExponentialHistogramDataPoint) bool {
+			id := baseIdentity
+			id.StartTimestamp = dp.StartTimestamp()
+			id.Attributes = dp.Attributes()
+
+			if dp.Flags().NoRecordedValue() {
+				// drop points with no value
+				return true
+			}
+
+			point := tracking.ValuePoint{
+				ObservedTimestamp: dp.Timestamp(),
+				ExpHistogramValue: &tracking.ExpHistogramPoint{
+					Count:               dp.Count(),
+					Sum:                 dp.Sum(),
+					Scale:               dp.Scale(),
+					ZeroCount:           dp.ZeroCount(),
+					ZeroThreshold:       dp.ZeroThreshold(),
+					PositiveOffsetIndex: dp.Positive().Offset(),
+					NegativeOffsetIndex: dp.Negative().Offset(),
+					PosBuckets:          dp.Positive().BucketCounts().AsRaw(),
+					NegBuckets:          dp.Negative().BucketCounts().AsRaw(),
+				},
+			}
+
+			trackingPoint := tracking.MetricPoint{
+				Identity: id,
+				Value:    point,
+			}
+			delta, valid := ctdp.deltaCalculator.Convert(trackingPoint)
+
+			if valid {
+				dp.SetStartTimestamp(delta.StartTimestamp)
+				dp.SetCount(delta.ExpHistogramValue.Count)
+				if dp.HasSum() && !math.IsNaN(dp.Sum()) {
+					dp.SetSum(delta.ExpHistogramValue.Sum)
+				}
+				dp.Positive().BucketCounts().FromRaw(delta.ExpHistogramValue.PosBuckets)
+				dp.Negative().BucketCounts().FromRaw(delta.ExpHistogramValue.NegBuckets)
+				dp.SetZeroCount(delta.ExpHistogramValue.ZeroCount)
+				dp.RemoveMin()
+				dp.RemoveMax()
+				return false
+			}
+
+			return !valid
+		})
+	}
+}