fix delta time

Author: shivanthzen

connector/spanmetricsconnector/connector.go

@@ -279,6 +279,16 @@ func (p *connectorImp) buildMetrics() pmetric.Metrics {
 		 * - For delta metrics: (T1, T2), (T2, T3), (T3, T4) ...
 		 */
 		deltaMetricKeys := make(map[metrics.Key]bool)
+		timeStampGenerator := func(mk metrics.Key, startTime pcommon.Timestamp) pcommon.Timestamp {
+			if p.config.GetAggregationTemporality() == pmetric.AggregationTemporalityDelta {
+				if lastTimestamp, ok := p.lastDeltaTimestamps.Get(mk); ok {
+					startTime = lastTimestamp
+				}
+				// Collect lastDeltaTimestamps keys that need to be updated. Metrics can share the same key, so defer the update.
+				deltaMetricKeys[mk] = true
+			}
+			return startTime
+		}
 
 		metricsNamespace := p.config.Namespace
 		if legacyMetricNamesFeatureGate.IsEnabled() && metricsNamespace == DefaultNamespace {
@@ -288,21 +298,21 @@ func (p *connectorImp) buildMetrics() pmetric.Metrics {
 		sums := rawMetrics.sums
 		metric := sm.Metrics().AppendEmpty()
 		metric.SetName(buildMetricName(metricsNamespace, metricNameCalls))
-		sums.BuildMetrics(metric, timestamp, p.config.GetAggregationTemporality())
+		sums.BuildMetrics(metric, timestamp, timeStampGenerator, p.config.GetAggregationTemporality())
 
 		if !p.config.Histogram.Disable {
 			histograms := rawMetrics.histograms
 			metric = sm.Metrics().AppendEmpty()
 			metric.SetName(buildMetricName(metricsNamespace, metricNameDuration))
 			metric.SetUnit(p.config.Histogram.Unit.String())
-			histograms.BuildMetrics(metric, timestamp, p.config.GetAggregationTemporality())
+			histograms.BuildMetrics(metric, timestamp, timeStampGenerator, p.config.GetAggregationTemporality())
 		}
 
 		events := rawMetrics.events
 		if p.events.Enabled {
 			metric = sm.Metrics().AppendEmpty()
 			metric.SetName(buildMetricName(metricsNamespace, metricNameEvents))
-			events.BuildMetrics(metric, timestamp, p.config.GetAggregationTemporality())
+			events.BuildMetrics(metric, timestamp, timeStampGenerator, p.config.GetAggregationTemporality())
 		}
 
 		for mk := range deltaMetricKeys {

connector/spanmetricsconnector/internal/metrics/metrics.go

@@ -15,7 +15,7 @@ type Key string
 
 type HistogramMetrics interface {
 	GetOrCreate(key Key, attributes pcommon.Map, startTimestamp pcommon.Timestamp) Histogram
-	BuildMetrics(pmetric.Metric, pcommon.Timestamp, pmetric.AggregationTemporality)
+	BuildMetrics(pmetric.Metric, pcommon.Timestamp, func(Key, pcommon.Timestamp) pcommon.Timestamp, pmetric.AggregationTemporality)
 	ClearExemplars()
 }
 
@@ -48,8 +48,7 @@ type explicitHistogram struct {
 
 	maxExemplarCount *int
 
-	startTimestamp    pcommon.Timestamp
-	lastSeenTimestamp pcommon.Timestamp
+	startTimestamp pcommon.Timestamp
 }
 
 type exponentialHistogram struct {
@@ -60,8 +59,7 @@ type exponentialHistogram struct {
 
 	maxExemplarCount *int
 
-	startTimestamp    pcommon.Timestamp
-	lastSeenTimestamp pcommon.Timestamp
+	startTimestamp pcommon.Timestamp
 }
 
 type generateStartTimestamp = func(Key) pcommon.Timestamp
@@ -91,31 +89,26 @@ func (m *explicitHistogramMetrics) GetOrCreate(key Key, attributes pcommon.Map,
 			bounds:           m.bounds,
 			bucketCounts:     make([]uint64, len(m.bounds)+1),
 			maxExemplarCount: m.maxExemplarCount,
+			startTimestamp:   startTimestamp,
 		}
 		m.metrics[key] = h
 	}
-
-	h.lastSeenTimestamp = startTimestamp
 	return h
 }
 
 func (m *explicitHistogramMetrics) BuildMetrics(
 	metric pmetric.Metric,
 	timestamp pcommon.Timestamp,
+	startTimeStampGenerator func(Key, pcommon.Timestamp) pcommon.Timestamp,
 	temporality pmetric.AggregationTemporality,
 ) {
 	metric.SetEmptyHistogram().SetAggregationTemporality(temporality)
 	dps := metric.Histogram().DataPoints()
 	dps.EnsureCapacity(len(m.metrics))
-	for _, h := range m.metrics {
+	for k, h := range m.metrics {
 		dp := dps.AppendEmpty()
-		var startTimeStamp pcommon.Timestamp
-		if temporality == pmetric.AggregationTemporalityDelta {
-			startTimeStamp = h.lastSeenTimestamp
-		} else {
-			startTimeStamp = h.startTimestamp
-		}
-		dp.SetStartTimestamp(startTimeStamp)
+		startTimestamp := startTimeStampGenerator(k, h.startTimestamp)
+		dp.SetStartTimestamp(startTimestamp)
 		dp.SetTimestamp(timestamp)
 		dp.ExplicitBounds().FromRaw(h.bounds)
 		dp.BucketCounts().FromRaw(h.bucketCounts)
@@ -149,31 +142,26 @@ func (m *exponentialHistogramMetrics) GetOrCreate(key Key, attributes pcommon.Ma
 			attributes:       attributes,
 			exemplars:        pmetric.NewExemplarSlice(),
 			maxExemplarCount: m.maxExemplarCount,
+			startTimestamp:   startTimeStamp,
 		}
 		m.metrics[key] = h
 	}
-
-	h.lastSeenTimestamp = startTimeStamp
 	return h
 }
 
 func (m *exponentialHistogramMetrics) BuildMetrics(
 	metric pmetric.Metric,
 	timestamp pcommon.Timestamp,
+	startTimeStampGenerator func(Key, pcommon.Timestamp) pcommon.Timestamp,
 	temporality pmetric.AggregationTemporality,
 ) {
 	metric.SetEmptyExponentialHistogram().SetAggregationTemporality(temporality)
 	dps := metric.ExponentialHistogram().DataPoints()
 	dps.EnsureCapacity(len(m.metrics))
-	for _, e := range m.metrics {
+	for k, e := range m.metrics {
 		dp := dps.AppendEmpty()
-		var startTimeStamp pcommon.Timestamp
-		if temporality == pmetric.AggregationTemporalityDelta {
-			startTimeStamp = e.lastSeenTimestamp
-		} else {
-			startTimeStamp = e.startTimestamp
-		}
-		dp.SetStartTimestamp(startTimeStamp)
+		startTimestamp := startTimeStampGenerator(k, e.startTimestamp)
+		dp.SetStartTimestamp(startTimestamp)
 		dp.SetTimestamp(timestamp)
 		expoHistToExponentialDataPoint(e.histogram, dp)
 		for i := 0; i < e.exemplars.Len(); i++ {
@@ -261,8 +249,7 @@ type Sum struct {
 	exemplars        pmetric.ExemplarSlice
 	maxExemplarCount *int
 
-	startTimestamp    pcommon.Timestamp
-	lastSeenTimestamp pcommon.Timestamp
+	startTimestamp pcommon.Timestamp
 }
 
 func (s *Sum) Add(value uint64) {
@@ -292,7 +279,6 @@ func (m *SumMetrics) GetOrCreate(key Key, attributes pcommon.Map, startTimestamp
 		}
 		m.metrics[key] = s
 	}
-	s.lastSeenTimestamp = startTimestamp
 	return s
 }
 
@@ -309,21 +295,17 @@ func (s *Sum) AddExemplar(traceID pcommon.TraceID, spanID pcommon.SpanID, value
 func (m *SumMetrics) BuildMetrics(
 	metric pmetric.Metric,
 	timestamp pcommon.Timestamp,
+	startTimeStampGenerator func(Key, pcommon.Timestamp) pcommon.Timestamp,
 	temporality pmetric.AggregationTemporality,
 ) {
 	metric.SetEmptySum().SetIsMonotonic(true)
 	metric.Sum().SetAggregationTemporality(temporality)
 
 	dps := metric.Sum().DataPoints()
 	dps.EnsureCapacity(len(m.metrics))
-	for _, s := range m.metrics {
+	for k, s := range m.metrics {
 		dp := dps.AppendEmpty()
-		var startTimeStamp pcommon.Timestamp
-		if temporality == pmetric.AggregationTemporalityDelta {
-			startTimeStamp = s.lastSeenTimestamp
-		} else {
-			startTimeStamp = s.startTimestamp
-		}
+		startTimeStamp := startTimeStampGenerator(k, s.startTimestamp)
 		dp.SetStartTimestamp(startTimeStamp)
 		dp.SetTimestamp(timestamp)
 		dp.SetIntValue(int64(s.count))