metricstarttimeprocessor: add subtractinitial strategy implementation

This change addresses open-telemetry#38379.

Author: ridwanmsharif

.chloggen/decouple-cache.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: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: metricstarttimeprocessor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Decouples the cache from the strategies for adjusting
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [38382]
+
+# (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:
+
+# 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: []

processor/metricstarttimeprocessor/internal/subtractinitial/adjuster.go

@@ -5,11 +5,14 @@ package subtractinitial // import "github.com/open-telemetry/opentelemetry-colle
 
 import (
 	"context"
-	"errors"
 	"time"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/metricstarttimeprocessor/internal/datapointstorage"
 )
 
 // Type is the value users can use to configure the subtract initial point adjuster.
@@ -19,19 +22,349 @@ import (
 const Type = "subtract_initial_point"
 
 type Adjuster struct {
-	set component.TelemetrySettings
+	referenceCache     *datapointstorage.Cache
+	previousValueCache *datapointstorage.Cache
+	set                component.TelemetrySettings
 }
 
 // NewAdjuster returns a new Adjuster which adjust metrics' start times based on the initial received points.
-func NewAdjuster(set component.TelemetrySettings, _ time.Duration) *Adjuster {
+func NewAdjuster(set component.TelemetrySettings, gcInterval time.Duration) *Adjuster {
 	return &Adjuster{
-		set: set,
+		referenceCache:     datapointstorage.NewCache(gcInterval),
+		previousValueCache: datapointstorage.NewCache(gcInterval),
+		set:                set,
 	}
 }
 
 // AdjustMetrics takes a sequence of metrics and adjust their start times based on the initial and
 // previous points in the timeseriesMap.
+// For each metric:
+// - Check if it exists in the map already.
+// - If it doesn't, save its value in the reference cache and move on
+// - If it does, find its reference cache value
+// - Add new entry to the result with the value - reference value
+// - When a reset is discovered, update the reference as well.
 func (a *Adjuster) AdjustMetrics(_ context.Context, metrics pmetric.Metrics) (pmetric.Metrics, error) {
-	// TODO(#38379): Implement the subtract_initial_point adjuster
-	return metrics, errors.New("not implemented")
+	// Create a copy of metrics to store the results
+	resultMetrics := pmetric.NewMetrics()
+	for i := 0; i < metrics.ResourceMetrics().Len(); i++ {
+		rm := metrics.ResourceMetrics().At(i)
+
+		// Copy over resource info to the result.
+		resResource := resultMetrics.ResourceMetrics().AppendEmpty()
+		resResource.SetSchemaUrl(rm.SchemaUrl())
+		rm.Resource().CopyTo(resResource.Resource())
+
+		attrHash := pdatautil.MapHash(rm.Resource().Attributes())
+		referenceTsm, _ := a.referenceCache.Get(attrHash)
+		previousValueTsm, _ := a.previousValueCache.Get(attrHash)
+
+		// The lock on the relevant timeseriesMap is held throughout the adjustment process to ensure that
+		// nothing else can modify the data used for adjustment.
+		referenceTsm.Lock()
+		previousValueTsm.Lock()
+		for j := 0; j < rm.ScopeMetrics().Len(); j++ {
+			ilm := rm.ScopeMetrics().At(j)
+
+			// Copy over scope info to the result.
+			resScope := resResource.ScopeMetrics().AppendEmpty()
+			resScope.SetSchemaUrl(ilm.SchemaUrl())
+			ilm.Scope().CopyTo(resScope.Scope())
+
+			for k := 0; k < ilm.Metrics().Len(); k++ {
+				metric := ilm.Metrics().At(k)
+
+				// Copy over metric info to the result.
+				resMetric := resScope.Metrics().AppendEmpty()
+				resMetric.SetName(metric.Name())
+				resMetric.SetDescription(metric.Description())
+				resMetric.SetUnit(metric.Unit())
+				metric.Metadata().CopyTo(resMetric.Metadata())
+
+				switch dataType := metric.Type(); dataType {
+				case pmetric.MetricTypeGauge:
+					// gauges don't need to be adjusted so no additional processing is necessary
+					metric.CopyTo(resMetric)
+
+				case pmetric.MetricTypeHistogram:
+					a.adjustMetricHistogram(referenceTsm, previousValueTsm, metric, resMetric.SetEmptyHistogram())
+
+				case pmetric.MetricTypeSummary:
+					a.adjustMetricSummary(referenceTsm, previousValueTsm, metric, resMetric.SetEmptySummary())
+
+				case pmetric.MetricTypeSum:
+					a.adjustMetricSum(referenceTsm, previousValueTsm, metric, resMetric.SetEmptySum())
+
+				case pmetric.MetricTypeExponentialHistogram:
+					a.adjustMetricExponentialHistogram(referenceTsm, previousValueTsm, metric, resMetric.SetEmptyExponentialHistogram())
+
+				case pmetric.MetricTypeEmpty:
+					fallthrough
+
+				default:
+					// this shouldn't happen
+					a.set.Logger.Info("Adjust - skipping unexpected point", zap.String("type", dataType.String()))
+				}
+			}
+		}
+		referenceTsm.Unlock()
+		previousValueTsm.Unlock()
+	}
+
+	return resultMetrics, nil
+}
+
+func (a *Adjuster) adjustMetricHistogram(referenceTsm, previousValueTsm *datapointstorage.TimeseriesMap, current pmetric.Metric, resHistogram pmetric.Histogram) {
+	resHistogram.SetAggregationTemporality(current.Histogram().AggregationTemporality())
+
+	histogram := current.Histogram()
+	if histogram.AggregationTemporality() != pmetric.AggregationTemporalityCumulative {
+		// Only dealing with CumulativeDistributions.
+		histogram.CopyTo(resHistogram)
+		return
+	}
+
+	currentPoints := histogram.DataPoints()
+	for i := 0; i < currentPoints.Len(); i++ {
+		currentDist := currentPoints.At(i)
+
+		referenceTsi, found := referenceTsm.Get(current, currentDist.Attributes())
+		if !found {
+			// initialize everything. Don't add the datapoint to the result.
+			referenceTsi.Histogram = pmetric.NewHistogramDataPoint()
+			currentDist.CopyTo(referenceTsi.Histogram)
+			continue
+		}
+
+		// Adjust the datapoint based on the reference value.
+		adjustedPoint := pmetric.NewHistogramDataPoint()
+		currentDist.CopyTo(adjustedPoint)
+		adjustedPoint.SetStartTimestamp(referenceTsi.Histogram.StartTimestamp())
+		if currentDist.Flags().NoRecordedValue() {
+			// TODO: Investigate why this does not reset.
+			tmp := resHistogram.DataPoints().AppendEmpty()
+			adjustedPoint.CopyTo(tmp)
+			continue
+		}
+		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) {
+			// reset re-initialize everything and use the non adjusted points start time.
+			referenceTsi.Histogram = pmetric.NewHistogramDataPoint()
+			referenceTsi.Histogram.SetStartTimestamp(currentDist.StartTimestamp())
+
+			tmp := resHistogram.DataPoints().AppendEmpty()
+			currentDist.CopyTo(tmp)
+			continue
+		}
+
+		// Update previous values with the current point.
+		adjustedPoint.CopyTo(previousTsi.Histogram)
+		tmp := resHistogram.DataPoints().AppendEmpty()
+		adjustedPoint.CopyTo(tmp)
+	}
+}
+
+func (a *Adjuster) adjustMetricExponentialHistogram(referenceTsm, previousValueTsm *datapointstorage.TimeseriesMap, current pmetric.Metric, resExpHistogram pmetric.ExponentialHistogram) {
+	resExpHistogram.SetAggregationTemporality(current.ExponentialHistogram().AggregationTemporality())
+
+	histogram := current.ExponentialHistogram()
+	if histogram.AggregationTemporality() != pmetric.AggregationTemporalityCumulative {
+		// Only dealing with CumulativeDistributions.
+		histogram.CopyTo(resExpHistogram)
+		return
+	}
+
+	currentPoints := histogram.DataPoints()
+	for i := 0; i < currentPoints.Len(); i++ {
+		currentDist := currentPoints.At(i)
+
+		referenceTsi, found := referenceTsm.Get(current, currentDist.Attributes())
+		if !found {
+			// initialize everything. Don't add the datapoint to the result.
+			referenceTsi.ExponentialHistogram = pmetric.NewExponentialHistogramDataPoint()
+			currentDist.CopyTo(referenceTsi.ExponentialHistogram)
+			continue
+		}
+
+		// Adjust the datapoint based on the reference value.
+		adjustedPoint := pmetric.NewExponentialHistogramDataPoint()
+		currentDist.CopyTo(adjustedPoint)
+		adjustedPoint.SetStartTimestamp(referenceTsi.ExponentialHistogram.StartTimestamp())
+		if currentDist.Flags().NoRecordedValue() {
+			// TODO: Investigate why this does not reset.
+			tmp := resExpHistogram.DataPoints().AppendEmpty()
+			adjustedPoint.CopyTo(tmp)
+			continue
+		}
+		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) {
+			// reset re-initialize everything and use the non adjusted points start time.
+			referenceTsi.ExponentialHistogram = pmetric.NewExponentialHistogramDataPoint()
+			referenceTsi.ExponentialHistogram.SetStartTimestamp(currentDist.StartTimestamp())
+
+			tmp := resExpHistogram.DataPoints().AppendEmpty()
+			currentDist.CopyTo(tmp)
+			continue
+		}
+
+		// Update previous values with the current point.
+		adjustedPoint.CopyTo(previousTsi.ExponentialHistogram)
+		tmp := resExpHistogram.DataPoints().AppendEmpty()
+		adjustedPoint.CopyTo(tmp)
+	}
+}
+
+func (a *Adjuster) adjustMetricSum(referenceTsm, previousValueTsm *datapointstorage.TimeseriesMap, current pmetric.Metric, resSum pmetric.Sum) {
+	resSum.SetAggregationTemporality(current.Sum().AggregationTemporality())
+	resSum.SetIsMonotonic(current.Sum().IsMonotonic())
+
+	currentPoints := current.Sum().DataPoints()
+	for i := 0; i < currentPoints.Len(); i++ {
+		currentDist := currentPoints.At(i)
+
+		referenceTsi, found := referenceTsm.Get(current, currentDist.Attributes())
+		if !found {
+			// initialize everything. Don't add the datapoint to the result.
+			referenceTsi.Number = pmetric.NewNumberDataPoint()
+			currentDist.CopyTo(referenceTsi.Number)
+			continue
+		}
+
+		// Adjust the datapoint based on the reference value.
+		adjustedPoint := pmetric.NewNumberDataPoint()
+		currentDist.CopyTo(adjustedPoint)
+		adjustedPoint.SetStartTimestamp(referenceTsi.Number.StartTimestamp())
+		if currentDist.Flags().NoRecordedValue() {
+			// TODO: Investigate why this does not reset.
+			tmp := resSum.DataPoints().AppendEmpty()
+			adjustedPoint.CopyTo(tmp)
+			continue
+		}
+		adjustedPoint.SetDoubleValue(adjustedPoint.DoubleValue() - referenceTsi.Number.DoubleValue())
+
+		previousTsi, found := previousValueTsm.Get(current, currentDist.Attributes())
+		if !found {
+			// First point after the reference. Not a reset.
+			previousTsi.Number = pmetric.NewNumberDataPoint()
+		} else if previousTsi.IsResetSum(adjustedPoint) {
+			// reset re-initialize everything and use the non adjusted points start time.
+			referenceTsi.Number = pmetric.NewNumberDataPoint()
+			referenceTsi.Number.SetStartTimestamp(currentDist.StartTimestamp())
+
+			tmp := resSum.DataPoints().AppendEmpty()
+			currentDist.CopyTo(tmp)
+			continue
+		}
+
+		// Update previous values with the current point.
+		adjustedPoint.CopyTo(previousTsi.Number)
+		tmp := resSum.DataPoints().AppendEmpty()
+		adjustedPoint.CopyTo(tmp)
+	}
+}
+
+func (a *Adjuster) adjustMetricSummary(referenceTsm, previousValueTsm *datapointstorage.TimeseriesMap, current pmetric.Metric, resSummary pmetric.Summary) {
+	currentPoints := current.Summary().DataPoints()
+	for i := 0; i < currentPoints.Len(); i++ {
+		currentDist := currentPoints.At(i)
+
+		referenceTsi, found := referenceTsm.Get(current, currentDist.Attributes())
+		if !found {
+			// initialize everything. Don't add the datapoint to the result.
+			referenceTsi.Summary = pmetric.NewSummaryDataPoint()
+			currentDist.CopyTo(referenceTsi.Summary)
+			continue
+		}
+
+		// Adjust the datapoint based on the reference value.
+		adjustedPoint := pmetric.NewSummaryDataPoint()
+		currentDist.CopyTo(adjustedPoint)
+		adjustedPoint.SetStartTimestamp(referenceTsi.Summary.StartTimestamp())
+		if currentDist.Flags().NoRecordedValue() {
+			// TODO: Investigate why this does not reset.
+			tmp := resSummary.DataPoints().AppendEmpty()
+			adjustedPoint.CopyTo(tmp)
+			continue
+		}
+		adjustedPoint.SetCount(adjustedPoint.Count() - referenceTsi.Summary.Count())
+		adjustedPoint.SetSum(adjustedPoint.Sum() - referenceTsi.Summary.Sum())
+
+		previousTsi, found := previousValueTsm.Get(current, currentDist.Attributes())
+		if !found {
+			// First point after the reference. Not a reset.
+			previousTsi.Summary = pmetric.NewSummaryDataPoint()
+		} else if previousTsi.IsResetSummary(adjustedPoint) {
+			// reset re-initialize everything and use the non adjusted points start time.
+			referenceTsi.Summary = pmetric.NewSummaryDataPoint()
+			referenceTsi.Summary.SetStartTimestamp(currentDist.StartTimestamp())
+
+			tmp := resSummary.DataPoints().AppendEmpty()
+			currentDist.CopyTo(tmp)
+			continue
+		}
+
+		// Update previous values with the current point.
+		adjustedPoint.CopyTo(previousTsi.Summary)
+		tmp := resSummary.DataPoints().AppendEmpty()
+		adjustedPoint.CopyTo(tmp)
+	}
+}
+
+// subtractHistogramDataPoint subtracts b from a.
+func subtractHistogramDataPoint(a, b pmetric.HistogramDataPoint) {
+	a.SetStartTimestamp(b.StartTimestamp())
+	a.SetCount(a.Count() - b.Count())
+	a.SetSum(a.Sum() - b.Sum())
+	aBuckets := a.BucketCounts()
+	bBuckets := b.BucketCounts()
+	if bBuckets.Len() != aBuckets.Len() {
+		// Post reset, the reference histogram will have no buckets.
+		return
+	}
+	newBuckets := make([]uint64, aBuckets.Len())
+	for i := 0; i < aBuckets.Len(); i++ {
+		newBuckets[i] = aBuckets.At(i) - bBuckets.At(i)
+	}
+	a.BucketCounts().FromRaw(newBuckets)
+}
+
+// subtractExponentialHistogramDataPoint subtracts b from a.
+func subtractExponentialHistogramDataPoint(a, b pmetric.ExponentialHistogramDataPoint) {
+	a.SetStartTimestamp(b.StartTimestamp())
+	a.SetCount(a.Count() - b.Count())
+	a.SetSum(a.Sum() - b.Sum())
+	a.SetZeroCount(a.ZeroCount() - b.ZeroCount())
+	if a.Positive().BucketCounts().Len() != b.Positive().BucketCounts().Len() ||
+		a.Negative().BucketCounts().Len() != b.Negative().BucketCounts().Len() {
+		// Post reset, the reference histogram will have no buckets.
+		return
+	}
+	a.Positive().BucketCounts().FromRaw(subtractExponentialBuckets(a.Positive(), b.Positive()))
+	a.Negative().BucketCounts().FromRaw(subtractExponentialBuckets(a.Negative(), b.Negative()))
+}
+
+// subtractExponentialBuckets subtracts b from a.
+func subtractExponentialBuckets(a, b pmetric.ExponentialHistogramDataPointBuckets) []uint64 {
+	newBuckets := make([]uint64, a.BucketCounts().Len())
+	offsetDiff := int(a.Offset() - b.Offset())
+	for i := 0; i < a.BucketCounts().Len(); i++ {
+		bOffset := i + offsetDiff
+		// if there is no corresponding bucket for the starting BucketCounts, don't normalize
+		if bOffset < 0 || bOffset >= b.BucketCounts().Len() {
+			newBuckets[i] = a.BucketCounts().At(i)
+		} else {
+			newBuckets[i] = a.BucketCounts().At(i) - b.BucketCounts().At(bOffset)
+		}
+	}
+	return newBuckets
 }