[exporter/prometheusexporter] accumulate delta temporality histograms

.chloggen/promexp-delta.yaml

@@ -0,0 +1,11 @@
+# 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: prometheusexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Accumulate histograms with delta temporality
+
+# One or more tracking issues related to the change
+issues: [4968]

exporter/prometheusexporter/accumulator.go

@@ -84,7 +84,7 @@ func (a *lastValueAccumulator) addMetric(metric pmetric.Metric, il pcommon.Instr
 	case pmetric.MetricTypeSum:
 		return a.accumulateSum(metric, il, resourceAttrs, now)
 	case pmetric.MetricTypeHistogram:
-		return a.accumulateDoubleHistogram(metric, il, resourceAttrs, now)
+		return a.accumulateHistogram(metric, il, resourceAttrs, now)
 	case pmetric.MetricTypeSummary:
 		return a.accumulateSummary(metric, il, resourceAttrs, now)
 	default:
@@ -161,19 +161,19 @@ func (a *lastValueAccumulator) accumulateGauge(metric pmetric.Metric, il pcommon
 }
 
 func (a *lastValueAccumulator) accumulateSum(metric pmetric.Metric, il pcommon.InstrumentationScope, resourceAttrs pcommon.Map, now time.Time) (n int) {
-	doubleSum := metric.Sum()
+	sum := metric.Sum()
 
 	// Drop metrics with unspecified aggregations
-	if doubleSum.AggregationTemporality() == pmetric.AggregationTemporalityUnspecified {
+	if sum.AggregationTemporality() == pmetric.AggregationTemporalityUnspecified {
 		return
 	}
 
 	// Drop non-monotonic and non-cumulative metrics
-	if doubleSum.AggregationTemporality() == pmetric.AggregationTemporalityDelta && !doubleSum.IsMonotonic() {
+	if sum.AggregationTemporality() == pmetric.AggregationTemporalityDelta && !sum.IsMonotonic() {
 		return
 	}
 
-	dps := doubleSum.DataPoints()
+	dps := sum.DataPoints()
 	for i := 0; i < dps.Len(); i++ {
 		ip := dps.At(i)
 
@@ -201,7 +201,7 @@ func (a *lastValueAccumulator) accumulateSum(metric pmetric.Metric, il pcommon.I
 		}
 
 		// Delta-to-Cumulative
-		if doubleSum.AggregationTemporality() == pmetric.AggregationTemporalityDelta && ip.StartTimestamp() == mv.value.Sum().DataPoints().At(0).Timestamp() {
+		if sum.AggregationTemporality() == pmetric.AggregationTemporalityDelta && ip.StartTimestamp() == mv.value.Sum().DataPoints().At(0).Timestamp() {
 			ip.SetStartTimestamp(mv.value.Sum().DataPoints().At(0).StartTimestamp())
 			switch ip.ValueType() {
 			case pmetric.NumberDataPointValueTypeInt:
@@ -218,45 +218,75 @@ func (a *lastValueAccumulator) accumulateSum(metric pmetric.Metric, il pcommon.I
 		a.registeredMetrics.Store(signature, &accumulatedValue{value: m, resourceAttrs: resourceAttrs, scope: il, updated: now})
 		n++
 	}
+
 	return
 }
 
-func (a *lastValueAccumulator) accumulateDoubleHistogram(metric pmetric.Metric, il pcommon.InstrumentationScope, resourceAttrs pcommon.Map, now time.Time) (n int) {
-	doubleHistogram := metric.Histogram()
-
-	// Drop metrics with non-cumulative aggregations
-	if doubleHistogram.AggregationTemporality() != pmetric.AggregationTemporalityCumulative {
-		return
-	}
+func (a *lastValueAccumulator) accumulateHistogram(metric pmetric.Metric, il pcommon.InstrumentationScope, resourceAttrs pcommon.Map, now time.Time) (n int) {
+	histogram := metric.Histogram()
+	a.logger.Debug("Accumulate histogram.....")
+	dps := histogram.DataPoints()
 
-	dps := doubleHistogram.DataPoints()
 	for i := 0; i < dps.Len(); i++ {
 		ip := dps.At(i)
 
-		signature := timeseriesSignature(il.Name(), metric, ip.Attributes(), resourceAttrs)
+		signature := timeseriesSignature(il.Name(), metric, ip.Attributes(), resourceAttrs) // uniquely idenity this time series you are accumulating for
 		if ip.Flags().NoRecordedValue() {
 			a.registeredMetrics.Delete(signature)
 			return 0
 		}
 
-		v, ok := a.registeredMetrics.Load(signature)
+		v, ok := a.registeredMetrics.Load(signature) // a accumulates metric values for all times series. Get value for particular time series
 		if !ok {
+			// first data point
+			a.logger.Debug("Accumulate first histogram data point")
 			m := copyMetricMetadata(metric)
 			ip.CopyTo(m.SetEmptyHistogram().DataPoints().AppendEmpty())
+			m.Histogram().SetAggregationTemporality(pmetric.AggregationTemporalityCumulative)
 			a.registeredMetrics.Store(signature, &accumulatedValue{value: m, resourceAttrs: resourceAttrs, scope: il, updated: now})
 			n++
 			continue
 		}
 		mv := v.(*accumulatedValue)
 
-		if ip.Timestamp().AsTime().Before(mv.value.Histogram().DataPoints().At(0).Timestamp().AsTime()) {
-			// only keep datapoint with latest timestamp
+		m := copyMetricMetadata(metric)
+		m.SetEmptyHistogram().SetAggregationTemporality(pmetric.AggregationTemporalityCumulative)
+
+		switch histogram.AggregationTemporality() {
+		case pmetric.AggregationTemporalityDelta:
+			pp := mv.value.Histogram().DataPoints().At(0) // previous aggregated value for time range
+			if ip.StartTimestamp().AsTime() != pp.Timestamp().AsTime() {
+				// treat misalgnment as restart and reset, or violation of single-writer principle and drop
+				a.logger.With(
+					zap.String("ip_start_time", ip.StartTimestamp().String()),
+					zap.String("pp_start_time", pp.StartTimestamp().String()),
+					zap.String("pp_timestamp", pp.Timestamp().String()),
+					zap.String("ip_timestamp", ip.Timestamp().String()),
+				).Warn("Misaligned starting timestamps")
+				if ip.StartTimestamp().AsTime().After(pp.Timestamp().AsTime()) {
+					a.logger.Debug("treating it like reset")
+					ip.CopyTo(m.Histogram().DataPoints().AppendEmpty())
+				} else {
+					a.logger.With(
+						zap.String("metric_name", metric.Name()),
+					).Warn("Dropped misaligned histogram datapoint")
+					continue
+				}
+			} else {
+				a.logger.Debug("Accumulate another histogram datapoint")
+				accumulateHistogramValues(pp, ip, m.Histogram().DataPoints().AppendEmpty())
+			}
+		case pmetric.AggregationTemporalityCumulative:
+			if ip.Timestamp().AsTime().Before(mv.value.Histogram().DataPoints().At(0).Timestamp().AsTime()) {
+				// only keep datapoint with latest timestamp
+				continue
+			}
+
+			ip.CopyTo(m.Histogram().DataPoints().AppendEmpty())
+		default:
+			// unsupported temporality
 			continue
 		}
-
-		m := copyMetricMetadata(metric)
-		ip.CopyTo(m.SetEmptyHistogram().DataPoints().AppendEmpty())
-		m.Histogram().SetAggregationTemporality(pmetric.AggregationTemporalityCumulative)
 		a.registeredMetrics.Store(signature, &accumulatedValue{value: m, resourceAttrs: resourceAttrs, scope: il, updated: now})
 		n++
 	}
@@ -316,3 +346,42 @@ func copyMetricMetadata(metric pmetric.Metric) pmetric.Metric {
 
 	return m
 }
+
+func accumulateHistogramValues(prev, current, dest pmetric.HistogramDataPoint) {
+	dest.SetStartTimestamp(prev.StartTimestamp())
+
+	older := prev
+	newer := current
+	if current.Timestamp().AsTime().Before(prev.Timestamp().AsTime()) {
+		older = current
+		newer = prev
+	}
+
+	newer.Attributes().CopyTo(dest.Attributes())
+	dest.SetTimestamp(newer.Timestamp())
+
+	// checking for bucket boundary alignment, optionally re-aggregate on newer boundaries
+	match := older.ExplicitBounds().Len() == newer.ExplicitBounds().Len()
+	for i := 0; match && i < newer.ExplicitBounds().Len(); i++ {
+		match = older.ExplicitBounds().At(i) == newer.ExplicitBounds().At(i)
+	}
+
+	if match {
+
+		dest.SetCount(newer.Count() + older.Count())
+		dest.SetSum(newer.Sum() + older.Sum())
+
+		counts := make([]uint64, newer.BucketCounts().Len())
+		for i := 0; i < newer.BucketCounts().Len(); i++ {
+			counts[i] = newer.BucketCounts().At(i) + older.BucketCounts().At(i)
+		}
+		dest.BucketCounts().FromRaw(counts)
+	} else {
+		// use new value if bucket bounds do not match
+		dest.SetCount(newer.Count())
+		dest.SetSum(newer.Sum())
+		dest.BucketCounts().FromRaw(newer.BucketCounts().AsRaw())
+	}
+
+	dest.ExplicitBounds().FromRaw(newer.ExplicitBounds().AsRaw())
+}