metricstarttimeprocessor: address edge case where reset happens immediately after the reference is cached

Signed-off-by: Ridwan Sharif <[email protected]>

Author: ridwanmsharif

Makefile

@@ -400,7 +400,7 @@ chlog-update: $(CHLOGGEN)
 .PHONY: genotelcontribcol
 genotelcontribcol: $(BUILDER)
 	./internal/buildscripts/ocb-add-replaces.sh otelcontribcol
-	$(BUILDER) --skip-compilation --config cmd/otelcontribcol/builder-config.yaml
+	$(BUILDER) --skip-compilation --config cmd/otelcontribcol/builder-config-replaced.yaml
 
 # Build the Collector executable.
 .PHONY: otelcontribcol
@@ -417,7 +417,7 @@ otelcontribcollite: genotelcontribcol
 .PHONY: genoteltestbedcol
 genoteltestbedcol: $(BUILDER)
 	./internal/buildscripts/ocb-add-replaces.sh oteltestbedcol
-	$(BUILDER) --skip-compilation --config cmd/oteltestbedcol/builder-config.yaml
+	$(BUILDER) --skip-compilation --config cmd/oteltestbedcol/builder-config-replaced.yaml
 
 # Build the Collector executable, with only components used in testbed.
 .PHONY: oteltestbedcol

processor/metricstarttimeprocessor/internal/datapointstorage/timeseries_map.go

@@ -101,20 +101,20 @@ func (tsm *TimeseriesMap) GC() {
 	tsm.Mark = false
 }
 
-// IsResetHistogram compares the given histogram datapoint h, to tsi.Histogram
+// IsResetHistogram compares the given histogram datapoint h, to ref
 // and determines whether the metric has been reset based on the values.  It is
 // a reset if any of the bucket boundaries have changed, if any of the bucket
 // counts have decreased or if the total sum or count have decreased.
