[processor/cumulativetodelta] Add metric type filter

processor/cumulativetodeltaprocessor/README.md

@@ -23,8 +23,8 @@ Configuration is specified through a list of metrics. The processor uses metric
 
 The following settings can be optionally configured:
 
-- `include`: List of metrics names, patterns or metric types to convert to delta.
-- `exclude`: List of metrics names, patterns or metric types to not convert to delta.  **If a metric name matches both include and exclude, exclude takes precedence.**
+- `include`: List of metrics names (case-insensitive), patterns or metric types to convert to delta. Valid values are: `sum`, `histogram`.
+- `exclude`: List of metrics names (case-insensitive), patterns or metric types to not convert to delta.  **If a metric name matches both include and exclude, exclude takes precedence.** Valid values are: `sum`, `histogram`.
 - `max_staleness`: The total time a state entry will live past the time it was last seen. Set to 0 to retain state indefinitely. Default: 0
 - `initial_value`: Handling of the first observed point for a given metric identity.
   When the collector (re)starts, there's no record of how much of a given cumulative counter has already been converted to delta values.

processor/cumulativetodeltaprocessor/config.go

@@ -5,7 +5,7 @@ package cumulativetodeltaprocessor // import "github.com/open-telemetry/opentele
 
 import (
 	"fmt"
-	"slices"
+	"golang.org/x/exp/maps"
 	"strings"
 	"time"
 
@@ -16,11 +16,13 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/cumulativetodeltaprocessor/internal/tracking"
 )
 
