filterprocessor for filtering (dropping) incoming metrics

processor/filterprocessor/config.go

@@ -0,0 +1,39 @@
+// Copyright 2020 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 filterprocessor
+
+import (
+	"go.opentelemetry.io/collector/config/configmodels"
+	"go.opentelemetry.io/collector/internal/processor/filtermetric"
+)
+
+// Config defines configuration for Resource processor.
+type Config struct {
+	configmodels.ProcessorSettings `mapstructure:",squash"`
+	Metrics                        MetricFilters `mapstructure:"metrics"`
+}
+
+// MetricFilter filters by Metric properties.
+type MetricFilters struct {
+	// Include match properties describe metrics that should be included in the Collector Service pipeline,
+	// all other metrics should be dropped from further processing.
+	// If both Include and Exclude are specified, Include filtering occurs first.
+	Include *filtermetric.MatchProperties `mapstructure:"include"`
+
+	// Exclude match properties describe metrics that should be excluded from the Collector Service pipeline,
+	// all other metrics should be included.
+	// If both Include and Exclude are specified, Include filtering occurs first.
+	Exclude *filtermetric.MatchProperties `mapstructure:"exclude"`
+}

processor/filterprocessor/config_test.go

@@ -0,0 +1,230 @@
+// Copyright 2020, 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 filterprocessor
+
+import (
+	"path"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"go.opentelemetry.io/collector/config"
+	"go.opentelemetry.io/collector/config/configmodels"
+	"go.opentelemetry.io/collector/internal/processor/filtermetric"
+	"go.opentelemetry.io/collector/internal/processor/filterset"
+	fsregexp "go.opentelemetry.io/collector/internal/processor/filterset/regexp"
+)
+
+// TestLoadingConfigRegexp tests loading testdata/config_strict.yaml
+func TestLoadingConfigStrict(t *testing.T) {
+	// list of filters used repeatedly on testdata/config_strict.yaml
+	testDataFilters := []string{
+		"hello_world",
+		"hello/world",
+	}
+
+	testDataMetricProperties := &filtermetric.MatchProperties{
+		Config: filterset.Config{
+			MatchType: filterset.Strict,
+		},
+		MetricNames: testDataFilters,
+	}
+
+	factories, err := config.ExampleComponents()
+	assert.Nil(t, err)
+
+	factory := &Factory{}
+	factories.Processors[configmodels.Type(typeStr)] = factory
+	config, err := config.LoadConfigFile(t, path.Join(".", "testdata", "config_strict.yaml"), factories)
+
+	assert.Nil(t, err)
+	require.NotNil(t, config)
+
+	tests := []struct {
+		filterName string
+		expCfg     *Config
+	}{
+		{
+			filterName: "filter/empty",
+			expCfg: &Config{
+				ProcessorSettings: configmodels.ProcessorSettings{
+					NameVal: "filter/empty",
+					TypeVal: typeStr,
+				},
+				Metrics: MetricFilters{
+					Include: &filtermetric.MatchProperties{
+						Config: filterset.Config{
+							MatchType: filterset.Strict,
+						},
+					},
+				},
+			},
+		}, {
+			filterName: "filter/include",
+			expCfg: &Config{
+				ProcessorSettings: configmodels.ProcessorSettings{
+					NameVal: "filter/include",
+					TypeVal: typeStr,
+				},
+				Metrics: MetricFilters{
+					Include: testDataMetricProperties,
+				},
+			},
+		}, {
+			filterName: "filter/exclude",
+			expCfg: &Config{
+				ProcessorSettings: configmodels.ProcessorSettings{
+					NameVal: "filter/exclude",
+					TypeVal: typeStr,
+				},
+				Metrics: MetricFilters{
+					Exclude: testDataMetricProperties,
+				},
+			},
+		}, {
+			filterName: "filter/includeexclude",
+			expCfg: &Config{
+				ProcessorSettings: configmodels.ProcessorSettings{
+					NameVal: "filter/includeexclude",
+					TypeVal: typeStr,
+				},
+				Metrics: MetricFilters{
+					Include: testDataMetricProperties,
+					Exclude: &filtermetric.MatchProperties{
+						Config: filterset.Config{
+							MatchType: filterset.Strict,
+						},
+						MetricNames: []string{"hello_world"},
+					},
+				},
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.filterName, func(t *testing.T) {
+			cfg := config.Processors[test.filterName]
+			assert.Equal(t, test.expCfg, cfg)
+		})
+	}
+}
+
+// TestLoadingConfigRegexp tests loading testdata/config_regexp.yaml
+func TestLoadingConfigRegexp(t *testing.T) {
+	// list of filters used repeatedly on testdata/config.yaml
+	testDataFilters := []string{
+		"prefix/.*",
+		"prefix_.*",
+		".*/suffix",
+		".*_suffix",
+		".*/contains/.*",
+		".*_contains_.*",
+		"full/name/match",
+		"full_name_match",
+	}
+
+	testDataMetricProperties := &filtermetric.MatchProperties{
+		Config: filterset.Config{
+			MatchType: filterset.Regexp,
+		},
+		MetricNames: testDataFilters,
+	}
+
+	factories, err := config.ExampleComponents()
+	assert.Nil(t, err)
+
+	factory := &Factory{}
+	factories.Processors[typeStr] = factory
+	config, err := config.LoadConfigFile(t, path.Join(".", "testdata", "config_regexp.yaml"), factories)
+
+	assert.Nil(t, err)
+	require.NotNil(t, config)
+
+	tests := []struct {
+		filterName string
+		expCfg     *Config
+	}{
+		{
+			filterName: "filter/include",
+			expCfg: &Config{
+				ProcessorSettings: configmodels.ProcessorSettings{
+					NameVal: "filter/include",
+					TypeVal: typeStr,
+				},
+				Metrics: MetricFilters{
+					Include: testDataMetricProperties,
+				},
+			},
+		}, {
+			filterName: "filter/exclude",
+			expCfg: &Config{
+				ProcessorSettings: configmodels.ProcessorSettings{
+					NameVal: "filter/exclude",
+					TypeVal: typeStr,
+				},
+				Metrics: MetricFilters{
+					Exclude: testDataMetricProperties,
+				},
+			},
+		}, {
+			filterName: "filter/unlimitedcache",
+			expCfg: &Config{
+				ProcessorSettings: configmodels.ProcessorSettings{
+					NameVal: "filter/unlimitedcache",
+					TypeVal: typeStr,
+				},
+				Metrics: MetricFilters{
+					Include: &filtermetric.MatchProperties{
+						Config: filterset.Config{
+							MatchType: filterset.Regexp,
+							RegexpConfig: &fsregexp.Config{
+								CacheEnabled: true,
+							},
+						},
+						MetricNames: testDataFilters,
+					},
+				},
+			},
+		}, {
+			filterName: "filter/limitedcache",
+			expCfg: &Config{
+				ProcessorSettings: configmodels.ProcessorSettings{
+					NameVal: "filter/limitedcache",
+					TypeVal: typeStr,
+				},
+				Metrics: MetricFilters{
+					Exclude: &filtermetric.MatchProperties{
+						Config: filterset.Config{
+							MatchType: filterset.Regexp,
+							RegexpConfig: &fsregexp.Config{
+								CacheEnabled:       true,
+								CacheMaxNumEntries: 10,
+							},
+						},
+						MetricNames: testDataFilters,
+					},
+				},
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.filterName, func(t *testing.T) {
+			cfg := config.Processors[test.filterName]
+			assert.Equal(t, test.expCfg, cfg)
+		})
+	}
+}

processor/filterprocessor/doc.go

@@ -0,0 +1,17 @@
+// Copyright 2019 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 filterprocessor implements a processor for filtering
+// (dropping) metrics and/or spans by various properties.
+package filterprocessor

processor/filterprocessor/factory.go

@@ -0,0 +1,70 @@
+// Copyright 2019 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 filterprocessor
+
+import (
+	"context"
+
+	"go.uber.org/zap"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/configerror"
+	"go.opentelemetry.io/collector/config/configmodels"
+	"go.opentelemetry.io/collector/consumer"
+)
+
+const (
+	// The value of "type" key in configuration.
+	typeStr = "filter"
+)
+
+// Factory is the factory for filter processor.
+type Factory struct {
+}
+
+// Type gets the type of the Option config created by this factory.
+func (f Factory) Type() configmodels.Type {
+	return typeStr
+}
+
+// CreateDefaultConfig creates the default configuration for processor.
+func (f Factory) CreateDefaultConfig() configmodels.Processor {
+	return &Config{
+		ProcessorSettings: configmodels.ProcessorSettings{
+			TypeVal: typeStr,
+			NameVal: typeStr,
+		},
+	}
+}
+
+// CreateTraceProcessor creates a trace processor based on this config.
+func (f *Factory) CreateTraceProcessor(
+	ctx context.Context,
+	params component.ProcessorCreateParams,
+	nextConsumer consumer.TraceConsumer,
+	cfg configmodels.Processor,
+) (component.TraceProcessor, error) {
+	return nil, configerror.ErrDataTypeIsNotSupported
+}
+
+// CreateMetricsProcessor creates a metrics processor based on this config.
+func (f Factory) CreateMetricsProcessor(
+	logger *zap.Logger,
+	nextConsumer consumer.MetricsConsumerOld,
+	cfg configmodels.Processor,
+) (component.MetricsProcessorOld, error) {
+	oCfg := cfg.(*Config)
+	return newFilterMetricProcessor(nextConsumer, oCfg)
+}