diff --git a/pkg/beholder/config.go b/pkg/beholder/config.go index e5454e62e3..79ef5b0d01 100644 --- a/pkg/beholder/config.go +++ b/pkg/beholder/config.go @@ -40,6 +40,9 @@ type Config struct { MetricReaderInterval time.Duration MetricRetryConfig *RetryConfig MetricViews []metric.View + // MetricViewsAttributeDenylist lists attribute keys dropped by the default + // global deny view (e.g. event_id). Empty skips metricviews.DefaultViews(). + MetricViewsAttributeDenylist []string // 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 diff --git a/pkg/beholder/config_test.go b/pkg/beholder/config_test.go index 5f998616ca..34c41740bd 100644 --- a/pkg/beholder/config_test.go +++ b/pkg/beholder/config_test.go @@ -55,7 +55,8 @@ func TestConfig(t *testing.T) { // Metric MetricReaderInterval: 1 * time.Second, MetricCompressor: "gzip", - MetricCardinalityLimit: 0, + MetricViewsAttributeDenylist: nil, + MetricCardinalityLimit: 0, // OTel metric exporter retry config MetricRetryConfig: nil, // Log diff --git a/pkg/beholder/meter_provider.go b/pkg/beholder/meter_provider.go index 4b4eba4c17..dcb19de410 100644 --- a/pkg/beholder/meter_provider.go +++ b/pkg/beholder/meter_provider.go @@ -2,15 +2,34 @@ package beholder import ( sdkmetric "go.opentelemetry.io/otel/sdk/metric" + + "github.com/smartcontractkit/chainlink-common/pkg/beholder/metricviews" ) -// 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. +// metricViews returns 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. +func (cfg Config) metricViews() []sdkmetric.View { + if len(cfg.MetricViewsAttributeDenylist) == 0 { + return cfg.MetricViews + } + return append(cfg.MetricViews, metricviews.DefaultViews(cfg.MetricViewsAttributeDenylist)...) +} + +// metricOptions returns cfg-derived sdkmetric.Option values (metric views and the +// SDK per-instrument cardinality limit). Callers append reader/resource options +// at the call site. func (cfg Config) metricOptions() []sdkmetric.Option { + views := cfg.metricViews() var opts []sdkmetric.Option - if len(cfg.MetricViews) > 0 { - opts = append(opts, sdkmetric.WithView(cfg.MetricViews...)) + if len(views) > 0 { + opts = append(opts, sdkmetric.WithView(views...)) } if cfg.MetricCardinalityLimit > 0 { opts = append(opts, sdkmetric.WithCardinalityLimit(cfg.MetricCardinalityLimit)) diff --git a/pkg/beholder/meter_provider_test.go b/pkg/beholder/meter_provider_test.go index fb9fa6f0df..62d82ea1d8 100644 --- a/pkg/beholder/meter_provider_test.go +++ b/pkg/beholder/meter_provider_test.go @@ -10,6 +10,8 @@ import ( "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 TestConfig_metricOptions_cardinalityLimit(t *testing.T) { @@ -48,3 +50,42 @@ func TestConfig_metricOptions_cardinalityLimit(t *testing.T) { } assert.Equal(t, int64(uniqueAttributes), total) } + +func TestConfig_metricViews_appendsDefaultsAfterCallerViews(t *testing.T) { + t.Parallel() + + callerView := sdkmetric.NewView( + sdkmetric.Instrument{Name: "custom_metric"}, + sdkmetric.Stream{}, + ) + cfg := DefaultConfig() + cfg.MetricViews = []sdkmetric.View{callerView} + + views := cfg.metricViews() + require.GreaterOrEqual(t, len(views), len(metricviews.DefaultViews(nil))+1) +} + +func TestConfig_metricViews_includesDenylistDefaultView(t *testing.T) { + t.Parallel() + + cfg := DefaultConfig() + cfg.MetricViewsAttributeDenylist = []string{"event_id"} + + views := cfg.metricViews() + require.Len(t, views, len(metricviews.DefaultViews(cfg.MetricViewsAttributeDenylist))) + require.NotEmpty(t, views) +} + +func TestConfig_metricViews_emptyDenylistSkipsDefaults(t *testing.T) { + t.Parallel() + + callerView := sdkmetric.NewView( + sdkmetric.Instrument{Name: "custom_metric"}, + sdkmetric.Stream{}, + ) + cfg := DefaultConfig() + cfg.MetricViews = []sdkmetric.View{callerView} + + views := cfg.metricViews() + require.Len(t, views, 1) +} diff --git a/pkg/beholder/metricviews/views.go b/pkg/beholder/metricviews/views.go new file mode 100644 index 0000000000..b5e1561480 --- /dev/null +++ b/pkg/beholder/metricviews/views.go @@ -0,0 +1,40 @@ +// 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.Config.metricViews); 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" +) + +// DefaultViews returns deny-only attribute-filter views appended after +// caller-supplied MetricViews by beholder.Config.metricViews. When denyKeys is +// empty, no views are returned and no attributes are stripped. +func DefaultViews(denyKeys []string) []sdkmetric.View { + if len(denyKeys) == 0 { + return nil + } + keys := make([]attribute.Key, len(denyKeys)) + for i, k := range denyKeys { + keys[i] = attribute.Key(k) + } + return []sdkmetric.View{ + sdkmetric.NewView( + sdkmetric.Instrument{Name: "*"}, + sdkmetric.Stream{AttributeFilter: attribute.NewDenyKeysFilter(keys...)}, + ), + } +} diff --git a/pkg/beholder/metricviews/views_test.go b/pkg/beholder/metricviews/views_test.go new file mode 100644 index 0000000000..733c12d31c --- /dev/null +++ b/pkg/beholder/metricviews/views_test.go @@ -0,0 +1,98 @@ +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_emptyDenylist(t *testing.T) { + t.Parallel() + assert.Nil(t, metricviews.DefaultViews(nil)) + + reader := sdkmetric.NewManualReader() + mp := sdkmetric.NewMeterProvider( + sdkmetric.WithReader(reader), + sdkmetric.WithView(metricviews.DefaultViews(nil)...), + ) + t.Cleanup(func() { _ = mp.Shutdown(context.Background()) }) + + meter := mp.Meter("test") + counter, err := meter.Int64Counter("some_metric_total") + require.NoError(t, err) + + counter.Add(context.Background(), 1, + metric.WithAttributes( + attribute.String("service", "foo"), + attribute.String("event_id", "ev-1"), + 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.Contains(t, keys, attribute.Key("event_id")) + assert.Contains(t, keys, attribute.Key("workflow_execution_id")) +} + +func TestDefaultViews_dropsDeniedAttributes(t *testing.T) { + t.Parallel() + + reader := sdkmetric.NewManualReader() + mp := sdkmetric.NewMeterProvider( + sdkmetric.WithReader(reader), + sdkmetric.WithView(metricviews.DefaultViews([]string{"event_id"})...), + ) + t.Cleanup(func() { _ = mp.Shutdown(context.Background()) }) + + meter := mp.Meter("test") + counter, err := meter.Int64Counter("some_metric_total") + require.NoError(t, err) + + counter.Add(context.Background(), 1, + metric.WithAttributes( + attribute.String("service", "foo"), + attribute.String("event_id", "ev-1"), + 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("event_id")) + assert.Contains(t, keys, attribute.Key("workflow_execution_id")) +} + +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 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/testdata/config-example.json b/pkg/beholder/testdata/config-example.json index a3cf77c6c0..b8e3d33d2f 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, + "MetricViewsAttributeDenylist": null, "MetricCardinalityLimit": 0, "MetricCompressor": "gzip", "MetricProducers": null, diff --git a/pkg/loop/config.go b/pkg/loop/config.go index e3e448e768..9b146acb20 100644 --- a/pkg/loop/config.go +++ b/pkg/loop/config.go @@ -83,7 +83,8 @@ const ( envTelemetryTraceCompressor = "CL_TELEMETRY_TRACE_COMPRESSOR" envTelemetryMetricCompressor = "CL_TELEMETRY_METRIC_COMPRESSOR" envTelemetryMetricCardinalityLimit = "CL_TELEMETRY_METRIC_CARDINALITY_LIMIT" - envTelemetryPrometheusBridgeEnabled = "CL_TELEMETRY_PROMETHEUS_BRIDGE_ENABLED" + envTelemetryMetricViewsAttributeDenylist = "CL_TELEMETRY_METRIC_VIEWS_ATTRIBUTE_DENYLIST" + envTelemetryPrometheusBridgeEnabled = "CL_TELEMETRY_PROMETHEUS_BRIDGE_ENABLED" envTelemetryPrometheusBridgePrefixes = "CL_TELEMETRY_PROMETHEUS_BRIDGE_PREFIXES" envTelemetryLogCompressor = "CL_TELEMETRY_LOG_COMPRESSOR" envMeterRecordsEnabled = "CL_METER_RECORDS_ENABLED" @@ -181,7 +182,10 @@ type EnvConfig struct { // distinguish "no opinion" (child applies its own default) from an // explicit 0, which disables the limit. TelemetryMetricCardinalityLimit *int - TelemetryPrometheusBridgeEnabled bool + // TelemetryMetricViewsAttributeDenylist lists attribute keys dropped by the + // default global deny view (e.g. event_id). Empty skips DefaultViews(). + TelemetryMetricViewsAttributeDenylist []string + TelemetryPrometheusBridgeEnabled bool TelemetryPrometheusBridgePrefixes []string TelemetryLogCompressor string MeterRecordsEnabled bool @@ -298,6 +302,9 @@ func (e *EnvConfig) AsCmdEnv() (env []string) { if e.TelemetryMetricCardinalityLimit != nil { add(envTelemetryMetricCardinalityLimit, strconv.Itoa(*e.TelemetryMetricCardinalityLimit)) } + if len(e.TelemetryMetricViewsAttributeDenylist) > 0 { + add(envTelemetryMetricViewsAttributeDenylist, strings.Join(e.TelemetryMetricViewsAttributeDenylist, ",")) + } add(envTelemetryPrometheusBridgeEnabled, strconv.FormatBool(e.TelemetryPrometheusBridgeEnabled)) add(envTelemetryPrometheusBridgePrefixes, strings.Join(e.TelemetryPrometheusBridgePrefixes, ",")) add(envTelemetryLogCompressor, e.TelemetryLogCompressor) @@ -551,6 +558,9 @@ func (e *EnvConfig) parse() error { defaultLimit := 100000 e.TelemetryMetricCardinalityLimit = &defaultLimit } + if v, ok := os.LookupEnv(envTelemetryMetricViewsAttributeDenylist); ok && v != "" { + e.TelemetryMetricViewsAttributeDenylist = splitCommaTrimmed(v) + } e.TelemetryPrometheusBridgeEnabled, err = getBool(envTelemetryPrometheusBridgeEnabled) if err != nil { return fmt.Errorf("failed to parse %s: %w", envTelemetryPrometheusBridgeEnabled, err) @@ -626,6 +636,17 @@ func getValidCollectorTarget() (string, error) { return tracingCollectorTarget, nil } +func splitCommaTrimmed(s string) []string { + parts := strings.Split(s, ",") + out := make([]string, 0, len(parts)) + for _, part := range parts { + if trimmed := strings.TrimSpace(part); trimmed != "" { + out = append(out, trimmed) + } + } + return out +} + func getMap(envKeyPrefix string) map[string]string { m := make(map[string]string) for _, env := range os.Environ() { diff --git a/pkg/loop/config_test.go b/pkg/loop/config_test.go index f263c272d3..f2ad95fb60 100644 --- a/pkg/loop/config_test.go +++ b/pkg/loop/config_test.go @@ -83,6 +83,7 @@ func TestEnvConfig_parse(t *testing.T) { envTelemetryEmitterExportMaxBatchSize: "100", envTelemetryEmitterMaxQueueSize: "1000", envTelemetryLogStreamingEnabled: "false", + envTelemetryMetricViewsAttributeDenylist: "event_id", envTelemetryPrometheusBridgeEnabled: "true", envTelemetryPrometheusBridgePrefixes: "foo,bar", envMeterRecordsEnabled: "true", @@ -214,8 +215,9 @@ var envCfgFull = EnvConfig{ TelemetryEmitterExportMaxBatchSize: 100, TelemetryEmitterMaxQueueSize: 1000, TelemetryLogStreamingEnabled: false, - TelemetryMetricCardinalityLimit: new(100000), - TelemetryPrometheusBridgeEnabled: true, + TelemetryMetricCardinalityLimit: new(100000), + TelemetryMetricViewsAttributeDenylist: []string{"event_id"}, + TelemetryPrometheusBridgeEnabled: true, TelemetryPrometheusBridgePrefixes: []string{"foo", "bar"}, MeterRecordsEnabled: true, MeterSnapshotsEnabled: false, @@ -289,6 +291,7 @@ func TestEnvConfig_AsCmdEnv(t *testing.T) { assert.Equal(t, "1000", got[envTelemetryEmitterMaxQueueSize]) assert.Equal(t, "false", got[envTelemetryLogStreamingEnabled]) assert.Equal(t, "100000", got[envTelemetryMetricCardinalityLimit]) + assert.Equal(t, "event_id", got[envTelemetryMetricViewsAttributeDenylist]) assert.Equal(t, "true", got[envTelemetryPrometheusBridgeEnabled]) assert.Equal(t, "foo,bar", got[envTelemetryPrometheusBridgePrefixes]) assert.Equal(t, "true", got[envMeterRecordsEnabled]) @@ -361,6 +364,43 @@ func TestEnvConfig_MetricCardinalityLimit_RoundTrip(t *testing.T) { }) } +// TestEnvConfig_MetricViewsAttributeDenylist_RoundTrip verifies that a +// comma-separated denylist survives an AsCmdEnv -> parse round trip. +func TestEnvConfig_MetricViewsAttributeDenylist_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("denylist propagates", func(t *testing.T) { + cfg := envCfgFull + cfg.TelemetryMetricViewsAttributeDenylist = []string{"event_id", "workflow_execution_id"} + setEnvFromCmdEnv(t, cfg.AsCmdEnv()) + + var parsed EnvConfig + require.NoError(t, parsed.parse()) + assert.Equal(t, []string{"event_id", "workflow_execution_id"}, parsed.TelemetryMetricViewsAttributeDenylist) + }) + + t.Run("unset leaves denylist nil", func(t *testing.T) { + cfg := envCfgFull + cfg.TelemetryMetricViewsAttributeDenylist = nil + env := cfg.AsCmdEnv() + for _, kv := range env { + require.NotContains(t, kv, envTelemetryMetricViewsAttributeDenylist+"=", "unset denylist must not be emitted") + } + setEnvFromCmdEnv(t, env) + + var parsed EnvConfig + require.NoError(t, parsed.parse()) + assert.Nil(t, parsed.TelemetryMetricViewsAttributeDenylist) + }) +} + 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 b2d495eeaf..8ddfd3f1d1 100644 --- a/pkg/loop/server.go +++ b/pkg/loop/server.go @@ -192,6 +192,7 @@ func (s *Server) start(opts ...ServerOpt) error { ChipIngressLogger: s.Logger, MetricCompressor: s.EnvConfig.TelemetryMetricCompressor, MetricCardinalityLimit: *s.EnvConfig.TelemetryMetricCardinalityLimit, + MetricViewsAttributeDenylist: s.EnvConfig.TelemetryMetricViewsAttributeDenylist, } if s.EnvConfig.TelemetryPrometheusBridgeEnabled {