From 906f411232852f87c10c09c7c15bf432b91f65e3 Mon Sep 17 00:00:00 2001 From: Bogdan Drutu Date: Wed, 23 Nov 2022 14:03:44 -0800 Subject: [PATCH] Adapt ottl configuration to same BoolExpr, remove duplicate code (#16446) Signed-off-by: Bogdan Drutu Signed-off-by: Bogdan Drutu --- .chloggen/ottelfilternext.yaml | 11 + internal/filter/expr/matcher.go | 9 +- .../attributesprocessor/attributes_log.go | 14 +- .../attributesprocessor/attributes_metric.go | 15 +- .../attributesprocessor/attributes_trace.go | 16 +- processor/filterprocessor/config.go | 21 +- processor/filterprocessor/factory.go | 6 +- .../internal/common/functions.go | 39 ---- .../internal/common/matcher.go | 34 --- .../filterprocessor/internal/common/parser.go | 107 ++++++++- processor/filterprocessor/logs.go | 68 ++---- processor/filterprocessor/logs_test.go | 3 +- processor/filterprocessor/metrics.go | 211 +++++++----------- processor/filterprocessor/metrics_test.go | 3 +- processor/filterprocessor/traces.go | 124 ++++------ processor/filterprocessor/traces_test.go | 3 +- processor/spanprocessor/span.go | 14 +- 17 files changed, 290 insertions(+), 408 deletions(-) create mode 100755 .chloggen/ottelfilternext.yaml delete mode 100644 processor/filterprocessor/internal/common/functions.go delete mode 100644 processor/filterprocessor/internal/common/matcher.go diff --git a/.chloggen/ottelfilternext.yaml b/.chloggen/ottelfilternext.yaml new file mode 100755 index 0000000000000..434f396431765 --- /dev/null +++ b/.chloggen/ottelfilternext.yaml @@ -0,0 +1,11 @@ +# 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: Adapt ottl configuration to same BoolExpr, remove duplicate code + +# One or more tracking issues related to the change +issues: [16446] diff --git a/internal/filter/expr/matcher.go b/internal/filter/expr/matcher.go index 1d3f29041f611..560f3542e4a63 100644 --- a/internal/filter/expr/matcher.go +++ b/internal/filter/expr/matcher.go @@ -54,5 +54,12 @@ func (om orMatcher[K]) Eval(ctx context.Context, tCtx K) (bool, error) { } func Or[K any](matchers ...BoolExpr[K]) BoolExpr[K] { - return orMatcher[K]{matchers: matchers} + switch len(matchers) { + case 0: + return nil + case 1: + return matchers[0] + default: + return orMatcher[K]{matchers: matchers} + } } diff --git a/processor/attributesprocessor/attributes_log.go b/processor/attributesprocessor/attributes_log.go index 256cb030e3b80..93bb9fd59d403 100644 --- a/processor/attributesprocessor/attributes_log.go +++ b/processor/attributesprocessor/attributes_log.go @@ -54,12 +54,14 @@ func (a *logAttributesProcessor) processLogs(ctx context.Context, ld plog.Logs) library := ils.Scope() for k := 0; k < logs.Len(); k++ { lr := logs.At(k) - skip, err := a.skipExpr.Eval(ctx, ottllog.NewTransformContext(lr, library, resource)) - if err != nil { - return ld, err - } - if skip { - continue + if a.skipExpr != nil { + skip, err := a.skipExpr.Eval(ctx, ottllog.NewTransformContext(lr, library, resource)) + if err != nil { + return ld, err + } + if skip { + continue + } } a.attrProc.Process(ctx, a.logger, lr.Attributes()) diff --git a/processor/attributesprocessor/attributes_metric.go b/processor/attributesprocessor/attributes_metric.go index b13a998caa771..10a77363db7fa 100644 --- a/processor/attributesprocessor/attributes_metric.go +++ b/processor/attributesprocessor/attributes_metric.go @@ -54,14 +54,15 @@ func (a *metricAttributesProcessor) processMetrics(ctx context.Context, md pmetr metrics := ils.Metrics() for k := 0; k < metrics.Len(); k++ { m := metrics.At(k) - skip, err := a.skipExpr.Eval(ctx, ottlmetric.NewTransformContext(m, scope, resource)) - if err != nil { - return md, err + if a.skipExpr != nil { + skip, err := a.skipExpr.Eval(ctx, ottlmetric.NewTransformContext(m, scope, resource)) + if err != nil { + return md, err + } + if skip { + continue + } } - if skip { - continue - } - a.processMetricAttributes(ctx, m) } } diff --git a/processor/attributesprocessor/attributes_trace.go b/processor/attributesprocessor/attributes_trace.go index ed170f5d1f7c4..96f901e6a297f 100644 --- a/processor/attributesprocessor/attributes_trace.go +++ b/processor/attributesprocessor/attributes_trace.go @@ -54,15 +54,15 @@ func (a *spanAttributesProcessor) processTraces(ctx context.Context, td ptrace.T scope := ils.Scope() for k := 0; k < spans.Len(); k++ { span := spans.At(k) - - skip, err := a.skipExpr.Eval(ctx, ottlspan.NewTransformContext(span, scope, resource)) - if err != nil { - return td, err - } - if skip { - continue + if a.skipExpr != nil { + skip, err := a.skipExpr.Eval(ctx, ottlspan.NewTransformContext(span, scope, resource)) + if err != nil { + return td, err + } + if skip { + continue + } } - a.attrProc.Process(ctx, a.logger, span.Attributes()) } } diff --git a/processor/filterprocessor/config.go b/processor/filterprocessor/config.go index ac9203adc7b11..ba976050a890d 100644 --- a/processor/filterprocessor/config.go +++ b/processor/filterprocessor/config.go @@ -23,17 +23,11 @@ import ( "go.opentelemetry.io/collector/config" "go.opentelemetry.io/collector/pdata/plog" "go.uber.org/multierr" - "go.uber.org/zap" "github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter/filterconfig" "github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter/filtermetric" "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/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" "github.com/open-telemetry/opentelemetry-collector-contrib/processor/filterprocessor/internal/common" ) @@ -291,32 +285,27 @@ func (cfg *Config) Validate() error { var errors error if cfg.Traces.SpanConditions != nil { - spanp := ottlspan.NewParser(common.Functions[ottlspan.TransformContext](), component.TelemetrySettings{Logger: zap.NewNop()}) - _, err := spanp.ParseStatements(common.PrepareConditionForParsing(cfg.Traces.SpanConditions)) + _, err := common.ParseSpan(cfg.Traces.SpanConditions, component.TelemetrySettings{}) errors = multierr.Append(errors, err) } if cfg.Traces.SpanEventConditions != nil { - spaneventp := ottlspanevent.NewParser(common.Functions[ottlspanevent.TransformContext](), component.TelemetrySettings{Logger: zap.NewNop()}) - _, err := spaneventp.ParseStatements(common.PrepareConditionForParsing(cfg.Traces.SpanEventConditions)) + _, err := common.ParseSpanEvent(cfg.Traces.SpanEventConditions, component.TelemetrySettings{}) errors = multierr.Append(errors, err) } if cfg.Metrics.MetricConditions != nil { - metricp := ottlmetric.NewParser(common.Functions[ottlmetric.TransformContext](), component.TelemetrySettings{Logger: zap.NewNop()}) - _, err := metricp.ParseStatements(common.PrepareConditionForParsing(cfg.Metrics.MetricConditions)) + _, err := common.ParseMetric(cfg.Metrics.MetricConditions, component.TelemetrySettings{}) errors = multierr.Append(errors, err) } if cfg.Metrics.DataPointConditions != nil { - datapointp := ottldatapoint.NewParser(common.Functions[ottldatapoint.TransformContext](), component.TelemetrySettings{Logger: zap.NewNop()}) - _, err := datapointp.ParseStatements(common.PrepareConditionForParsing(cfg.Metrics.DataPointConditions)) + _, err := common.ParseDataPoint(cfg.Metrics.DataPointConditions, component.TelemetrySettings{}) errors = multierr.Append(errors, err) } if cfg.Logs.LogConditions != nil { - logp := ottllog.NewParser(common.Functions[ottllog.TransformContext](), component.TelemetrySettings{Logger: zap.NewNop()}) - _, err := logp.ParseStatements(common.PrepareConditionForParsing(cfg.Logs.LogConditions)) + _, err := common.ParseLog(cfg.Logs.LogConditions, component.TelemetrySettings{}) errors = multierr.Append(errors, err) } diff --git a/processor/filterprocessor/factory.go b/processor/filterprocessor/factory.go index 08dfada1a17ab..fa89a05e12a8b 100644 --- a/processor/filterprocessor/factory.go +++ b/processor/filterprocessor/factory.go @@ -55,7 +55,7 @@ func createMetricsProcessor( cfg component.ProcessorConfig, nextConsumer consumer.Metrics, ) (component.MetricsProcessor, error) { - fp, err := newFilterMetricProcessor(set.Logger, cfg.(*Config)) + fp, err := newFilterMetricProcessor(set.TelemetrySettings, cfg.(*Config)) if err != nil { return nil, err } @@ -74,7 +74,7 @@ func createLogsProcessor( cfg component.ProcessorConfig, nextConsumer consumer.Logs, ) (component.LogsProcessor, error) { - fp, err := newFilterLogsProcessor(set.Logger, cfg.(*Config)) + fp, err := newFilterLogsProcessor(set.TelemetrySettings, cfg.(*Config)) if err != nil { return nil, err } @@ -93,7 +93,7 @@ func createTracesProcessor( cfg component.ProcessorConfig, nextConsumer consumer.Traces, ) (component.TracesProcessor, error) { - fp, err := newFilterSpansProcessor(set.Logger, cfg.(*Config)) + fp, err := newFilterSpansProcessor(set.TelemetrySettings, cfg.(*Config)) if err != nil { return nil, err } diff --git a/processor/filterprocessor/internal/common/functions.go b/processor/filterprocessor/internal/common/functions.go deleted file mode 100644 index d85e50ce4d1b4..0000000000000 --- a/processor/filterprocessor/internal/common/functions.go +++ /dev/null @@ -1,39 +0,0 @@ -// Copyright The OpenTelemetry Authors -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package common // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/filterprocessor/internal/common" - -import ( - "context" - - "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl" - "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/ottlfuncs" -) - -func Functions[K any]() map[string]interface{} { - return map[string]interface{}{ - "TraceID": ottlfuncs.TraceID[K], - "SpanID": ottlfuncs.SpanID[K], - "IsMatch": ottlfuncs.IsMatch[K], - "Concat": ottlfuncs.Concat[K], - "Split": ottlfuncs.Split[K], - "Int": ottlfuncs.Int[K], - "ConvertCase": ottlfuncs.ConvertCase[K], - "drop": func() (ottl.ExprFunc[K], error) { - return func(context.Context, K) (interface{}, error) { - return true, nil - }, nil - }, - } -} diff --git a/processor/filterprocessor/internal/common/matcher.go b/processor/filterprocessor/internal/common/matcher.go deleted file mode 100644 index c1c620b97d073..0000000000000 --- a/processor/filterprocessor/internal/common/matcher.go +++ /dev/null @@ -1,34 +0,0 @@ -// Copyright The OpenTelemetry Authors -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package common // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/filterprocessor/internal/common" - -import ( - "context" - - "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl" -) - -func CheckConditions[K any](ctx context.Context, tCtx K, statements []*ottl.Statement[K]) (bool, error) { - for _, statement := range statements { - _, metCondition, err := statement.Execute(ctx, tCtx) - if err != nil { - return false, err - } - if metCondition { - return true, nil - } - } - return false, nil -} diff --git a/processor/filterprocessor/internal/common/parser.go b/processor/filterprocessor/internal/common/parser.go index 820cb7be1d672..2a77f472473d3 100644 --- a/processor/filterprocessor/internal/common/parser.go +++ b/processor/filterprocessor/internal/common/parser.go @@ -14,14 +14,109 @@ package common // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/filterprocessor/internal/common" -import "fmt" +import ( + "context" -const functionWithCondition = "drop() where %v" + "go.opentelemetry.io/collector/component" -func PrepareConditionForParsing(conditions []string) []string { - validStatements := make([]string, len(conditions)) + "github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter/expr" + "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" + "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/ottlfuncs" +) + +func ParseSpan(conditions []string, set component.TelemetrySettings) (expr.BoolExpr[ottlspan.TransformContext], error) { + statmentsStr := conditionsToStatements(conditions) + parser := ottlspan.NewParser(functions[ottlspan.TransformContext](), set) + statements, err := parser.ParseStatements(statmentsStr) + if err != nil { + return nil, err + } + return statementsToExpr(statements), nil +} + +func ParseSpanEvent(conditions []string, set component.TelemetrySettings) (expr.BoolExpr[ottlspanevent.TransformContext], error) { + statmentsStr := conditionsToStatements(conditions) + parser := ottlspanevent.NewParser(functions[ottlspanevent.TransformContext](), set) + statements, err := parser.ParseStatements(statmentsStr) + if err != nil { + return nil, err + } + return statementsToExpr(statements), nil +} + +func ParseLog(conditions []string, set component.TelemetrySettings) (expr.BoolExpr[ottllog.TransformContext], error) { + statmentsStr := conditionsToStatements(conditions) + parser := ottllog.NewParser(functions[ottllog.TransformContext](), set) + statements, err := parser.ParseStatements(statmentsStr) + if err != nil { + return nil, err + } + return statementsToExpr(statements), nil +} + +func ParseMetric(conditions []string, set component.TelemetrySettings) (expr.BoolExpr[ottlmetric.TransformContext], error) { + statmentsStr := conditionsToStatements(conditions) + parser := ottlmetric.NewParser(functions[ottlmetric.TransformContext](), set) + statements, err := parser.ParseStatements(statmentsStr) + if err != nil { + return nil, err + } + return statementsToExpr(statements), nil +} + +func ParseDataPoint(conditions []string, set component.TelemetrySettings) (expr.BoolExpr[ottldatapoint.TransformContext], error) { + statmentsStr := conditionsToStatements(conditions) + parser := ottldatapoint.NewParser(functions[ottldatapoint.TransformContext](), set) + statements, err := parser.ParseStatements(statmentsStr) + if err != nil { + return nil, err + } + return statementsToExpr(statements), nil +} + +func conditionsToStatements(conditions []string) []string { + statements := make([]string, len(conditions)) for i, condition := range conditions { - validStatements[i] = fmt.Sprintf(functionWithCondition, condition) + statements[i] = "drop() where " + condition + } + return statements +} + +type statementExpr[K any] struct { + statement *ottl.Statement[K] +} + +func (se statementExpr[K]) Eval(ctx context.Context, tCtx K) (bool, error) { + _, ret, err := se.statement.Execute(ctx, tCtx) + return ret, err +} + +func statementsToExpr[K any](statements []*ottl.Statement[K]) expr.BoolExpr[K] { + var rets []expr.BoolExpr[K] + for _, statement := range statements { + rets = append(rets, statementExpr[K]{statement: statement}) + } + return expr.Or(rets...) +} + +func functions[K any]() map[string]interface{} { + return map[string]interface{}{ + "TraceID": ottlfuncs.TraceID[K], + "SpanID": ottlfuncs.SpanID[K], + "IsMatch": ottlfuncs.IsMatch[K], + "Concat": ottlfuncs.Concat[K], + "Split": ottlfuncs.Split[K], + "Int": ottlfuncs.Int[K], + "ConvertCase": ottlfuncs.ConvertCase[K], + "drop": func() (ottl.ExprFunc[K], error) { + return func(context.Context, K) (interface{}, error) { + return true, nil + }, nil + }, } - return validStatements } diff --git a/processor/filterprocessor/logs.go b/processor/filterprocessor/logs.go index adeafd8dce557..0f77e4ca26089 100644 --- a/processor/filterprocessor/logs.go +++ b/processor/filterprocessor/logs.go @@ -22,36 +22,26 @@ import ( "go.opentelemetry.io/collector/pdata/plog" "go.opentelemetry.io/collector/processor/processorhelper" "go.uber.org/multierr" - "go.uber.org/zap" "github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter/expr" "github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter/filterconfig" "github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter/filterlog" - "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/filterprocessor/internal/common" ) type filterLogProcessor struct { - cfg *Config - skipExpr expr.BoolExpr[ottllog.TransformContext] - logger *zap.Logger - logConditions []*ottl.Statement[ottllog.TransformContext] + skipExpr expr.BoolExpr[ottllog.TransformContext] } -func newFilterLogsProcessor(logger *zap.Logger, cfg *Config) (*filterLogProcessor, error) { +func newFilterLogsProcessor(set component.TelemetrySettings, cfg *Config) (*filterLogProcessor, error) { if cfg.Logs.LogConditions != nil { - logp := ottllog.NewParser(common.Functions[ottllog.TransformContext](), component.TelemetrySettings{Logger: zap.NewNop()}) - statements, err := logp.ParseStatements(common.PrepareConditionForParsing(cfg.Logs.LogConditions)) + skipExpr, err := common.ParseLog(cfg.Logs.LogConditions, set) if err != nil { return nil, err } - return &filterLogProcessor{ - cfg: cfg, - logger: logger, - logConditions: statements, - }, nil + return &filterLogProcessor{skipExpr: skipExpr}, nil } cfgMatch := filterconfig.MatchConfig{} @@ -68,51 +58,20 @@ func newFilterLogsProcessor(logger *zap.Logger, cfg *Config) (*filterLogProcesso return nil, fmt.Errorf("failed to build skip matcher: %w", err) } - return &filterLogProcessor{ - cfg: cfg, - skipExpr: skipExpr, - logger: logger, - }, nil + return &filterLogProcessor{skipExpr: skipExpr}, nil } -func (flp *filterLogProcessor) processLogs(ctx context.Context, logs plog.Logs) (plog.Logs, error) { - filteringLogs := flp.logConditions != nil - - if filteringLogs { - var errors error - logs.ResourceLogs().RemoveIf(func(rlogs plog.ResourceLogs) bool { - rlogs.ScopeLogs().RemoveIf(func(slogs plog.ScopeLogs) bool { - slogs.LogRecords().RemoveIf(func(log plog.LogRecord) bool { - tCtx := ottllog.NewTransformContext(log, slogs.Scope(), rlogs.Resource()) - metCondition, err := common.CheckConditions(ctx, tCtx, flp.logConditions) - if err != nil { - errors = multierr.Append(errors, err) - } - return metCondition - }) - return slogs.LogRecords().Len() == 0 - }) - return rlogs.ScopeLogs().Len() == 0 - }) - - if errors != nil { - return logs, errors - } - if logs.ResourceLogs().Len() == 0 { - return logs, processorhelper.ErrSkipProcessingData - } - return logs, nil +func (flp *filterLogProcessor) processLogs(ctx context.Context, ld plog.Logs) (plog.Logs, error) { + if flp.skipExpr == nil { + return ld, nil } - rLogs := logs.ResourceLogs() - var errors error - rLogs.RemoveIf(func(rl plog.ResourceLogs) bool { + ld.ResourceLogs().RemoveIf(func(rl plog.ResourceLogs) bool { resource := rl.Resource() rl.ScopeLogs().RemoveIf(func(sl plog.ScopeLogs) bool { scope := sl.Scope() lrs := sl.LogRecords() - lrs.RemoveIf(func(lr plog.LogRecord) bool { skip, err := flp.skipExpr.Eval(ctx, ottllog.NewTransformContext(lr, scope, resource)) if err != nil { @@ -128,11 +87,10 @@ func (flp *filterLogProcessor) processLogs(ctx context.Context, logs plog.Logs) }) if errors != nil { - return logs, errors + return ld, errors } - if rLogs.Len() == 0 { - return logs, processorhelper.ErrSkipProcessingData + if ld.ResourceLogs().Len() == 0 { + return ld, processorhelper.ErrSkipProcessingData } - - return logs, nil + return ld, nil } diff --git a/processor/filterprocessor/logs_test.go b/processor/filterprocessor/logs_test.go index 68eb48cb12e5e..e81134a670374 100644 --- a/processor/filterprocessor/logs_test.go +++ b/processor/filterprocessor/logs_test.go @@ -28,7 +28,6 @@ import ( "go.opentelemetry.io/collector/pdata/pcommon" "go.opentelemetry.io/collector/pdata/plog" "go.opentelemetry.io/collector/processor/processorhelper" - "go.uber.org/zap" "github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter/filterconfig" ) @@ -749,7 +748,7 @@ func TestFilterLogProcessorWithOTTL(t *testing.T) { } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - processor, err := newFilterLogsProcessor(zap.NewNop(), &Config{Logs: LogFilters{LogConditions: tt.conditions}}) + processor, err := newFilterLogsProcessor(componenttest.NewNopTelemetrySettings(), &Config{Logs: LogFilters{LogConditions: tt.conditions}}) assert.NoError(t, err) got, err := processor.processLogs(context.Background(), constructLogs()) diff --git a/processor/filterprocessor/metrics.go b/processor/filterprocessor/metrics.go index a44fa28d72e76..0a76ac5e69c59 100644 --- a/processor/filterprocessor/metrics.go +++ b/processor/filterprocessor/metrics.go @@ -29,7 +29,6 @@ import ( "github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter/filtermatcher" "github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter/filtermetric" "github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter/filterset" - "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/ottlmetric" "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlresource" @@ -37,48 +36,38 @@ import ( ) type filterMetricProcessor struct { - cfg *Config - skipResourceExpr expr.BoolExpr[ottlresource.TransformContext] - skipMetricExpr expr.BoolExpr[ottlmetric.TransformContext] - logger *zap.Logger - metricConditions []*ottl.Statement[ottlmetric.TransformContext] - dataPointConditions []*ottl.Statement[ottldatapoint.TransformContext] + skipResourceExpr expr.BoolExpr[ottlresource.TransformContext] + skipMetricExpr expr.BoolExpr[ottlmetric.TransformContext] + skipDataPointExpr expr.BoolExpr[ottldatapoint.TransformContext] } -func newFilterMetricProcessor(logger *zap.Logger, cfg *Config) (*filterMetricProcessor, error) { +func newFilterMetricProcessor(set component.TelemetrySettings, cfg *Config) (*filterMetricProcessor, error) { + var err error + fsp := &filterMetricProcessor{} if cfg.Metrics.MetricConditions != nil || cfg.Metrics.DataPointConditions != nil { - fsp := &filterMetricProcessor{ - cfg: cfg, - logger: logger, - } - if cfg.Metrics.MetricConditions != nil { - metricp := ottlmetric.NewParser(common.Functions[ottlmetric.TransformContext](), component.TelemetrySettings{Logger: zap.NewNop()}) - statements, err := metricp.ParseStatements(common.PrepareConditionForParsing(cfg.Metrics.MetricConditions)) + fsp.skipMetricExpr, err = common.ParseMetric(cfg.Metrics.MetricConditions, set) if err != nil { return nil, err } - fsp.metricConditions = statements } if cfg.Metrics.DataPointConditions != nil { - datapointp := ottldatapoint.NewParser(common.Functions[ottldatapoint.TransformContext](), component.TelemetrySettings{Logger: zap.NewNop()}) - statements, err := datapointp.ParseStatements(common.PrepareConditionForParsing(cfg.Metrics.DataPointConditions)) + fsp.skipDataPointExpr, err = common.ParseDataPoint(cfg.Metrics.DataPointConditions, set) if err != nil { return nil, err } - fsp.dataPointConditions = statements } return fsp, nil } - skipResourceExpr, err := newSkipResExpr(cfg.Metrics.Include, cfg.Metrics.Exclude) + fsp.skipResourceExpr, err = newSkipResExpr(cfg.Metrics.Include, cfg.Metrics.Exclude) if err != nil { return nil, err } - skipMetricExpr, err := filtermetric.NewSkipExpr(cfg.Metrics.Include, cfg.Metrics.Exclude) + fsp.skipMetricExpr, err = filtermetric.NewSkipExpr(cfg.Metrics.Include, cfg.Metrics.Exclude) if err != nil { return nil, err } @@ -105,7 +94,7 @@ func newFilterMetricProcessor(logger *zap.Logger, cfg *Config) (*filterMetricPro excludeResourceAttributes = cfg.Metrics.Exclude.ResourceAttributes } - logger.Info( + set.Logger.Info( "Metric filter configured", zap.String("include match_type", includeMatchType), zap.Strings("include expressions", includeExpressions), @@ -117,121 +106,75 @@ func newFilterMetricProcessor(logger *zap.Logger, cfg *Config) (*filterMetricPro zap.Any("exclude metrics with resource attributes", excludeResourceAttributes), ) - return &filterMetricProcessor{ - cfg: cfg, - skipResourceExpr: skipResourceExpr, - skipMetricExpr: skipMetricExpr, - logger: logger, - }, nil + return fsp, nil } // processMetrics filters the given metrics based off the filterMetricProcessor's filters. -func (fmp *filterMetricProcessor) processMetrics(ctx context.Context, pdm pmetric.Metrics) (pmetric.Metrics, error) { - filteringMetrics := fmp.metricConditions != nil - filteringDataPoints := fmp.dataPointConditions != nil - - if filteringMetrics || filteringDataPoints { - var errors error - pdm.ResourceMetrics().RemoveIf(func(rmetrics pmetric.ResourceMetrics) bool { - rmetrics.ScopeMetrics().RemoveIf(func(smetrics pmetric.ScopeMetrics) bool { - smetrics.Metrics().RemoveIf(func(metric pmetric.Metric) bool { - if filteringMetrics { - tCtx := ottlmetric.NewTransformContext(metric, smetrics.Scope(), rmetrics.Resource()) - metCondition, err := common.CheckConditions(ctx, tCtx, fmp.metricConditions) - if err != nil { - errors = multierr.Append(errors, err) - } - if metCondition { - return true - } - } - if filteringDataPoints { - switch metric.Type() { - case pmetric.MetricTypeSum: - err := fmp.handleNumberDataPoints(ctx, metric.Sum().DataPoints(), metric, smetrics.Metrics(), smetrics.Scope(), rmetrics.Resource()) - if err != nil { - errors = multierr.Append(errors, err) - } - return metric.Sum().DataPoints().Len() == 0 - case pmetric.MetricTypeGauge: - err := fmp.handleNumberDataPoints(ctx, metric.Gauge().DataPoints(), metric, smetrics.Metrics(), smetrics.Scope(), rmetrics.Resource()) - if err != nil { - errors = multierr.Append(errors, err) - } - return metric.Gauge().DataPoints().Len() == 0 - case pmetric.MetricTypeHistogram: - err := fmp.handleHistogramDataPoints(ctx, metric.Histogram().DataPoints(), metric, smetrics.Metrics(), smetrics.Scope(), rmetrics.Resource()) - if err != nil { - errors = multierr.Append(errors, err) - } - return metric.Histogram().DataPoints().Len() == 0 - case pmetric.MetricTypeExponentialHistogram: - err := fmp.handleExponetialHistogramDataPoints(ctx, metric.ExponentialHistogram().DataPoints(), metric, smetrics.Metrics(), smetrics.Scope(), rmetrics.Resource()) - if err != nil { - errors = multierr.Append(errors, err) - } - return metric.ExponentialHistogram().DataPoints().Len() == 0 - case pmetric.MetricTypeSummary: - err := fmp.handleSummaryDataPoints(ctx, metric.Summary().DataPoints(), metric, smetrics.Metrics(), smetrics.Scope(), rmetrics.Resource()) - if err != nil { - errors = multierr.Append(errors, err) - } - return metric.Summary().DataPoints().Len() == 0 - default: - return false - } - } - return false - }) - return smetrics.Metrics().Len() == 0 - }) - return rmetrics.ScopeMetrics().Len() == 0 - }) - - if errors != nil { - return pdm, errors - } - if pdm.ResourceMetrics().Len() == 0 { - return pdm, processorhelper.ErrSkipProcessingData - } - return pdm, nil +func (fmp *filterMetricProcessor) processMetrics(ctx context.Context, md pmetric.Metrics) (pmetric.Metrics, error) { + if fmp.skipResourceExpr == nil && fmp.skipMetricExpr == nil && fmp.skipDataPointExpr == nil { + return md, nil } var errors error - pdm.ResourceMetrics().RemoveIf(func(rm pmetric.ResourceMetrics) bool { - resource := rm.Resource() - skip, err := fmp.skipResourceExpr.Eval(ctx, ottlresource.NewTransformContext(resource)) - if err != nil { - errors = multierr.Append(errors, err) - return false - } - if skip { - return true + md.ResourceMetrics().RemoveIf(func(rmetrics pmetric.ResourceMetrics) bool { + resource := rmetrics.Resource() + if fmp.skipResourceExpr != nil { + skip, err := fmp.skipResourceExpr.Eval(ctx, ottlresource.NewTransformContext(resource)) + if err != nil { + errors = multierr.Append(errors, err) + return false + } + if skip { + return true + } } - - rm.ScopeMetrics().RemoveIf(func(ilm pmetric.ScopeMetrics) bool { - scope := ilm.Scope() - ilm.Metrics().RemoveIf(func(m pmetric.Metric) bool { - skip, err = fmp.skipMetricExpr.Eval(ctx, ottlmetric.NewTransformContext(m, scope, resource)) - if err != nil { - errors = multierr.Append(errors, err) - return false + rmetrics.ScopeMetrics().RemoveIf(func(smetrics pmetric.ScopeMetrics) bool { + scope := smetrics.Scope() + smetrics.Metrics().RemoveIf(func(metric pmetric.Metric) bool { + if fmp.skipMetricExpr != nil { + skip, err := fmp.skipMetricExpr.Eval(ctx, ottlmetric.NewTransformContext(metric, scope, resource)) + if err != nil { + errors = multierr.Append(errors, err) + } + if skip { + return true + } + } + if fmp.skipDataPointExpr != nil { + switch metric.Type() { + case pmetric.MetricTypeSum: + errors = multierr.Append(errors, fmp.handleNumberDataPoints(ctx, metric.Sum().DataPoints(), metric, smetrics.Metrics(), scope, resource)) + return metric.Sum().DataPoints().Len() == 0 + case pmetric.MetricTypeGauge: + errors = multierr.Append(errors, fmp.handleNumberDataPoints(ctx, metric.Gauge().DataPoints(), metric, smetrics.Metrics(), scope, resource)) + return metric.Gauge().DataPoints().Len() == 0 + case pmetric.MetricTypeHistogram: + errors = multierr.Append(errors, fmp.handleHistogramDataPoints(ctx, metric.Histogram().DataPoints(), metric, smetrics.Metrics(), scope, resource)) + return metric.Histogram().DataPoints().Len() == 0 + case pmetric.MetricTypeExponentialHistogram: + errors = multierr.Append(errors, fmp.handleExponetialHistogramDataPoints(ctx, metric.ExponentialHistogram().DataPoints(), metric, smetrics.Metrics(), scope, resource)) + return metric.ExponentialHistogram().DataPoints().Len() == 0 + case pmetric.MetricTypeSummary: + errors = multierr.Append(errors, fmp.handleSummaryDataPoints(ctx, metric.Summary().DataPoints(), metric, smetrics.Metrics(), scope, resource)) + return metric.Summary().DataPoints().Len() == 0 + default: + return false + } } - return skip + return false }) - // Filter out empty ScopeMetrics - return ilm.Metrics().Len() == 0 + return smetrics.Metrics().Len() == 0 }) - // Filter out empty ResourceMetrics - return rm.ScopeMetrics().Len() == 0 + return rmetrics.ScopeMetrics().Len() == 0 }) - if pdm.ResourceMetrics().Len() == 0 { - return pdm, processorhelper.ErrSkipProcessingData - } + if errors != nil { - return pdm, errors + return md, errors + } + if md.ResourceMetrics().Len() == 0 { + return md, processorhelper.ErrSkipProcessingData } - return pdm, nil + return md, nil } func newSkipResExpr(include *filtermetric.MatchProperties, exclude *filtermetric.MatchProperties) (expr.BoolExpr[ottlresource.TransformContext], error) { @@ -282,13 +225,12 @@ func newResExpr(mp *filtermetric.MatchProperties) (expr.BoolExpr[ottlresource.Tr func (fmp *filterMetricProcessor) handleNumberDataPoints(ctx context.Context, dps pmetric.NumberDataPointSlice, metric pmetric.Metric, metrics pmetric.MetricSlice, is pcommon.InstrumentationScope, resource pcommon.Resource) error { var errors error dps.RemoveIf(func(datapoint pmetric.NumberDataPoint) bool { - tCtx := ottldatapoint.NewTransformContext(datapoint, metric, metrics, is, resource) - metCondition, err := common.CheckConditions(ctx, tCtx, fmp.dataPointConditions) + skip, err := fmp.skipDataPointExpr.Eval(ctx, ottldatapoint.NewTransformContext(datapoint, metric, metrics, is, resource)) if err != nil { errors = multierr.Append(errors, err) return false } - return metCondition + return skip }) return errors } @@ -296,13 +238,12 @@ func (fmp *filterMetricProcessor) handleNumberDataPoints(ctx context.Context, dp func (fmp *filterMetricProcessor) handleHistogramDataPoints(ctx context.Context, dps pmetric.HistogramDataPointSlice, metric pmetric.Metric, metrics pmetric.MetricSlice, is pcommon.InstrumentationScope, resource pcommon.Resource) error { var errors error dps.RemoveIf(func(datapoint pmetric.HistogramDataPoint) bool { - tCtx := ottldatapoint.NewTransformContext(datapoint, metric, metrics, is, resource) - metCondition, err := common.CheckConditions(ctx, tCtx, fmp.dataPointConditions) + skip, err := fmp.skipDataPointExpr.Eval(ctx, ottldatapoint.NewTransformContext(datapoint, metric, metrics, is, resource)) if err != nil { errors = multierr.Append(errors, err) return false } - return metCondition + return skip }) return errors } @@ -310,13 +251,12 @@ func (fmp *filterMetricProcessor) handleHistogramDataPoints(ctx context.Context, func (fmp *filterMetricProcessor) handleExponetialHistogramDataPoints(ctx context.Context, dps pmetric.ExponentialHistogramDataPointSlice, metric pmetric.Metric, metrics pmetric.MetricSlice, is pcommon.InstrumentationScope, resource pcommon.Resource) error { var errors error dps.RemoveIf(func(datapoint pmetric.ExponentialHistogramDataPoint) bool { - tCtx := ottldatapoint.NewTransformContext(datapoint, metric, metrics, is, resource) - metCondition, err := common.CheckConditions(ctx, tCtx, fmp.dataPointConditions) + skip, err := fmp.skipDataPointExpr.Eval(ctx, ottldatapoint.NewTransformContext(datapoint, metric, metrics, is, resource)) if err != nil { errors = multierr.Append(errors, err) return false } - return metCondition + return skip }) return errors } @@ -324,13 +264,12 @@ func (fmp *filterMetricProcessor) handleExponetialHistogramDataPoints(ctx contex func (fmp *filterMetricProcessor) handleSummaryDataPoints(ctx context.Context, dps pmetric.SummaryDataPointSlice, metric pmetric.Metric, metrics pmetric.MetricSlice, is pcommon.InstrumentationScope, resource pcommon.Resource) error { var errors error dps.RemoveIf(func(datapoint pmetric.SummaryDataPoint) bool { - tCtx := ottldatapoint.NewTransformContext(datapoint, metric, metrics, is, resource) - metCondition, err := common.CheckConditions(ctx, tCtx, fmp.dataPointConditions) + skip, err := fmp.skipDataPointExpr.Eval(ctx, ottldatapoint.NewTransformContext(datapoint, metric, metrics, is, resource)) if err != nil { errors = multierr.Append(errors, err) return false } - return metCondition + return skip }) return errors } diff --git a/processor/filterprocessor/metrics_test.go b/processor/filterprocessor/metrics_test.go index de5d343a7dd88..02ffc67aa0608 100644 --- a/processor/filterprocessor/metrics_test.go +++ b/processor/filterprocessor/metrics_test.go @@ -29,7 +29,6 @@ import ( "go.opentelemetry.io/collector/pdata/pcommon" "go.opentelemetry.io/collector/pdata/pmetric" "go.opentelemetry.io/collector/processor/processorhelper" - "go.uber.org/zap" "github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/goldendataset" "github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter/filterconfig" @@ -695,7 +694,7 @@ func TestFilterMetricProcessorWithOTTL(t *testing.T) { } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - processor, err := newFilterMetricProcessor(zap.NewNop(), &Config{Metrics: tt.conditions}) + processor, err := newFilterMetricProcessor(componenttest.NewNopTelemetrySettings(), &Config{Metrics: tt.conditions}) assert.NoError(t, err) got, err := processor.processMetrics(context.Background(), constructMetrics()) diff --git a/processor/filterprocessor/traces.go b/processor/filterprocessor/traces.go index 362edfada212b..1d6bed1529000 100644 --- a/processor/filterprocessor/traces.go +++ b/processor/filterprocessor/traces.go @@ -25,52 +25,36 @@ import ( "github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter/expr" "github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter/filterspan" - "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/pkg/ottl/contexts/ottlspanevent" "github.com/open-telemetry/opentelemetry-collector-contrib/processor/filterprocessor/internal/common" ) type filterSpanProcessor struct { - cfg *Config - skipExpr expr.BoolExpr[ottlspan.TransformContext] - logger *zap.Logger - spanConditions []*ottl.Statement[ottlspan.TransformContext] - spanEventConditions []*ottl.Statement[ottlspanevent.TransformContext] + skipSpanExpr expr.BoolExpr[ottlspan.TransformContext] + skipSpanEventExpr expr.BoolExpr[ottlspanevent.TransformContext] } -func newFilterSpansProcessor(logger *zap.Logger, cfg *Config) (*filterSpanProcessor, error) { +func newFilterSpansProcessor(set component.TelemetrySettings, cfg *Config) (*filterSpanProcessor, error) { + var err error + fsp := &filterSpanProcessor{} if cfg.Traces.SpanConditions != nil || cfg.Traces.SpanEventConditions != nil { - fsp := &filterSpanProcessor{ - cfg: cfg, - logger: logger, - } - if cfg.Traces.SpanConditions != nil { - spanp := ottlspan.NewParser(common.Functions[ottlspan.TransformContext](), component.TelemetrySettings{Logger: zap.NewNop()}) - statements, err := spanp.ParseStatements(common.PrepareConditionForParsing(cfg.Traces.SpanConditions)) + fsp.skipSpanExpr, err = common.ParseSpan(cfg.Traces.SpanConditions, set) if err != nil { return nil, err } - fsp.spanConditions = statements } - if cfg.Traces.SpanEventConditions != nil { - spaneventp := ottlspanevent.NewParser(common.Functions[ottlspanevent.TransformContext](), component.TelemetrySettings{Logger: zap.NewNop()}) - statements, err := spaneventp.ParseStatements(common.PrepareConditionForParsing(cfg.Traces.SpanEventConditions)) + fsp.skipSpanEventExpr, err = common.ParseSpanEvent(cfg.Traces.SpanEventConditions, set) if err != nil { return nil, err } - fsp.spanEventConditions = statements } return fsp, nil } - if cfg.Spans.Include == nil && cfg.Spans.Exclude == nil { - return nil, nil - } - - skipExpr, err := filterspan.NewSkipExpr(&cfg.Spans) + fsp.skipSpanExpr, err = filterspan.NewSkipExpr(&cfg.Spans) if err != nil { return nil, err } @@ -84,90 +68,60 @@ func newFilterSpansProcessor(logger *zap.Logger, cfg *Config) (*filterSpanProces excludeMatchType = string(cfg.Spans.Exclude.MatchType) } - logger.Info( + set.Logger.Info( "Span filter configured", zap.String("ID", cfg.ID().String()), zap.String("[Include] match_type", includeMatchType), zap.String("[Exclude] match_type", excludeMatchType), ) - return &filterSpanProcessor{ - cfg: cfg, - skipExpr: skipExpr, - logger: logger, - }, nil + return fsp, nil } // processTraces filters the given spans of a traces based off the filterSpanProcessor's filters. -func (fsp *filterSpanProcessor) processTraces(ctx context.Context, pdt ptrace.Traces) (ptrace.Traces, error) { - filteringSpans := fsp.spanConditions != nil - filteringSpanEvents := fsp.spanEventConditions != nil - - if filteringSpans || filteringSpanEvents { - var errors error - pdt.ResourceSpans().RemoveIf(func(rspans ptrace.ResourceSpans) bool { - rspans.ScopeSpans().RemoveIf(func(sspans ptrace.ScopeSpans) bool { - sspans.Spans().RemoveIf(func(span ptrace.Span) bool { - if filteringSpans { - tCtx := ottlspan.NewTransformContext(span, sspans.Scope(), rspans.Resource()) - metCondition, err := common.CheckConditions(ctx, tCtx, fsp.spanConditions) - if err != nil { - errors = multierr.Append(errors, err) - return false - } - if metCondition { - return true - } - } - if filteringSpanEvents { - span.Events().RemoveIf(func(spanEvent ptrace.SpanEvent) bool { - tCtx := ottlspanevent.NewTransformContext(spanEvent, span, sspans.Scope(), rspans.Resource()) - metCondition, err := common.CheckConditions(ctx, tCtx, fsp.spanEventConditions) - if err != nil { - errors = multierr.Append(errors, err) - return false - } - return metCondition - }) - } - return false - }) - return sspans.Spans().Len() == 0 - }) - return rspans.ScopeSpans().Len() == 0 - }) - - if errors != nil { - return pdt, errors - } - if pdt.ResourceSpans().Len() == 0 { - return pdt, processorhelper.ErrSkipProcessingData - } - return pdt, nil +func (fsp *filterSpanProcessor) processTraces(ctx context.Context, td ptrace.Traces) (ptrace.Traces, error) { + if fsp.skipSpanExpr == nil && fsp.skipSpanEventExpr == nil { + return td, nil } var errors error - pdt.ResourceSpans().RemoveIf(func(rs ptrace.ResourceSpans) bool { + td.ResourceSpans().RemoveIf(func(rs ptrace.ResourceSpans) bool { resource := rs.Resource() rs.ScopeSpans().RemoveIf(func(ss ptrace.ScopeSpans) bool { scope := ss.Scope() ss.Spans().RemoveIf(func(span ptrace.Span) bool { - skip, err := fsp.skipExpr.Eval(ctx, ottlspan.NewTransformContext(span, scope, resource)) - if err != nil { - errors = multierr.Append(errors, err) - return false + if fsp.skipSpanExpr != nil { + skip, err := fsp.skipSpanExpr.Eval(ctx, ottlspan.NewTransformContext(span, scope, resource)) + if err != nil { + errors = multierr.Append(errors, err) + return false + } + if skip { + return true + } } - return skip + if fsp.skipSpanEventExpr != nil { + span.Events().RemoveIf(func(spanEvent ptrace.SpanEvent) bool { + skip, err := fsp.skipSpanEventExpr.Eval(ctx, ottlspanevent.NewTransformContext(spanEvent, span, scope, resource)) + if err != nil { + errors = multierr.Append(errors, err) + return false + } + return skip + }) + } + return false }) return ss.Spans().Len() == 0 }) return rs.ScopeSpans().Len() == 0 }) + if errors != nil { - return pdt, errors + return td, errors } - if pdt.ResourceSpans().Len() == 0 { - return pdt, processorhelper.ErrSkipProcessingData + if td.ResourceSpans().Len() == 0 { + return td, processorhelper.ErrSkipProcessingData } - return pdt, nil + return td, nil } diff --git a/processor/filterprocessor/traces_test.go b/processor/filterprocessor/traces_test.go index f0775aa385dd7..af11339c081c8 100644 --- a/processor/filterprocessor/traces_test.go +++ b/processor/filterprocessor/traces_test.go @@ -28,7 +28,6 @@ import ( "go.opentelemetry.io/collector/pdata/pcommon" "go.opentelemetry.io/collector/pdata/ptrace" "go.opentelemetry.io/collector/processor/processorhelper" - "go.uber.org/zap" "github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter/filterconfig" "github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter/filterset" @@ -262,7 +261,7 @@ func TestFilterTraceProcessorWithOTTL(t *testing.T) { } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - processor, err := newFilterSpansProcessor(zap.NewNop(), &Config{Traces: tt.conditions}) + processor, err := newFilterSpansProcessor(componenttest.NewNopTelemetrySettings(), &Config{Traces: tt.conditions}) assert.NoError(t, err) got, err := processor.processTraces(context.Background(), constructTraces()) diff --git a/processor/spanprocessor/span.go b/processor/spanprocessor/span.go index 5c985c71ef664..f66371cff91c3 100644 --- a/processor/spanprocessor/span.go +++ b/processor/spanprocessor/span.go @@ -89,12 +89,14 @@ func (sp *spanProcessor) processTraces(ctx context.Context, td ptrace.Traces) (p scope := ils.Scope() for k := 0; k < spans.Len(); k++ { span := spans.At(k) - skip, err := sp.skipExpr.Eval(ctx, ottlspan.NewTransformContext(span, scope, resource)) - if err != nil { - return td, err - } - if skip { - continue + if sp.skipExpr != nil { + skip, err := sp.skipExpr.Eval(ctx, ottlspan.NewTransformContext(span, scope, resource)) + if err != nil { + return td, err + } + if skip { + continue + } } sp.processFromAttributes(span) sp.processToAttributes(span)