diff --git a/pkg/beholder/client.go b/pkg/beholder/client.go index cf66c39306..73579eccc7 100644 --- a/pkg/beholder/client.go +++ b/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 diff --git a/pkg/beholder/config.go b/pkg/beholder/config.go index 09ef907ed7..e5454e62e3 100644 --- a/pkg/beholder/config.go +++ b/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 diff --git a/pkg/beholder/config_test.go b/pkg/beholder/config_test.go index a287af70b3..5f998616ca 100644 --- a/pkg/beholder/config_test.go +++ b/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 diff --git a/pkg/beholder/httpclient.go b/pkg/beholder/httpclient.go index 4bc3c78db6..7553a68ece 100644 --- a/pkg/beholder/httpclient.go +++ b/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 } diff --git a/pkg/beholder/meter_provider.go b/pkg/beholder/meter_provider.go new file mode 100644 index 0000000000..4b4eba4c17 --- /dev/null +++ b/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 +} diff --git a/pkg/beholder/meter_provider_test.go b/pkg/beholder/meter_provider_test.go new file mode 100644 index 0000000000..fb9fa6f0df --- /dev/null +++ b/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) +} diff --git a/pkg/beholder/noop.go b/pkg/beholder/noop.go index 5c506d527e..97ca69178a 100644 --- a/pkg/beholder/noop.go +++ b/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 diff --git a/pkg/beholder/testdata/config-example.json b/pkg/beholder/testdata/config-example.json index c8aacc6baf..a3cf77c6c0 100644 --- a/pkg/beholder/testdata/config-example.json +++ b/pkg/beholder/testdata/config-example.json @@ -32,6 +32,7 @@ "MetricReaderInterval": 1000000000, "MetricRetryConfig": null, "MetricViews": null, + "MetricCardinalityLimit": 0, "MetricCompressor": "gzip", "MetricProducers": null, "ChipIngressEmitterEnabled": false, diff --git a/pkg/loop/config.go b/pkg/loop/config.go index 58d956bbfc..e3e448e768 100644 --- a/pkg/loop/config.go +++ b/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) diff --git a/pkg/loop/config_test.go b/pkg/loop/config_test.go index 4487868aa1..f263c272d3 100644 --- a/pkg/loop/config_test.go +++ b/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 { @@ -205,6 +214,7 @@ var envCfgFull = EnvConfig{ TelemetryEmitterExportMaxBatchSize: 100, TelemetryEmitterMaxQueueSize: 1000, TelemetryLogStreamingEnabled: false, + TelemetryMetricCardinalityLimit: new(100000), TelemetryPrometheusBridgeEnabled: true, TelemetryPrometheusBridgePrefixes: []string{"foo", "bar"}, MeterRecordsEnabled: true, @@ -278,6 +288,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 +309,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 = new(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 = new(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") diff --git a/pkg/loop/server.go b/pkg/loop/server.go index fca2fda761..b2d495eeaf 100644 --- a/pkg/loop/server.go +++ b/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 {