-var validMetricTypes = []string{
-	strings.ToLower(pmetric.MetricTypeSum.String()),
-	strings.ToLower(pmetric.MetricTypeHistogram.String()),
+var validMetricTypes = map[string]bool{
+	strings.ToLower(pmetric.MetricTypeSum.String()):       true,
+	strings.ToLower(pmetric.MetricTypeHistogram.String()): true,
 }
 
+var validMetricTypeList = maps.Keys(validMetricTypes)
+
 // Config defines the configuration for the processor.
 type Config struct {
 	// MaxStaleness is the total time a state entry will live past the time it was last seen. Set to 0 to retain state indefinitely.
@@ -63,23 +65,21 @@ func (config *Config) Validate() error {
 		return fmt.Errorf("metrics must be supplied if match_type is set")
 	}
 
-	for i, metricType := range config.Exclude.MetricTypes {
-		config.Exclude.MetricTypes[i] = strings.ToLower(metricType)
-		if !slices.Contains(validMetricTypes, config.Exclude.MetricTypes[i]) {
+	for _, metricType := range config.Exclude.MetricTypes {
+		if valid := validMetricTypes[strings.ToLower(metricType)]; !valid {
 			return fmt.Errorf(
 				"found invalid metric type in exclude.metric_types: %s. Valid values are [%s]",
 				metricType,
-				strings.Join(validMetricTypes, ","),
+				strings.Join(validMetricTypeList, ","),
 			)
 		}
 	}
-	for i, metricType := range config.Include.MetricTypes {
-		config.Include.MetricTypes[i] = strings.ToLower(metricType)
-		if !slices.Contains(validMetricTypes, config.Include.MetricTypes[i]) {
+	for _, metricType := range config.Include.MetricTypes {
+		if valid := validMetricTypes[strings.ToLower(metricType)]; !valid {
 			return fmt.Errorf(
 				"found invalid metric type in include.metric_types: %s. Valid values are [%s]",
 				metricType,
-				strings.Join(validMetricTypes, ","),
+				strings.Join(validMetricTypeList, ","),
 			)
 		}
 	}

processor/cumulativetodeltaprocessor/go.mod

@@ -15,6 +15,7 @@ require (
 	go.opentelemetry.io/collector/processor/processortest v0.112.0
 	go.uber.org/goleak v1.3.0
 	go.uber.org/zap v1.27.0
+	golang.org/x/exp v0.0.0-20240506185415-9bf2ced13842
 )
 
 require (

processor/cumulativetodeltaprocessor/processor.go

@@ -5,12 +5,10 @@ package cumulativetodeltaprocessor // import "github.com/open-telemetry/opentele
 
 import (
 	"context"
-	"math"
-	"slices"
-	"strings"
-
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.uber.org/zap"
+	"math"
+	"strings"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter/filterset"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/cumulativetodeltaprocessor/internal/tracking"
@@ -19,8 +17,8 @@ import (
 type cumulativeToDeltaProcessor struct {
 	includeFS          filterset.FilterSet
 	excludeFS          filterset.FilterSet
-	includeMetricTypes []string
-	excludeMetricTypes []string
+	includeMetricTypes map[pmetric.MetricType]bool
+	excludeMetricTypes map[pmetric.MetricType]bool
 	logger             *zap.Logger
 	deltaCalculator    *tracking.MetricTracker
 	cancelFunc         context.CancelFunc
@@ -32,8 +30,8 @@ func newCumulativeToDeltaProcessor(config *Config, logger *zap.Logger) *cumulati
 		logger:             logger,
 		deltaCalculator:    tracking.NewMetricTracker(ctx, logger, config.MaxStaleness, config.InitialValue),
 		cancelFunc:         cancel,
-		includeMetricTypes: config.Include.MetricTypes,
-		excludeMetricTypes: config.Exclude.MetricTypes,
+		includeMetricTypes: getMetricTypeFilter(config.Include.MetricTypes),
+		excludeMetricTypes: getMetricTypeFilter(config.Exclude.MetricTypes),
 	}
 	if len(config.Include.Metrics) > 0 {
 		p.includeFS, _ = filterset.CreateFilterSet(config.Include.Metrics, &config.Include.Config)
@@ -44,6 +42,20 @@ func newCumulativeToDeltaProcessor(config *Config, logger *zap.Logger) *cumulati
 	return p
 }
 
+func getMetricTypeFilter(types []string) map[pmetric.MetricType]bool {
+	res := map[pmetric.MetricType]bool{}
+	for _, t := range types {
+		switch strings.ToLower(t) {
+		case strings.ToLower(pmetric.MetricTypeSum.String()):
+			res[pmetric.MetricTypeSum] = true
+		case strings.ToLower(pmetric.MetricTypeHistogram.String()):
+			res[pmetric.MetricTypeHistogram] = true
+		default:
+		}
+	}
+	return res
+}
+
 // processMetrics implements the ProcessMetricsFunc type.
 func (ctdp *cumulativeToDeltaProcessor) processMetrics(_ context.Context, md pmetric.Metrics) (pmetric.Metrics, error) {
 	md.ResourceMetrics().RemoveIf(func(rm pmetric.ResourceMetrics) bool {
@@ -119,9 +131,9 @@ func (ctdp *cumulativeToDeltaProcessor) shutdown(context.Context) error {
 
 func (ctdp *cumulativeToDeltaProcessor) shouldConvertMetric(metric pmetric.Metric) bool {
 	return (ctdp.includeFS == nil || ctdp.includeFS.Matches(metric.Name())) &&
-		(len(ctdp.includeMetricTypes) == 0 || slices.Contains(ctdp.includeMetricTypes, strings.ToLower(metric.Type().String()))) &&
+		(len(ctdp.includeMetricTypes) == 0 || ctdp.includeMetricTypes[metric.Type()]) &&
 		(ctdp.excludeFS == nil || !ctdp.excludeFS.Matches(metric.Name())) &&
-		(len(ctdp.excludeMetricTypes) == 0 || !slices.Contains(ctdp.excludeMetricTypes, strings.ToLower(metric.Type().String())))
+		(len(ctdp.excludeMetricTypes) == 0 || !ctdp.excludeMetricTypes[metric.Type()])
 }
 
 func (ctdp *cumulativeToDeltaProcessor) convertNumberDataPoints(dps pmetric.NumberDataPointSlice, baseIdentity tracking.MetricIdentity) {