[connector/routing] Add ability to route logs by request context

.chloggen/routing-connector-by-request.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: 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.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [19738]
+
+# (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: []

connector/routingconnector/README.md

@@ -33,15 +33,34 @@ If you are not already familiar with connectors, you may find it helpful to firs
 The following settings are available:
 
 - `table (required)`: the routing table for this connector.
-- `table.context (optional, default: resource)`: the [OTTL Context] in which the statement will be evaluated. Currently, only `resource` and `log` are supported.
+- `table.context (optional, default: resource)`: the [OTTL Context] in which the statement will be evaluated. Currently, only `resource`, `log`, and `request` are supported.
 - `table.statement`: the routing condition provided as the [OTTL] statement. Required if `table.condition` is not provided.
 - `table.condition`: the routing condition provided as the [OTTL] condition. Required if `table.statement` is not provided.
 - `table.pipelines (required)`: the list of pipelines to use when the routing condition is met.
 - `default_pipelines (optional)`: contains the list of pipelines to use when a record does not meet any of specified conditions.
 - `error_mode (optional)`: determines how errors returned from OTTL statements are handled. Valid values are `propagate`, `ignore` and `silent`. If `ignore` or `silent` is used and a statement's condition has an error then the payload will be routed to the default pipelines. When `silent` is used the error is not logged. If not supplied, `propagate` is used.
 - `match_once (optional, default: false)`: determines whether the connector matches multiple statements or not. If enabled, the payload will be routed to the first pipeline in the `table` whose routing condition is met. May only be `false` when used with `resource` context.
 
