diff --git a/pkg/beholder/client.go b/pkg/beholder/client.go index cf66c39306..017004f47d 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 := appendMeterProviderOptions(cfg, 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..2b550799af 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 @@ -40,6 +40,11 @@ type Config struct { MetricReaderInterval time.Duration MetricRetryConfig *RetryConfig MetricViews []metric.View + // MetricViewsDisabled skips DefaultViews attribute filtering (for tests). + MetricViewsDisabled bool + // MetricCardinalityLimit sets the SDK per-instrument attribute-set limit (0 = disabled). + // DefaultConfig uses 10000 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 +132,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 +142,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: 10000, // OTel metric exporter retry config MetricRetryConfig: defaultRetryConfig.Copy(), // Log @@ -148,8 +154,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 +179,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 +193,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..b2e72cc8b6 100644 --- a/pkg/beholder/config_test.go +++ b/pkg/beholder/config_test.go @@ -53,8 +53,10 @@ 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", + MetricViewsDisabled: false, + MetricCardinalityLimit: 0, // OTel metric exporter retry config MetricRetryConfig: nil, // Log diff --git a/pkg/beholder/httpclient.go b/pkg/beholder/httpclient.go index 4bc3c78db6..05f1f0fbbd 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 := appendMeterProviderOptions(config, 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..a25e857d26 --- /dev/null +++ b/pkg/beholder/meter_provider.go @@ -0,0 +1,37 @@ +package beholder + +import ( + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + + "github.com/smartcontractkit/chainlink-common/pkg/beholder/metricviews" +) + +// mergeMetricViews builds the view list passed to sdkmetric.WithView. +// +// Final order: +// 1. cfg.MetricViews — caller overrides (e.g. chainlink metricViews() histogram buckets) +// 2. metricviews.DefaultViews() — attribute filters for high-cardinality labels +// +// Caller views must come first: when multiple views match an instrument and +// resolve to the same output stream (same name/description/unit/kind), the SDK +// keeps only the first in registration order and drops the rest. If the default +// "*" denylist ran first, caller histogram-bucket views would be silently dropped. +// +// A consequence is that filters do not compose: once a caller view wins for a +// stream, the default attribute filter no longer applies to it. That is acceptable +// because caller views target metrics that do not emit the denied labels; capability +// trigger metrics carry no caller view and fall through to the defaults below. +func mergeMetricViews(cfg Config) []sdkmetric.View { + if cfg.MetricViewsDisabled { + return cfg.MetricViews + } + return append(append([]sdkmetric.View{}, cfg.MetricViews...), metricviews.DefaultViews()...) +} + +func appendMeterProviderOptions(cfg Config, opts ...sdkmetric.Option) []sdkmetric.Option { + opts = append(opts, sdkmetric.WithView(mergeMetricViews(cfg)...)) + 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..c1a7ae9949 --- /dev/null +++ b/pkg/beholder/meter_provider_test.go @@ -0,0 +1,67 @@ +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" + + "github.com/smartcontractkit/chainlink-common/pkg/beholder/metricviews" +) + +func TestAppendMeterProviderOptions_cardinalityLimit(t *testing.T) { + t.Parallel() + + const ( + uniqueAttributes = 10 + limit = 5 + ) + + reader := sdkmetric.NewManualReader() + cfg := DefaultConfig() + cfg.MetricCardinalityLimit = limit + cfg.MetricViewsDisabled = true + + mpOpts := appendMeterProviderOptions(cfg, 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) +} + +func TestMergeMetricViews_prependsDefaults(t *testing.T) { + t.Parallel() + + views := mergeMetricViews(Config{ + MetricViews: []sdkmetric.View{ + sdkmetric.NewView( + sdkmetric.Instrument{Name: "custom_metric"}, + sdkmetric.Stream{}, + ), + }, + }) + require.GreaterOrEqual(t, len(views), len(metricviews.DefaultViews())+1) +} diff --git a/pkg/beholder/metricviews/perworkflow_histograms.go b/pkg/beholder/metricviews/perworkflow_histograms.go new file mode 100644 index 0000000000..4814815083 --- /dev/null +++ b/pkg/beholder/metricviews/perworkflow_histograms.go @@ -0,0 +1,98 @@ +package metricviews + +import sdkmetric "go.opentelemetry.io/otel/sdk/metric" + +const perWorkflowInstrumentGlob = "*.PerWorkflow.*" + +// OTel SDK default histogram boundaries (15 values → 16 Prometheus buckets). +// PerWorkflow limit metrics from pkg/settings/limits use the default because +// they do not pass metric.WithExplicitBucketBoundaries at creation time. +var ( + perWorkflowBytesBoundaries = []float64{ + 0, 1e3, 1e4, 1e5, 1e6, 1e7, 1e8, + } + perWorkflowSecondsBoundaries = []float64{ + 0, 1, 10, 60, 300, 900, 3600, + } + // perWorkflowGasBoundaries covers pkg/settings/cresettings ChainWrite gas + // limit defaults (Solana 300_000, Aptos 2_000_000, EVM 5_000_000, up to + // 50_000_000 for per-chain-selector overrides) without collapsing them + // into the +Inf overflow bucket. + perWorkflowGasBoundaries = []float64{ + 0, 1e5, 5e5, 1e6, 5e6, 1e7, 5e7, + } + // perWorkflowCountBoundaries is the fallback for PerWorkflow histograms + // whose unit is neither "By", "s", nor "{gas}" (e.g. dimensionless counts). + perWorkflowCountBoundaries = []float64{ + 0, 1, 10, 100, 1e3, 1e4, 1e5, + } +) + +// perWorkflowHistogramViews returns bucket-boundary overrides for +// *.PerWorkflow.* histograms, keyed by unit. Each view also carries +// globalHighCardinalityDeny so the attribute deny-list travels with the +// bucket override: see the ordering note on DefaultViews for why a view +// registered here would otherwise silently bypass the deny-filter views +// below it. +// +// The unit-less (count) view matches any unit and only acts as a fallback +// because the more specific By/s/gas views are registered first: the OTel +// SDK dedupes views that resolve to the same stream identity (name/ +// description/unit/kind) and keeps only the first match, so an instrument +// with unit "By" is already claimed by perWorkflowBytesBoundaries by the +// time the count view is evaluated for it. +func perWorkflowHistogramViews() []sdkmetric.View { + return []sdkmetric.View{ + sdkmetric.NewView( + sdkmetric.Instrument{ + Name: perWorkflowInstrumentGlob, + Kind: sdkmetric.InstrumentKindHistogram, + Unit: "By", + }, + sdkmetric.Stream{ + Aggregation: sdkmetric.AggregationExplicitBucketHistogram{ + Boundaries: perWorkflowBytesBoundaries, + }, + AttributeFilter: globalHighCardinalityDeny, + }, + ), + sdkmetric.NewView( + sdkmetric.Instrument{ + Name: perWorkflowInstrumentGlob, + Kind: sdkmetric.InstrumentKindHistogram, + Unit: "s", + }, + sdkmetric.Stream{ + Aggregation: sdkmetric.AggregationExplicitBucketHistogram{ + Boundaries: perWorkflowSecondsBoundaries, + }, + AttributeFilter: globalHighCardinalityDeny, + }, + ), + sdkmetric.NewView( + sdkmetric.Instrument{ + Name: perWorkflowInstrumentGlob, + Kind: sdkmetric.InstrumentKindHistogram, + Unit: "{gas}", + }, + sdkmetric.Stream{ + Aggregation: sdkmetric.AggregationExplicitBucketHistogram{ + Boundaries: perWorkflowGasBoundaries, + }, + AttributeFilter: globalHighCardinalityDeny, + }, + ), + sdkmetric.NewView( + sdkmetric.Instrument{ + Name: perWorkflowInstrumentGlob, + Kind: sdkmetric.InstrumentKindHistogram, + }, + sdkmetric.Stream{ + Aggregation: sdkmetric.AggregationExplicitBucketHistogram{ + Boundaries: perWorkflowCountBoundaries, + }, + AttributeFilter: globalHighCardinalityDeny, + }, + ), + } +} diff --git a/pkg/beholder/metricviews/views.go b/pkg/beholder/metricviews/views.go new file mode 100644 index 0000000000..291d79cfa1 --- /dev/null +++ b/pkg/beholder/metricviews/views.go @@ -0,0 +1,73 @@ +// Package metricviews defines Beholder's default OTel metric views for +// cardinality control: PerWorkflow histogram bucket reduction and +// attribute-filter deny/allow lists. +// +// Callers (e.g. chainlink core/cmd/shell.go via beholder.Config.MetricViews) +// may supply additional views—typically histogram bucket Aggregation overrides +// for specific instrument names. Beholder merges caller views before these +// defaults (see beholder.mergeMetricViews); callers do not need to invoke +// DefaultViews themselves. +// +// An instrument may match multiple views. When several matching views resolve +// to the same output stream (same name/description/unit/kind), the SDK keeps +// only the first in registration order and logs a duplicate-stream warning for +// the rest; attribute filters and aggregations do not compose across them. +// Because Beholder registers caller views ahead of these defaults, a matching +// caller view wins and the default attribute filter for that stream is dropped. +package metricviews + +import ( + "go.opentelemetry.io/otel/attribute" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" +) + +const ( + baseTriggerInstrumentGlob = "capabilities_base_trigger_*" + stoppedResendingInstrument = "capabilities_base_trigger_stopped_resending_timestamp" +) + +var ( + globalHighCardinalityDeny = attribute.NewDenyKeysFilter( + attribute.Key("event_id"), + attribute.Key("trigger_id"), + attribute.Key("workflow_execution_id"), + ) + + baseTriggerAllow = attribute.NewAllowKeysFilter( + attribute.Key("capability_id"), + attribute.Key("reason"), + attribute.Key("outcome"), + ) + + stoppedResendingAllow = attribute.NewAllowKeysFilter( + attribute.Key("capability_id"), + attribute.Key("trigger_id"), + ) +) + +// DefaultViews returns views appended after caller-supplied MetricViews by +// beholder.mergeMetricViews. PerWorkflow histogram bucket views are registered +// first so they apply to CRE limit metrics (chainlink does not supply caller +// views for those names); each carries globalHighCardinalityDeny directly on +// its Stream mask so claiming the stream identity for a bucket override does +// not also bypass the deny-filter views below (see the dedup note above). +// Attribute-filter views follow; within that group, more specific instrument +// matchers precede the global "*" catch-all. +func DefaultViews() []sdkmetric.View { + views := perWorkflowHistogramViews() + views = append(views, + sdkmetric.NewView( + sdkmetric.Instrument{Name: stoppedResendingInstrument}, + sdkmetric.Stream{AttributeFilter: stoppedResendingAllow}, + ), + sdkmetric.NewView( + sdkmetric.Instrument{Name: baseTriggerInstrumentGlob}, + sdkmetric.Stream{AttributeFilter: baseTriggerAllow}, + ), + sdkmetric.NewView( + sdkmetric.Instrument{Name: "*"}, + sdkmetric.Stream{AttributeFilter: globalHighCardinalityDeny}, + ), + ) + return views +} diff --git a/pkg/beholder/metricviews/views_test.go b/pkg/beholder/metricviews/views_test.go new file mode 100644 index 0000000000..9185c579e6 --- /dev/null +++ b/pkg/beholder/metricviews/views_test.go @@ -0,0 +1,324 @@ +package metricviews_test + +import ( + "context" + "slices" + "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" + + "github.com/smartcontractkit/chainlink-common/pkg/beholder/metricviews" +) + +func TestDefaultViews_dropsEventIDFromBaseTriggerRetry(t *testing.T) { + t.Parallel() + + reader := sdkmetric.NewManualReader() + mp := sdkmetric.NewMeterProvider( + sdkmetric.WithReader(reader), + sdkmetric.WithView(metricviews.DefaultViews()...), + ) + t.Cleanup(func() { _ = mp.Shutdown(context.Background()) }) + + meter := mp.Meter("test") + counter, err := meter.Int64Counter("capabilities_base_trigger_retry_total") + require.NoError(t, err) + + counter.Add(context.Background(), 1, + metric.WithAttributes( + attribute.String("capability_id", "cap-a"), + attribute.String("trigger_id", "trig-a"), + attribute.String("event_id", "ev-1"), + ), + ) + + var rm metricdata.ResourceMetrics + require.NoError(t, reader.Collect(context.Background(), &rm)) + + keys := attributeKeysFromSum(t, rm) + assert.Contains(t, keys, attribute.Key("capability_id")) + assert.NotContains(t, keys, attribute.Key("trigger_id")) + assert.NotContains(t, keys, attribute.Key("event_id")) +} + +func TestDefaultViews_dropsHighCardinalityKeysGlobally(t *testing.T) { + t.Parallel() + + reader := sdkmetric.NewManualReader() + mp := sdkmetric.NewMeterProvider( + sdkmetric.WithReader(reader), + sdkmetric.WithView(metricviews.DefaultViews()...), + ) + t.Cleanup(func() { _ = mp.Shutdown(context.Background()) }) + + meter := mp.Meter("test") + counter, err := meter.Int64Counter("some_other_metric_total") + require.NoError(t, err) + + counter.Add(context.Background(), 1, + metric.WithAttributes( + attribute.String("service", "foo"), + attribute.String("trigger_id", "trig-a"), + attribute.String("workflow_execution_id", "exec-1"), + ), + ) + + var rm metricdata.ResourceMetrics + require.NoError(t, reader.Collect(context.Background(), &rm)) + + keys := attributeKeysFromSum(t, rm) + assert.Contains(t, keys, attribute.Key("service")) + assert.NotContains(t, keys, attribute.Key("trigger_id")) + assert.NotContains(t, keys, attribute.Key("workflow_execution_id")) +} + +func TestDefaultViews_stoppedResendingDropsHighCardinalityKeys(t *testing.T) { + t.Parallel() + + reader := sdkmetric.NewManualReader() + mp := sdkmetric.NewMeterProvider( + sdkmetric.WithReader(reader), + sdkmetric.WithView(metricviews.DefaultViews()...), + ) + t.Cleanup(func() { _ = mp.Shutdown(context.Background()) }) + + meter := mp.Meter("test") + gauge, err := meter.Int64Gauge("capabilities_base_trigger_stopped_resending_timestamp") + require.NoError(t, err) + + gauge.Record(context.Background(), 123, + metric.WithAttributes( + attribute.String("capability_id", "cap-a"), + attribute.String("trigger_id", "trig-a"), + attribute.String("event_id", "ev-1"), + attribute.Int("attempts", 20), + ), + ) + + var rm metricdata.ResourceMetrics + require.NoError(t, reader.Collect(context.Background(), &rm)) + + keys := attributeKeysFromGauge(t, rm) + assert.Contains(t, keys, attribute.Key("capability_id")) + assert.Contains(t, keys, attribute.Key("trigger_id")) + assert.NotContains(t, keys, attribute.Key("event_id")) + assert.NotContains(t, keys, attribute.Key("attempts")) +} + +func TestDefaultViews_count(t *testing.T) { + t.Parallel() + assert.Len(t, metricviews.DefaultViews(), 7) +} + +func TestDefaultViews_perWorkflowHistogramBuckets(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + instrument string + unit string + record func(t *testing.T, meter metric.Meter) + wantBounds []float64 + }{ + { + name: "bytes usage", + instrument: "bound.PerWorkflow.WASMBinarySizeLimit.usage", + unit: "By", + record: func(t *testing.T, meter metric.Meter) { + t.Helper() + h, err := meter.Int64Histogram("bound.PerWorkflow.WASMBinarySizeLimit.usage", metric.WithUnit("By")) + require.NoError(t, err) + h.Record(context.Background(), 512*1024, metric.WithAttributes(attribute.String("workflow", "wf-1"))) + }, + wantBounds: []float64{0, 1e3, 1e4, 1e5, 1e6, 1e7, 1e8}, + }, + { + name: "seconds runtime", + instrument: "time.PerWorkflow.ExecutionTimeout.runtime", + unit: "s", + record: func(t *testing.T, meter metric.Meter) { + t.Helper() + h, err := meter.Float64Histogram("time.PerWorkflow.ExecutionTimeout.runtime", metric.WithUnit("s")) + require.NoError(t, err) + h.Record(context.Background(), 42.5, metric.WithAttributes(attribute.String("workflow", "wf-1"))) + }, + wantBounds: []float64{0, 1, 10, 60, 300, 900, 3600}, + }, + { + name: "gas usage", + instrument: "bound.PerWorkflow.ChainWrite.EVM.GasLimit.usage", + unit: "{gas}", + record: func(t *testing.T, meter metric.Meter) { + t.Helper() + h, err := meter.Int64Histogram("bound.PerWorkflow.ChainWrite.EVM.GasLimit.usage", metric.WithUnit("{gas}")) + require.NoError(t, err) + // pkg/settings/cresettings ChainWrite gas defaults: Solana + // 300_000, Aptos 2_000_000, EVM 5_000_000, up to 50_000_000 + // for per-chain-selector overrides. All must land in a finite + // bucket, not overflow to +Inf. + for _, gas := range []int64{300_000, 2_000_000, 5_000_000, 10_000_000, 50_000_000} { + h.Record(context.Background(), gas, metric.WithAttributes(attribute.String("workflow", "wf-1"))) + } + }, + wantBounds: []float64{0, 1e5, 5e5, 1e6, 5e6, 1e7, 5e7}, + }, + { + name: "count usage", + instrument: "bound.PerWorkflow.TriggerSubscriptionLimit.usage", + record: func(t *testing.T, meter metric.Meter) { + t.Helper() + h, err := meter.Int64Histogram("bound.PerWorkflow.TriggerSubscriptionLimit.usage") + require.NoError(t, err) + h.Record(context.Background(), 3, metric.WithAttributes(attribute.String("workflow", "wf-1"))) + }, + wantBounds: []float64{0, 1, 10, 100, 1e3, 1e4, 1e5}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + reader := sdkmetric.NewManualReader() + mp := sdkmetric.NewMeterProvider( + sdkmetric.WithReader(reader), + sdkmetric.WithView(metricviews.DefaultViews()...), + ) + t.Cleanup(func() { _ = mp.Shutdown(context.Background()) }) + + meter := mp.Meter("test") + tt.record(t, meter) + + var rm metricdata.ResourceMetrics + require.NoError(t, reader.Collect(context.Background(), &rm)) + + bounds := histogramBounds(t, rm, tt.instrument) + assert.Equal(t, tt.wantBounds, bounds) + assert.Len(t, bounds, 7, "expected 7 boundaries (8 Prometheus buckets including +Inf)") + + if tt.name == "gas usage" { + counts := histogramBucketCounts(t, rm, tt.instrument) + overflow := counts[len(counts)-1] + assert.Zero(t, overflow, "gas observations up to the documented ChainWrite defaults must not collapse into the +Inf bucket") + } + }) + } +} + +// TestDefaultViews_perWorkflowHistogramDropsHighCardinalityKeys guards +// against the bucket-override view for a PerWorkflow histogram claiming the +// stream identity ahead of the global "*" deny-filter view and, as a result, +// silently bypassing it (the SDK dedupes matching views by stream identity +// and keeps only the first match's Stream mask). The bucket-boundary views +// must carry the deny filter themselves. +func TestDefaultViews_perWorkflowHistogramDropsHighCardinalityKeys(t *testing.T) { + t.Parallel() + + reader := sdkmetric.NewManualReader() + mp := sdkmetric.NewMeterProvider( + sdkmetric.WithReader(reader), + sdkmetric.WithView(metricviews.DefaultViews()...), + ) + t.Cleanup(func() { _ = mp.Shutdown(context.Background()) }) + + meter := mp.Meter("test") + h, err := meter.Int64Histogram("bound.PerWorkflow.WASMBinarySizeLimit.usage", metric.WithUnit("By")) + require.NoError(t, err) + h.Record(context.Background(), 512*1024, metric.WithAttributes( + attribute.String("workflow_execution_id", "wf-exec-1"), + attribute.String("event_id", "evt-1"), + attribute.String("workflow", "wf-1"), + )) + + var rm metricdata.ResourceMetrics + require.NoError(t, reader.Collect(context.Background(), &rm)) + + require.Len(t, rm.ScopeMetrics, 1) + require.Len(t, rm.ScopeMetrics[0].Metrics, 1) + data, ok := rm.ScopeMetrics[0].Metrics[0].Data.(metricdata.Histogram[int64]) + require.True(t, ok) + require.Len(t, data.DataPoints, 1) + + keys := keysFromSet(data.DataPoints[0].Attributes) + assert.Contains(t, keys, attribute.Key("workflow")) + assert.NotContains(t, keys, attribute.Key("workflow_execution_id")) + assert.NotContains(t, keys, attribute.Key("event_id")) +} + +func histogramBounds(t *testing.T, rm metricdata.ResourceMetrics, name string) []float64 { + t.Helper() + require.Len(t, rm.ScopeMetrics, 1) + for _, m := range rm.ScopeMetrics[0].Metrics { + if m.Name != name { + continue + } + switch data := m.Data.(type) { + case metricdata.Histogram[int64]: + require.Len(t, data.DataPoints, 1) + return data.DataPoints[0].Bounds + case metricdata.Histogram[float64]: + require.Len(t, data.DataPoints, 1) + return data.DataPoints[0].Bounds + default: + t.Fatalf("unexpected metric data type %T for %s", m.Data, name) + } + } + t.Fatalf("metric %q not found", name) + return nil +} + +func histogramBucketCounts(t *testing.T, rm metricdata.ResourceMetrics, name string) []uint64 { + t.Helper() + require.Len(t, rm.ScopeMetrics, 1) + for _, m := range rm.ScopeMetrics[0].Metrics { + if m.Name != name { + continue + } + switch data := m.Data.(type) { + case metricdata.Histogram[int64]: + require.Len(t, data.DataPoints, 1) + return data.DataPoints[0].BucketCounts + case metricdata.Histogram[float64]: + require.Len(t, data.DataPoints, 1) + return data.DataPoints[0].BucketCounts + default: + t.Fatalf("unexpected metric data type %T for %s", m.Data, name) + } + } + t.Fatalf("metric %q not found", name) + return nil +} + +func attributeKeysFromSum(t *testing.T, rm metricdata.ResourceMetrics) []attribute.Key { + t.Helper() + require.Len(t, rm.ScopeMetrics, 1) + require.Len(t, rm.ScopeMetrics[0].Metrics, 1) + sum, ok := rm.ScopeMetrics[0].Metrics[0].Data.(metricdata.Sum[int64]) + require.True(t, ok) + require.Len(t, sum.DataPoints, 1) + return keysFromSet(sum.DataPoints[0].Attributes) +} + +func attributeKeysFromGauge(t *testing.T, rm metricdata.ResourceMetrics) []attribute.Key { + t.Helper() + require.Len(t, rm.ScopeMetrics, 1) + require.Len(t, rm.ScopeMetrics[0].Metrics, 1) + gauge, ok := rm.ScopeMetrics[0].Metrics[0].Data.(metricdata.Gauge[int64]) + require.True(t, ok) + require.Len(t, gauge.DataPoints, 1) + return keysFromSet(gauge.DataPoints[0].Attributes) +} + +func keysFromSet(set attribute.Set) []attribute.Key { + keys := make([]attribute.Key, 0, set.Len()) + for _, kv := range set.ToSlice() { + keys = append(keys, kv.Key) + } + slices.Sort(keys) + return keys +} diff --git a/pkg/beholder/noop.go b/pkg/beholder/noop.go index 5c506d527e..3b05f0a868 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 := appendMeterProviderOptions(cfg.Config, 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..db2a37a1d7 100644 --- a/pkg/beholder/testdata/config-example.json +++ b/pkg/beholder/testdata/config-example.json @@ -32,6 +32,8 @@ "MetricReaderInterval": 1000000000, "MetricRetryConfig": null, "MetricViews": null, + "MetricViewsDisabled": false, + "MetricCardinalityLimit": 0, "MetricCompressor": "gzip", "MetricProducers": null, "ChipIngressEmitterEnabled": false, diff --git a/pkg/loop/config.go b/pkg/loop/config.go index 58d956bbfc..cfc273654a 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,6 +177,7 @@ type EnvConfig struct { TelemetryLogMaxQueueSize int TelemetryTraceCompressor string TelemetryMetricCompressor string + TelemetryMetricCardinalityLimit int TelemetryPrometheusBridgeEnabled bool TelemetryPrometheusBridgePrefixes []string TelemetryLogCompressor string @@ -290,6 +292,9 @@ func (e *EnvConfig) AsCmdEnv() (env []string) { add(envTelemetryLogMaxQueueSize, strconv.Itoa(e.TelemetryLogMaxQueueSize)) add(envTelemetryTraceCompressor, e.TelemetryTraceCompressor) add(envTelemetryMetricCompressor, e.TelemetryMetricCompressor) + if e.TelemetryMetricCardinalityLimit != 0 { + add(envTelemetryMetricCardinalityLimit, strconv.Itoa(e.TelemetryMetricCardinalityLimit)) + } add(envTelemetryPrometheusBridgeEnabled, strconv.FormatBool(e.TelemetryPrometheusBridgeEnabled)) add(envTelemetryPrometheusBridgePrefixes, strings.Join(e.TelemetryPrometheusBridgePrefixes, ",")) add(envTelemetryLogCompressor, e.TelemetryLogCompressor) @@ -530,6 +535,15 @@ 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) + } + e.TelemetryMetricCardinalityLimit = limit + } else { + e.TelemetryMetricCardinalityLimit = 10000 + } 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..bd07a2ac97 100644 --- a/pkg/loop/config_test.go +++ b/pkg/loop/config_test.go @@ -205,6 +205,7 @@ var envCfgFull = EnvConfig{ TelemetryEmitterExportMaxBatchSize: 100, TelemetryEmitterMaxQueueSize: 1000, TelemetryLogStreamingEnabled: false, + TelemetryMetricCardinalityLimit: 10000, TelemetryPrometheusBridgeEnabled: true, TelemetryPrometheusBridgePrefixes: []string{"foo", "bar"}, MeterRecordsEnabled: true, @@ -278,6 +279,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, "10000", got[envTelemetryMetricCardinalityLimit]) assert.Equal(t, "true", got[envTelemetryPrometheusBridgeEnabled]) assert.Equal(t, "foo,bar", got[envTelemetryPrometheusBridgePrefixes]) assert.Equal(t, "true", got[envMeterRecordsEnabled]) diff --git a/pkg/loop/server.go b/pkg/loop/server.go index fca2fda761..d807f46ce2 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 {