Add Beholder OTel SDK metric cardinality limit

pkg/beholder/client.go

@@ -549,11 +549,11 @@ func newMeterProvider(cfg Config, resource *sdkresource.Resource, auth Auth, cre
 	for _, p := range cfg.MetricProducers {
 		readerOpts = append(readerOpts, sdkmetric.WithProducer(p))
 	}
-	return sdkmetric.NewMeterProvider(
+	mpOpts := append(cfg.metricOptions(),
 		sdkmetric.WithReader(sdkmetric.NewPeriodicReader(exporter, readerOpts...)),
 		sdkmetric.WithResource(resource),
-		sdkmetric.WithView(cfg.MetricViews...),
-	), nil
+	)
+	return sdkmetric.NewMeterProvider(mpOpts...), nil
 }
 
 // newLoggerOpts creates options for a logger exporter

pkg/beholder/config.go

@@ -26,7 +26,7 @@ type Config struct {
 	EmitterMaxQueueSize       int
 	// EmitterBatchProcessor controls custom-message export mode:
 	// true = batched async export; false = immediate per-record export.
-	EmitterBatchProcessor     bool
+	EmitterBatchProcessor bool
 
 	// OTel Trace
 	TraceSampleRatio  float64
@@ -39,7 +39,10 @@ type Config struct {
 	// OTel Metric
 	MetricReaderInterval time.Duration
 	MetricRetryConfig    *RetryConfig
-	MetricViews          []metric.View
+	MetricViews []metric.View
+	// MetricCardinalityLimit sets the SDK per-instrument attribute-set limit (0 = disabled).
+	// DefaultConfig uses 100000 as a production safety valve for high-cardinality workloads.
+	MetricCardinalityLimit int
 	// MetricCompressor sets the gRPC compressor for metrics. Valid values: "gzip" (default), "none".
 	MetricCompressor string
 	MetricProducers  []metric.Producer // For example, a prometheus bridge
@@ -127,7 +130,7 @@ func DefaultConfig() Config {
 		EmitterExportInterval:     1 * time.Second,
 		EmitterMaxQueueSize:       2048,
 		// Keep batched export enabled by default for throughput.
-		EmitterBatchProcessor:     true,
+		EmitterBatchProcessor: true,
 		// OTel message log exporter retry config
 		LogRetryConfig: defaultRetryConfig.Copy(),
 		// Trace
@@ -137,8 +140,9 @@ func DefaultConfig() Config {
 		// OTel trace exporter retry config
 		TraceRetryConfig: defaultRetryConfig.Copy(),
 		// Metric
-		MetricReaderInterval: 1 * time.Second,
-		MetricCompressor:     "gzip",
+		MetricReaderInterval:   1 * time.Second,
+		MetricCompressor:       "gzip",
+		MetricCardinalityLimit: 100000,
 		// OTel metric exporter retry config
 		MetricRetryConfig: defaultRetryConfig.Copy(),
 		// Log
@@ -148,8 +152,8 @@ func DefaultConfig() Config {
 		LogMaxQueueSize:       2048,
 		LogBatchProcessor:     true,
 		LogStreamingEnabled:   true, // Enable logs streaming by default
-		LogLevel:      zapcore.InfoLevel,
-		LogCompressor: "gzip",
+		LogLevel:              zapcore.InfoLevel,
+		LogCompressor:         "gzip",
 		// Chip Ingress Batch Emitter
 		ChipIngressBatchEmitterEnabled: false,
 		ChipIngressBufferSize:          1000,
@@ -173,6 +177,7 @@ func TestDefaultConfig() Config {
 	config.LogRetryConfig.MaxElapsedTime = 0    // Retry is disabled
 	config.TraceRetryConfig.MaxElapsedTime = 0  // Retry is disabled
 	config.MetricRetryConfig.MaxElapsedTime = 0 // Retry is disabled
+	config.MetricCardinalityLimit = 0           // Disable overflow aggregation in unit tests
 	// Auth disabled for testing (TTL=0 means static auth mode)
 	config.AuthHeadersTTL = 0
 	return config
@@ -186,6 +191,7 @@ func TestDefaultConfigHTTPClient() Config {
 	config.LogBatchProcessor = false
 	config.OtelExporterGRPCEndpoint = ""
 	config.OtelExporterHTTPEndpoint = "localhost:4318"
+	config.MetricCardinalityLimit = 0
 	// Auth disabled for testing (TTL=0 means static auth mode)
 	config.AuthHeadersTTL = 0
 	return config

pkg/beholder/config_test.go

@@ -53,8 +53,9 @@ func TestConfig(t *testing.T) {
 		// OTel trace exporter retry config
 		TraceRetryConfig: nil,
 		// Metric
-		MetricReaderInterval: 1 * time.Second,
-		MetricCompressor:     "gzip",
+		MetricReaderInterval:   1 * time.Second,
+		MetricCompressor:       "gzip",
+		MetricCardinalityLimit: 0,
 		// OTel metric exporter retry config
 		MetricRetryConfig: nil,
 		// Log

pkg/beholder/httpclient.go

@@ -285,14 +285,13 @@ func newHTTPMeterProvider(config Config, resource *sdkresource.Resource, tlsConf
 		return nil, err
 	}
 
-	mp := sdkmetric.NewMeterProvider(
+	mpOpts := append(config.metricOptions(),
 		sdkmetric.WithReader(
 			sdkmetric.NewPeriodicReader(
 				exporter,
 				sdkmetric.WithInterval(config.MetricReaderInterval), // Default is 10s
 			)),
 		sdkmetric.WithResource(resource),
-		sdkmetric.WithView(config.MetricViews...),
 	)
-	return mp, nil
+	return sdkmetric.NewMeterProvider(mpOpts...), nil
 }

pkg/beholder/meter_provider.go

@@ -0,0 +1,19 @@
+package beholder
+
+import (
+	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+)
+
+// metricOptions returns cfg-derived sdkmetric.Option values (caller MetricViews
+// and the SDK per-instrument cardinality limit). Callers append reader/resource
+// options at the call site.
+func (cfg Config) metricOptions() []sdkmetric.Option {
+	var opts []sdkmetric.Option
+	if len(cfg.MetricViews) > 0 {
+		opts = append(opts, sdkmetric.WithView(cfg.MetricViews...))
+	}
+	if cfg.MetricCardinalityLimit > 0 {
+		opts = append(opts, sdkmetric.WithCardinalityLimit(cfg.MetricCardinalityLimit))
+	}
+	return opts
+}

pkg/beholder/meter_provider_test.go

@@ -0,0 +1,50 @@
+package beholder
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+)
+
+func TestConfig_metricOptions_cardinalityLimit(t *testing.T) {
+	t.Parallel()
+
+	const (
+		uniqueAttributes = 10
+		limit            = 5
+	)
+
+	reader := sdkmetric.NewManualReader()
+	cfg := DefaultConfig()
+	cfg.MetricCardinalityLimit = limit
+
+	mpOpts := append(cfg.metricOptions(), sdkmetric.WithReader(reader))
+	mp := sdkmetric.NewMeterProvider(mpOpts...)
+	t.Cleanup(func() { _ = mp.Shutdown(context.Background()) })
+
+	meter := mp.Meter("test")
+	counter, err := meter.Int64Counter("overflow_test_total")
+	require.NoError(t, err)
+
+	for i := range uniqueAttributes {
+		counter.Add(context.Background(), 1, metric.WithAttributes(attribute.Int("key", i)))
+	}
+
+	var rm metricdata.ResourceMetrics
+	require.NoError(t, reader.Collect(context.Background(), &rm))
+
+	sum := rm.ScopeMetrics[0].Metrics[0].Data.(metricdata.Sum[int64])
+	assert.Len(t, sum.DataPoints, limit)
+
+	var total int64
+	for _, dp := range sum.DataPoints {
+		total += dp.Value
+	}
+	assert.Equal(t, int64(uniqueAttributes), total)
+}

pkg/beholder/noop.go

@@ -109,13 +109,14 @@ func NewWriterClient(w io.Writer) (*Client, error) {
 	if err != nil {
 		return NewNoopClient(), err
 	}
-	meterProvider := sdkmetric.NewMeterProvider(
+	mpOpts := append(cfg.Config.metricOptions(),
 		sdkmetric.WithReader(
 			sdkmetric.NewPeriodicReader(
 				metricExporter,
 				sdkmetric.WithInterval(100*time.Millisecond), // Default is 10s
 			)),
 	)
+	meterProvider := sdkmetric.NewMeterProvider(mpOpts...)
 	meter := meterProvider.Meter(defaultPackageName)
 
 	// MessageEmitter

pkg/beholder/testdata/config-example.json

@@ -32,6 +32,7 @@
   "MetricReaderInterval": 1000000000,
   "MetricRetryConfig": null,
   "MetricViews": null,
+  "MetricCardinalityLimit": 0,
   "MetricCompressor": "gzip",
   "MetricProducers": null,
   "ChipIngressEmitterEnabled": false,

pkg/loop/config.go

@@ -82,6 +82,7 @@ const (
 	envTelemetryLogMaxQueueSize           = "CL_TELEMETRY_LOG_MAX_QUEUE_SIZE"
 	envTelemetryTraceCompressor           = "CL_TELEMETRY_TRACE_COMPRESSOR"
 	envTelemetryMetricCompressor          = "CL_TELEMETRY_METRIC_COMPRESSOR"
+	envTelemetryMetricCardinalityLimit    = "CL_TELEMETRY_METRIC_CARDINALITY_LIMIT"
 	envTelemetryPrometheusBridgeEnabled   = "CL_TELEMETRY_PROMETHEUS_BRIDGE_ENABLED"
 	envTelemetryPrometheusBridgePrefixes  = "CL_TELEMETRY_PROMETHEUS_BRIDGE_PREFIXES"
 	envTelemetryLogCompressor             = "CL_TELEMETRY_LOG_COMPRESSOR"
@@ -176,11 +177,15 @@ type EnvConfig struct {
 	TelemetryLogMaxQueueSize           int
 	TelemetryTraceCompressor           string
 	TelemetryMetricCompressor          string
-	TelemetryPrometheusBridgeEnabled   bool
-	TelemetryPrometheusBridgePrefixes  []string
-	TelemetryLogCompressor             string
-	MeterRecordsEnabled                bool
-	MeterSnapshotsEnabled              bool
+	// TelemetryMetricCardinalityLimit is nil when unset, so AsCmdEnv can
+	// distinguish "no opinion" (child applies its own default) from an
+	// explicit 0, which disables the limit.
+	TelemetryMetricCardinalityLimit   *int
+	TelemetryPrometheusBridgeEnabled  bool
+	TelemetryPrometheusBridgePrefixes []string
+	TelemetryLogCompressor            string
+	MeterRecordsEnabled               bool
+	MeterSnapshotsEnabled             bool
 
 	// MeterProduct / MeterTenant / MeterNumericTenantID / MeterEnvironment /
 	// MeterZone / MeterNodeID are
@@ -290,6 +295,9 @@ func (e *EnvConfig) AsCmdEnv() (env []string) {
 	add(envTelemetryLogMaxQueueSize, strconv.Itoa(e.TelemetryLogMaxQueueSize))
 	add(envTelemetryTraceCompressor, e.TelemetryTraceCompressor)
 	add(envTelemetryMetricCompressor, e.TelemetryMetricCompressor)
+	if e.TelemetryMetricCardinalityLimit != nil {
+		add(envTelemetryMetricCardinalityLimit, strconv.Itoa(*e.TelemetryMetricCardinalityLimit))
+	}
 	add(envTelemetryPrometheusBridgeEnabled, strconv.FormatBool(e.TelemetryPrometheusBridgeEnabled))
 	add(envTelemetryPrometheusBridgePrefixes, strings.Join(e.TelemetryPrometheusBridgePrefixes, ","))
 	add(envTelemetryLogCompressor, e.TelemetryLogCompressor)
@@ -530,6 +538,19 @@ func (e *EnvConfig) parse() error {
 		}
 		e.TelemetryTraceCompressor = os.Getenv(envTelemetryTraceCompressor)
 		e.TelemetryMetricCompressor = os.Getenv(envTelemetryMetricCompressor)
+		if v, ok := os.LookupEnv(envTelemetryMetricCardinalityLimit); ok {
+			limit, err := strconv.Atoi(v)
+			if err != nil {
+				return fmt.Errorf("failed to parse %s: %w", envTelemetryMetricCardinalityLimit, err)
+			}
+			if limit < 0 {
+				return fmt.Errorf("failed to parse %s: value %d must not be negative (0 disables the limit)", envTelemetryMetricCardinalityLimit, limit)
+			}
+			e.TelemetryMetricCardinalityLimit = &limit
+		} else {
+			defaultLimit := 100000
+			e.TelemetryMetricCardinalityLimit = &defaultLimit
+		}
 		e.TelemetryPrometheusBridgeEnabled, err = getBool(envTelemetryPrometheusBridgeEnabled)
 		if err != nil {
 			return fmt.Errorf("failed to parse %s: %w", envTelemetryPrometheusBridgeEnabled, err)

pkg/loop/config_test.go

@@ -126,6 +126,15 @@ func TestEnvConfig_parse(t *testing.T) {
 			},
 			expectError: true,
 		},
+		{
+			name: "CL_TELEMETRY_METRIC_CARDINALITY_LIMIT negative value rejected",
+			envVars: map[string]string{
+				envPromPort:                        "8080",
+				envTelemetryEnabled:                "true",
+				envTelemetryMetricCardinalityLimit: "-1",
+			},
+			expectError: true,
+		},
 	}
 
 	for _, tc := range cases {
@@ -147,6 +156,8 @@ func TestEnvConfig_parse(t *testing.T) {
 	}
 }
 
+func ptr[T any](v T) *T { return &v }
+
 var envCfgFull = EnvConfig{
 	AppID: "app-id",
 
@@ -205,6 +216,7 @@ var envCfgFull = EnvConfig{
 	TelemetryEmitterExportMaxBatchSize: 100,
 	TelemetryEmitterMaxQueueSize:       1000,
 	TelemetryLogStreamingEnabled:       false,
+	TelemetryMetricCardinalityLimit:    ptr(100000),
 	TelemetryPrometheusBridgeEnabled:   true,
 	TelemetryPrometheusBridgePrefixes:  []string{"foo", "bar"},
 	MeterRecordsEnabled:                true,
@@ -278,6 +290,7 @@ func TestEnvConfig_AsCmdEnv(t *testing.T) {
 	assert.Equal(t, "100", got[envTelemetryEmitterExportMaxBatchSize])
 	assert.Equal(t, "1000", got[envTelemetryEmitterMaxQueueSize])
 	assert.Equal(t, "false", got[envTelemetryLogStreamingEnabled])
+	assert.Equal(t, "100000", got[envTelemetryMetricCardinalityLimit])
 	assert.Equal(t, "true", got[envTelemetryPrometheusBridgeEnabled])
 	assert.Equal(t, "foo,bar", got[envTelemetryPrometheusBridgePrefixes])
 	assert.Equal(t, "true", got[envMeterRecordsEnabled])
@@ -298,6 +311,58 @@ func TestEnvConfig_AsCmdEnv(t *testing.T) {
 	assert.JSONEq(t, `{"foo":"bar"}`, got[envCRESettingsDefault])
 }
 
+// TestEnvConfig_MetricCardinalityLimit_RoundTrip verifies that an explicit
+// disable (0) survives a AsCmdEnv -> parse round trip the same way an
+// explicit positive limit does, and that leaving the field unset lets the
+// child apply its own default instead of forcing a value.
+func TestEnvConfig_MetricCardinalityLimit_RoundTrip(t *testing.T) {
+	setEnvFromCmdEnv := func(t *testing.T, env []string) {
+		t.Helper()
+		for _, kv := range env {
+			pair := strings.SplitN(kv, "=", 2)
+			require.Len(t, pair, 2)
+			t.Setenv(pair[0], pair[1])
+		}
+	}
+
+	t.Run("explicit disable propagates as 0", func(t *testing.T) {
+		cfg := envCfgFull
+		cfg.TelemetryMetricCardinalityLimit = ptr(0)
+		setEnvFromCmdEnv(t, cfg.AsCmdEnv())
+
+		var parsed EnvConfig
+		require.NoError(t, parsed.parse())
+		require.NotNil(t, parsed.TelemetryMetricCardinalityLimit)
+		assert.Equal(t, 0, *parsed.TelemetryMetricCardinalityLimit)
+	})
+
+	t.Run("explicit positive limit propagates", func(t *testing.T) {
+		cfg := envCfgFull
+		cfg.TelemetryMetricCardinalityLimit = ptr(500)
+		setEnvFromCmdEnv(t, cfg.AsCmdEnv())
+
+		var parsed EnvConfig
+		require.NoError(t, parsed.parse())
+		require.NotNil(t, parsed.TelemetryMetricCardinalityLimit)
+		assert.Equal(t, 500, *parsed.TelemetryMetricCardinalityLimit)
+	})
+
+	t.Run("unset falls back to child default", func(t *testing.T) {
+		cfg := envCfgFull
+		cfg.TelemetryMetricCardinalityLimit = nil
+		env := cfg.AsCmdEnv()
+		for _, kv := range env {
+			require.NotContains(t, kv, envTelemetryMetricCardinalityLimit+"=", "unset limit must not be emitted")
+		}
+		setEnvFromCmdEnv(t, env)
+
+		var parsed EnvConfig
+		require.NoError(t, parsed.parse())
+		require.NotNil(t, parsed.TelemetryMetricCardinalityLimit)
+		assert.Equal(t, 100000, *parsed.TelemetryMetricCardinalityLimit)
+	})
+}
+
 func TestGetMap(t *testing.T) {
 	t.Setenv("TEST_PREFIX_KEY1", "value1")
 	t.Setenv("TEST_PREFIX_KEY2", "value2")

pkg/loop/server.go

@@ -191,6 +191,7 @@ func (s *Server) start(opts ...ServerOpt) error {
 			ChipIngressBatchEmitterEnabled: s.EnvConfig.ChipIngressBatchEmitterEnabled,
 			ChipIngressLogger:              s.Logger,
 			MetricCompressor:               s.EnvConfig.TelemetryMetricCompressor,
+			MetricCardinalityLimit:         *s.EnvConfig.TelemetryMetricCardinalityLimit,
 		}
 
 		if s.EnvConfig.TelemetryPrometheusBridgeEnabled {