-### Examples
+### 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.
+
+### Supported [OTTL] functions
+
+- [IsMatch](../../pkg/ottl/ottlfuncs/README.md#IsMatch)
+- [delete_key](../../pkg/ottl/ottlfuncs/README.md#delete_key)
+- [delete_matching_keys](../../pkg/ottl/ottlfuncs/README.md#delete_matching_keys)
+
+## Additional Settings
+
+The full list of settings exposed for this connector are documented [here](./config.go) with detailed sample configuration files:
+
+- [logs](./testdata/config/logs.yaml)
+- [metrics](./testdata/config/metrics.yaml)
+- [traces](./testdata/config/traces.yaml)
+
+## Examples
 
 Route traces based on an attribute:
 
@@ -94,6 +113,48 @@ service:
       exporters: [jaeger/ecorp]
 ```
 
+Route logs based on tenant:
+
+```yaml
+receivers:
+    otlp:
+
+exporters:
+  file/other:
+    path: ./other.log
+  file/acme:
+    path: ./acme.log
+  file/ecorp:
+    path: ./ecorp.log
+
+connectors:
+  routing:
+    match_once: true
+    default_pipelines: [logs/other]
+    table:
+      - context: request
+        condition: reqeust["X-Tenant"] == "acme"
+        pipelines: [logs/acme]
+      - context: request
+        condition: reqeust["X-Tenant"] == "ecorp"
+        pipelines: [logs/ecorp]
+
+service:
+  pipelines:
+    logs/in:
+      receivers: [otlp]
+      exporters: [routing]
+    logs/acme:
+      receivers: [routing]
+      exporters: [file/acme]
+    logs/ecorp:
+      receivers: [routing]
+      exporters: [file/ecorp]
+    logs/other:
+      receivers: [routing]
+      exporters: [file/other]
+```
+
 Route logs based on region:
 
 ```yaml
@@ -180,29 +241,56 @@ service:
       exporters: [file/service2]
 ```
 
-A signal may get matched by routing conditions of more than one routing table entry. In this case, the signal will be routed to all pipelines of matching routes.
-Respectively, if none of the routing conditions met, then a signal is routed to default pipelines.
-
-## Differences between the Routing Connector and Routing Processor
+Route all low level logs to cheap storage. Route the remainder based on tenant:
 
-- The connector will only route using [OTTL] statements which can only be applied to resource attributes. It does not support matching on context values at this time.
-- The connector routes to pipelines, not exporters as the processor does.
+```yaml
+receivers:
+    otlp:
 
-### Supported [OTTL] functions
+exporters:
+  file/cheap:
+    path: ./cheap.log
+  file/acme:
+    path: ./acme.log
+  file/ecorp:
+    path: ./ecorp.log
 
-- [IsMatch](../../pkg/ottl/ottlfuncs/README.md#IsMatch)
-- [delete_key](../../pkg/ottl/ottlfuncs/README.md#delete_key)
-- [delete_matching_keys](../../pkg/ottl/ottlfuncs/README.md#delete_matching_keys)
+connectors:
+  routing:
+    match_once: true
+    table:
+      - context: log
+        condition: severity_number < SEVERITY_NUMBER_ERROR
+        pipelines: [logs/cheap]
+      - context: request
+        condition: reqeust["X-Tenant"] == "acme"
+        pipelines: [logs/acme]
+      - context: request
+        condition: reqeust["X-Tenant"] == "ecorp"
+        pipelines: [logs/ecorp]
 
-## Additional Settings
+service:
+  pipelines:
+    logs/in:
+      receivers: [otlp]
+      exporters: [routing]
+    logs/cheap:
+      receivers: [routing]
+      exporters: [file/cheap]
+    logs/acme:
+      receivers: [routing]
+      exporters: [file/acme]
+    logs/ecorp:
+      receivers: [routing]
+      exporters: [file/ecorp]
+```
 
-The full list of settings exposed for this connector are documented [here](./config.go) with detailed sample configuration files:
+## Differences between the Routing Connector and Routing Processor
 
-- [logs](./testdata/config/logs.yaml)
-- [metrics](./testdata/config/metrics.yaml)
-- [traces](./testdata/config/traces.yaml)
+- Routing on context values is only supported for logs at this time.
+- The connector routes to pipelines, not exporters as the processor does.
 
 [Connectors README]:https://github.com/open-telemetry/opentelemetry-collector/blob/main/connector/README.md
 
 [OTTL]: https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/pkg/ottl/README.md
-[OTTL Context]: https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/pkg/ottl/LANGUAGE.md#contexts
+[OTTL Context]: https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/pkg/ottl/LANGUAGE.md#contexts

connector/routingconnector/config.go

@@ -5,6 +5,7 @@ package routingconnector // import "github.com/open-telemetry/opentelemetry-coll
 
 import (
 	"errors"
+	"fmt"
 
 	"go.opentelemetry.io/collector/pipeline"
 
@@ -59,32 +60,37 @@ func (c *Config) Validate() error {
 		if item.Statement == "" && item.Condition == "" {
 			return errNoConditionOrStatement
 		}
-
 		if item.Statement != "" && item.Condition != "" {
 			return errConditionAndStatement
 		}
-
 		if len(item.Pipelines) == 0 {
 			return errNoPipelines
 		}
 
 		switch item.Context {
 		case "", "resource": // ok
-		case "log":
+		case "request":
+			if item.Statement != "" || item.Condition == "" {
+				return fmt.Errorf("%q context requires a 'condition'", item.Context)
+			}
+			if _, err := parseRequestCondition(item.Condition); err != nil {
+				return err
+			}
+			fallthrough
+		case "log": // ok
 			if !c.MatchOnce {
-				return errors.New("log context is not supported with match_once: false")
+				return fmt.Errorf(`%q context is not supported with "match_once: false"`, item.Context)
 			}
 		default:
 			return errors.New("invalid context: " + item.Context)
 		}
 	}
-
 	return nil
 }
 
 // RoutingTableItem specifies how data should be routed to the different pipelines
 type RoutingTableItem struct {
-	// One of "resource" or "log" (other OTTL contexts will be added in the future)
+	// One of "request", "resource", "log" (other OTTL contexts will be added in the future)
 	// Optional. Default "resource".
 	Context string `mapstructure:"context"`
 

connector/routingconnector/config_test.go

@@ -232,7 +232,37 @@ func TestValidateConfig(t *testing.T) {
 					},
 				},
 			},
-			error: "log context is not supported with match_once: false",
+			error: `"log" context is not supported with "match_once: false"`,
+		},
+		{
+			name: "request context with statement",
+			config: &Config{
+				Table: []RoutingTableItem{
+					{
+						Context:   "request",
+						Statement: `route() where attributes["attr"] == "acme"`,
+						Pipelines: []pipeline.ID{
+							pipeline.NewIDWithName(pipeline.SignalTraces, "otlp"),
+						},
+					},
+				},
+			},
+			error: `"request" context requires a 'condition'`,
+		},
+		{
+			name: "request context with invalid condition",
+			config: &Config{
+				Table: []RoutingTableItem{
+					{
+						Context:   "request",
+						Condition: `attributes["attr"] == "acme"`,
+						Pipelines: []pipeline.ID{
+							pipeline.NewIDWithName(pipeline.SignalTraces, "otlp"),
+						},
+					},
+				},
+			},
+			error: `condition must have format 'request["<name>"] <comparator> <value>'`,
 		},
 	}
 

connector/routingconnector/go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl v0.112.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest v0.112.0
 	github.com/stretchr/testify v1.9.0
+	go.opentelemetry.io/collector/client v1.18.0
 	go.opentelemetry.io/collector/component v0.112.0
 	go.opentelemetry.io/collector/confmap v1.18.0
 	go.opentelemetry.io/collector/connector v0.112.0
@@ -17,6 +18,8 @@ require (
 	go.opentelemetry.io/collector/pipeline v0.112.0
 	go.uber.org/goleak v1.3.0
 	go.uber.org/zap v1.27.0
+	google.golang.org/grpc v1.67.1
+	gopkg.in/yaml.v3 v3.0.1
 )
 
 require (
@@ -68,10 +71,8 @@ require (
 	golang.org/x/sys v0.26.0 // indirect
 	golang.org/x/text v0.19.0 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240822170219-fc7c04adadcd // indirect
-	google.golang.org/grpc v1.67.1 // indirect
 	google.golang.org/protobuf v1.35.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
-	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl => ../../pkg/ottl

connector/routingconnector/logs.go

@@ -73,9 +73,15 @@ func (c *logsConnector) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
 func (c *logsConnector) switchLogs(ctx context.Context, ld plog.Logs) error {
 	groups := make(map[consumer.Logs]plog.Logs)
 	var errs error
-	for _, route := range c.router.routeSlice {
+	for i := 0; i < len(c.router.routeSlice) && ld.LogRecordCount() > 0; i++ {
+		route := c.router.routeSlice[i]
 		matchedLogs := plog.NewLogs()
 		switch route.statementContext {
+		case "request":
+			if route.requestCondition.matchRequest(ctx) {
+				groupAll(groups, route.consumer, ld)
+				ld = plog.NewLogs() // all logs have been routed
+			}
 		case "", "resource":
 			plogutil.MoveResourcesIf(ld, matchedLogs,
 				func(rl plog.ResourceLogs) bool {

connector/routingconnector/logs_test.go

@@ -475,6 +475,13 @@ func TestLogsConnectorCapabilities(t *testing.T) {
 
 func TestLogsConnectorDetailed(t *testing.T) {
 	testCases := []string{
+		filepath.Join("testdata", "logs", "request_context", "match_any_value"),
+		filepath.Join("testdata", "logs", "request_context", "match_grpc_value"),
+		filepath.Join("testdata", "logs", "request_context", "match_http_value"),
+		filepath.Join("testdata", "logs", "request_context", "match_http_value2"),
+		filepath.Join("testdata", "logs", "request_context", "match_no_grpc_value"),
+		filepath.Join("testdata", "logs", "request_context", "match_no_http_value"),
+		filepath.Join("testdata", "logs", "request_context", "no_request_values"),
 		filepath.Join("testdata", "logs", "resource_context", "all_match_first_only"),
 		filepath.Join("testdata", "logs", "resource_context", "all_match_last_only"),
 		filepath.Join("testdata", "logs", "resource_context", "all_match_once"),
@@ -489,8 +496,12 @@ func TestLogsConnectorDetailed(t *testing.T) {
 		filepath.Join("testdata", "logs", "log_context", "with_resource_condition"),
 		filepath.Join("testdata", "logs", "log_context", "with_scope_condition"),
 		filepath.Join("testdata", "logs", "log_context", "with_resource_and_scope_conditions"),
-		filepath.Join("testdata", "logs", "mixed_context", "match_resource_then_logs"),
+		filepath.Join("testdata", "logs", "mixed_context", "match_logs_then_grpc_request"),
+		filepath.Join("testdata", "logs", "mixed_context", "match_logs_then_http_request"),
 		filepath.Join("testdata", "logs", "mixed_context", "match_logs_then_resource"),
+		filepath.Join("testdata", "logs", "mixed_context", "match_resource_then_grpc_request"),
+		filepath.Join("testdata", "logs", "mixed_context", "match_resource_then_http_request"),
+		filepath.Join("testdata", "logs", "mixed_context", "match_resource_then_logs"),
 	}
 
 	for _, tt := range testCases {
@@ -539,10 +550,17 @@ func TestLogsConnectorDetailed(t *testing.T) {
 				t.Fatalf("Error reading sink_default.yaml: %v", readErr)
 			}
 
+			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.ReadLogs(filepath.Join(tt, "input.yaml"))
 			require.NoError(t, readErr)
 
-			require.NoError(t, conn.ConsumeLogs(context.Background(), input))
+			require.NoError(t, conn.ConsumeLogs(ctx, input))
 
 			if expected0 == nil {
 				assert.Empty(t, sink0.AllLogs(), "sink0 should be empty")