fix: Snapshot processor preserves metric aggregation temporality (PIPE-315) (#2541)

preserve metric temporality

Author: Caleb-Hurshman

internal/report/snapshot/filter_metrics.go

@@ -139,6 +139,10 @@ func filterGauge(g pmetric.Gauge, queryMatchesResource, queryMatchesName bool, s
 func filterSum(s pmetric.Sum, queryMatchesResource, queryMatchesName bool, searchQuery *string, minimumTimestamp *time.Time) pmetric.Sum {
 	filteredSum := pmetric.NewSum()
 
+	// Copy aggregation temporality and monotonic flag from original sum
+	filteredSum.SetAggregationTemporality(s.AggregationTemporality())
+	filteredSum.SetIsMonotonic(s.IsMonotonic())
+
 	dps := s.DataPoints()
 	for i := 0; i < dps.Len(); i++ {
 		dp := dps.At(i)
@@ -153,6 +157,9 @@ func filterSum(s pmetric.Sum, queryMatchesResource, queryMatchesName bool, searc
 func filterHistogram(h pmetric.Histogram, queryMatchesResource, queryMatchesName bool, searchQuery *string, minimumTimestamp *time.Time) pmetric.Histogram {
 	filteredHistogram := pmetric.NewHistogram()
 
+	// Copy aggregation temporality from original histogram
+	filteredHistogram.SetAggregationTemporality(h.AggregationTemporality())
+
 	dps := h.DataPoints()
 	for i := 0; i < dps.Len(); i++ {
 		dp := dps.At(i)
@@ -167,6 +174,9 @@ func filterHistogram(h pmetric.Histogram, queryMatchesResource, queryMatchesName
 func filterExponentialHistogram(eh pmetric.ExponentialHistogram, queryMatchesResource, queryMatchesName bool, searchQuery *string, minimumTimestamp *time.Time) pmetric.ExponentialHistogram {
 	filteredExponentialHistogram := pmetric.NewExponentialHistogram()
 
+	// Copy aggregation temporality from original exponential histogram
+	filteredExponentialHistogram.SetAggregationTemporality(eh.AggregationTemporality())
+
 	dps := eh.DataPoints()
 	for i := 0; i < dps.Len(); i++ {
 		dp := dps.At(i)

internal/report/snapshot/testdata/metrics/after/filters-timestamp.yaml

@@ -12,6 +12,7 @@ resourceMetrics:
           - description: Filesystem bytes used.
             name: system.filesystem.usage
             sum:
+              aggregationTemporality: 2
               dataPoints:
                 - asInt: "220672"
                   attributes:

internal/report/snapshot/testdata/metrics/after/matches-attr-key-exp-histogram.yaml

@@ -3,6 +3,7 @@ resourceMetrics:
     scopeMetrics:
       - metrics:
           - exponentialHistogram:
+              aggregationTemporality: 2
               dataPoints:
                 - attributes:
                     - key: prod-machine

internal/report/snapshot/testdata/metrics/after/matches-attr-key-histogram.yaml

@@ -3,6 +3,7 @@ resourceMetrics:
     scopeMetrics:
       - metrics:
           - histogram:
+              aggregationTemporality: 2
               dataPoints:
                 - attributes:
                     - key: prod-machine

internal/report/snapshot/testdata/metrics/after/matches-attr-key-sum.yaml

@@ -12,6 +12,7 @@ resourceMetrics:
           - description: Filesystem bytes used.
             name: system.filesystem.usage
             sum:
+              aggregationTemporality: 2
               dataPoints:
                 - asInt: "8717185024"
                   attributes:

internal/report/snapshot/testdata/metrics/after/matches-attr-val-exp-histogram.yaml

@@ -3,6 +3,7 @@ resourceMetrics:
     scopeMetrics:
       - metrics:
           - exponentialHistogram:
+              aggregationTemporality: 2
               dataPoints:
                 - attributes:
                     - key: dev-machine

internal/report/snapshot/testdata/metrics/after/matches-attr-val-histogram.yaml

@@ -3,6 +3,7 @@ resourceMetrics:
     scopeMetrics:
       - metrics:
           - histogram:
+              aggregationTemporality: 2
               dataPoints:
                 - attributes:
                     - key: dev-machine

internal/report/snapshot/testdata/metrics/after/matches-attr-val-sum.yaml

@@ -12,6 +12,7 @@ resourceMetrics:
           - description: Filesystem bytes used.
             name: system.filesystem.usage
             sum:
+              aggregationTemporality: 2
               dataPoints:
                 - asInt: "8717185024"
                   attributes:

processor/snapshotprocessor/processor_test.go

@@ -228,6 +228,97 @@ func TestProcess_Traces(t *testing.T) {
 	require.Equal(t, "reportSnapshot", mockOpamp.sentMessageType)
 }
 
+func TestProcess_Metrics_PreservesTemporalityWithFiltering(t *testing.T) {
+	factory := NewFactory()
+	sink := &consumertest.MetricsSink{}
+
+	pSet := processortest.NewNopSettings(componentType)
+	p, err := factory.CreateMetrics(context.Background(), pSet, factory.CreateDefaultConfig(), sink)
+	require.NoError(t, err)
+
+	mockOpamp := &mockOpAMPExtension{
+		msgChan: make(chan *protobufs.CustomMessage, 1),
+	}
+
+	mockHost := &mockHost{
+		extensions: map[component.ID]component.Component{
+			component.MustNewID("opamp"): mockOpamp,
+		},
+	}
+
+	require.NoError(t, p.Start(context.Background(), mockHost))
+	t.Cleanup(func() {
+		require.NoError(t, p.Shutdown(context.Background()))
+	})
+
+	// Load test metrics with different aggregation temporalities
+	m, err := golden.ReadMetrics(filepath.Join("testdata", "metrics", "temporality-metrics.yaml"))
+	require.NoError(t, err)
+
+	require.NoError(t, p.ConsumeMetrics(context.Background(), m))
+
+	// Request buffer with search query (this triggers the filtering code path where the bug occurred)
+	reqPayload := fmt.Sprintf(`{"processor":%q,"pipeline_type":"metrics","session_id":"filtering-test","search_query":"transmit"}`, pSet.ID)
+
+	cm := &protobufs.CustomMessage{
+		Capability: "com.bindplane.snapshot",
+		Type:       "requestSnapshot",
+		Data:       []byte(reqPayload),
+	}
+
+	mockOpamp.msgChan <- cm
+
+	// Wait for response
+	require.Eventually(t, func() bool {
+		return mockOpamp.GotMessage()
+	}, 5*time.Second, 100*time.Millisecond)
+
+	// Parse the actual response
+	var actualMessageContents map[string]any
+	err = json.Unmarshal(gunzipBytes(t, mockOpamp.sentMessage), &actualMessageContents)
+	require.NoError(t, err)
+
+	// Verify filtering worked and only "transmit" metric is present
+	telemetryPayload := actualMessageContents["telemetry_payload"].(map[string]any)
+	resourceMetrics := telemetryPayload["resourceMetrics"].([]any)
+	require.Len(t, resourceMetrics, 1, "Should have one resource metric after filtering")
+
+	firstResource := resourceMetrics[0].(map[string]any)
+	scopeMetrics := firstResource["scopeMetrics"].([]any)
+	require.Len(t, scopeMetrics, 1, "Should have one scope metric")
+
+	firstScope := scopeMetrics[0].(map[string]any)
+	metrics := firstScope["metrics"].([]any)
+	require.Len(t, metrics, 1, "Should have one metric matching 'transmit' filter")
+
+	// Verify the filtered metric is the correct one and has preserved aggregation temporality
+	filteredMetric := metrics[0].(map[string]any)
+	require.Equal(t, "system.network.io", filteredMetric["name"])
+
+	sum := filteredMetric["sum"].(map[string]any)
+	require.Equal(t, float64(2), sum["aggregationTemporality"], "Aggregation temporality should be preserved as Cumulative (2) even after filtering")
+	require.Equal(t, true, sum["isMonotonic"], "IsMonotonic should be preserved after filtering")
+
+	// Verify the data point attributes contain "transmit"
+	dataPoints := sum["dataPoints"].([]any)
+	require.Len(t, dataPoints, 1, "Should have one data point")
+
+	dataPoint := dataPoints[0].(map[string]any)
+	attributes := dataPoint["attributes"].([]any)
+	foundTransmit := false
+	for _, attrAny := range attributes {
+		attr := attrAny.(map[string]any)
+		if attr["key"] == "direction" {
+			value := attr["value"].(map[string]any)
+			if value["stringValue"] == "transmit" {
+				foundTransmit = true
+				break
+			}
+		}
+	}
+	require.True(t, foundTransmit, "Filtered metric should contain 'transmit' attribute")
+}
+
 // mockHost for component.Host
 type mockHost struct {
 	extensions map[component.ID]component.Component

processor/snapshotprocessor/testdata/metrics/temporality-metrics.yaml

@@ -0,0 +1,61 @@
+resourceMetrics:
+  - resource:
+      attributes:
+        - key: host.name
+          value:
+            stringValue: test-host
+        - key: service.name
+          value:
+            stringValue: temporality-test
+    schemaUrl: https://opentelemetry.io/schemas/1.9.0
+    scopeMetrics:
+      - metrics:
+          # Delta temporality metric
+          - description: Request count with delta temporality.
+            name: http.server.requests
+            sum:
+              aggregationTemporality: 1
+              isMonotonic: true
+              dataPoints:
+                - asInt: "42"
+                  attributes:
+                    - key: method
+                      value:
+                        stringValue: GET
+                    - key: status_code
+                      value:
+                        stringValue: "200"
+                  startTimeUnixNano: "1000000"
+                  timeUnixNano: "2000000"
+          # Cumulative temporality metric
+          - description: Total bytes processed with cumulative temporality.
+            name: system.network.io
+            sum:
+              aggregationTemporality: 2
+              isMonotonic: true
+              dataPoints:
+                - asInt: "1024000"
+                  attributes:
+                    - key: direction
+                      value:
+                        stringValue: transmit
+                    - key: device
+                      value:
+                        stringValue: eth0
+                  startTimeUnixNano: "1000000"
+                  timeUnixNano: "2000000"
+          # Gauge metric (no aggregation temporality)
+          - description: Current memory usage.
+            name: system.memory.usage
+            gauge:
+              dataPoints:
+                - asDouble: 85.5
+                  attributes:
+                    - key: state
+                      value:
+                        stringValue: used
+                  startTimeUnixNano: "1000000"
+                  timeUnixNano: "2000000"
+        scope:
+          name: otelcol/test
+          version: v1.0.0