[processor/filterprocessor] Ability to filter spans

CHANGELOG.md

@@ -1,7 +1,8 @@
 # Changelog
 
 ## Unreleased
-
+- `filterprocessor`: Ability to filter `Spans` (#6341)
+- 
 ## 🛑 Breaking changes 🛑
 - `transformprocessor`: `metric.is_monotonic` is now accessed via a bool literal instead of a string. (#10473)
 

internal/coreinternal/processor/filterspan/filterspan.go

@@ -112,7 +112,9 @@ func SkipSpan(include Matcher, exclude Matcher, span ptrace.Span, resource pcomm
 func (mp *propertiesMatcher) MatchSpan(span ptrace.Span, resource pcommon.Resource, library pcommon.InstrumentationScope) bool {
 	// If a set of properties was not in the mp, all spans are considered to match on that property
 	if mp.serviceFilters != nil {
+		// Check resource and spans for service.name
 		serviceName := serviceNameForResource(resource)
+
 		if !mp.serviceFilters.Matches(serviceName) {
 			return false
 		}
@@ -131,6 +133,5 @@ func serviceNameForResource(resource pcommon.Resource) string {
 	if !found {
 		return "<nil-service-name>"
 	}
-
-	return service.StringVal()
+	return service.AsString()
 }

internal/coreinternal/processor/filterspan/filterspan_test.go

@@ -247,12 +247,12 @@ func TestSpan_Matching_True(t *testing.T) {
 
 func TestServiceNameForResource(t *testing.T) {
 	td := testdata.GenerateTracesOneSpanNoResource()
-	require.Equal(t, serviceNameForResource(td.ResourceSpans().At(0).Resource()), "<nil-service-name>")
+	name := serviceNameForResource(td.ResourceSpans().At(0).Resource())
+	require.Equal(t, name, "<nil-service-name>")
 
 	td = testdata.GenerateTracesOneSpan()
 	resource := td.ResourceSpans().At(0).Resource()
-	require.Equal(t, serviceNameForResource(resource), "<nil-service-name>")
+	name = serviceNameForResource(resource)
+	require.Equal(t, name, "<nil-service-name>")
 
-	resource.Attributes().InsertString(conventions.AttributeServiceName, "test-service")
-	require.Equal(t, serviceNameForResource(resource), "test-service")
 }

processor/filterprocessor/README.md

@@ -1,19 +1,23 @@
 # Filter Processor
 
+- Supported pipeline types: logs, metrics, spans
+=======
 | Status                   |                   |
 | ------------------------ | ----------------- |
 | Stability                | [alpha]           |
-| Supported pipeline types | metrics, logs     |
+| Supported pipeline types | metrics, logs, traces  |
 | Distributions            | [core], [contrib] |
 
 The filter processor can be configured to include or exclude:
 
+
 - logs, based on resource attributes using the `strict` or `regexp` match types
 - metrics based on metric name in the case of the `strict` or `regexp` match types,
   or based on other metric attributes in the case of the `expr` match type.
   Please refer to [config.go](./config.go) for the config spec.
+- Spans based on tags, resources, and names, all with full regex support
 
-It takes a pipeline type, of which `logs` and `metrics` are supported, followed
+It takes a pipeline type, of which `logs` `metrics`, and `traces` are supported, followed
 by an action:
 
 - `include`: Any names NOT matching filters are excluded from remainder of pipeline
@@ -209,6 +213,43 @@ processors:
 
 In case the no metric names are provided, `matric_names` being empty, the filtering is only done at resource level.
 
+### Filter Spans from Traces
+
+* This pipeline is able to drop spans and whole traces 
+* Note: If this drops a parent span, it does not search out it's children leading to a missing Span in your trace visualization
+
+See the documentation in the [attribute processor](../attributesprocessor/README.md) for syntax
+
+For spans, one of Services, SpanNames, Attributes, Resources or Libraries must be specified with a
+non-empty value for a valid configuration.
+
+```yaml
+processors:
+  filter:
+    spans:
+      include:
+        match_type: strict
+        services:
+          - app_3
+      exclude:
+        match_type: regex
+        services:
+          - app_1
+          - app_2
+        span_names:
+          - hello_world
+          - hello/world
+        attributes:
+          - Key: container.name
+            Value: (app_container_1|app_container_2)
+        libraries:
+          - Name: opentelemetry
+            Version: 0.0-beta
+        resources:
+          - Key: container.host
+            Value: (localhost|127.0.0.1)
+```
+
 [alpha]:https://github.com/open-telemetry/opentelemetry-collector#alpha
 [contrib]:https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol-contrib
 [core]:https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol

processor/filterprocessor/config.go

@@ -20,6 +20,7 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/processor/filterconfig"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/processor/filtermetric"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/processor/filterset"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/processor/filterset/regexp"
 )
 
 // Config defines configuration for Resource processor.
@@ -29,6 +30,8 @@ type Config struct {
 	Metrics MetricFilters `mapstructure:"metrics"`
 
 	Logs LogFilters `mapstructure:"logs"`
+
+	Spans SpanFilters `mapstructure:"spans"`
 }
 
 // MetricFilters filters by Metric properties.
@@ -42,6 +45,22 @@ type MetricFilters struct {
 	// all other metrics should be included.
 	// If both Include and Exclude are specified, Include filtering occurs first.
 	Exclude *filtermetric.MatchProperties `mapstructure:"exclude"`
+
+	// RegexpConfig specifies options for the Regexp match type
+	RegexpConfig *regexp.Config `mapstructure:"regexp"`
+}
+
+// SpanFilters filters by Span attributes and various other fields, Regexp config is per matcher
+type SpanFilters struct {
+	// Include match properties describe spans that should be included in the Collector Service pipeline,
+	// all other spans should be dropped from further processing.
+	// If both Include and Exclude are specified, Include filtering occurs first.
+	Include *filterconfig.MatchProperties `mapstructure:"include"`
+
+	// Exclude match properties describe spans that should be excluded from the Collector Service pipeline,
+	// all other spans should be included.
+	// If both Include and Exclude are specified, Include filtering occurs first.
+	Exclude *filterconfig.MatchProperties `mapstructure:"exclude"`
 }
 
 // LogFilters filters by Log properties.

processor/filterprocessor/config_test.go

@@ -15,6 +15,7 @@
 package filterprocessor
 
 import (
+	"path"
 	"path/filepath"
 	"testing"
 
@@ -26,6 +27,7 @@ import (
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/processor/filterconfig"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/processor/filtermetric"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/processor/filterset"
 	fsregexp "github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/processor/filterset/regexp"
 )
 
@@ -272,6 +274,52 @@ func TestLoadingConfigRegexp(t *testing.T) {
 	}
 }
 
+func TestLoadingSpans(t *testing.T) {
+	factories, err := componenttest.NopFactories()
+	require.NoError(t, err)
+	factory := NewFactory()
+	factories.Processors[typeStr] = factory
+	cfg, err := servicetest.LoadConfigAndValidate(path.Join(".", "testdata", "config_traces.yaml"), factories)
+	require.NoError(t, err)
+	require.NotNil(t, cfg)
+
+	tests := []struct {
+		expCfg config.Processor
+	}{
+		{
+			expCfg: &Config{
+				ProcessorSettings: config.NewProcessorSettings(config.NewComponentIDWithName(typeStr, "spans")),
+				Spans: SpanFilters{
+					Include: &filterconfig.MatchProperties{
+						Config: filterset.Config{
+							MatchType: filterset.Strict,
+						},
+						Services: []string{"test", "test2"},
+						Attributes: []filterconfig.Attribute{
+							{Key: "should_include", Value: "(true|probably_true)"},
+						},
+					},
+					Exclude: &filterconfig.MatchProperties{
+						Config: filterset.Config{
+							MatchType: filterset.Regexp,
+						},
+						Attributes: []filterconfig.Attribute{
+							{Key: "should_exclude", Value: "(probably_false|false)"},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.expCfg.ID().String(), func(t *testing.T) {
+			cfg := cfg.Processors[test.expCfg.ID()]
+			assert.Equal(t, test.expCfg, cfg)
+		})
+	}
+}
+
 func TestLoadingConfigExpr(t *testing.T) {
 	factories, err := componenttest.NopFactories()
 	require.NoError(t, err)

processor/filterprocessor/factory.go

@@ -37,6 +37,7 @@ func NewFactory() component.ProcessorFactory {
 		createDefaultConfig,
 		component.WithMetricsProcessor(createMetricsProcessor),
 		component.WithLogsProcessor(createLogsProcessor),
+		component.WithTracesProcessor(createTracesProcessor),
 	)
 }
 
@@ -79,3 +80,20 @@ func createLogsProcessor(
 		fp.ProcessLogs,
 		processorhelper.WithCapabilities(processorCapabilities))
 }
+
+func createTracesProcessor(
+	_ context.Context,
+	set component.ProcessorCreateSettings,
+	cfg config.Processor,
+	nextConsumer consumer.Traces,
+) (component.TracesProcessor, error) {
+	fp, err := newFilterSpansProcessor(set.Logger, cfg.(*Config))
+	if err != nil {
+		return nil, err
+	}
+	return processorhelper.NewTracesProcessor(
+		cfg,
+		nextConsumer,
+		fp.processTraces,
+		processorhelper.WithCapabilities(processorCapabilities))
+}

processor/filterprocessor/factory_test.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"fmt"
 	"path/filepath"
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -70,6 +71,12 @@ func TestCreateProcessors(t *testing.T) {
 		}, {
 			configName: "config_logs_record_attributes_regexp.yaml",
 			succeed:    true,
+		}, {
+			configName: "config_traces.yaml",
+			succeed:    true,
+		}, {
+			configName: "config_traces_invalid.yaml",
+			succeed:    false,
 		},
 	}
 
@@ -87,13 +94,22 @@ func TestCreateProcessors(t *testing.T) {
 				factory := NewFactory()
 
 				tp, tErr := factory.CreateTracesProcessor(context.Background(), componenttest.NewNopProcessorCreateSettings(), cfg, consumertest.NewNop())
-				// Not implemented error
-				assert.NotNil(t, tErr)
-				assert.Nil(t, tp)
-
 				mp, mErr := factory.CreateMetricsProcessor(context.Background(), componenttest.NewNopProcessorCreateSettings(), cfg, consumertest.NewNop())
-				assert.Equal(t, test.succeed, mp != nil)
-				assert.Equal(t, test.succeed, mErr == nil)
+				if strings.Contains(test.configName, "traces") {
+					assert.Equal(t, test.succeed, tp != nil)
+					assert.Equal(t, test.succeed, tErr == nil)
+
+					assert.NotNil(t, mp)
+					assert.Nil(t, mErr)
+				} else {
+					// Should not break configs with no trace data
+					assert.NotNil(t, tp)
+					assert.Nil(t, tErr)
+
+					assert.Equal(t, test.succeed, mp != nil)
+					assert.Equal(t, test.succeed, mErr == nil)
+				}
+
 			})
 		}
 	}