Add cardinality limit for spanmetrics

Signed-off-by: Israel Blancas <[email protected]>

Author: iblancasa

.chloggen/38990.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# 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: spanmetricsconnector
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add new `aggregation_cardinality_limit` configuration option to limit the number of unique combinations of dimensions that will be tracked for metrics aggregation.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [38990]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

connector/spanmetricsconnector/README.md

@@ -122,6 +122,7 @@ The following settings can be optionally configured:
   - `enabled`: (default: `false`): enabling will add the events metric.
   - `dimensions`: (mandatory if `enabled`) the list of the span's event attributes to add as dimensions to the events metric, which will be included _on top of_ the common and configured `dimensions` for span and resource attributes.
 - `resource_metrics_key_attributes`: Filter the resource attributes used to produce the resource metrics key map hash. Use this in case changing resource attributes (e.g. process id) are breaking counter metrics.
+- `aggregation_cardinality_limit` (default: `0`): Defines the maximum number of unique combinations of dimensions that will be tracked for metrics aggregation. When the limit is reached, additional unique combinations will be dropped but registered under a new entry with `otel.metric.overflow="true"`. A value of `0` means no limit is applied.
 
 The feature gate `connector.spanmetrics.legacyMetricNames` (disabled by default) controls the connector to use legacy metric names.
 

connector/spanmetricsconnector/config.go

@@ -88,6 +88,8 @@ type Config struct {
 	Events EventsConfig `mapstructure:"events"`
 
 	IncludeInstrumentationScope []string `mapstructure:"include_instrumentation_scope"`
+
+	AggregationCardinalityLimit int `mapstructure:"aggregation_cardinality_limit"`
 }
 
 type HistogramConfig struct {
@@ -157,6 +159,10 @@ func (c Config) Validate() error {
 		)
 	}
 
+	if c.AggregationCardinalityLimit < 0 {
+		return fmt.Errorf("invalid aggregation_cardinality_limit: %v, the limit should be positive", c.AggregationCardinalityLimit)
+	}
+
 	return nil
 }
 

connector/spanmetricsconnector/config_test.go

@@ -288,3 +288,126 @@ func TestValidateEventDimensions(t *testing.T) {
 		})
 	}
 }
