[processor/filter] Create With*Functions factory options to register custom OTTL funcs programatically. (#40933)

Co-authored-by: Matthew Wear <[email protected]>
Co-authored-by: Evan Bradley <[email protected]>

Author: 3 people

.chloggen/fcovalente-filterprocessor-with-functions.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: filterprocessor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Create `With*Functions` factory options to provide custom OTTL functions for logs, metrics or traces to the resulting filter processor.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [40948]
+
+# (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: [api]

processor/filterprocessor/config.go

@@ -18,6 +18,11 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter/filterset"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter/filterset/regexp"
 	"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/pkg/ottl/contexts/ottllog"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlmetric"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlspan"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlspanevent"
 )
 
 // Config defines configuration for Resource processor.
@@ -36,6 +41,12 @@ type Config struct {
 	Spans filterconfig.MatchConfig `mapstructure:"spans"`
 
 	Traces TraceFilters `mapstructure:"traces"`
+
+	dataPointFunctions map[string]ottl.Factory[ottldatapoint.TransformContext]
+	logFunctions       map[string]ottl.Factory[ottllog.TransformContext]
+	metricFunctions    map[string]ottl.Factory[ottlmetric.TransformContext]
+	spanEventFunctions map[string]ottl.Factory[ottlspanevent.TransformContext]
+	spanFunctions      map[string]ottl.Factory[ottlspan.TransformContext]
 }
 
 // MetricFilters filters by Metric properties.
@@ -279,27 +290,27 @@ func (cfg *Config) Validate() error {
 	var errors error
 
 	if cfg.Traces.SpanConditions != nil {
-		_, err := filterottl.NewBoolExprForSpan(cfg.Traces.SpanConditions, filterottl.StandardSpanFuncs(), ottl.PropagateError, component.TelemetrySettings{Logger: zap.NewNop()})
+		_, err := filterottl.NewBoolExprForSpan(cfg.Traces.SpanConditions, cfg.spanFunctions, ottl.PropagateError, component.TelemetrySettings{Logger: zap.NewNop()})
 		errors = multierr.Append(errors, err)
 	}
 
 	if cfg.Traces.SpanEventConditions != nil {
-		_, err := filterottl.NewBoolExprForSpanEvent(cfg.Traces.SpanEventConditions, filterottl.StandardSpanEventFuncs(), ottl.PropagateError, component.TelemetrySettings{Logger: zap.NewNop()})
+		_, err := filterottl.NewBoolExprForSpanEvent(cfg.Traces.SpanEventConditions, cfg.spanEventFunctions, ottl.PropagateError, component.TelemetrySettings{Logger: zap.NewNop()})
 		errors = multierr.Append(errors, err)
 	}
 
 	if cfg.Metrics.MetricConditions != nil {
-		_, err := filterottl.NewBoolExprForMetric(cfg.Metrics.MetricConditions, filterottl.StandardMetricFuncs(), ottl.PropagateError, component.TelemetrySettings{Logger: zap.NewNop()})
+		_, err := filterottl.NewBoolExprForMetric(cfg.Metrics.MetricConditions, cfg.metricFunctions, ottl.PropagateError, component.TelemetrySettings{Logger: zap.NewNop()})
 		errors = multierr.Append(errors, err)
 	}
 
 	if cfg.Metrics.DataPointConditions != nil {
-		_, err := filterottl.NewBoolExprForDataPoint(cfg.Metrics.DataPointConditions, filterottl.StandardDataPointFuncs(), ottl.PropagateError, component.TelemetrySettings{Logger: zap.NewNop()})
+		_, err := filterottl.NewBoolExprForDataPoint(cfg.Metrics.DataPointConditions, cfg.dataPointFunctions, ottl.PropagateError, component.TelemetrySettings{Logger: zap.NewNop()})
 		errors = multierr.Append(errors, err)
 	}
 
 	if cfg.Logs.LogConditions != nil {
-		_, err := filterottl.NewBoolExprForLog(cfg.Logs.LogConditions, filterottl.StandardLogFuncs(), ottl.PropagateError, component.TelemetrySettings{Logger: zap.NewNop()})
+		_, err := filterottl.NewBoolExprForLog(cfg.Logs.LogConditions, cfg.logFunctions, ottl.PropagateError, component.TelemetrySettings{Logger: zap.NewNop()})
 		errors = multierr.Append(errors, err)
 	}
 

processor/filterprocessor/config_test.go

@@ -21,6 +21,25 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/filterprocessor/internal/metadata"
 )
 
+func assertConfigContainsDefaultFunctions(t *testing.T, config Config) {
+	t.Helper()
+	for _, f := range DefaultLogFunctions() {
+		assert.Contains(t, config.logFunctions, f.Name(), "missing log function %v", f.Name())
+	}
+	for _, f := range DefaultDataPointFunctions() {
+		assert.Contains(t, config.dataPointFunctions, f.Name(), "missing data point function %v", f.Name())
+	}
+	for _, f := range DefaultMetricFunctions() {
+		assert.Contains(t, config.metricFunctions, f.Name(), "missing metric function %v", f.Name())
+	}
+	for _, f := range DefaultSpanFunctions() {
+		assert.Contains(t, config.spanFunctions, f.Name(), "missing span function %v", f.Name())
+	}
+	for _, f := range DefaultSpanEventFunctions() {
+		assert.Contains(t, config.spanEventFunctions, f.Name(), "missing span event function %v", f.Name())
+	}
+}
+
 // TestLoadingConfigRegexp tests loading testdata/config_strict.yaml
 func TestLoadingConfigStrict(t *testing.T) {
 	// list of filters used repeatedly on testdata/config_strict.yaml
@@ -91,7 +110,8 @@ func TestLoadingConfigStrict(t *testing.T) {
 			require.NoError(t, sub.Unmarshal(cfg))
 
 			assert.NoError(t, xconfmap.Validate(cfg))
-			assert.Equal(t, tt.expected, cfg)
+			assert.EqualExportedValues(t, tt.expected, cfg)
+			assertConfigContainsDefaultFunctions(t, *cfg.(*Config))
 		})
 	}
 }
