[connector/routing] Supports matching the statement only once. (open-…

…telemetry#28888)

**Description:** <Describe what has changed.>
<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
Adding a feature: routing connector supports matching the statement only
once

**Link to tracking Issue:** <Issue number if applicable> open-telemetry#26353

**Testing:** <Describe what testing was performed and which tests were
added.>
Basic tests included for the functionality.

**Documentation:** <Describe the documentation added.>

---------

Co-authored-by: Daniel Jaglowski <[email protected]>

.chloggen/feat_routing_match_once.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: routingconnector supports matching the statement only once
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [26353]
+
+# (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: [user]

connector/routingconnector/README.md

@@ -36,6 +36,7 @@ The following settings are available:
 - `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 `ignored` 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.
 
 Example:
 
@@ -55,12 +56,23 @@ connectors:
  routing:
  default_pipelines: [traces/jaeger]
  error_mode: ignore
+ match_once: false
  table:
  - statement: route() where attributes["X-Tenant"] == "acme"
  pipelines: [traces/jaeger-acme]
  - statement: delete_key(attributes, "X-Tenant") where IsMatch(attributes["X-Tenant"], ".*corp")
  pipelines: [traces/jaeger-ecorp]
 
+ routing/match_once:
+ default_pipelines: [traces/jaeger]
+ error_mode: ignore
+ match_once: true
+ table:
+ - statement: route() where attributes["X-Tenant"] == "acme"
+ pipelines: [traces/jaeger-acme]
+ - statement: route() where attributes["X-Tenant"] == ".*acme"
+ pipelines: [traces/jaeger-ecorp]
+
 service:
  pipelines:
  traces/in:

connector/routingconnector/config.go

@@ -39,6 +39,10 @@ type Config struct {
  // Table contains the routing table for this processor.
  // Required.
  Table []RoutingTableItem `mapstructure:"table"`
+
+ // MatchOnce determines whether the connector matches multiple statements.
+ // Optional.
+ MatchOnce bool `mapstructure:"match_once"`
 }
 
 // Validate checks if the processor configuration is valid.

connector/routingconnector/logs.go

@@ -72,7 +72,7 @@ func (c *logsConnector) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
  rtx := ottlresource.NewTransformContext(rlogs.Resource())
 
  noRoutesMatch := true
- for _, route := range c.router.routes {
+ for _, route := range c.router.routeSlice {
  _, isMatch, err := route.statement.Execute(ctx, rtx)
  if err != nil {
  if c.config.ErrorMode == ottl.PropagateError {
@@ -84,6 +84,9 @@ func (c *logsConnector) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
  if isMatch {
  noRoutesMatch = false
  c.group(groups, route.consumer, rlogs)
+ if c.config.MatchOnce {
+ break
+ }
  }
 
  }

connector/routingconnector/logs_test.go

@@ -229,6 +229,158 @@ func TestLogsAreCorrectlySplitPerResourceAttributeWithOTTL(t *testing.T) {
  })
 }
 
+func TestLogsAreCorrectlyMatchOnceWithOTTL(t *testing.T) {
+ logsDefault := component.NewIDWithName(component.DataTypeLogs, "default")
+ logs0 := component.NewIDWithName(component.DataTypeLogs, "0")
+ logs1 := component.NewIDWithName(component.DataTypeLogs, "1")
+
+ cfg := &Config{
+ DefaultPipelines: []component.ID{logsDefault},
+ Table: []RoutingTableItem{
+ {
+ Statement: `route() where IsMatch(attributes["X-Tenant"], ".*acme") == true`,
+ Pipelines: []component.ID{logs0},
+ },
+ {
+ Statement: `route() where IsMatch(attributes["X-Tenant"], "_acme") == true`,
+ Pipelines: []component.ID{logs1},
+ },
+ {
+ Statement: `route() where attributes["X-Tenant"] == "ecorp"`,
+ Pipelines: []component.ID{logsDefault, logs0},
+ },
+ },
+ MatchOnce: true,
+ }
+
+ var defaultSink, sink0, sink1 consumertest.LogsSink
+
+ router := connectortest.NewLogsRouter(
+ connectortest.WithLogsSink(logsDefault, &defaultSink),
+ connectortest.WithLogsSink(logs0, &sink0),
+ connectortest.WithLogsSink(logs1, &sink1),
+ )
+
+ resetSinks := func() {
+ defaultSink.Reset()
+ sink0.Reset()
+ sink1.Reset()
+ }
+
+ factory := NewFactory()
+ conn, err := factory.CreateLogsToLogs(
+ context.Background(),
+ connectortest.NewNopCreateSettings(),
+ cfg,
+ router.(consumer.Logs),
+ )
+
+ require.NoError(t, err)
+ require.NotNil(t, conn)
+ require.NoError(t, conn.Start(context.Background(), componenttest.NewNopHost()))
+ defer func() {
+ assert.NoError(t, conn.Shutdown(context.Background()))
+ }()
+
+ t.Run("logs matched by no expressions", func(t *testing.T) {
+ resetSinks()
+
+ l := plog.NewLogs()
+ rl := l.ResourceLogs().AppendEmpty()
+ rl.Resource().Attributes().PutStr("X-Tenant", "something-else")
+ rl.ScopeLogs().AppendEmpty().LogRecords().AppendEmpty()
+
+ require.NoError(t, conn.ConsumeLogs(context.Background(), l))
+
+ assert.Len(t, defaultSink.AllLogs(), 1)
+ assert.Len(t, sink0.AllLogs(), 0)
+ assert.Len(t, sink1.AllLogs(), 0)
+ })
+
+ t.Run("logs matched one expression", func(t *testing.T) {
+ resetSinks()
+
+ l := plog.NewLogs()
+
+ rl := l.ResourceLogs().AppendEmpty()
+ rl.Resource().Attributes().PutStr("X-Tenant", "xacme")
+ rl.ScopeLogs().AppendEmpty().LogRecords().AppendEmpty()
+
+ require.NoError(t, conn.ConsumeLogs(context.Background(), l))
+
+ assert.Len(t, defaultSink.AllLogs(), 0)
+ assert.Len(t, sink0.AllLogs(), 1)
+ assert.Len(t, sink1.AllLogs(), 0)
+ })
+
+ t.Run("logs matched by two expressions, but sinks to one", func(t *testing.T) {
+ resetSinks()
+
+ l := plog.NewLogs()
+
+ rl := l.ResourceLogs().AppendEmpty()
+ rl.Resource().Attributes().PutStr("X-Tenant", "x_acme")
+ rl.ScopeLogs().AppendEmpty().LogRecords().AppendEmpty()
+
+ rl = l.ResourceLogs().AppendEmpty()
+ rl.Resource().Attributes().PutStr("X-Tenant", "_acme")
+ rl.ScopeLogs().AppendEmpty().LogRecords().AppendEmpty()
+
+ require.NoError(t, conn.ConsumeLogs(context.Background(), l))
+
+ assert.Len(t, defaultSink.AllLogs(), 0)
+ assert.Len(t, sink0.AllLogs(), 1)
+ assert.Len(t, sink1.AllLogs(), 0)
+
+ assert.Equal(t, sink0.AllLogs()[0].LogRecordCount(), 2)
+ })
+
+ t.Run("one log matched by multiple expressions, other matched none", func(t *testing.T) {
+ resetSinks()
+
+ l := plog.NewLogs()
+
+ rl := l.ResourceLogs().AppendEmpty()
+ rl.Resource().Attributes().PutStr("X-Tenant", "_acme")
+ rl.ScopeLogs().AppendEmpty().LogRecords().AppendEmpty()
+
+ rl = l.ResourceLogs().AppendEmpty()
+ rl.Resource().Attributes().PutStr("X-Tenant", "something-else")
+ rl.ScopeLogs().AppendEmpty().LogRecords().AppendEmpty()
+
+ require.NoError(t, conn.ConsumeLogs(context.Background(), l))
+
+ assert.Len(t, defaultSink.AllLogs(), 1)
+ assert.Len(t, sink0.AllLogs(), 1)
+ assert.Len(t, sink1.AllLogs(), 0)
+
+ rlog := defaultSink.AllLogs()[0].ResourceLogs().At(0)
+ attr, ok := rlog.Resource().Attributes().Get("X-Tenant")
+ assert.True(t, ok, "routing attribute must exists")
+ assert.Equal(t, attr.AsString(), "something-else")
+ })
+
+ t.Run("logs matched by one expression, multiple pipelines", func(t *testing.T) {
+ resetSinks()
+
+ l := plog.NewLogs()
+
+ rl := l.ResourceLogs().AppendEmpty()
+ rl.Resource().Attributes().PutStr("X-Tenant", "ecorp")
+ rl.ScopeLogs().AppendEmpty().LogRecords().AppendEmpty()
+
+ require.NoError(t, conn.ConsumeLogs(context.Background(), l))
+
+ assert.Len(t, defaultSink.AllLogs(), 1)
+ assert.Len(t, sink0.AllLogs(), 1)
+ assert.Len(t, sink1.AllLogs(), 0)
+
+ assert.Equal(t, defaultSink.AllLogs()[0].LogRecordCount(), 1)
+ assert.Equal(t, sink0.AllLogs()[0].LogRecordCount(), 1)
+ assert.Equal(t, defaultSink.AllLogs(), sink0.AllLogs())
+ })
+}
+
 func TestLogsResourceAttributeDroppedByOTTL(t *testing.T) {
  logsDefault := component.NewIDWithName(component.DataTypeLogs, "default")
  logsOther := component.NewIDWithName(component.DataTypeLogs, "other")

connector/routingconnector/metrics.go

@@ -72,7 +72,7 @@ func (c *metricsConnector) ConsumeMetrics(ctx context.Context, md pmetric.Metric
  rtx := ottlresource.NewTransformContext(rmetrics.Resource())
 
  noRoutesMatch := true
- for _, route := range c.router.routes {
+ for _, route := range c.router.routeSlice {
  _, isMatch, err := route.statement.Execute(ctx, rtx)
  if err != nil {
  if c.config.ErrorMode == ottl.PropagateError {
@@ -84,6 +84,9 @@ func (c *metricsConnector) ConsumeMetrics(ctx context.Context, md pmetric.Metric
  if isMatch {
  noRoutesMatch = false
  c.group(groups, route.consumer, rmetrics)
+ if c.config.MatchOnce {
+ break
+ }
  }
 
  }