-func (tsi *TimeseriesInfo) IsResetHistogram(h pmetric.HistogramDataPoint) bool {
-	if h.Count() < tsi.Histogram.Count() {
+func IsResetHistogram(h, ref pmetric.HistogramDataPoint) bool {
+	if h.Count() < ref.Count() {
 		return true
 	}
-	if h.Sum() < tsi.Histogram.Sum() {
+	if h.Sum() < ref.Sum() {
 		return true
 	}
 
 	// Guard against bucket boundaries changes.
-	refBounds := tsi.Histogram.ExplicitBounds().AsRaw()
+	refBounds := ref.ExplicitBounds().AsRaw()
 	hBounds := h.ExplicitBounds().AsRaw()
 	if len(refBounds) != len(hBounds) {
 		return true
@@ -126,69 +126,69 @@ func (tsi *TimeseriesInfo) IsResetHistogram(h pmetric.HistogramDataPoint) bool {
 	}
 
 	// We need to check individual buckets to make sure the counts are all increasing.
-	if tsi.Histogram.BucketCounts().Len() != h.BucketCounts().Len() {
+	if ref.BucketCounts().Len() != h.BucketCounts().Len() {
 		return true
 	}
-	for i := range tsi.Histogram.BucketCounts().Len() {
-		if h.BucketCounts().At(i) < tsi.Histogram.BucketCounts().At(i) {
+	for i := range ref.BucketCounts().Len() {
+		if h.BucketCounts().At(i) < ref.BucketCounts().At(i) {
 			return true
 		}
 	}
 	return false
 }
 
 // IsResetExponentialHistogram compares the given exponential histogram
-// datapoint eh, to tsi.ExponentialHistogram and determines whether the metric
+// datapoint eh, to ref and determines whether the metric
 // has been reset based on the values.  It is a reset if any of the bucket
 // boundaries have changed, if any of the bucket counts have decreased or if the
 // total sum or count have decreased.
-func (tsi *TimeseriesInfo) IsResetExponentialHistogram(eh pmetric.ExponentialHistogramDataPoint) bool {
+func IsResetExponentialHistogram(eh, ref pmetric.ExponentialHistogramDataPoint) bool {
 	// Same as the histogram implementation
-	if eh.Count() < tsi.ExponentialHistogram.Count() {
+	if eh.Count() < ref.Count() {
 		return true
 	}
-	if eh.Sum() < tsi.ExponentialHistogram.Sum() {
+	if eh.Sum() < ref.Sum() {
 		return true
 	}
 
 	// Guard against bucket boundaries changes.
-	if tsi.ExponentialHistogram.Scale() != eh.Scale() {
+	if ref.Scale() != eh.Scale() {
 		return true
 	}
 
 	// We need to check individual buckets to make sure the counts are all increasing.
-	if tsi.ExponentialHistogram.Positive().BucketCounts().Len() != eh.Positive().BucketCounts().Len() {
+	if ref.Positive().BucketCounts().Len() != eh.Positive().BucketCounts().Len() {
 		return true
 	}
-	for i := range tsi.ExponentialHistogram.Positive().BucketCounts().Len() {
-		if eh.Positive().BucketCounts().At(i) < tsi.ExponentialHistogram.Positive().BucketCounts().At(i) {
+	for i := range ref.Positive().BucketCounts().Len() {
+		if eh.Positive().BucketCounts().At(i) < ref.Positive().BucketCounts().At(i) {
 			return true
 		}
 	}
-	if tsi.ExponentialHistogram.Negative().BucketCounts().Len() != eh.Negative().BucketCounts().Len() {
+	if ref.Negative().BucketCounts().Len() != eh.Negative().BucketCounts().Len() {
 		return true
 	}
-	for i := range tsi.ExponentialHistogram.Negative().BucketCounts().Len() {
-		if eh.Negative().BucketCounts().At(i) < tsi.ExponentialHistogram.Negative().BucketCounts().At(i) {
+	for i := range ref.Negative().BucketCounts().Len() {
+		if eh.Negative().BucketCounts().At(i) < ref.Negative().BucketCounts().At(i) {
 			return true
 		}
 	}
 
 	return false
 }
 
-// IsResetSummary compares the given summary datapoint s to tsi.Summary and
+// IsResetSummary compares the given summary datapoint s to ref and
 // determines whether the metric has been reset based on the values.  It is a
 // reset if the count or sum has decreased.
-func (tsi *TimeseriesInfo) IsResetSummary(s pmetric.SummaryDataPoint) bool {
-	return s.Count() < tsi.Summary.Count() || s.Sum() < tsi.Summary.Sum()
+func IsResetSummary(s, ref pmetric.SummaryDataPoint) bool {
+	return s.Count() < ref.Count() || s.Sum() < ref.Sum()
 }
 
-// IsResetSum compares the given number datapoint s to tsi.Number and determines
+// IsResetSum compares the given number datapoint s to ref and determines
 // whether the metric has been reset based on the values.  It is a reset if the
 // value has decreased.
-func (tsi *TimeseriesInfo) IsResetSum(s pmetric.NumberDataPoint) bool {
-	return s.DoubleValue() < tsi.Number.DoubleValue()
+func IsResetSum(s, ref pmetric.NumberDataPoint) bool {
+	return s.DoubleValue() < ref.DoubleValue()
 }
 
 func newTimeseriesMap() *TimeseriesMap {

processor/metricstarttimeprocessor/internal/datapointstorage/timeseries_map_test.go

@@ -316,7 +316,7 @@ func TestTimeseriesInfo_IsResetHistogram(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			tsi := tt.setupTsi()
 			h := tt.setupH()
-			assert.Equal(t, tt.expectedReset, tsi.IsResetHistogram(h))
+			assert.Equal(t, tt.expectedReset, IsResetHistogram(h, tsi.Histogram))
 		})
 	}
 }
@@ -518,7 +518,7 @@ func TestTimeseriesInfo_IsResetExponentialHistogram(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			tsi := tt.setupTsi()
 			eh := tt.setupEh()
-			assert.Equal(t, tt.expectedReset, tsi.IsResetExponentialHistogram(eh))
+			assert.Equal(t, tt.expectedReset, IsResetExponentialHistogram(eh, tsi.ExponentialHistogram))
 		})
 	}
 }
@@ -604,7 +604,7 @@ func TestTimeseriesInfo_IsResetSummary(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			tsi := tt.setupTsi()
 			s := tt.setupS()
-			assert.Equal(t, tt.expectedReset, tsi.IsResetSummary(s))
+			assert.Equal(t, tt.expectedReset, IsResetSummary(s, tsi.Summary))
 		})
 	}
 }
@@ -652,7 +652,7 @@ func TestTimeseriesInfo_IsResetSum(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			tsi := tt.setupTsi()
 			s := tt.setupS()
-			assert.Equal(t, tt.expectedReset, tsi.IsResetSum(s))
+			assert.Equal(t, tt.expectedReset, IsResetSum(s, tsi.Number))
 		})
 	}
 }

processor/metricstarttimeprocessor/internal/subtractinitial/adjuster.go

@@ -20,6 +20,9 @@ import (
 // The subtract initial point adjuster sets the start time of all points in a series by:
 //   - Dropping the initial point, and recording its value and timestamp.
 //   - Subtracting the initial point from all subsequent points, and using the timestamp of the initial point as the start timestamp.
+//
+// Note that when a reset is detected (eg: value of a counter is decreasing) - the strategy will set the
+// start time of the reset point as point timestamp - 1ms.
 const Type = "subtract_initial_point"
 
 type Adjuster struct {
@@ -155,29 +158,33 @@ func adjustMetricHistogram(referenceTsm, previousValueTsm *datapointstorage.Time
 			adjustedPoint.CopyTo(resHistogram.DataPoints().AppendEmpty())
 			continue
 		}
+		isReset := datapointstorage.IsResetHistogram(adjustedPoint, referenceTsi.Histogram)
 		subtractHistogramDataPoint(adjustedPoint, referenceTsi.Histogram)
 
 		previousTsi, found := previousValueTsm.Get(current, currentDist.Attributes())
-		if !found {
-			// First point after the reference. Not a reset.
-			previousTsi.Histogram = pmetric.NewHistogramDataPoint()
-		} else if previousTsi.IsResetHistogram(adjustedPoint) {
+		if isReset || (found && datapointstorage.IsResetHistogram(adjustedPoint, previousTsi.Histogram)) {
 			// reset re-initialize everything and use the non adjusted points start time.
 			resetStartTimeStamp := pcommon.NewTimestampFromTime(currentDist.StartTimestamp().AsTime().Add(-1 * time.Millisecond))
 			currentDist.SetStartTimestamp(resetStartTimeStamp)
 
+			// Update the reference value with the current point.
 			referenceTsi.Histogram = pmetric.NewHistogramDataPoint()
+			previousTsi.Histogram = pmetric.NewHistogramDataPoint()
 			referenceTsi.Histogram.SetStartTimestamp(currentDist.StartTimestamp())
+			currentDist.ExplicitBounds().CopyTo(referenceTsi.Histogram.ExplicitBounds())
+			referenceTsi.Histogram.BucketCounts().FromRaw(make([]uint64, currentDist.BucketCounts().Len()))
 
-			tmp := resHistogram.DataPoints().AppendEmpty()
-			currentDist.CopyTo(tmp)
+			currentDist.CopyTo(resHistogram.DataPoints().AppendEmpty())
+			currentDist.CopyTo(previousTsi.Histogram)
 			continue
+		} else if !found {
+			// First point after the reference. Not a reset.
+			previousTsi.Histogram = pmetric.NewHistogramDataPoint()
 		}
 
 		// Update previous values with the current point.
 		adjustedPoint.CopyTo(previousTsi.Histogram)
-		tmp := resHistogram.DataPoints().AppendEmpty()
-		adjustedPoint.CopyTo(tmp)
+		adjustedPoint.CopyTo(resHistogram.DataPoints().AppendEmpty())
 	}
 }
 
@@ -217,29 +224,34 @@ func adjustMetricExponentialHistogram(referenceTsm, previousValueTsm *datapoints
 			adjustedPoint.CopyTo(resExpHistogram.DataPoints().AppendEmpty())
 			continue
 		}
+
+		isReset := datapointstorage.IsResetExponentialHistogram(adjustedPoint, referenceTsi.ExponentialHistogram)
 		subtractExponentialHistogramDataPoint(adjustedPoint, referenceTsi.ExponentialHistogram)
 
 		previousTsi, found := previousValueTsm.Get(current, currentDist.Attributes())
-		if !found {
-			// First point after the reference. Not a reset.
-			previousTsi.ExponentialHistogram = pmetric.NewExponentialHistogramDataPoint()
-		} else if previousTsi.IsResetExponentialHistogram(adjustedPoint) {
+		if isReset || (found && datapointstorage.IsResetExponentialHistogram(adjustedPoint, previousTsi.ExponentialHistogram)) {
 			// reset re-initialize everything and use the non adjusted points start time.
 			resetStartTimeStamp := pcommon.NewTimestampFromTime(currentDist.StartTimestamp().AsTime().Add(-1 * time.Millisecond))
 			currentDist.SetStartTimestamp(resetStartTimeStamp)
 
 			referenceTsi.ExponentialHistogram = pmetric.NewExponentialHistogramDataPoint()
+			previousTsi.ExponentialHistogram = pmetric.NewExponentialHistogramDataPoint()
 			referenceTsi.ExponentialHistogram.SetStartTimestamp(currentDist.StartTimestamp())
+			referenceTsi.ExponentialHistogram.SetScale(currentDist.Scale())
+			referenceTsi.ExponentialHistogram.Positive().BucketCounts().FromRaw(make([]uint64, currentDist.Positive().BucketCounts().Len()))
+			referenceTsi.ExponentialHistogram.Negative().BucketCounts().FromRaw(make([]uint64, currentDist.Negative().BucketCounts().Len()))
 
-			tmp := resExpHistogram.DataPoints().AppendEmpty()
-			currentDist.CopyTo(tmp)
+			currentDist.CopyTo(resExpHistogram.DataPoints().AppendEmpty())
+			currentDist.CopyTo(previousTsi.ExponentialHistogram)
 			continue
+		} else if !found {
+			// First point after the reference. Not a reset.
+			previousTsi.ExponentialHistogram = pmetric.NewExponentialHistogramDataPoint()
 		}
 
 		// Update previous values with the current point.
 		adjustedPoint.CopyTo(previousTsi.ExponentialHistogram)
-		tmp := resExpHistogram.DataPoints().AppendEmpty()
-		adjustedPoint.CopyTo(tmp)
+		adjustedPoint.CopyTo(resExpHistogram.DataPoints().AppendEmpty())
 	}
 }
 
@@ -278,29 +290,30 @@ func adjustMetricSum(referenceTsm, previousValueTsm *datapointstorage.Timeseries
 			adjustedPoint.CopyTo(resSum.DataPoints().AppendEmpty())
 			continue
 		}
+		isReset := datapointstorage.IsResetSum(adjustedPoint, referenceTsi.Number)
 		adjustedPoint.SetDoubleValue(adjustedPoint.DoubleValue() - referenceTsi.Number.DoubleValue())
 
 		previousTsi, found := previousValueTsm.Get(current, currentSum.Attributes())
-		if !found {
-			// First point after the reference. Not a reset.
-			previousTsi.Number = pmetric.NewNumberDataPoint()
-		} else if previousTsi.IsResetSum(adjustedPoint) {
+		if isReset || (found && datapointstorage.IsResetSum(adjustedPoint, previousTsi.Number)) {
 			// reset re-initialize everything and use the non adjusted points start time.
 			resetStartTimeStamp := pcommon.NewTimestampFromTime(currentSum.StartTimestamp().AsTime().Add(-1 * time.Millisecond))
 			currentSum.SetStartTimestamp(resetStartTimeStamp)
 
 			referenceTsi.Number = pmetric.NewNumberDataPoint()
+			previousTsi.Number = pmetric.NewNumberDataPoint()
 			referenceTsi.Number.SetStartTimestamp(currentSum.StartTimestamp())
 
-			tmp := resSum.DataPoints().AppendEmpty()
-			currentSum.CopyTo(tmp)
+			currentSum.CopyTo(resSum.DataPoints().AppendEmpty())
+			currentSum.CopyTo(previousTsi.Number)
 			continue
+		} else if !found {
+			// First point after the reference. Not a reset.
+			previousTsi.Number = pmetric.NewNumberDataPoint()
 		}
 
 		// Update previous values with the current point.
 		adjustedPoint.CopyTo(previousTsi.Number)
-		tmp := resSum.DataPoints().AppendEmpty()
-		adjustedPoint.CopyTo(tmp)
+		adjustedPoint.CopyTo(resSum.DataPoints().AppendEmpty())
 	}
 }
 
@@ -331,30 +344,32 @@ func adjustMetricSummary(referenceTsm, previousValueTsm *datapointstorage.Timese
 			adjustedPoint.CopyTo(resSummary.DataPoints().AppendEmpty())
 			continue
 		}
+
+		isReset := datapointstorage.IsResetSummary(adjustedPoint, referenceTsi.Summary)
 		adjustedPoint.SetCount(adjustedPoint.Count() - referenceTsi.Summary.Count())
 		adjustedPoint.SetSum(adjustedPoint.Sum() - referenceTsi.Summary.Sum())
 
 		previousTsi, found := previousValueTsm.Get(current, currentSummary.Attributes())
-		if !found {
-			// First point after the reference. Not a reset.
-			previousTsi.Summary = pmetric.NewSummaryDataPoint()
-		} else if previousTsi.IsResetSummary(adjustedPoint) {
+		if isReset || (found && datapointstorage.IsResetSummary(adjustedPoint, previousTsi.Summary)) {
 			// reset re-initialize everything and use the non adjusted points start time.
 			resetStartTimeStamp := pcommon.NewTimestampFromTime(currentSummary.StartTimestamp().AsTime().Add(-1 * time.Millisecond))
 			currentSummary.SetStartTimestamp(resetStartTimeStamp)
 
 			referenceTsi.Summary = pmetric.NewSummaryDataPoint()
+			previousTsi.Summary = pmetric.NewSummaryDataPoint()
 			referenceTsi.Summary.SetStartTimestamp(currentSummary.StartTimestamp())
 
-			tmp := resSummary.DataPoints().AppendEmpty()
-			currentSummary.CopyTo(tmp)
+			currentSummary.CopyTo(resSummary.DataPoints().AppendEmpty())
+			currentSummary.CopyTo(previousTsi.Summary)
 			continue
+		} else if !found {
+			// First point after the reference. Not a reset.
+			previousTsi.Summary = pmetric.NewSummaryDataPoint()
 		}
 
 		// Update previous values with the current point.
 		adjustedPoint.CopyTo(previousTsi.Summary)
-		tmp := resSummary.DataPoints().AppendEmpty()
-		adjustedPoint.CopyTo(tmp)
+		adjustedPoint.CopyTo(resSummary.DataPoints().AppendEmpty())
 	}
 }