@@ -173,7 +193,8 @@ func TestLoadingConfigStrictLogs(t *testing.T) {
 			require.NoError(t, sub.Unmarshal(cfg))
 
 			assert.NoError(t, xconfmap.Validate(cfg))
-			assert.Equal(t, tt.expected, cfg)
+			assert.EqualExportedValues(t, tt.expected, cfg)
+			assertConfigContainsDefaultFunctions(t, *cfg.(*Config))
 		})
 	}
 }
@@ -235,7 +256,8 @@ func TestLoadingConfigSeverityLogsStrict(t *testing.T) {
 			require.NoError(t, sub.Unmarshal(cfg))
 
 			assert.NoError(t, xconfmap.Validate(cfg))
-			assert.Equal(t, tt.expected, cfg)
+			assert.EqualExportedValues(t, tt.expected, cfg)
+			assertConfigContainsDefaultFunctions(t, *cfg.(*Config))
 		})
 	}
 }
@@ -297,7 +319,8 @@ func TestLoadingConfigSeverityLogsRegexp(t *testing.T) {
 			require.NoError(t, sub.Unmarshal(cfg))
 
 			assert.NoError(t, xconfmap.Validate(cfg))
-			assert.Equal(t, tt.expected, cfg)
+			assert.EqualExportedValues(t, tt.expected, cfg)
+			assertConfigContainsDefaultFunctions(t, *cfg.(*Config))
 		})
 	}
 }
@@ -359,7 +382,8 @@ func TestLoadingConfigBodyLogsStrict(t *testing.T) {
 			require.NoError(t, sub.Unmarshal(cfg))
 
 			assert.NoError(t, xconfmap.Validate(cfg))
-			assert.Equal(t, tt.expected, cfg)
+			assert.EqualExportedValues(t, tt.expected, cfg)
+			assertConfigContainsDefaultFunctions(t, *cfg.(*Config))
 		})
 	}
 }
@@ -421,7 +445,8 @@ func TestLoadingConfigBodyLogsRegexp(t *testing.T) {
 			require.NoError(t, sub.Unmarshal(cfg))
 
 			assert.NoError(t, xconfmap.Validate(cfg))
-			assert.Equal(t, tt.expected, cfg)
+			assert.EqualExportedValues(t, tt.expected, cfg)
+			assertConfigContainsDefaultFunctions(t, *cfg.(*Config))
 		})
 	}
 }
@@ -486,7 +511,8 @@ func TestLoadingConfigMinSeverityNumberLogs(t *testing.T) {
 			require.NoError(t, sub.Unmarshal(cfg))
 
 			assert.NoError(t, xconfmap.Validate(cfg))
-			assert.Equal(t, tt.expected, cfg)
+			assert.EqualExportedValues(t, tt.expected, cfg)
+			assertConfigContainsDefaultFunctions(t, *cfg.(*Config))
 		})
 	}
 }
@@ -575,7 +601,8 @@ func TestLoadingConfigRegexp(t *testing.T) {
 			require.NoError(t, sub.Unmarshal(cfg))
 
 			assert.NoError(t, xconfmap.Validate(cfg))
-			assert.Equal(t, tt.expected, cfg)
+			assert.EqualExportedValues(t, tt.expected, cfg)
+			assertConfigContainsDefaultFunctions(t, *cfg.(*Config))
 		})
 	}
 }
@@ -625,7 +652,8 @@ func TestLoadingSpans(t *testing.T) {
 			require.NoError(t, sub.Unmarshal(cfg))
 
 			assert.NoError(t, xconfmap.Validate(cfg))
-			assert.Equal(t, tt.expected, cfg)
+			assert.EqualExportedValues(t, tt.expected, cfg)
+			assertConfigContainsDefaultFunctions(t, *cfg.(*Config))
 		})
 	}
 }
@@ -710,7 +738,8 @@ func TestLoadingConfigExpr(t *testing.T) {
 			require.NoError(t, sub.Unmarshal(cfg))
 
 			assert.NoError(t, xconfmap.Validate(cfg))
-			assert.Equal(t, tt.expected, cfg)
+			assert.EqualExportedValues(t, tt.expected, cfg)
+			assertConfigContainsDefaultFunctions(t, *cfg.(*Config))
 		})
 	}
 }
@@ -922,7 +951,8 @@ func TestLoadingConfigOTTL(t *testing.T) {
 				}
 			} else {
 				assert.NoError(t, xconfmap.Validate(cfg))
-				assert.Equal(t, tt.expected, cfg)
+				assert.EqualExportedValues(t, tt.expected, cfg)
+				assertConfigContainsDefaultFunctions(t, *cfg.(*Config))
 			}
 		})
 	}