[processor/routing] Add error_mode configuration option

.chloggen/rp-error-mode.yaml

@@ -0,0 +1,16 @@
+# 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: routingprocessor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Adds new `error_mode` configuration option that allows specifying how errors returned by OTTL statements should be handled.
+
+# One or more tracking issues related to the change
+issues: [19147]
+
+# (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 a condition errors when using `ignore` the payload will be routed to the default exporter.

processor/routingprocessor/README.md

@@ -60,15 +60,17 @@ To configure the routing processor with [OTTL] routing conditions use the follow
 - `table (required)`: the routing table for this processor.
 - `table.statement (required)`: the routing condition provided as the [OTTL] statement.
 - `table.exporters (required)`: the list of exporters to use when the routing condition is met.
-- `default_exporters (optional)`: contains the list of exporters to use when a record
-does not meet any of specified conditions.
+- `default_exporters (optional)`: contains the list of exporters 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 `ignore` and `propagate`. If `ignored` is used and a statement's condition has an error then the payload will be routed to the default exporter.  If not supplied, `propagate` is used.
+
 
 ```yaml
 
 processors:
   routing:
     default_exporters:
     - jaeger
+    error_mode: ignore
     table:
       - statement: route() where resource.attributes["X-Tenant"] == "acme"
         exporters: [jaeger/acme]

processor/routingprocessor/config.go

@@ -18,6 +18,8 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl"
 )
 
 var (
@@ -55,6 +57,14 @@ type Config struct {
 	// Optional.
 	DropRoutingResourceAttribute bool `mapstructure:"drop_resource_routing_attribute"`
 
+	// ErrorMode determines how the processor reacts to errors that occur while processing an OTTL condition.
+	// Valid values are `ignore` and `propagate`.
+	// `ignore` means the processor ignores errors returned by conditions and continues on to the next condition. This is the recommended mode.
+	// If `ignored` is used and a statement's condition has an error then the payload will be routed to the default exporter.
+	// `propagate` means the processor returns the error up the pipeline.  This will result in the payload being dropped from the collector.
+	// The default value is `propagate`.
+	ErrorMode ottl.ErrorMode `mapstructure:"error_mode"`
+
 	// Table contains the routing table for this processor.
 	// Required.
 	Table []RoutingTableItem `mapstructure:"table"`

processor/routingprocessor/config_test.go

@@ -22,6 +22,8 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/confmap/confmaptest"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl"
 )
 
 func TestLoadConfig(t *testing.T) {
@@ -37,6 +39,7 @@ func TestLoadConfig(t *testing.T) {
 				DefaultExporters: []string{"otlp"},
 				AttributeSource:  "context",
 				FromAttribute:    "X-Tenant",
+				ErrorMode:        ottl.PropagateError,
 				Table: []RoutingTableItem{
 					{
 						Value:     "acme",
@@ -56,6 +59,7 @@ func TestLoadConfig(t *testing.T) {
 				DefaultExporters: []string{"logging/default"},
 				AttributeSource:  "context",
 				FromAttribute:    "X-Custom-Metrics-Header",
+				ErrorMode:        ottl.PropagateError,
 				Table: []RoutingTableItem{
 					{
 						Value:     "acme",
@@ -75,6 +79,7 @@ func TestLoadConfig(t *testing.T) {
 				DefaultExporters: []string{"logging/default"},
 				AttributeSource:  "context",
 				FromAttribute:    "X-Custom-Logs-Header",
+				ErrorMode:        ottl.PropagateError,
 				Table: []RoutingTableItem{
 					{
 						Value:     "acme",
@@ -94,6 +99,7 @@ func TestLoadConfig(t *testing.T) {
 				DefaultExporters: []string{"jaeger"},
 				AttributeSource:  resourceAttributeSource,
 				FromAttribute:    "X-Tenant",
+				ErrorMode:        ottl.IgnoreError,
 				Table: []RoutingTableItem{
 					{
 						Value:     "acme",
@@ -107,6 +113,7 @@ func TestLoadConfig(t *testing.T) {
 			id:         component.NewIDWithName(typeStr, "ottl"),
 			expected: &Config{
 				DefaultExporters: []string{"jaeger"},
+				ErrorMode:        ottl.PropagateError,
 				Table: []RoutingTableItem{
 					{
 						Statement: "route() where resource.attributes[\"X-Tenant\"] == \"acme\"",

processor/routingprocessor/factory.go

@@ -21,6 +21,8 @@ import (
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/processor"
 	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl"
 )
 
 const (
@@ -44,6 +46,7 @@ func NewFactory() processor.Factory {
 func createDefaultConfig() component.Config {
 	return &Config{
 		AttributeSource: defaultAttributeSource,
+		ErrorMode:       ottl.PropagateError,
 	}
 }
 

processor/routingprocessor/logs.go

@@ -26,6 +26,7 @@ import (
 	"go.uber.org/multierr"
 	"go.uber.org/zap"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottllog"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/routingprocessor/internal/common"
 )
@@ -106,7 +107,11 @@ func (p *logProcessor) route(ctx context.Context, l plog.Logs) error {
 		for key, route := range p.router.routes {
 			_, isMatch, err := route.statement.Execute(ctx, ltx)
 			if err != nil {
-				return err
+				if p.config.ErrorMode == ottl.PropagateError {
+					return err
+				}
+				p.group("", groups, p.router.defaultExporters, rlogs)
+				continue
 			}
 			if !isMatch {
 				matchCount--

processor/routingprocessor/metrics.go

@@ -26,6 +26,7 @@ import (
 	"go.uber.org/multierr"
 	"go.uber.org/zap"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottldatapoint"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/routingprocessor/internal/common"
 )
@@ -108,7 +109,11 @@ func (p *metricsProcessor) route(ctx context.Context, tm pmetric.Metrics) error
 		for key, route := range p.router.routes {
 			_, isMatch, err := route.statement.Execute(ctx, mtx)
 			if err != nil {
-				return err
+				if p.config.ErrorMode == ottl.PropagateError {
+					return err
+				}
+				p.group("", groups, p.router.defaultExporters, rmetrics)
+				continue
 			}
 			if !isMatch {
 				matchCount--

processor/routingprocessor/testdata/config.yaml

@@ -3,6 +3,7 @@ routing:
     - jaeger
   attribute_source: resource
   from_attribute: X-Tenant
+  error_mode: ignore
   table:
     - value: acme
       exporters:

processor/routingprocessor/traces.go

@@ -26,6 +26,7 @@ import (
 	"go.uber.org/multierr"
 	"go.uber.org/zap"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlspan"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/routingprocessor/internal/common"
 )
@@ -106,7 +107,11 @@ func (p *tracesProcessor) route(ctx context.Context, t ptrace.Traces) error {
 		for key, route := range p.router.routes {
 			_, isMatch, err := route.statement.Execute(ctx, stx)
 			if err != nil {
-				return err
+				if p.config.ErrorMode == ottl.PropagateError {
+					return err
+				}
+				p.group("", groups, p.router.defaultExporters, rspans)
+				continue
 			}
 			if !isMatch {
 				matchCount--