[chore][connector/routing] Add ability to route metrics and traces by…

… request context

.chloggen/routing-connector-by-request.yaml

@@ -7,7 +7,7 @@ change_type: enhancement
 component: routingconnector
 
 # A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
-note: Add ability to route logs by request metadata.
+note: Add ability to route by request metadata.
 
 # Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
 issues: [19738]

connector/routingconnector/README.md

@@ -44,7 +44,6 @@ The following settings are available:
 ### Limitations
 
 - The `match_once` setting is only supported when using the `resource` context. If any routes use `log` or `request` context, `match_once` must be set to `true`.
-- The `request` context is only supported for logs at this time.
 - The `request` context requires use of the `condition` setting, and relies on a very limited grammar. Conditions must be in the form of `request["key"] == "value"` or `request["key"] != "value"`. (In the future, this grammar may be expanded to support more complex conditions.)
 
 ### Supported [OTTL] functions

connector/routingconnector/metrics.go

@@ -74,6 +74,11 @@ func (c *metricsConnector) switchMetrics(ctx context.Context, md pmetric.Metrics
 		route := c.router.routeSlice[i]
 		matchedMetrics := pmetric.NewMetrics()
 		switch route.statementContext {
+		case "request":
+			if route.requestCondition.matchRequest(ctx) {
+				groupAllMetrics(groups, route.consumer, md)
+				md = pmetric.NewMetrics() // all metrics have been routed
+			}
 		case "", "resource":
 			pmetricutil.MoveResourcesIf(md, matchedMetrics,
 				func(rs pmetric.ResourceMetrics) bool {

connector/routingconnector/metrics_test.go

@@ -504,12 +504,21 @@ func TestMetricsConnectorCapabilities(t *testing.T) {
 
 func TestMetricsConnectorDetailed(t *testing.T) {
 	testCases := []string{
+		filepath.Join("testdata", "metrics", "request_context", "match_any_value"),
+		filepath.Join("testdata", "metrics", "request_context", "match_grpc_value"),
+		filepath.Join("testdata", "metrics", "request_context", "match_http_value"),
+		filepath.Join("testdata", "metrics", "request_context", "match_http_value2"),
+		filepath.Join("testdata", "metrics", "request_context", "match_no_grpc_value"),
+		filepath.Join("testdata", "metrics", "request_context", "match_no_http_value"),
+		filepath.Join("testdata", "metrics", "request_context", "no_request_values"),
 		filepath.Join("testdata", "metrics", "resource_context", "all_match_first_only"),
 		filepath.Join("testdata", "metrics", "resource_context", "all_match_last_only"),
 		filepath.Join("testdata", "metrics", "resource_context", "all_match_once"),
 		filepath.Join("testdata", "metrics", "resource_context", "each_matches_one"),
 		filepath.Join("testdata", "metrics", "resource_context", "match_none_with_default"),
 		filepath.Join("testdata", "metrics", "resource_context", "match_none_without_default"),
+		filepath.Join("testdata", "metrics", "mixed_context", "match_resource_then_grpc_request"),
+		filepath.Join("testdata", "metrics", "mixed_context", "match_resource_then_http_request"),
 	}
 
 	for _, tt := range testCases {
@@ -539,10 +548,17 @@ func TestMetricsConnectorDetailed(t *testing.T) {
 			)
 			require.NoError(t, err)
 
+			ctx := context.Background()
+			if ctxFromFile, readErr := createContextFromFile(t, filepath.Join(tt, "request.yaml")); readErr == nil {
+				ctx = ctxFromFile
+			} else if !os.IsNotExist(readErr) {
+				t.Fatalf("Error reading request.yaml: %v", readErr)
+			}
+
 			input, readErr := golden.ReadMetrics(filepath.Join("testdata", "metrics", "input.yaml"))
 			require.NoError(t, readErr)
 
-			require.NoError(t, conn.ConsumeMetrics(context.Background(), input))
+			require.NoError(t, conn.ConsumeMetrics(ctx, input))
 
 			assertExpected := func(actual []pmetric.Metrics, filePath string) {
 				expected, err := golden.ReadMetrics(filePath)

connector/routingconnector/testdata/metrics/mixed_context/match_resource_then_grpc_request/config.yaml

@@ -0,0 +1,13 @@
+routing:
+  match_once: true
+  default_pipelines:
+    - metrics/default
+  table:
+    - context: resource
+      condition: attributes["resourceName"] == "resourceA"
+      pipelines:
+        - metrics/0
+    - context: request
+      condition: request["X-Tenant"] == "acme"
+      pipelines:
+        - metrics/1

connector/routingconnector/testdata/metrics/mixed_context/match_resource_then_grpc_request/request.yaml

@@ -0,0 +1,2 @@
+grpc:
+  X-Tenant: acme

connector/routingconnector/testdata/metrics/mixed_context/match_resource_then_grpc_request/sink_0.yaml

@@ -0,0 +1,195 @@
+resourceMetrics:
+  - resource:
+      attributes:
+        - key: resourceName
+          value:
+            stringValue: resourceA
+        - key: resourceNameAgain
+          value:
+            stringValue: resourceA
+    schemaUrl: https://opentelemetry.io/schemas/1.6.1
+    scopeMetrics:
+      - attributes:
+          - key: scopeName
+            value:
+              stringValue: scopeA
+          - key: scopeNameAgain
+            value:
+              stringValue: scopeA
+        metrics:
+          - name: sumMonotonicCumulative
+            sum:
+              aggregationTemporality: 2 # AGGREGATION_TEMPORALITY_CUMULATIVE
+              isMonotonic: true
+              dataPoints:
+                - attributes:
+                    - key: dataPointName
+                      value:
+                        stringValue: dataPointA
+                    - key: dataPointNameAgain
+                      value:
+                        stringValue: dataPointA
+                  asInt: "101"
+                - attributes:
+                    - key: dataPointName
+                      value:
+                        stringValue: dataPointB
+                    - key: dataPointNameAgain
+                      value:
+                        stringValue: dataPointB
+                  asInt: "102"
+          - name: sumNonmonotonicCumulative
+            sum:
+              aggregationTemporality: 2 # AGGREGATION_TEMPORALITY_CUMULATIVE
+              isMonotonic: false
+              dataPoints:
+                - attributes:
+                    - key: dataPointName
+                      value:
+                        stringValue: dataPointA
+                    - key: dataPointNameAgain
+                      value:
+                        stringValue: dataPointA
+                  asInt: "102"
+                - attributes:
+                    - key: dataPointName
+                      value:
+                        stringValue: dataPointB
+                    - key: dataPointNameAgain
+                      value:
+                        stringValue: dataPointB
+                  asInt: "101"
+          - name: sumMonotonicDelta
+            sum:
+              aggregationTemporality: 1 # AGGREGATION_TEMPORALITY_DELTA
+              isMonotonic: true
+              dataPoints:
+                - attributes:
+                    - key: dataPointName
+                      value:
+                        stringValue: dataPointA
+                    - key: dataPointNameAgain
+                      value:
+                        stringValue: dataPointA
+                  asInt: "1"
+                - attributes:
+                    - key: dataPointName
+                      value:
+                        stringValue: dataPointB
+                    - key: dataPointNameAgain
+                      value:
+                        stringValue: dataPointB
+                  asInt: "2"
+          - name: sumNonmonotonicDelta
+            sum:
+              aggregationTemporality: 1 # AGGREGATION_TEMPORALITY_DELTA
+              isMonotonic: false
+              dataPoints:
+                - attributes:
+                    - key: dataPointName
+                      value:
+                        stringValue: dataPointA
+                    - key: dataPointNameAgain
+                      value:
+                        stringValue: dataPointA
+                  asInt: "2"
+                - attributes:
+                    - key: dataPointName
+                      value:
+                        stringValue: dataPointB
+                    - key: dataPointNameAgain
+                      value:
+                        stringValue: dataPointB
+                  asInt: "1"
+      - attributes:
+          - key: scopeName
+            value:
+              stringValue: scopeB
+          - key: scopeNameAgain
+            value:
+              stringValue: scopeB
+        metrics:
+          - name: sumMonotonicCumulative
+            sum:
+              aggregationTemporality: 2 # AGGREGATION_TEMPORALITY_CUMULATIVE
+              isMonotonic: true
+              dataPoints:
+                - attributes:
+                    - key: dataPointName
+                      value:
+                        stringValue: dataPointA
+                    - key: dataPointNameAgain
+                      value:
+                        stringValue: dataPointA
+                  asInt: "101"
+                - attributes:
+                    - key: dataPointName
+                      value:
+                        stringValue: dataPointB
+                    - key: dataPointNameAgain
+                      value:
+                        stringValue: dataPointB
+                  asInt: "102"
+          - name: sumNonmonotonicCumulative
+            sum:
+              aggregationTemporality: 2 # AGGREGATION_TEMPORALITY_CUMULATIVE
+              isMonotonic: false
+              dataPoints:
+                - attributes:
+                    - key: dataPointName
+                      value:
+                        stringValue: dataPointA
+                    - key: dataPointNameAgain
+                      value:
+                        stringValue: dataPointA
+                  asInt: "102"
+                - attributes:
+                    - key: dataPointName
+                      value:
+                        stringValue: dataPointB
+                    - key: dataPointNameAgain
+                      value:
+                        stringValue: dataPointB
+                  asInt: "101"
+          - name: sumMonotonicDelta
+            sum:
+              aggregationTemporality: 1 # AGGREGATION_TEMPORALITY_DELTA
+              isMonotonic: true
+              dataPoints:
+                - attributes:
+                    - key: dataPointName
+                      value:
+                        stringValue: dataPointA
+                    - key: dataPointNameAgain
+                      value:
+                        stringValue: dataPointA
+                  asInt: "1"
+                - attributes:
+                    - key: dataPointName
+                      value:
+                        stringValue: dataPointB
+                    - key: dataPointNameAgain
+                      value:
+                        stringValue: dataPointB
+                  asInt: "2"
+          - name: sumNonmonotonicDelta
+            sum:
+              aggregationTemporality: 1 # AGGREGATION_TEMPORALITY_DELTA
+              isMonotonic: false
+              dataPoints:
+                - attributes:
+                    - key: dataPointName
+                      value:
+                        stringValue: dataPointA
+                    - key: dataPointNameAgain
+                      value:
+                        stringValue: dataPointA
+                  asInt: "2"
+                - attributes:
+                    - key: dataPointName
+                      value:
+                        stringValue: dataPointB
+                    - key: dataPointNameAgain
+                      value:
+                        stringValue: dataPointB
+                  asInt: "1"