From 36560c30ac37cc47287647ccf795c88deab6d8f3 Mon Sep 17 00:00:00 2001 From: Pavel <177363085+pkcll@users.noreply.github.com> Date: Mon, 6 Jul 2026 22:11:22 -0400 Subject: [PATCH 1/6] Limit Beholder metric cardinality --- pkg/beholder/client.go | 6 +- pkg/beholder/config.go | 20 ++- pkg/beholder/config_test.go | 6 +- pkg/beholder/httpclient.go | 5 +- pkg/beholder/meter_provider.go | 22 ++++ pkg/beholder/meter_provider_test.go | 67 ++++++++++ pkg/beholder/metricviews/views.go | 51 ++++++++ pkg/beholder/metricviews/views_test.go | 145 ++++++++++++++++++++++ pkg/beholder/noop.go | 3 +- pkg/beholder/testdata/config-example.json | 2 + pkg/loop/config.go | 12 ++ pkg/loop/config_test.go | 2 + pkg/loop/server.go | 1 + 13 files changed, 327 insertions(+), 15 deletions(-) create mode 100644 pkg/beholder/meter_provider.go create mode 100644 pkg/beholder/meter_provider_test.go create mode 100644 pkg/beholder/metricviews/views.go create mode 100644 pkg/beholder/metricviews/views_test.go 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..243cf40462 --- /dev/null +++ b/pkg/beholder/meter_provider.go @@ -0,0 +1,22 @@ +package beholder + +import ( + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + + "github.com/smartcontractkit/chainlink-common/pkg/beholder/metricviews" +) + +func mergeMetricViews(cfg Config) []sdkmetric.View { + if cfg.MetricViewsDisabled { + return cfg.MetricViews + } + return append(metricviews.DefaultViews(), cfg.MetricViews...) +} + +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/views.go b/pkg/beholder/metricviews/views.go new file mode 100644 index 0000000000..8e557bccfa --- /dev/null +++ b/pkg/beholder/metricviews/views.go @@ -0,0 +1,51 @@ +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"), + attribute.Key("event_id"), // kept: low-cardinality vs retry; required by stopped-resending alert + ) +) + +// DefaultViews returns cardinality-limiting views prepended before caller-supplied +// MetricViews. More specific instrument matchers are registered first so they take +// precedence over the global catch-all denylist. +func DefaultViews() []sdkmetric.View { + return []sdkmetric.View{ + 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}, + ), + } +} diff --git a/pkg/beholder/metricviews/views_test.go b/pkg/beholder/metricviews/views_test.go new file mode 100644 index 0000000000..d0acbe8e0f --- /dev/null +++ b/pkg/beholder/metricviews/views_test.go @@ -0,0 +1,145 @@ +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_stoppedResendingDropsAttempts(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.Contains(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(), 3) +} + +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..7e58fff228 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,7 @@ func (e *EnvConfig) AsCmdEnv() (env []string) { add(envTelemetryLogMaxQueueSize, strconv.Itoa(e.TelemetryLogMaxQueueSize)) add(envTelemetryTraceCompressor, e.TelemetryTraceCompressor) add(envTelemetryMetricCompressor, e.TelemetryMetricCompressor) + add(envTelemetryMetricCardinalityLimit, strconv.Itoa(e.TelemetryMetricCardinalityLimit)) add(envTelemetryPrometheusBridgeEnabled, strconv.FormatBool(e.TelemetryPrometheusBridgeEnabled)) add(envTelemetryPrometheusBridgePrefixes, strings.Join(e.TelemetryPrometheusBridgePrefixes, ",")) add(envTelemetryLogCompressor, e.TelemetryLogCompressor) @@ -530,6 +533,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 { From 7fb61920ec17fd570eddbacd11c6a63c3740ee7a Mon Sep 17 00:00:00 2001 From: Pavel <177363085+pkcll@users.noreply.github.com> Date: Mon, 6 Jul 2026 22:22:29 -0400 Subject: [PATCH 2/6] Drop event ID from stopped-resending metrics --- pkg/beholder/metricviews/views.go | 1 - pkg/beholder/metricviews/views_test.go | 4 ++-- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/pkg/beholder/metricviews/views.go b/pkg/beholder/metricviews/views.go index 8e557bccfa..21b3c380cd 100644 --- a/pkg/beholder/metricviews/views.go +++ b/pkg/beholder/metricviews/views.go @@ -26,7 +26,6 @@ var ( stoppedResendingAllow = attribute.NewAllowKeysFilter( attribute.Key("capability_id"), attribute.Key("trigger_id"), - attribute.Key("event_id"), // kept: low-cardinality vs retry; required by stopped-resending alert ) ) diff --git a/pkg/beholder/metricviews/views_test.go b/pkg/beholder/metricviews/views_test.go index d0acbe8e0f..28ac5e826c 100644 --- a/pkg/beholder/metricviews/views_test.go +++ b/pkg/beholder/metricviews/views_test.go @@ -77,7 +77,7 @@ func TestDefaultViews_dropsHighCardinalityKeysGlobally(t *testing.T) { assert.NotContains(t, keys, attribute.Key("workflow_execution_id")) } -func TestDefaultViews_stoppedResendingDropsAttempts(t *testing.T) { +func TestDefaultViews_stoppedResendingDropsHighCardinalityKeys(t *testing.T) { t.Parallel() reader := sdkmetric.NewManualReader() @@ -106,7 +106,7 @@ func TestDefaultViews_stoppedResendingDropsAttempts(t *testing.T) { keys := attributeKeysFromGauge(t, rm) assert.Contains(t, keys, attribute.Key("capability_id")) assert.Contains(t, keys, attribute.Key("trigger_id")) - assert.Contains(t, keys, attribute.Key("event_id")) + assert.NotContains(t, keys, attribute.Key("event_id")) assert.NotContains(t, keys, attribute.Key("attempts")) } From ca0b2b4d142d0d851b49fc8bf94185846d3d1830 Mon Sep 17 00:00:00 2001 From: Pavel <177363085+pkcll@users.noreply.github.com> Date: Mon, 6 Jul 2026 22:54:52 -0400 Subject: [PATCH 3/6] Fix cardinality-limit rollout defeat and view-shadowing bugs AsCmdEnv always emitted CL_TELEMETRY_METRIC_CARDINALITY_LIMIT, so callers that don't set the field explicitly forced 0 and disabled the 10000 default in EnvConfig.parse. Only emit the env var when non-zero. mergeMetricViews prepended the default cardinality-limiting views (including a catch-all "*" view) before caller-supplied MetricViews. Since the OTel SDK dedupes views by stream identity and keeps the first match, this silently shadowed caller customizations such as histogram buckets set via WithOtelViews. Append defaults after caller views instead. --- pkg/beholder/meter_provider.go | 7 ++++++- pkg/loop/config.go | 4 +++- 2 files changed, 9 insertions(+), 2 deletions(-) diff --git a/pkg/beholder/meter_provider.go b/pkg/beholder/meter_provider.go index 243cf40462..150144d0cd 100644 --- a/pkg/beholder/meter_provider.go +++ b/pkg/beholder/meter_provider.go @@ -10,7 +10,12 @@ func mergeMetricViews(cfg Config) []sdkmetric.View { if cfg.MetricViewsDisabled { return cfg.MetricViews } - return append(metricviews.DefaultViews(), cfg.MetricViews...) + // Caller-supplied views must be evaluated before the default cardinality-limiting + // views: the OTel SDK dedupes views that resolve to the same stream identity + // (name/description/unit/kind) and keeps only the first match, so a caller + // customizing a stream (e.g. histogram buckets via WithOtelViews) would + // otherwise be silently shadowed by the default catch-all view. + return append(append([]sdkmetric.View{}, cfg.MetricViews...), metricviews.DefaultViews()...) } func appendMeterProviderOptions(cfg Config, opts ...sdkmetric.Option) []sdkmetric.Option { diff --git a/pkg/loop/config.go b/pkg/loop/config.go index 7e58fff228..cfc273654a 100644 --- a/pkg/loop/config.go +++ b/pkg/loop/config.go @@ -292,7 +292,9 @@ func (e *EnvConfig) AsCmdEnv() (env []string) { add(envTelemetryLogMaxQueueSize, strconv.Itoa(e.TelemetryLogMaxQueueSize)) add(envTelemetryTraceCompressor, e.TelemetryTraceCompressor) add(envTelemetryMetricCompressor, e.TelemetryMetricCompressor) - add(envTelemetryMetricCardinalityLimit, strconv.Itoa(e.TelemetryMetricCardinalityLimit)) + 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) From 8deb12372ee7ee77afcfb5fe2f45da43c53e34f1 Mon Sep 17 00:00:00 2001 From: Pavel <177363085+pkcll@users.noreply.github.com> Date: Mon, 6 Jul 2026 23:27:39 -0400 Subject: [PATCH 4/6] docs(beholder): clarify metric view merge order and SDK dedup semantics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Correct the metricviews package/DefaultViews and mergeMetricViews comments to describe the actual OTel SDK behavior: an instrument may match multiple views, and dedup is by resolved stream identity (name/description/unit/kind) keeping the first in registration order — not "first matching view for the instrument". Explains why caller views precede defaults and why attribute filters do not compose. Co-authored-by: Cursor --- pkg/beholder/meter_provider.go | 20 +++++++++++++++----- pkg/beholder/metricviews/views.go | 28 +++++++++++++++++++++++++--- 2 files changed, 40 insertions(+), 8 deletions(-) diff --git a/pkg/beholder/meter_provider.go b/pkg/beholder/meter_provider.go index 150144d0cd..a25e857d26 100644 --- a/pkg/beholder/meter_provider.go +++ b/pkg/beholder/meter_provider.go @@ -6,15 +6,25 @@ import ( "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 } - // Caller-supplied views must be evaluated before the default cardinality-limiting - // views: the OTel SDK dedupes views that resolve to the same stream identity - // (name/description/unit/kind) and keeps only the first match, so a caller - // customizing a stream (e.g. histogram buckets via WithOtelViews) would - // otherwise be silently shadowed by the default catch-all view. return append(append([]sdkmetric.View{}, cfg.MetricViews...), metricviews.DefaultViews()...) } diff --git a/pkg/beholder/metricviews/views.go b/pkg/beholder/metricviews/views.go index 21b3c380cd..0fce7ab149 100644 --- a/pkg/beholder/metricviews/views.go +++ b/pkg/beholder/metricviews/views.go @@ -1,3 +1,18 @@ +// Package metricviews defines Beholder's default OTel metric views for +// cardinality-limiting attribute filters. +// +// 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 ( @@ -29,9 +44,16 @@ var ( ) ) -// DefaultViews returns cardinality-limiting views prepended before caller-supplied -// MetricViews. More specific instrument matchers are registered first so they take -// precedence over the global catch-all denylist. +// DefaultViews returns attribute-filter views appended after caller-supplied +// MetricViews by beholder.mergeMetricViews. Within this slice, more specific +// instrument matchers precede the global "*" catch-all: an instrument like +// capabilities_base_trigger_stopped_resending_timestamp matches all three, and +// since these views resolve to the same stream identity the SDK keeps the first +// in registration order. Ordering most-specific-first gives it precedence: +// +// 1. capabilities_base_trigger_stopped_resending_timestamp — allow capability_id, trigger_id +// 2. capabilities_base_trigger_* — allow capability_id, reason, outcome +// 3. * — deny event_id, trigger_id, workflow_execution_id func DefaultViews() []sdkmetric.View { return []sdkmetric.View{ sdkmetric.NewView( From 1f04f83e82d362926e4fc2d491912108fbc20d3a Mon Sep 17 00:00:00 2001 From: Pavel <177363085+pkcll@users.noreply.github.com> Date: Mon, 6 Jul 2026 23:54:27 -0400 Subject: [PATCH 5/6] =?UTF-8?q?feat(beholder):=20reduce=20PerWorkflow=20hi?= =?UTF-8?q?stogram=20buckets=2016=E2=86=928?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add DefaultViews aggregation overrides for *.PerWorkflow.* histograms emitted by pkg/settings/limits. Uses 7 explicit boundaries (8 Prometheus buckets) per unit class: bytes (By), seconds (s), and dimensionless counts. CRE Limits-Workflow dashboards query histogram_quantile(1, ...) on these metrics, so coarser buckets preserve max usage vs limit panels while halving le cardinality on the dominant series families. Co-authored-by: Cursor --- .../metricviews/perworkflow_histograms.go | 54 +++++++++++ pkg/beholder/metricviews/views.go | 24 +++-- pkg/beholder/metricviews/views_test.go | 95 ++++++++++++++++++- 3 files changed, 159 insertions(+), 14 deletions(-) create mode 100644 pkg/beholder/metricviews/perworkflow_histograms.go diff --git a/pkg/beholder/metricviews/perworkflow_histograms.go b/pkg/beholder/metricviews/perworkflow_histograms.go new file mode 100644 index 0000000000..cba7bba446 --- /dev/null +++ b/pkg/beholder/metricviews/perworkflow_histograms.go @@ -0,0 +1,54 @@ +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, + } + perWorkflowCountBoundaries = []float64{ + 0, 1, 10, 100, 1e3, 1e4, 1e5, + } +) + +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, + }}, + ), + sdkmetric.NewView( + sdkmetric.Instrument{ + Name: perWorkflowInstrumentGlob, + Kind: sdkmetric.InstrumentKindHistogram, + Unit: "s", + }, + sdkmetric.Stream{Aggregation: sdkmetric.AggregationExplicitBucketHistogram{ + Boundaries: perWorkflowSecondsBoundaries, + }}, + ), + sdkmetric.NewView( + sdkmetric.Instrument{ + Name: perWorkflowInstrumentGlob, + Kind: sdkmetric.InstrumentKindHistogram, + }, + sdkmetric.Stream{Aggregation: sdkmetric.AggregationExplicitBucketHistogram{ + Boundaries: perWorkflowCountBoundaries, + }}, + ), + } +} diff --git a/pkg/beholder/metricviews/views.go b/pkg/beholder/metricviews/views.go index 0fce7ab149..2d6fbd2cb5 100644 --- a/pkg/beholder/metricviews/views.go +++ b/pkg/beholder/metricviews/views.go @@ -1,5 +1,6 @@ // Package metricviews defines Beholder's default OTel metric views for -// cardinality-limiting attribute filters. +// 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 @@ -44,18 +45,14 @@ var ( ) ) -// DefaultViews returns attribute-filter views appended after caller-supplied -// MetricViews by beholder.mergeMetricViews. Within this slice, more specific -// instrument matchers precede the global "*" catch-all: an instrument like -// capabilities_base_trigger_stopped_resending_timestamp matches all three, and -// since these views resolve to the same stream identity the SDK keeps the first -// in registration order. Ordering most-specific-first gives it precedence: -// -// 1. capabilities_base_trigger_stopped_resending_timestamp — allow capability_id, trigger_id -// 2. capabilities_base_trigger_* — allow capability_id, reason, outcome -// 3. * — deny event_id, trigger_id, workflow_execution_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). Attribute-filter views follow; within that group, +// more specific instrument matchers precede the global "*" catch-all. func DefaultViews() []sdkmetric.View { - return []sdkmetric.View{ + views := perWorkflowHistogramViews() + views = append(views, sdkmetric.NewView( sdkmetric.Instrument{Name: stoppedResendingInstrument}, sdkmetric.Stream{AttributeFilter: stoppedResendingAllow}, @@ -68,5 +65,6 @@ func DefaultViews() []sdkmetric.View { 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 index 28ac5e826c..5fe9ac11e6 100644 --- a/pkg/beholder/metricviews/views_test.go +++ b/pkg/beholder/metricviews/views_test.go @@ -112,7 +112,100 @@ func TestDefaultViews_stoppedResendingDropsHighCardinalityKeys(t *testing.T) { func TestDefaultViews_count(t *testing.T) { t.Parallel() - assert.Len(t, metricviews.DefaultViews(), 3) + assert.Len(t, metricviews.DefaultViews(), 6) +} + +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: "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)") + }) + } +} + +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 attributeKeysFromSum(t *testing.T, rm metricdata.ResourceMetrics) []attribute.Key { From 2c5845304baa31b99583142431fa6add40249689 Mon Sep 17 00:00:00 2001 From: Pavel <177363085+pkcll@users.noreply.github.com> Date: Tue, 7 Jul 2026 13:22:21 -0400 Subject: [PATCH 6/6] Fix attribute-filter shadowing and gas bucket overflow in PerWorkflow views The bucket-boundary views for *.PerWorkflow.* histograms claimed the OTel stream identity ahead of the global deny-filter view, so the SDK's first-match-wins dedup silently dropped the deny filter for those metrics. Attach globalHighCardinalityDeny directly to each bucket-override Stream mask so it travels with the aggregation override. Also add a {gas}-unit-specific view: PerWorkflow gas limit histograms (ChainWrite defaults up to 50_000_000) were falling into the unit-less count-boundaries view (max 1e5), collapsing nearly all observations into the +Inf bucket. --- .../metricviews/perworkflow_histograms.go | 62 +++++++++++-- pkg/beholder/metricviews/views.go | 7 +- pkg/beholder/metricviews/views_test.go | 88 ++++++++++++++++++- 3 files changed, 145 insertions(+), 12 deletions(-) diff --git a/pkg/beholder/metricviews/perworkflow_histograms.go b/pkg/beholder/metricviews/perworkflow_histograms.go index cba7bba446..4814815083 100644 --- a/pkg/beholder/metricviews/perworkflow_histograms.go +++ b/pkg/beholder/metricviews/perworkflow_histograms.go @@ -14,11 +14,33 @@ var ( 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( @@ -27,9 +49,12 @@ func perWorkflowHistogramViews() []sdkmetric.View { Kind: sdkmetric.InstrumentKindHistogram, Unit: "By", }, - sdkmetric.Stream{Aggregation: sdkmetric.AggregationExplicitBucketHistogram{ - Boundaries: perWorkflowBytesBoundaries, - }}, + sdkmetric.Stream{ + Aggregation: sdkmetric.AggregationExplicitBucketHistogram{ + Boundaries: perWorkflowBytesBoundaries, + }, + AttributeFilter: globalHighCardinalityDeny, + }, ), sdkmetric.NewView( sdkmetric.Instrument{ @@ -37,18 +62,37 @@ func perWorkflowHistogramViews() []sdkmetric.View { Kind: sdkmetric.InstrumentKindHistogram, Unit: "s", }, - sdkmetric.Stream{Aggregation: sdkmetric.AggregationExplicitBucketHistogram{ - Boundaries: perWorkflowSecondsBoundaries, - }}, + 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, }, - sdkmetric.Stream{Aggregation: sdkmetric.AggregationExplicitBucketHistogram{ - Boundaries: perWorkflowCountBoundaries, - }}, ), } } diff --git a/pkg/beholder/metricviews/views.go b/pkg/beholder/metricviews/views.go index 2d6fbd2cb5..291d79cfa1 100644 --- a/pkg/beholder/metricviews/views.go +++ b/pkg/beholder/metricviews/views.go @@ -48,8 +48,11 @@ var ( // 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). Attribute-filter views follow; within that group, -// more specific instrument matchers precede the global "*" catch-all. +// 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, diff --git a/pkg/beholder/metricviews/views_test.go b/pkg/beholder/metricviews/views_test.go index 5fe9ac11e6..9185c579e6 100644 --- a/pkg/beholder/metricviews/views_test.go +++ b/pkg/beholder/metricviews/views_test.go @@ -112,7 +112,7 @@ func TestDefaultViews_stoppedResendingDropsHighCardinalityKeys(t *testing.T) { func TestDefaultViews_count(t *testing.T) { t.Parallel() - assert.Len(t, metricviews.DefaultViews(), 6) + assert.Len(t, metricviews.DefaultViews(), 7) } func TestDefaultViews_perWorkflowHistogramBuckets(t *testing.T) { @@ -149,6 +149,24 @@ func TestDefaultViews_perWorkflowHistogramBuckets(t *testing.T) { }, 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", @@ -182,10 +200,56 @@ func TestDefaultViews_perWorkflowHistogramBuckets(t *testing.T) { 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) @@ -208,6 +272,28 @@ func histogramBounds(t *testing.T, rm metricdata.ResourceMetrics, name string) [ 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)