diff --git a/pkg/beholder/client.go b/pkg/beholder/client.go index cf66c39306..829c27fff3 100644 --- a/pkg/beholder/client.go +++ b/pkg/beholder/client.go @@ -61,6 +61,9 @@ type Client struct { // lazySigner allows updating the keystore after client initialization. lazySigner *lazySigner + // fullResource retains all resource attributes for the info gauge export. + fullResource *sdkresource.Resource + // OnClose OnClose func() error } @@ -111,10 +114,12 @@ type otlploggrpcFactory func(options ...otlploggrpc.Option) (sdklog.Exporter, er // NewGRPCClient creates a GRPC based beholder Client. Use NewClient to create a client from a Config which will pick // the best client type from the Config. func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (*Client, error) { - baseResource, err := newOtelResource(cfg) + resources, err := buildOtelResources(cfg) if err != nil { return nil, err } + fullResource := resources.Full + metricResource := resources.Metric creds := insecure.NewCredentials() if !cfg.InsecureConnection && cfg.CACertFile != "" { creds, err = credentials.NewClientTLSFromFile(cfg.CACertFile, "") @@ -133,14 +138,14 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (*Client, erro } // Tracer - tracerProvider, err := newTracerProvider(cfg, baseResource, auth, creds) + tracerProvider, err := newTracerProvider(cfg, fullResource, auth, creds) if err != nil { return nil, err } tracer := tracerProvider.Tracer(defaultPackageName) // Meter - meterProvider, err := newMeterProvider(cfg, baseResource, auth, creds) + meterProvider, err := newMeterProvider(cfg, metricResource, auth, creds) if err != nil { return nil, err } @@ -161,7 +166,7 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (*Client, erro if !cfg.LogStreamingEnabled { loggerProvider = BeholderNoopLoggerProvider() } else { - loggerOpts, err := newLoggerProviderOpts(cfg, baseResource, sharedLogExporter) + loggerOpts, err := newLoggerProviderOpts(cfg, fullResource, sharedLogExporter) if err != nil { return nil, err } @@ -170,7 +175,7 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (*Client, erro logger := loggerProvider.Logger(defaultPackageName) // Message emitter - messageLoggerOpts, err := newMessageLoggerProviderOpts(cfg, baseResource, sharedLogExporter) + messageLoggerOpts, err := newMessageLoggerProviderOpts(cfg, fullResource, sharedLogExporter) if err != nil { return nil, err } @@ -265,9 +270,15 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (*Client, erro MeterProvider: meterProvider, MessageLoggerProvider: messageLoggerProvider, lazySigner: signer, + fullResource: fullResource, OnClose: onClose, } c.initService(lggr, batchEmitterService) + if cfg.metricResourceFilteringEnabled() { + if err := c.RecordFullResourceAttributesMetric(context.Background()); err != nil { + return nil, err + } + } return c, nil } @@ -331,49 +342,6 @@ func (c *Client) IsSignerSet() bool { return c.lazySigner != nil && c.lazySigner.IsSet() } -func newOtelResource(cfg Config) (resource *sdkresource.Resource, err error) { - extraResources, err := sdkresource.New( - context.Background(), - sdkresource.WithOS(), - sdkresource.WithContainer(), - sdkresource.WithHost(), - ) - if err != nil { - return nil, err - } - resource, err = sdkresource.Merge( - sdkresource.Default(), - extraResources, - ) - if err != nil { - return nil, err - } - - // Add csa public key resource attribute - csaPublicKeyHex := "not-configured" - if len(cfg.AuthPublicKeyHex) > 0 { - csaPublicKeyHex = cfg.AuthPublicKeyHex - } - csaPublicKeyAttr := attribute.String("csa_public_key", csaPublicKeyHex) - resource, err = sdkresource.Merge( - sdkresource.NewSchemaless(csaPublicKeyAttr), - resource, - ) - if err != nil { - return nil, err - } - - // Add custom resource attributes - resource, err = sdkresource.Merge( - sdkresource.NewSchemaless(cfg.ResourceAttributes...), - resource, - ) - if err != nil { - return nil, err - } - return -} - // RecordConfig records the beholder config as a metric. func (c *Client) RecordConfigMetric(ctx context.Context) error { configGauge, configAttrs, err := createConfigMetric(c.Meter, c.Config) @@ -384,6 +352,22 @@ func (c *Client) RecordConfigMetric(ctx context.Context) error { return nil } +// RecordFullResourceAttributesMetric records all full-resource attributes on a +// single info gauge. The metric resource itself remains reduced when enabled. +func (c *Client) RecordFullResourceAttributesMetric(ctx context.Context) error { + resourceGauge, err := c.Meter.Int64Gauge( + "beholder.otel.resource_attributes.full", + otelmetric.WithDescription("Full OTel resource attributes before metric-resource filtering"), + otelmetric.WithUnit("{info}"), + ) + if err != nil { + return fmt.Errorf("failed to create full resource attributes metric: %w", err) + } + + resourceGauge.Record(ctx, 1, otelmetric.WithAttributes(resourceAttributesAsKV(c.fullResource)...)) + return nil +} + // createConfigMetric creates a configuration info metric with Beholder settings as attributes. func createConfigMetric(meter otelmetric.Meter, cfg Config) (otelmetric.Int64Gauge, []attribute.KeyValue, error) { configGauge, err := meter.Int64Gauge( @@ -437,6 +421,12 @@ func createConfigMetric(meter otelmetric.Meter, cfg Config) (otelmetric.Int64Gau "metric_reader_interval", cfg.MetricReaderInterval.String()), attribute.String( "metric_compressor", cfg.MetricCompressor), + attribute.Bool( + "reduced_metric_resource_attributes_enabled", + cfg.ReducedMetricResourceAttributesEnabled), + attribute.Bool( + "exclude_volatile_resource_attributes_from_metrics_enabled", + cfg.ExcludeVolatileResourceAttributesFromMetricsEnabled), } return configGauge, configAttrs, nil diff --git a/pkg/beholder/config.go b/pkg/beholder/config.go index 09ef907ed7..f9256d228d 100644 --- a/pkg/beholder/config.go +++ b/pkg/beholder/config.go @@ -19,6 +19,12 @@ type Config struct { // OTel Resource ResourceAttributes []attribute.KeyValue + // ReducedMetricResourceAttributesEnabled uses a slim OTel resource on the metric + // export path while traces and logs keep the full resource. + ReducedMetricResourceAttributesEnabled bool + // ExcludeVolatileResourceAttributesFromMetricsEnabled strips volatile resource + // attributes (process.pid, k8s.pod.*, etc.) from the metric resource only. + ExcludeVolatileResourceAttributesFromMetricsEnabled bool // Message Emitter EmitterExportTimeout time.Duration EmitterExportInterval time.Duration @@ -26,7 +32,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 @@ -127,7 +133,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 @@ -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, diff --git a/pkg/beholder/httpclient.go b/pkg/beholder/httpclient.go index 4bc3c78db6..5d5d1bd4f7 100644 --- a/pkg/beholder/httpclient.go +++ b/pkg/beholder/httpclient.go @@ -37,10 +37,12 @@ func newCertFromFile(certFile string) (*x509.CertPool, error) { // NewHTTPClient creates a HTTP based beholder Client. Use NewClient to create a client from a Config which will pick // the best client type from the Config. func NewHTTPClient(cfg Config, otlploghttpNew otlploghttpFactory) (*Client, error) { - baseResource, err := newOtelResource(cfg) + resources, err := buildOtelResources(cfg) if err != nil { return nil, err } + fullResource := resources.Full + metricResource := resources.Metric var tlsConfig *tls.Config if !cfg.InsecureConnection { tlsConfig = &tls.Config{ @@ -106,7 +108,7 @@ func NewHTTPClient(cfg Config, otlploghttpNew otlploghttpFactory) (*Client, erro } loggerResource, err := sdkresource.Merge( sdkresource.NewSchemaless(loggerAttributes...), - baseResource, + fullResource, ) if err != nil { return nil, err @@ -124,14 +126,14 @@ func NewHTTPClient(cfg Config, otlploghttpNew otlploghttpFactory) (*Client, erro logger := loggerProvider.Logger(defaultPackageName) // Tracer - tracerProvider, err := newHTTPTracerProvider(cfg, baseResource, tlsConfig) + tracerProvider, err := newHTTPTracerProvider(cfg, fullResource, tlsConfig) if err != nil { return nil, err } tracer := tracerProvider.Tracer(defaultPackageName) // Meter - meterProvider, err := newHTTPMeterProvider(cfg, baseResource, tlsConfig) + meterProvider, err := newHTTPMeterProvider(cfg, metricResource, tlsConfig) if err != nil { return nil, err } @@ -168,7 +170,7 @@ func NewHTTPClient(cfg Config, otlploghttpNew otlploghttpFactory) (*Client, erro } messageLoggerResource, err := sdkresource.Merge( sdkresource.NewSchemaless(messageAttributes...), - baseResource, + fullResource, ) if err != nil { return nil, err @@ -203,6 +205,7 @@ func NewHTTPClient(cfg Config, otlploghttpNew otlploghttpFactory) (*Client, erro MeterProvider: meterProvider, MessageLoggerProvider: messageLoggerProvider, lazySigner: nil, + fullResource: fullResource, OnClose: onClose, } lggr := cfg.ChipIngressLogger @@ -210,6 +213,11 @@ func NewHTTPClient(cfg Config, otlploghttpNew otlploghttpFactory) (*Client, erro lggr = pkglogger.Nop() } c.initService(lggr, nil) + if cfg.metricResourceFilteringEnabled() { + if err := c.RecordFullResourceAttributesMetric(context.Background()); err != nil { + return nil, err + } + } return c, nil } diff --git a/pkg/beholder/resource.go b/pkg/beholder/resource.go new file mode 100644 index 0000000000..672f6031e0 --- /dev/null +++ b/pkg/beholder/resource.go @@ -0,0 +1,191 @@ +package beholder + +import ( + "context" + "strings" + + "go.opentelemetry.io/otel/attribute" + sdkresource "go.opentelemetry.io/otel/sdk/resource" +) + +const ( + attrKeyCSAPublicKey = "csa_public_key" +) + +// ResourcePair holds full and metric-scoped OTel resources. +type ResourcePair struct { + Full *sdkresource.Resource + Metric *sdkresource.Resource +} + +var ( + reducedMetricResourceExactDenylist = map[string]struct{}{ + "service.sha": {}, + "service.shortversion": {}, + "package_name": {}, + } + + reducedMetricResourcePrefixDenylist = []string{ + "os.", + "host.", + "container.", + } + + volatileResourceAttributeDenylist = map[string]struct{}{ + "process.pid": {}, + "service.instance.id": {}, + "container.id": {}, + "host.id": {}, + "host.name": {}, + } + + volatileResourceAttributePrefixDenylist = []string{"k8s.pod."} +) + +func buildOtelResources(cfg Config) (ResourcePair, error) { + full, err := buildFullOtelResource(cfg) + if err != nil { + return ResourcePair{}, err + } + + if !cfg.metricResourceFilteringEnabled() { + return ResourcePair{Full: full, Metric: full}, nil + } + + metric, err := buildMetricOtelResource(cfg, full) + if err != nil { + return ResourcePair{}, err + } + + return ResourcePair{Full: full, Metric: metric}, nil +} + +func (cfg Config) metricResourceFilteringEnabled() bool { + return cfg.ReducedMetricResourceAttributesEnabled || + cfg.ExcludeVolatileResourceAttributesFromMetricsEnabled +} + +func buildFullOtelResource(cfg Config) (*sdkresource.Resource, error) { + extraResources, err := sdkresource.New( + context.Background(), + sdkresource.WithOS(), + sdkresource.WithContainer(), + sdkresource.WithHost(), + ) + if err != nil { + return nil, err + } + + resource, err := sdkresource.Merge( + sdkresource.Default(), + extraResources, + ) + if err != nil { + return nil, err + } + + resource, err = sdkresource.Merge( + sdkresource.NewSchemaless(identityResourceAttributes(cfg)...), + resource, + ) + if err != nil { + return nil, err + } + + resource, err = sdkresource.Merge( + sdkresource.NewSchemaless(cfg.ResourceAttributes...), + resource, + ) + if err != nil { + return nil, err + } + + return resource, nil +} + +func buildMetricOtelResource(cfg Config, full *sdkresource.Resource) (*sdkresource.Resource, error) { + attrs := resourceAttributesAsKV(full) + if cfg.ReducedMetricResourceAttributesEnabled { + attrs = filterResourceAttributes(attrs, reducedResourceAttributeFilter()) + } + + if cfg.ExcludeVolatileResourceAttributesFromMetricsEnabled { + attrs = filterResourceAttributes(attrs, volatileResourceAttributeFilter()) + } + + return sdkresource.NewSchemaless(attrs...), nil +} + +func identityResourceAttributes(cfg Config) []attribute.KeyValue { + csaPublicKeyHex := "not-configured" + if len(cfg.AuthPublicKeyHex) > 0 { + csaPublicKeyHex = cfg.AuthPublicKeyHex + } + + attrs := []attribute.KeyValue{ + attribute.String(attrKeyCSAPublicKey, csaPublicKeyHex), + } + + return attrs +} + +type resourceAttributeFilter func(key string) bool + +func reducedResourceAttributeFilter() resourceAttributeFilter { + return func(key string) bool { + if _, denied := reducedMetricResourceExactDenylist[key]; denied { + return true + } + for _, prefix := range reducedMetricResourcePrefixDenylist { + if strings.HasPrefix(key, prefix) { + return true + } + } + return false + } +} + +func volatileResourceAttributeFilter() resourceAttributeFilter { + return func(key string) bool { + if _, denied := volatileResourceAttributeDenylist[key]; denied { + return true + } + for _, prefix := range volatileResourceAttributePrefixDenylist { + if strings.HasPrefix(key, prefix) { + return true + } + } + return false + } +} + +func filterResourceAttributes( + attrs []attribute.KeyValue, + deny resourceAttributeFilter, +) []attribute.KeyValue { + if deny == nil { + return attrs + } + + filtered := make([]attribute.KeyValue, 0, len(attrs)) + for _, attr := range attrs { + if deny(string(attr.Key)) { + continue + } + filtered = append(filtered, attr) + } + return filtered +} + +func resourceAttributesAsKV(res *sdkresource.Resource) []attribute.KeyValue { + if res == nil { + return nil + } + + var attrs []attribute.KeyValue + iter := res.Iter() + for iter.Next() { + attrs = append(attrs, iter.Attribute()) + } + return attrs +} diff --git a/pkg/beholder/resource_test.go b/pkg/beholder/resource_test.go new file mode 100644 index 0000000000..b7fdd2422e --- /dev/null +++ b/pkg/beholder/resource_test.go @@ -0,0 +1,207 @@ +package beholder + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/attribute" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" +) + +func TestBuildOtelResources(t *testing.T) { + baseConfig := func() Config { + return Config{ + AuthPublicKeyHex: "csa-key", + ResourceAttributes: []attribute.KeyValue{ + attribute.String("service.name", "chainlink"), + attribute.String("node_id", "node-1"), + attribute.String("service.version", "1.2.3"), + attribute.String("service.sha", "abcdef"), + attribute.String("service.shortversion", "1.2.3@abcdef"), + attribute.String("package_name", "beholder"), + attribute.Int("process.pid", 1234), + attribute.String("service.instance.id", "instance-1"), + attribute.String("k8s.pod.owner", "deployment-1"), + attribute.String("custom.attribute", "keep"), + }, + } + } + + cases := []struct { + name string + reduced, volatile bool + wantMetricKeys []string + wantAbsentMetricKeys []string + }{ + { + name: "default is backwards compatible", + wantMetricKeys: []string{ + "service.sha", "package_name", "process.pid", attrKeyCSAPublicKey, "node_id", + }, + }, + { + name: "reduced removes detector and redundant attributes", + reduced: true, + wantMetricKeys: []string{ + "service.name", "service.version", "custom.attribute", "process.pid", "service.instance.id", + attrKeyCSAPublicKey, "node_id", + }, + wantAbsentMetricKeys: []string{ + "service.sha", "service.shortversion", "package_name", + }, + }, + { + name: "volatile removes process scoped attributes", + volatile: true, + wantMetricKeys: []string{ + "service.sha", "package_name", attrKeyCSAPublicKey, "node_id", + }, + wantAbsentMetricKeys: []string{ + "process.pid", "service.instance.id", "k8s.pod.owner", + }, + }, + { + name: "reduced and volatile combine filtering", + reduced: true, + volatile: true, + wantMetricKeys: []string{ + "service.name", "service.version", "custom.attribute", attrKeyCSAPublicKey, "node_id", + }, + wantAbsentMetricKeys: []string{ + "service.sha", "service.shortversion", "package_name", "process.pid", "service.instance.id", "k8s.pod.owner", + }, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + cfg := baseConfig() + cfg.ReducedMetricResourceAttributesEnabled = tc.reduced + cfg.ExcludeVolatileResourceAttributesFromMetricsEnabled = tc.volatile + + resources, err := buildOtelResources(cfg) + require.NoError(t, err) + + fullAttrs := resourceAttributeMap(resources.Full) + metricAttrs := resourceAttributeMap(resources.Metric) + assert.Contains(t, fullAttrs, "process.pid") + assert.Contains(t, fullAttrs, "service.instance.id") + for _, key := range tc.wantMetricKeys { + assert.Contains(t, metricAttrs, key) + } + for _, key := range tc.wantAbsentMetricKeys { + assert.NotContains(t, metricAttrs, key) + } + }) + } +} + +func TestReducedMetricResourcePreservesFullAttributesAndIdentity(t *testing.T) { + cfg := Config{ + ReducedMetricResourceAttributesEnabled: true, + AuthPublicKeyHex: "authoritative-csa-key", + ResourceAttributes: []attribute.KeyValue{ + attribute.String(attrKeyCSAPublicKey, "caller-csa-key"), + attribute.String("node_id", "configured-node-id"), + attribute.String("k8s.pod.extra", "pod-attribute"), + }, + } + resources, err := buildOtelResources(cfg) + require.NoError(t, err) + + fullAttrs := resourceAttributeMap(resources.Full) + metricAttrs := resourceAttributeMap(resources.Metric) + assert.Equal(t, "authoritative-csa-key", fullAttrs[attrKeyCSAPublicKey].AsString()) + assert.Equal(t, "authoritative-csa-key", metricAttrs[attrKeyCSAPublicKey].AsString()) + assert.Equal(t, "configured-node-id", fullAttrs["node_id"].AsString()) + assert.Equal(t, "configured-node-id", metricAttrs["node_id"].AsString()) + assert.Equal(t, fullAttrs["service.name"], metricAttrs["service.name"]) + assert.NotEmpty(t, metricAttrs["service.name"].AsString()) + assert.Contains(t, metricAttrs, "k8s.pod.extra") + + cfg.ExcludeVolatileResourceAttributesFromMetricsEnabled = true + resources, err = buildOtelResources(cfg) + require.NoError(t, err) + assert.NotContains(t, resourceAttributeMap(resources.Metric), "k8s.pod.extra") +} + +func TestMetricResourceFilteringEnabled(t *testing.T) { + for _, tc := range []struct { + name string + cfg Config + want bool + }{ + {name: "default", want: false}, + { + name: "reduced resource enabled", + cfg: Config{ReducedMetricResourceAttributesEnabled: true}, + want: true, + }, + { + name: "volatile attributes excluded", + cfg: Config{ExcludeVolatileResourceAttributesFromMetricsEnabled: true}, + want: true, + }, + } { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.want, tc.cfg.metricResourceFilteringEnabled()) + }) + } +} + +func TestRecordFullResourceAttributesMetric(t *testing.T) { + cfg := Config{ + ReducedMetricResourceAttributesEnabled: true, + AuthPublicKeyHex: "csa-key", + ResourceAttributes: []attribute.KeyValue{ + attribute.String("service.name", "chainlink"), + attribute.String("node_id", "node-1"), + attribute.Int("process.pid", 1234), + }, + } + resources, err := buildOtelResources(cfg) + require.NoError(t, err) + + reader := sdkmetric.NewManualReader() + provider := sdkmetric.NewMeterProvider( + sdkmetric.WithReader(reader), + sdkmetric.WithResource(resources.Metric), + ) + t.Cleanup(func() { require.NoError(t, provider.Shutdown(context.Background())) }) + + client := Client{ + Meter: provider.Meter(defaultPackageName), + fullResource: resources.Full, + } + require.NoError(t, client.RecordFullResourceAttributesMetric(context.Background())) + + var metrics metricdata.ResourceMetrics + require.NoError(t, reader.Collect(context.Background(), &metrics)) + + for _, scopeMetrics := range metrics.ScopeMetrics { + for _, metric := range scopeMetrics.Metrics { + if metric.Name != "beholder.otel.resource_attributes.full" { + continue + } + gauge, ok := metric.Data.(metricdata.Gauge[int64]) + require.True(t, ok) + require.Len(t, gauge.DataPoints, 1) + assert.Equal(t, int64(1), gauge.DataPoints[0].Value) + _, hasPID := gauge.DataPoints[0].Attributes.Value("process.pid") + assert.True(t, hasPID) + return + } + } + t.Fatal("full resource attributes metric not found") +} + +func resourceAttributeMap(resource interface{ Attributes() []attribute.KeyValue }) map[string]attribute.Value { + attrs := make(map[string]attribute.Value) + for _, attr := range resource.Attributes() { + attrs[string(attr.Key)] = attr.Value + } + return attrs +} diff --git a/pkg/beholder/testdata/config-example.json b/pkg/beholder/testdata/config-example.json index c8aacc6baf..b4757d5903 100644 --- a/pkg/beholder/testdata/config-example.json +++ b/pkg/beholder/testdata/config-example.json @@ -19,6 +19,8 @@ } } ], + "ReducedMetricResourceAttributesEnabled": false, + "ExcludeVolatileResourceAttributesFromMetricsEnabled": false, "EmitterExportTimeout": 1000000000, "EmitterExportInterval": 1000000000, "EmitterExportMaxBatchSize": 512, diff --git a/pkg/loop/config.go b/pkg/loop/config.go index 58d956bbfc..4f8de14057 100644 --- a/pkg/loop/config.go +++ b/pkg/loop/config.go @@ -59,40 +59,42 @@ const ( envTracingAttribute = "CL_TRACING_ATTRIBUTE_" envTracingTLSCertPath = "CL_TRACING_TLS_CERT_PATH" - envTelemetryEnabled = "CL_TELEMETRY_ENABLED" - envTelemetryEndpoint = "CL_TELEMETRY_ENDPOINT" - envTelemetryInsecureConn = "CL_TELEMETRY_INSECURE_CONNECTION" - envTelemetryCACertFile = "CL_TELEMETRY_CA_CERT_FILE" - envTelemetryAttribute = "CL_TELEMETRY_ATTRIBUTE_" - envTelemetryTraceSampleRatio = "CL_TELEMETRY_TRACE_SAMPLE_RATIO" - envTelemetryAuthHeader = "CL_TELEMETRY_AUTH_HEADER" - envTelemetryAuthPubKeyHex = "CL_TELEMETRY_AUTH_PUB_KEY_HEX" - envTelemetryAuthHeadersTTL = "CL_TELEMETRY_AUTH_HEADERS_TTL" - envTelemetryEmitterBatchProcessor = "CL_TELEMETRY_EMITTER_BATCH_PROCESSOR" - envTelemetryEmitterExportTimeout = "CL_TELEMETRY_EMITTER_EXPORT_TIMEOUT" - envTelemetryEmitterExportInterval = "CL_TELEMETRY_EMITTER_EXPORT_INTERVAL" - envTelemetryEmitterExportMaxBatchSize = "CL_TELEMETRY_EMITTER_EXPORT_MAX_BATCH_SIZE" - envTelemetryEmitterMaxQueueSize = "CL_TELEMETRY_EMITTER_MAX_QUEUE_SIZE" - envTelemetryLogStreamingEnabled = "CL_TELEMETRY_LOG_STREAMING_ENABLED" - envTelemetryLogLevel = "CL_TELEMETRY_LOG_LEVEL" - envTelemetryLogBatchProcessor = "CL_TELEMETRY_LOG_BATCH_PROCESSOR" - envTelemetryLogExportTimeout = "CL_TELEMETRY_LOG_EXPORT_TIMEOUT" - envTelemetryLogExportMaxBatchSize = "CL_TELEMETRY_LOG_EXPORT_MAX_BATCH_SIZE" - envTelemetryLogExportInterval = "CL_TELEMETRY_LOG_EXPORT_INTERVAL" - envTelemetryLogMaxQueueSize = "CL_TELEMETRY_LOG_MAX_QUEUE_SIZE" - envTelemetryTraceCompressor = "CL_TELEMETRY_TRACE_COMPRESSOR" - envTelemetryMetricCompressor = "CL_TELEMETRY_METRIC_COMPRESSOR" - envTelemetryPrometheusBridgeEnabled = "CL_TELEMETRY_PROMETHEUS_BRIDGE_ENABLED" - envTelemetryPrometheusBridgePrefixes = "CL_TELEMETRY_PROMETHEUS_BRIDGE_PREFIXES" - envTelemetryLogCompressor = "CL_TELEMETRY_LOG_COMPRESSOR" - envMeterRecordsEnabled = "CL_METER_RECORDS_ENABLED" - envMeterSnapshotsEnabled = "CL_METER_SNAPSHOTS_ENABLED" - envMeterProduct = "CL_METER_PRODUCT" - envMeterTenant = "CL_METER_TENANT" - envMeterNumericTenantID = "CL_METER_NUMERIC_TENANT_ID" - envMeterEnvironment = "CL_METER_ENVIRONMENT" - envMeterZone = "CL_METER_ZONE" - envMeterNodeID = "CL_METER_NODE_ID" + envTelemetryEnabled = "CL_TELEMETRY_ENABLED" + envTelemetryEndpoint = "CL_TELEMETRY_ENDPOINT" + envTelemetryInsecureConn = "CL_TELEMETRY_INSECURE_CONNECTION" + envTelemetryCACertFile = "CL_TELEMETRY_CA_CERT_FILE" + envTelemetryAttribute = "CL_TELEMETRY_ATTRIBUTE_" + envTelemetryTraceSampleRatio = "CL_TELEMETRY_TRACE_SAMPLE_RATIO" + envTelemetryAuthHeader = "CL_TELEMETRY_AUTH_HEADER" + envTelemetryAuthPubKeyHex = "CL_TELEMETRY_AUTH_PUB_KEY_HEX" + envTelemetryAuthHeadersTTL = "CL_TELEMETRY_AUTH_HEADERS_TTL" + envTelemetryEmitterBatchProcessor = "CL_TELEMETRY_EMITTER_BATCH_PROCESSOR" + envTelemetryEmitterExportTimeout = "CL_TELEMETRY_EMITTER_EXPORT_TIMEOUT" + envTelemetryEmitterExportInterval = "CL_TELEMETRY_EMITTER_EXPORT_INTERVAL" + envTelemetryEmitterExportMaxBatchSize = "CL_TELEMETRY_EMITTER_EXPORT_MAX_BATCH_SIZE" + envTelemetryEmitterMaxQueueSize = "CL_TELEMETRY_EMITTER_MAX_QUEUE_SIZE" + envTelemetryLogStreamingEnabled = "CL_TELEMETRY_LOG_STREAMING_ENABLED" + envTelemetryLogLevel = "CL_TELEMETRY_LOG_LEVEL" + envTelemetryLogBatchProcessor = "CL_TELEMETRY_LOG_BATCH_PROCESSOR" + envTelemetryLogExportTimeout = "CL_TELEMETRY_LOG_EXPORT_TIMEOUT" + envTelemetryLogExportMaxBatchSize = "CL_TELEMETRY_LOG_EXPORT_MAX_BATCH_SIZE" + envTelemetryLogExportInterval = "CL_TELEMETRY_LOG_EXPORT_INTERVAL" + envTelemetryLogMaxQueueSize = "CL_TELEMETRY_LOG_MAX_QUEUE_SIZE" + envTelemetryTraceCompressor = "CL_TELEMETRY_TRACE_COMPRESSOR" + envTelemetryMetricCompressor = "CL_TELEMETRY_METRIC_COMPRESSOR" + envTelemetryReducedMetricResourceAttributes = "CL_TELEMETRY_REDUCED_METRIC_RESOURCE_ATTRIBUTES_ENABLED" + envTelemetryExcludeVolatileResourceAttributes = "CL_TELEMETRY_EXCLUDE_VOLATILE_RESOURCE_ATTRIBUTES_FROM_METRICS_ENABLED" + envTelemetryPrometheusBridgeEnabled = "CL_TELEMETRY_PROMETHEUS_BRIDGE_ENABLED" + envTelemetryPrometheusBridgePrefixes = "CL_TELEMETRY_PROMETHEUS_BRIDGE_PREFIXES" + envTelemetryLogCompressor = "CL_TELEMETRY_LOG_COMPRESSOR" + envMeterRecordsEnabled = "CL_METER_RECORDS_ENABLED" + envMeterSnapshotsEnabled = "CL_METER_SNAPSHOTS_ENABLED" + envMeterProduct = "CL_METER_PRODUCT" + envMeterTenant = "CL_METER_TENANT" + envMeterNumericTenantID = "CL_METER_NUMERIC_TENANT_ID" + envMeterEnvironment = "CL_METER_ENVIRONMENT" + envMeterZone = "CL_METER_ZONE" + envMeterNodeID = "CL_METER_NODE_ID" envChipIngressEndpoint = "CL_CHIP_INGRESS_ENDPOINT" envChipIngressInsecureConnection = "CL_CHIP_INGRESS_INSECURE_CONNECTION" @@ -162,25 +164,27 @@ type EnvConfig struct { TelemetryAuthHeadersTTL time.Duration // TelemetryEmitterBatchProcessor maps to beholder Config.EmitterBatchProcessor // (batched async custom-message export vs immediate per-record export). - TelemetryEmitterBatchProcessor bool - TelemetryEmitterExportTimeout time.Duration - TelemetryEmitterExportInterval time.Duration - TelemetryEmitterExportMaxBatchSize int - TelemetryEmitterMaxQueueSize int - TelemetryLogStreamingEnabled bool - TelemetryLogLevel zapcore.Level - TelemetryLogBatchProcessor bool - TelemetryLogExportTimeout time.Duration - TelemetryLogExportMaxBatchSize int - TelemetryLogExportInterval time.Duration - TelemetryLogMaxQueueSize int - TelemetryTraceCompressor string - TelemetryMetricCompressor string - TelemetryPrometheusBridgeEnabled bool - TelemetryPrometheusBridgePrefixes []string - TelemetryLogCompressor string - MeterRecordsEnabled bool - MeterSnapshotsEnabled bool + TelemetryEmitterBatchProcessor bool + TelemetryEmitterExportTimeout time.Duration + TelemetryEmitterExportInterval time.Duration + TelemetryEmitterExportMaxBatchSize int + TelemetryEmitterMaxQueueSize int + TelemetryLogStreamingEnabled bool + TelemetryLogLevel zapcore.Level + TelemetryLogBatchProcessor bool + TelemetryLogExportTimeout time.Duration + TelemetryLogExportMaxBatchSize int + TelemetryLogExportInterval time.Duration + TelemetryLogMaxQueueSize int + TelemetryTraceCompressor string + TelemetryMetricCompressor string + TelemetryReducedMetricResourceAttributesEnabled bool + TelemetryExcludeVolatileResourceAttributesFromMetricsEnabled bool + TelemetryPrometheusBridgeEnabled bool + TelemetryPrometheusBridgePrefixes []string + TelemetryLogCompressor string + MeterRecordsEnabled bool + MeterSnapshotsEnabled bool // MeterProduct / MeterTenant / MeterNumericTenantID / MeterEnvironment / // MeterZone / MeterNodeID are @@ -290,6 +294,8 @@ func (e *EnvConfig) AsCmdEnv() (env []string) { add(envTelemetryLogMaxQueueSize, strconv.Itoa(e.TelemetryLogMaxQueueSize)) add(envTelemetryTraceCompressor, e.TelemetryTraceCompressor) add(envTelemetryMetricCompressor, e.TelemetryMetricCompressor) + add(envTelemetryReducedMetricResourceAttributes, strconv.FormatBool(e.TelemetryReducedMetricResourceAttributesEnabled)) + add(envTelemetryExcludeVolatileResourceAttributes, strconv.FormatBool(e.TelemetryExcludeVolatileResourceAttributesFromMetricsEnabled)) add(envTelemetryPrometheusBridgeEnabled, strconv.FormatBool(e.TelemetryPrometheusBridgeEnabled)) add(envTelemetryPrometheusBridgePrefixes, strings.Join(e.TelemetryPrometheusBridgePrefixes, ",")) add(envTelemetryLogCompressor, e.TelemetryLogCompressor) @@ -530,6 +536,14 @@ func (e *EnvConfig) parse() error { } e.TelemetryTraceCompressor = os.Getenv(envTelemetryTraceCompressor) e.TelemetryMetricCompressor = os.Getenv(envTelemetryMetricCompressor) + e.TelemetryReducedMetricResourceAttributesEnabled, err = getBool(envTelemetryReducedMetricResourceAttributes) + if err != nil { + return fmt.Errorf("failed to parse %s: %w", envTelemetryReducedMetricResourceAttributes, err) + } + e.TelemetryExcludeVolatileResourceAttributesFromMetricsEnabled, err = getBool(envTelemetryExcludeVolatileResourceAttributes) + if err != nil { + return fmt.Errorf("failed to parse %s: %w", envTelemetryExcludeVolatileResourceAttributes, err) + } 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..dfdca27214 100644 --- a/pkg/loop/config_test.go +++ b/pkg/loop/config_test.go @@ -68,31 +68,33 @@ func TestEnvConfig_parse(t *testing.T) { envTracingTLSCertPath: "internal/test/fixtures/client.pem", envTracingAttribute + "XYZ": "value", - envTelemetryEnabled: "true", - envTelemetryEndpoint: "example.com/beholder", - envTelemetryInsecureConn: "true", - envTelemetryCACertFile: "foo/bar", - envTelemetryAttribute + "foo": "bar", - envTelemetryAttribute + "baz": "42", - envTelemetryTraceSampleRatio: "0.42", - envTelemetryAuthHeader + "header-key": "header-value", - envTelemetryAuthPubKeyHex: "pub-key-hex", - envTelemetryEmitterBatchProcessor: "true", - envTelemetryEmitterExportTimeout: "1s", - envTelemetryEmitterExportInterval: "2s", - envTelemetryEmitterExportMaxBatchSize: "100", - envTelemetryEmitterMaxQueueSize: "1000", - envTelemetryLogStreamingEnabled: "false", - envTelemetryPrometheusBridgeEnabled: "true", - envTelemetryPrometheusBridgePrefixes: "foo,bar", - envMeterRecordsEnabled: "true", - envMeterSnapshotsEnabled: "false", - envMeterProduct: "cre-mainline", - envMeterTenant: "mainline", - envMeterNumericTenantID: "42", - envMeterEnvironment: "production", - envMeterZone: "wf-zone-a", - envMeterNodeID: "csa-pubkey-1", + envTelemetryEnabled: "true", + envTelemetryEndpoint: "example.com/beholder", + envTelemetryInsecureConn: "true", + envTelemetryCACertFile: "foo/bar", + envTelemetryAttribute + "foo": "bar", + envTelemetryAttribute + "baz": "42", + envTelemetryTraceSampleRatio: "0.42", + envTelemetryAuthHeader + "header-key": "header-value", + envTelemetryAuthPubKeyHex: "pub-key-hex", + envTelemetryEmitterBatchProcessor: "true", + envTelemetryEmitterExportTimeout: "1s", + envTelemetryEmitterExportInterval: "2s", + envTelemetryEmitterExportMaxBatchSize: "100", + envTelemetryEmitterMaxQueueSize: "1000", + envTelemetryLogStreamingEnabled: "false", + envTelemetryReducedMetricResourceAttributes: "true", + envTelemetryExcludeVolatileResourceAttributes: "true", + envTelemetryPrometheusBridgeEnabled: "true", + envTelemetryPrometheusBridgePrefixes: "foo,bar", + envMeterRecordsEnabled: "true", + envMeterSnapshotsEnabled: "false", + envMeterProduct: "cre-mainline", + envMeterTenant: "mainline", + envMeterNumericTenantID: "42", + envMeterEnvironment: "production", + envMeterZone: "wf-zone-a", + envMeterNodeID: "csa-pubkey-1", envChipIngressEndpoint: "chip-ingress.example.com:50051", envChipIngressInsecureConnection: "true", @@ -191,30 +193,32 @@ var envCfgFull = EnvConfig{ TracingSamplingRatio: 1.0, TracingTLSCertPath: "internal/test/fixtures/client.pem", - TelemetryEnabled: true, - TelemetryEndpoint: "example.com/beholder", - TelemetryInsecureConnection: true, - TelemetryCACertFile: "foo/bar", - TelemetryAttributes: OtelAttributes{"foo": "bar", "baz": "42"}, - TelemetryTraceSampleRatio: 0.42, - TelemetryAuthHeaders: map[string]string{"header-key": "header-value"}, - TelemetryAuthPubKeyHex: "pub-key-hex", - TelemetryEmitterBatchProcessor: true, - TelemetryEmitterExportTimeout: 1 * time.Second, - TelemetryEmitterExportInterval: 2 * time.Second, - TelemetryEmitterExportMaxBatchSize: 100, - TelemetryEmitterMaxQueueSize: 1000, - TelemetryLogStreamingEnabled: false, - TelemetryPrometheusBridgeEnabled: true, - TelemetryPrometheusBridgePrefixes: []string{"foo", "bar"}, - MeterRecordsEnabled: true, - MeterSnapshotsEnabled: false, - MeterProduct: "cre-mainline", - MeterTenant: "mainline", - MeterNumericTenantID: "42", - MeterEnvironment: "production", - MeterZone: "wf-zone-a", - MeterNodeID: "csa-pubkey-1", + TelemetryEnabled: true, + TelemetryEndpoint: "example.com/beholder", + TelemetryInsecureConnection: true, + TelemetryCACertFile: "foo/bar", + TelemetryAttributes: OtelAttributes{"foo": "bar", "baz": "42"}, + TelemetryTraceSampleRatio: 0.42, + TelemetryAuthHeaders: map[string]string{"header-key": "header-value"}, + TelemetryAuthPubKeyHex: "pub-key-hex", + TelemetryEmitterBatchProcessor: true, + TelemetryEmitterExportTimeout: 1 * time.Second, + TelemetryEmitterExportInterval: 2 * time.Second, + TelemetryEmitterExportMaxBatchSize: 100, + TelemetryEmitterMaxQueueSize: 1000, + TelemetryLogStreamingEnabled: false, + TelemetryReducedMetricResourceAttributesEnabled: true, + TelemetryExcludeVolatileResourceAttributesFromMetricsEnabled: true, + TelemetryPrometheusBridgeEnabled: true, + TelemetryPrometheusBridgePrefixes: []string{"foo", "bar"}, + MeterRecordsEnabled: true, + MeterSnapshotsEnabled: false, + MeterProduct: "cre-mainline", + MeterTenant: "mainline", + MeterNumericTenantID: "42", + MeterEnvironment: "production", + MeterZone: "wf-zone-a", + MeterNodeID: "csa-pubkey-1", ChipIngressEndpoint: "chip-ingress.example.com:50051", ChipIngressInsecureConnection: true, @@ -278,6 +282,8 @@ 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, "true", got[envTelemetryReducedMetricResourceAttributes]) + assert.Equal(t, "true", got[envTelemetryExcludeVolatileResourceAttributes]) 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..c095c59634 100644 --- a/pkg/loop/server.go +++ b/pkg/loop/server.go @@ -166,31 +166,33 @@ func (s *Server) start(opts ...ServerOpt) error { attributes = tracingConfig.Attributes() } beholderCfg := beholder.Config{ - InsecureConnection: s.EnvConfig.TelemetryInsecureConnection, - CACertFile: s.EnvConfig.TelemetryCACertFile, - OtelExporterGRPCEndpoint: s.EnvConfig.TelemetryEndpoint, - ResourceAttributes: append(attributes, s.EnvConfig.TelemetryAttributes.AsStringAttributes()...), - TraceSampleRatio: s.EnvConfig.TelemetryTraceSampleRatio, - TraceCompressor: s.EnvConfig.TelemetryTraceCompressor, - EmitterBatchProcessor: s.EnvConfig.TelemetryEmitterBatchProcessor, - EmitterExportTimeout: s.EnvConfig.TelemetryEmitterExportTimeout, - EmitterExportInterval: s.EnvConfig.TelemetryEmitterExportInterval, - EmitterExportMaxBatchSize: s.EnvConfig.TelemetryEmitterExportMaxBatchSize, - EmitterMaxQueueSize: s.EnvConfig.TelemetryEmitterMaxQueueSize, - LogStreamingEnabled: s.EnvConfig.TelemetryLogStreamingEnabled, - LogLevel: s.EnvConfig.TelemetryLogLevel, - LogBatchProcessor: s.EnvConfig.TelemetryLogBatchProcessor, - LogExportTimeout: s.EnvConfig.TelemetryLogExportTimeout, - LogExportMaxBatchSize: s.EnvConfig.TelemetryLogExportMaxBatchSize, - LogExportInterval: s.EnvConfig.TelemetryLogExportInterval, - LogMaxQueueSize: s.EnvConfig.TelemetryLogMaxQueueSize, - LogCompressor: s.EnvConfig.TelemetryLogCompressor, - ChipIngressEmitterEnabled: s.EnvConfig.ChipIngressEndpoint != "", - ChipIngressEmitterGRPCEndpoint: s.EnvConfig.ChipIngressEndpoint, - ChipIngressInsecureConnection: s.EnvConfig.ChipIngressInsecureConnection, - ChipIngressBatchEmitterEnabled: s.EnvConfig.ChipIngressBatchEmitterEnabled, - ChipIngressLogger: s.Logger, - MetricCompressor: s.EnvConfig.TelemetryMetricCompressor, + InsecureConnection: s.EnvConfig.TelemetryInsecureConnection, + CACertFile: s.EnvConfig.TelemetryCACertFile, + OtelExporterGRPCEndpoint: s.EnvConfig.TelemetryEndpoint, + ResourceAttributes: append(attributes, s.EnvConfig.TelemetryAttributes.AsStringAttributes()...), + TraceSampleRatio: s.EnvConfig.TelemetryTraceSampleRatio, + TraceCompressor: s.EnvConfig.TelemetryTraceCompressor, + EmitterBatchProcessor: s.EnvConfig.TelemetryEmitterBatchProcessor, + EmitterExportTimeout: s.EnvConfig.TelemetryEmitterExportTimeout, + EmitterExportInterval: s.EnvConfig.TelemetryEmitterExportInterval, + EmitterExportMaxBatchSize: s.EnvConfig.TelemetryEmitterExportMaxBatchSize, + EmitterMaxQueueSize: s.EnvConfig.TelemetryEmitterMaxQueueSize, + LogStreamingEnabled: s.EnvConfig.TelemetryLogStreamingEnabled, + LogLevel: s.EnvConfig.TelemetryLogLevel, + LogBatchProcessor: s.EnvConfig.TelemetryLogBatchProcessor, + LogExportTimeout: s.EnvConfig.TelemetryLogExportTimeout, + LogExportMaxBatchSize: s.EnvConfig.TelemetryLogExportMaxBatchSize, + LogExportInterval: s.EnvConfig.TelemetryLogExportInterval, + LogMaxQueueSize: s.EnvConfig.TelemetryLogMaxQueueSize, + LogCompressor: s.EnvConfig.TelemetryLogCompressor, + ChipIngressEmitterEnabled: s.EnvConfig.ChipIngressEndpoint != "", + ChipIngressEmitterGRPCEndpoint: s.EnvConfig.ChipIngressEndpoint, + ChipIngressInsecureConnection: s.EnvConfig.ChipIngressInsecureConnection, + ChipIngressBatchEmitterEnabled: s.EnvConfig.ChipIngressBatchEmitterEnabled, + ChipIngressLogger: s.Logger, + MetricCompressor: s.EnvConfig.TelemetryMetricCompressor, + ReducedMetricResourceAttributesEnabled: s.EnvConfig.TelemetryReducedMetricResourceAttributesEnabled, + ExcludeVolatileResourceAttributesFromMetricsEnabled: s.EnvConfig.TelemetryExcludeVolatileResourceAttributesFromMetricsEnabled, } if s.EnvConfig.TelemetryPrometheusBridgeEnabled {