+
+func TestConfigValidate(t *testing.T) {
+	tests := []struct {
+		name        string
+		config      Config
+		expectedErr string
+	}{
+		{
+			name: "valid config",
+			config: Config{
+				DimensionsCacheSize:      1000,
+				ResourceMetricsCacheSize: 1000,
+				MetricsFlushInterval:     60 * time.Second,
+				Histogram: HistogramConfig{
+					Explicit: &ExplicitHistogramConfig{
+						Buckets: []time.Duration{10 * time.Millisecond},
+					},
+				},
+			},
+		},
+		{
+			name: "invalid dimensions cache size",
+			config: Config{
+				DimensionsCacheSize:      -1,
+				ResourceMetricsCacheSize: 1000,
+				MetricsFlushInterval:     60 * time.Second,
+			},
+			expectedErr: "invalid cache size: -1, the maximum number of the items in the cache should be positive",
+		},
+		{
+			name: "invalid metrics flush interval",
+			config: Config{
+				DimensionsCacheSize:      1000,
+				ResourceMetricsCacheSize: 1000,
+				MetricsFlushInterval:     -1 * time.Second,
+			},
+			expectedErr: "invalid metrics_flush_interval: -1s, the duration should be positive",
+		},
+		{
+			name: "invalid metrics expiration",
+			config: Config{
+				DimensionsCacheSize:      1000,
+				ResourceMetricsCacheSize: 1000,
+				MetricsFlushInterval:     60 * time.Second,
+				MetricsExpiration:        -1 * time.Second,
+			},
+			expectedErr: "invalid metrics_expiration: -1s, the duration should be positive",
+		},
+		{
+			name: "invalid delta timestamp cache size",
+			config: Config{
+				DimensionsCacheSize:      1000,
+				ResourceMetricsCacheSize: 1000,
+				MetricsFlushInterval:     60 * time.Second,
+				AggregationTemporality:   delta,
+				TimestampCacheSize:       new(int), // zero value
+			},
+			expectedErr: "invalid delta timestamp cache size: 0, the maximum number of the items in the cache should be positive",
+		},
+		{
+			name: "invalid aggregation cardinality limit",
+			config: Config{
+				DimensionsCacheSize:         1000,
+				ResourceMetricsCacheSize:    1000,
+				MetricsFlushInterval:        60 * time.Second,
+				AggregationCardinalityLimit: -1,
+			},
+			expectedErr: "invalid aggregation_cardinality_limit: -1, the limit should be positive",
+		},
+		{
+			name: "both explicit and exponential histogram",
+			config: Config{
+				DimensionsCacheSize:      1000,
+				ResourceMetricsCacheSize: 1000,
+				MetricsFlushInterval:     60 * time.Second,
+				Histogram: HistogramConfig{
+					Explicit: &ExplicitHistogramConfig{
+						Buckets: []time.Duration{10 * time.Millisecond},
+					},
+					Exponential: &ExponentialHistogramConfig{
+						MaxSize: 10,
+					},
+				},
+			},
+			expectedErr: "use either `explicit` or `exponential` buckets histogram",
+		},
+		{
+			name: "duplicate dimension name",
+			config: Config{
+				DimensionsCacheSize:      1000,
+				ResourceMetricsCacheSize: 1000,
+				MetricsFlushInterval:     60 * time.Second,
+				Dimensions: []Dimension{
+					{Name: "service.name"},
+				},
+			},
+			expectedErr: "failed validating dimensions: duplicate dimension name service.name",
+		},
+		{
+			name: "events enabled with no dimensions",
+			config: Config{
+				DimensionsCacheSize:      1000,
+				ResourceMetricsCacheSize: 1000,
+				MetricsFlushInterval:     60 * time.Second,
+				Events: EventsConfig{
+					Enabled: true,
+				},
+			},
+			expectedErr: "failed validating event dimensions: no dimensions configured for events",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := tt.config.Validate()
+			if tt.expectedErr != "" {
+				assert.ErrorContains(t, err, tt.expectedErr)
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+}

connector/spanmetricsconnector/connector.go

@@ -44,6 +44,9 @@ const (
 	metricNameEvents   = "events"
 
 	defaultUnit = metrics.Milliseconds
+
+	// https://github.com/open-telemetry/opentelemetry-go/blob/3ae002c3caf3e44387f0554dfcbbde2c5aab7909/sdk/metric/internal/aggregate/limit.go#L11C36-L11C50
+	overflowKey = "otel.metric.overflow"
 )
 
 type connectorImp struct {
@@ -401,11 +404,33 @@ func (p *connectorImp) aggregateMetrics(traces ptrace.Traces) {
 				}
 				key := p.buildKey(serviceName, span, p.dimensions, resourceAttr)
 
-				attributes, ok := p.metricKeyToDimensions.Get(key)
-				if !ok {
-					attributes = p.buildAttributes(serviceName, span, resourceAttr, p.dimensions, ils.Scope())
-					p.metricKeyToDimensions.Add(key, attributes)
+				var attributes pcommon.Map
+
+				// Note: we check cardinality limit here for sums metrics but it is the same
+				// for histograms because both use the same key and attributes.
+				if rm.sums.IsCardinalityLimitReached() {
+					attributes = pcommon.NewMap()
+					for _, d := range p.dimensions {
+						if v, exists := utilattri.GetDimensionValue(d, span.Attributes(), resourceAttr); exists {
+							v.CopyTo(attributes.PutEmpty(d.Name))
+						}
+					}
+					attributes.PutBool(overflowKey, true)
+				} else {
+					var cached bool
+					attributes, cached = p.metricKeyToDimensions.Get(key)
+					if !cached {
+						attributes = p.buildAttributes(
+							serviceName,
+							span,
+							resourceAttr,
+							p.dimensions,
+							ils.Scope(),
+						)
+						p.metricKeyToDimensions.Add(key, attributes)
+					}
 				}
+
 				if !p.config.Histogram.Disable {
 					// aggregate histogram metrics
 					h := histograms.GetOrCreate(key, attributes)
@@ -427,16 +452,30 @@ func (p *connectorImp) aggregateMetrics(traces ptrace.Traces) {
 						eDimensions = append(eDimensions, p.eDimensions...)
 
 						rscAndEventAttrs := pcommon.NewMap()
+
 						rscAndEventAttrs.EnsureCapacity(resourceAttr.Len() + event.Attributes().Len())
 						resourceAttr.CopyTo(rscAndEventAttrs)
-						event.Attributes().CopyTo(rscAndEventAttrs)
+						// We cannot use CopyTo because it overrides the existing keys.
+						event.Attributes().Range(func(k string, v pcommon.Value) bool {
+							rscAndEventAttrs.PutStr(k, v.Str())
+							return true
+						})
 
 						eKey := p.buildKey(serviceName, span, eDimensions, rscAndEventAttrs)
-						eAttributes, ok := p.metricKeyToDimensions.Get(eKey)
-						if !ok {
-							eAttributes = p.buildAttributes(serviceName, span, rscAndEventAttrs, eDimensions, ils.Scope())
-							p.metricKeyToDimensions.Add(eKey, eAttributes)
+
+						var eAttributes pcommon.Map
+						if rm.events.IsCardinalityLimitReached() {
+							eAttributes = pcommon.NewMap()
+							rscAndEventAttrs.CopyTo(eAttributes)
+							eAttributes.PutBool(overflowKey, true)
+						} else {
+							eAttributes, ok = p.metricKeyToDimensions.Get(eKey)
+							if !ok {
+								eAttributes = p.buildAttributes(serviceName, span, rscAndEventAttrs, eDimensions, ils.Scope())
+								p.metricKeyToDimensions.Add(eKey, eAttributes)
+							}
 						}
+
 						e := events.GetOrCreate(eKey, eAttributes)
 						if p.config.Exemplars.Enabled && !span.TraceID().IsEmpty() {
 							e.AddExemplar(span.TraceID(), span.SpanID(), duration)
@@ -481,8 +520,8 @@ func (p *connectorImp) getOrCreateResourceMetrics(attr pcommon.Map, startTimesta
 	if !ok {
 		v = &resourceMetrics{
 			histograms:     initHistogramMetrics(p.config),
-			sums:           metrics.NewSumMetrics(p.config.Exemplars.MaxPerDataPoint),
-			events:         metrics.NewSumMetrics(p.config.Exemplars.MaxPerDataPoint),
+			sums:           metrics.NewSumMetrics(p.config.Exemplars.MaxPerDataPoint, p.config.AggregationCardinalityLimit),
+			events:         metrics.NewSumMetrics(p.config.Exemplars.MaxPerDataPoint, p.config.AggregationCardinalityLimit),
 			attributes:     attr,
 			startTimestamp: startTimestamp,
 		}
@@ -507,7 +546,13 @@ func contains(elements []string, value string) bool {
 	return false
 }
 
-func (p *connectorImp) buildAttributes(serviceName string, span ptrace.Span, resourceAttrs pcommon.Map, dimensions []utilattri.Dimension, instrumentationScope pcommon.InstrumentationScope) pcommon.Map {
+func (p *connectorImp) buildAttributes(
+	serviceName string,
+	span ptrace.Span,
+	resourceAttrs pcommon.Map,
+	dimensions []utilattri.Dimension,
+	instrumentationScope pcommon.InstrumentationScope,
+) pcommon.Map {
 	attr := pcommon.NewMap()
 	attr.EnsureCapacity(4 + len(dimensions))
 	if !contains(p.config.ExcludeDimensions, serviceNameKey) {
@@ -523,22 +568,26 @@ func (p *connectorImp) buildAttributes(serviceName string, span ptrace.Span, res
 		attr.PutStr(statusCodeKey, traceutil.StatusCodeStr(span.Status().Code()))
 	}
 
-	for _, d := range dimensions {
-		if v, ok := utilattri.GetDimensionValue(d, span.Attributes(), resourceAttrs); ok {
-			v.CopyTo(attr.PutEmpty(d.Name))
-		}
-	}
-
 	if contains(p.config.IncludeInstrumentationScope, instrumentationScope.Name()) && instrumentationScope.Name() != "" {
 		attr.PutStr(instrumentationScopeNameKey, instrumentationScope.Name())
 		if instrumentationScope.Version() != "" {
 			attr.PutStr(instrumentationScopeVersionKey, instrumentationScope.Version())
 		}
 	}
 
+	addResourceAttributes(&attr, dimensions, span, resourceAttrs)
+
 	return attr
 }
 
+func addResourceAttributes(attrs *pcommon.Map, dimensions []utilattri.Dimension, span ptrace.Span, resourceAttrs pcommon.Map) {
+	for _, d := range dimensions {
+		if v, ok := utilattri.GetDimensionValue(d, span.Attributes(), resourceAttrs); ok {
+			v.CopyTo(attrs.PutEmpty(d.Name))
+		}
+	}
+}
+
 func concatDimensionValue(dest *bytes.Buffer, value string, prefixSep bool) {
 	if prefixSep {
 		dest.WriteString(metricKeySeparator)
@@ -553,6 +602,7 @@ func concatDimensionValue(dest *bytes.Buffer, value string, prefixSep bool) {
 // The metric key is a simple concatenation of dimension values, delimited by a null character.
 func (p *connectorImp) buildKey(serviceName string, span ptrace.Span, optionalDims []utilattri.Dimension, resourceOrEventAttrs pcommon.Map) metrics.Key {
 	p.keyBuf.Reset()
+
 	if !contains(p.config.ExcludeDimensions, serviceNameKey) {
 		concatDimensionValue(p.keyBuf, serviceName, false)
 	}