[processor/cumulativetodelta] Add include/exclude configuration

CHANGELOG.md

@@ -18,8 +18,8 @@
 - `windowsperfcountersreceiver`: Move code into separate package for use in other windowsperfcounter receivers (#9108) 
 - `datadogexporter`: Add `host_metadata` configuration section to configure host metadata export (#9100)
 - `cmd/mdatagen`: Update documentation generated for attributes to list enumerated values and show the "value" that will be visible on metrics when it is different from the attribute key in metadata.yaml (#8983)
-
 - `routingprocessor`: add option to drop resource attribute used for routing (#8990)
+- `cumulativetodeltaprocessor`: add new include/exclude configuration options with regex support (#8952)
 
 ### 🛑 Breaking changes 🛑
 

processor/cumulativetodeltaprocessor/README.md

@@ -1,33 +1,74 @@
 # Cumulative to Delta Processor
+
 **Status: under development; Not recommended for production usage.**
 
 Supported pipeline types: metrics
 
 ## Description
 
-The cumulative to delta processor (`cumulativetodeltaprocessor`) converts cumulative sum metrics to cumulative delta. 
+The cumulative to delta processor (`cumulativetodeltaprocessor`) converts cumulative sum metrics to cumulative delta. Non-monotonic sums are excluded.  
 
 ## Configuration
 
 Configuration is specified through a list of metrics. The processor uses metric names to identify a set of cumulative metrics and converts them from cumulative to delta.
 
 The following settings can be optionally configured:
 
-- `metrics`: The processor uses metric names to identify a set of cumulative metrics and converts them to delta.
+- `include`: List of metrics names or patterns to convert to delta.
+- `exclude`: List of metrics names or patterns to not convert to delta.  **If a metric name matches both include and exclude, exclude takes precedence.**
 - `max_stale`: The total time a state entry will live past the time it was last seen. Set to 0 to retain state indefinitely. Default: 0
+- `metrics`: Deprecated. The processor uses metric names to identify a set of cumulative metrics and converts them to delta.
+
+If neither include nor exclude are supplied, no filtering is applied.
+
+#### Examples
+
+```yaml
+processors:
+    # processor name: cumulativetodelta
+    cumulativetodelta:
+
+        # list the exact cumulative sum metrics to convert to delta
+        include:
+            metrics:
+                - <metric_1_name>
+                - <metric_2_name>
+                .
+                .
+                - <metric_n_name>
+            match_type: strict
+```
+
+```yaml
+processors:
+    # processor name: cumulativetodelta
+    cumulativetodelta:
 
-#### Example
+        # Convert cumulative sum metrics to delta 
+        # if and only if 'metric' is in the name 
+        include:
+            metrics:
+                - "*metric*"
+            match_type: regexp
+```
 
 ```yaml
 processors:
     # processor name: cumulativetodelta
     cumulativetodelta:
 
-        # list the cumulative sum metrics to convert to delta
-        metrics:
-            - <metric_1_name>
-            - <metric_2_name>
-            .
-            .
-            - <metric_n_name>
+        # Convert cumulative sum metrics to delta 
+        # if and only if 'metric' is not in the name 
+        exclude:
+            metrics:
+                - "*metric*"
+            match_type: regexp
+```
+
+```yaml
+processors:
+    # processor name: cumulativetodelta
+    cumulativetodelta:
+        # If include/exclude are not specified
+        # convert all cumulative sum metrics to delta
 ```

processor/cumulativetodeltaprocessor/config.go

@@ -19,26 +19,50 @@ import (
 	"time"
 
 	"go.opentelemetry.io/collector/config"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/processor/filterset"
 )
 
 // Config defines the configuration for the processor.
 type Config struct {
 	config.ProcessorSettings `mapstructure:",squash"` // squash ensures fields are correctly decoded in embedded struct
 
-	// List of cumulative metrics to convert to delta.
-	Metrics []string `mapstructure:"metrics"`
-
 	// MaxStaleness is the total time a state entry will live past the time it was last seen. Set to 0 to retain state indefinitely.
 	MaxStaleness time.Duration `mapstructure:"max_staleness"`
+
+	// Deprecated: List of cumulative metrics to convert to delta.
+	// Cannot be used with Include/Exclude.
+	Metrics []string `mapstructure:"metrics"`
+
+	// Include specifies a filter on the metrics that should be converted.
+	// Exclude specifies a filter on the metrics that should not be converted.
+	// If neither `include` nor `exclude` are set, all metrics will be converted.
+	// Cannot be used with deprecated Metrics config option.
+	Include MatchMetrics `mapstructure:"include"`
+	Exclude MatchMetrics `mapstructure:"exclude"`
+}
+
+type MatchMetrics struct {
+	filterset.Config `mapstructure:",squash"`
+
+	Metrics []string `mapstructure:"metrics"`
 }
 
 var _ config.Processor = (*Config)(nil)
 
 // Validate checks whether the input configuration has all of the required fields for the processor.
 // An error is returned if there are any invalid inputs.
 func (config *Config) Validate() error {
-	if len(config.Metrics) == 0 {
-		return fmt.Errorf("metric names are missing")
+	if len(config.Metrics) > 0 && (len(config.Include.Metrics) > 0 || len(config.Exclude.Metrics) > 0) {
+		return fmt.Errorf("metrics and include/exclude cannot be used at the same time")
+	}
+	if (len(config.Include.Metrics) > 0 && len(config.Include.MatchType) == 0) ||
+		(len(config.Exclude.Metrics) > 0 && len(config.Exclude.MatchType) == 0) {
+		return fmt.Errorf("match_type must be set if metrics are supplied")
+	}
+	if (len(config.Include.MatchType) > 0 && len(config.Include.Metrics) == 0) ||
+		(len(config.Exclude.MatchType) > 0 && len(config.Exclude.Metrics) == 0) {
+		return fmt.Errorf("metrics must be supplied if match_type is set")
 	}
 	return nil
 }

processor/cumulativetodeltaprocessor/config_test.go

@@ -25,6 +25,8 @@ import (
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/service/servicetest"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/processor/filterset"
 )
 
 const configFile = "config.yaml"
@@ -46,9 +48,25 @@ func TestLoadingFullConfig(t *testing.T) {
 		{
 			expCfg: &Config{
 				ProcessorSettings: config.NewProcessorSettings(config.NewComponentID(typeStr)),
-				Metrics: []string{
-					"metric1",
-					"metric2",
+				Include: MatchMetrics{
+					Metrics: []string{
+						"metric1",
+						"metric2",
+					},
+					Config: filterset.Config{
+						MatchType:    "strict",
+						RegexpConfig: nil,
+					},
+				},
+				Exclude: MatchMetrics{
+					Metrics: []string{
+						"metric3",
+						"metric4",
+					},
+					Config: filterset.Config{
+						MatchType:    "strict",
+						RegexpConfig: nil,
+					},
 				},
 				MaxStaleness: 10 * time.Second,
 			},
@@ -73,10 +91,24 @@ func TestValidateConfig(t *testing.T) {
 			configName: "config.yaml",
 			succeed:    true,
 		},
+		{
+			configName:   "config_invalid_combo.yaml",
+			succeed:      false,
+			errorMessage: "metrics and include/exclude cannot be used at the same time",
+		},
+		{
+			configName:   "config_missing_match_type.yaml",
+			succeed:      false,
+			errorMessage: "match_type must be set if metrics are supplied",
+		},
 		{
 			configName:   "config_missing_name.yaml",
 			succeed:      false,
-			errorMessage: "metric names are missing",
+			errorMessage: "metrics must be supplied if match_type is set",
+		},
+		{
+			configName: "config_empty.yaml",
+			succeed:    true,
 		},
 	}
 

processor/cumulativetodeltaprocessor/go.mod

@@ -3,6 +3,7 @@ module github.com/open-telemetry/opentelemetry-collector-contrib/processor/cumul
 go 1.17
 
 require (
+	github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal v0.48.0
 	github.com/stretchr/testify v1.7.1
 	go.opentelemetry.io/collector v0.48.0
 	go.opentelemetry.io/collector/model v0.48.0
@@ -12,24 +13,24 @@ require (
 
 require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/fsnotify/fsnotify v1.5.1 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/knadh/koanf v1.4.0 // indirect
-	github.com/kr/pretty v0.3.0 // indirect
+	github.com/kr/text v0.2.0 // indirect
 	github.com/mitchellh/copystructure v1.2.0 // indirect
 	github.com/mitchellh/mapstructure v1.4.3 // indirect
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
-	github.com/pelletier/go-toml v1.9.4 // indirect
-	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/rogpeppe/go-internal v1.6.1 // indirect
 	github.com/spf13/cast v1.4.1 // indirect
 	go.opencensus.io v0.23.0 // indirect
 	go.opentelemetry.io/otel v1.6.3 // indirect
 	go.opentelemetry.io/otel/metric v0.29.0 // indirect
 	go.opentelemetry.io/otel/trace v1.6.3 // indirect
 	go.uber.org/atomic v1.9.0 // indirect
 	go.uber.org/multierr v1.8.0 // indirect
-	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
 )
+
+replace github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal => ../../internal/coreinternal