beholder: set compression config option explicitly to gzip

Author: pkcll

pkg/beholder/client.go

@@ -28,7 +28,7 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/chipingress"
 )
 
-const grpcCompressorGzip = "gzip"
+const defaultGRPCCompressor = "gzip"
 
 type Emitter interface {
 	// Sends message with bytes and attributes to OTel Collector
@@ -364,8 +364,12 @@ func newTracerProvider(config Config, resource *sdkresource.Resource, auth Auth,
 	// No auth
 	default:
 	}
-	if !config.TraceCompressionDisabled {
-		exporterOpts = append(exporterOpts, otlptracegrpc.WithCompressor(grpcCompressorGzip))
+	switch compressor := config.TraceCompressor; compressor {
+	case "disabled":
+	case "":
+		exporterOpts = append(exporterOpts, otlptracegrpc.WithCompressor(defaultGRPCCompressor))
+	default:
+		exporterOpts = append(exporterOpts, otlptracegrpc.WithCompressor(compressor))
 	}
 	if config.TraceRetryConfig != nil {
 		// NOTE: By default, the retry is enabled in the OTel SDK
@@ -406,8 +410,12 @@ func newMeterProvider(cfg Config, resource *sdkresource.Resource, auth Auth, cre
 		otlpmetricgrpc.WithTLSCredentials(creds),
 		otlpmetricgrpc.WithEndpoint(cfg.OtelExporterGRPCEndpoint),
 	}
-	if !cfg.MetricCompressionDisabled {
-		opts = append(opts, otlpmetricgrpc.WithCompressor(grpcCompressorGzip))
+	switch compressor := cfg.MetricCompressor; compressor {
+	case "disabled":
+	case "":
+		opts = append(opts, otlpmetricgrpc.WithCompressor(defaultGRPCCompressor))
+	default:
+		opts = append(opts, otlpmetricgrpc.WithCompressor(compressor))
 	}
 
 	switch {
@@ -459,8 +467,12 @@ func newLoggerOpts(cfg Config, auth Auth, creds credentials.TransportCredentials
 		otlploggrpc.WithEndpoint(cfg.OtelExporterGRPCEndpoint),
 		otlploggrpc.WithDialOption(grpc.WithStatsHandler(otelgrpc.NewClientHandler(otelOpts...))),
 	}
-	if !cfg.LogCompressionDisabled {
-		opts = append(opts, otlploggrpc.WithCompressor(grpcCompressorGzip))
+	switch compressor := cfg.LogCompressor; compressor {
+	case "disabled":
+	case "":
+		opts = append(opts, otlploggrpc.WithCompressor(defaultGRPCCompressor))
+	default:
+		opts = append(opts, otlploggrpc.WithCompressor(compressor))
 	}
 	// Log exporter auth
 	switch {

pkg/beholder/config.go

@@ -29,15 +29,15 @@ type Config struct {
 	TraceBatchTimeout time.Duration
 	TraceSpanExporter trace.SpanExporter // optional additional exporter
 	TraceRetryConfig  *RetryConfig
-	// TraceCompressionDisabled disables gRPC compression for traces (compression uses "gzip" by default).
-	TraceCompressionDisabled bool
+	// TraceCompressor sets the gRPC compressor for traces. Valid values: "gzip" (default), "disabled".
+	TraceCompressor string
 
 	// OTel Metric
 	MetricReaderInterval time.Duration
 	MetricRetryConfig    *RetryConfig
 	MetricViews          []metric.View
-	// MetricCompressionDisabled disables gRPC compression for metrics (compression uses "gzip" by default).
-	MetricCompressionDisabled bool
+	// MetricCompressor sets the gRPC compressor for metrics. Valid values: "gzip" (default), "disabled".
+	MetricCompressor string
 
 	// Custom Events via Chip Ingress Emitter
 	ChipIngressEmitterEnabled      bool
@@ -54,8 +54,8 @@ type Config struct {
 	LogRetryConfig      *RetryConfig
 	LogStreamingEnabled bool          // Enable logs streaming to the OTel log exporter
 	LogLevel            zapcore.Level // Log level for telemetry streaming
-	// LogCompressionDisabled disables gRPC compression for logs (compression uses "gzip" by default).
-	LogCompressionDisabled bool
+	// LogCompressor sets the gRPC compressor for logs. Valid values: "gzip" (default), "disabled".
+	LogCompressor string
 
 	// Auth
 	// AuthHeaders serves two purposes:
@@ -116,10 +116,12 @@ func DefaultConfig() Config {
 		// Trace
 		TraceSampleRatio:  1,
 		TraceBatchTimeout: 1 * time.Second,
+		TraceCompressor:   "gzip",
 		// OTel trace exporter retry config
 		TraceRetryConfig: defaultRetryConfig.Copy(),
 		// Metric
 		MetricReaderInterval: 1 * time.Second,
+		MetricCompressor:     "gzip",
 		// OTel metric exporter retry config
 		MetricRetryConfig: defaultRetryConfig.Copy(),
 		// Log
@@ -130,6 +132,7 @@ func DefaultConfig() Config {
 		LogBatchProcessor:     true,
 		LogStreamingEnabled:   true, // Enable logs streaming by default
 		LogLevel:              zapcore.InfoLevel,
+		LogCompressor:         "gzip",
 		// Auth (defaults to static auth mode with TTL=0)
 		AuthHeadersTTL: 0,
 	}

pkg/beholder/config_test.go

@@ -36,10 +36,12 @@ func ExampleConfig() {
 		// Trace
 		TraceSampleRatio:  1,
 		TraceBatchTimeout: 1 * time.Second,
+		TraceCompressor:   "gzip",
 		// OTel trace exporter retry config
 		TraceRetryConfig: nil,
 		// Metric
 		MetricReaderInterval: 1 * time.Second,
+		MetricCompressor:     "gzip",
 		// OTel metric exporter retry config
 		MetricRetryConfig: nil,
 		// Log
@@ -50,6 +52,7 @@ func ExampleConfig() {
 		LogBatchProcessor:     true,
 		LogStreamingEnabled:   false,             // Disable streaming logs by default
 		LogLevel:              zapcore.InfoLevel, // Default log level
+		LogCompressor:         "gzip",
 		// Auth
 		AuthPublicKeyHex: "",
 		AuthHeaders:      map[string]string{},
@@ -64,6 +67,6 @@ func ExampleConfig() {
 	}
 	fmt.Printf("%+v\n", *config.LogRetryConfig)
 	// Output:
-	// {InsecureConnection:true CACertFile: OtelExporterGRPCEndpoint:localhost:4317 OtelExporterHTTPEndpoint:localhost:4318 ResourceAttributes:[{Key:package_name Value:{vtype:4 numeric:0 stringly:beholder slice:<nil>}} {Key:sender Value:{vtype:4 numeric:0 stringly:beholderclient slice:<nil>}}] EmitterExportTimeout:1s EmitterExportInterval:1s EmitterExportMaxBatchSize:512 EmitterMaxQueueSize:2048 EmitterBatchProcessor:true TraceSampleRatio:1 TraceBatchTimeout:1s TraceSpanExporter:<nil> TraceRetryConfig:<nil> TraceCompressionDisabled:false MetricReaderInterval:1s MetricRetryConfig:<nil> MetricViews:[] MetricCompressionDisabled:false ChipIngressEmitterEnabled:false ChipIngressEmitterGRPCEndpoint: ChipIngressInsecureConnection:false LogExportTimeout:1s LogExportInterval:1s LogExportMaxBatchSize:512 LogMaxQueueSize:2048 LogBatchProcessor:true LogRetryConfig:<nil> LogStreamingEnabled:false LogLevel:info LogCompressionDisabled:false AuthHeaders:map[] AuthHeadersTTL:0s AuthKeySigner:<nil> AuthPublicKeyHex:}
+	// {InsecureConnection:true CACertFile: OtelExporterGRPCEndpoint:localhost:4317 OtelExporterHTTPEndpoint:localhost:4318 ResourceAttributes:[{Key:package_name Value:{vtype:4 numeric:0 stringly:beholder slice:<nil>}} {Key:sender Value:{vtype:4 numeric:0 stringly:beholderclient slice:<nil>}}] EmitterExportTimeout:1s EmitterExportInterval:1s EmitterExportMaxBatchSize:512 EmitterMaxQueueSize:2048 EmitterBatchProcessor:true TraceSampleRatio:1 TraceBatchTimeout:1s TraceSpanExporter:<nil> TraceRetryConfig:<nil> TraceCompressor:gzip MetricReaderInterval:1s MetricRetryConfig:<nil> MetricViews:[] MetricCompressor:gzip ChipIngressEmitterEnabled:false ChipIngressEmitterGRPCEndpoint: ChipIngressInsecureConnection:false LogExportTimeout:1s LogExportInterval:1s LogExportMaxBatchSize:512 LogMaxQueueSize:2048 LogBatchProcessor:true LogRetryConfig:<nil> LogStreamingEnabled:false LogLevel:info LogCompressor:gzip AuthHeaders:map[] AuthHeadersTTL:0s AuthKeySigner:<nil> AuthPublicKeyHex:}
 	// {InitialInterval:5s MaxInterval:30s MaxElapsedTime:1m0s}
 }

pkg/loop/config.go

@@ -70,9 +70,9 @@ const (
 	envTelemetryLogExportMaxBatchSize     = "CL_TELEMETRY_LOG_EXPORT_MAX_BATCH_SIZE"
 	envTelemetryLogExportInterval         = "CL_TELEMETRY_LOG_EXPORT_INTERVAL"
 	envTelemetryLogMaxQueueSize           = "CL_TELEMETRY_LOG_MAX_QUEUE_SIZE"
-	envTelemetryTraceCompressionDisabled  = "CL_TELEMETRY_TRACE_COMPRESSION_DISABLED"
-	envTelemetryMetricCompressionDisabled = "CL_TELEMETRY_METRIC_COMPRESSION_DISABLED"
-	envTelemetryLogCompressionDisabled    = "CL_TELEMETRY_LOG_COMPRESSION_DISABLED"
+	envTelemetryTraceCompressor           = "CL_TELEMETRY_TRACE_COMPRESSOR"
+	envTelemetryMetricCompressor          = "CL_TELEMETRY_METRIC_COMPRESSOR"
+	envTelemetryLogCompressor             = "CL_TELEMETRY_LOG_COMPRESSOR"
 
 	envChipIngressEndpoint           = "CL_CHIP_INGRESS_ENDPOINT"
 	envChipIngressInsecureConnection = "CL_CHIP_INGRESS_INSECURE_CONNECTION"
@@ -136,9 +136,9 @@ type EnvConfig struct {
 	TelemetryLogExportMaxBatchSize     int
 	TelemetryLogExportInterval         time.Duration
 	TelemetryLogMaxQueueSize           int
-	TelemetryTraceCompressionDisabled  bool
-	TelemetryMetricCompressionDisabled bool
-	TelemetryLogCompressionDisabled    bool
+	TelemetryTraceCompressor           string
+	TelemetryMetricCompressor          string
+	TelemetryLogCompressor             string
 
 	ChipIngressEndpoint           string
 	ChipIngressInsecureConnection bool
@@ -215,9 +215,9 @@ func (e *EnvConfig) AsCmdEnv() (env []string) {
 	add(envTelemetryLogExportMaxBatchSize, strconv.Itoa(e.TelemetryLogExportMaxBatchSize))
 	add(envTelemetryLogExportInterval, e.TelemetryLogExportInterval.String())
 	add(envTelemetryLogMaxQueueSize, strconv.Itoa(e.TelemetryLogMaxQueueSize))
-	add(envTelemetryTraceCompressionDisabled, strconv.FormatBool(e.TelemetryTraceCompressionDisabled))
-	add(envTelemetryMetricCompressionDisabled, strconv.FormatBool(e.TelemetryMetricCompressionDisabled))
-	add(envTelemetryLogCompressionDisabled, strconv.FormatBool(e.TelemetryLogCompressionDisabled))
+	add(envTelemetryTraceCompressor, e.TelemetryTraceCompressor)
+	add(envTelemetryMetricCompressor, e.TelemetryMetricCompressor)
+	add(envTelemetryLogCompressor, e.TelemetryLogCompressor)
 
 	add(envChipIngressEndpoint, e.ChipIngressEndpoint)
 	add(envChipIngressInsecureConnection, strconv.FormatBool(e.ChipIngressInsecureConnection))
@@ -415,18 +415,9 @@ func (e *EnvConfig) parse() error {
 		if err != nil {
 			return fmt.Errorf("failed to parse %s: %w", envTelemetryLogMaxQueueSize, err)
 		}
-		e.TelemetryTraceCompressionDisabled, err = getBool(envTelemetryTraceCompressionDisabled)
-		if err != nil {
-			return fmt.Errorf("failed to parse %s: %w", envTelemetryTraceCompressionDisabled, err)
-		}
-		e.TelemetryMetricCompressionDisabled, err = getBool(envTelemetryMetricCompressionDisabled)
-		if err != nil {
-			return fmt.Errorf("failed to parse %s: %w", envTelemetryMetricCompressionDisabled, err)
-		}
-		e.TelemetryLogCompressionDisabled, err = getBool(envTelemetryLogCompressionDisabled)
-		if err != nil {
-			return fmt.Errorf("failed to parse %s: %w", envTelemetryLogCompressionDisabled, err)
-		}
+		e.TelemetryTraceCompressor = os.Getenv(envTelemetryTraceCompressor)
+		e.TelemetryMetricCompressor = os.Getenv(envTelemetryMetricCompressor)
+		e.TelemetryLogCompressor = os.Getenv(envTelemetryLogCompressor)
 		// Optional
 		e.ChipIngressEndpoint = os.Getenv(envChipIngressEndpoint)
 		e.ChipIngressInsecureConnection, err = getBool(envChipIngressInsecureConnection)

pkg/loop/server.go

@@ -155,7 +155,7 @@ func (s *Server) start(opts ...ServerOpt) error {
 			OtelExporterGRPCEndpoint:       s.EnvConfig.TelemetryEndpoint,
 			ResourceAttributes:             append(attributes, s.EnvConfig.TelemetryAttributes.AsStringAttributes()...),
 			TraceSampleRatio:               s.EnvConfig.TelemetryTraceSampleRatio,
-			TraceCompressionDisabled:       s.EnvConfig.TelemetryTraceCompressionDisabled,
+			TraceCompressor:                s.EnvConfig.TelemetryTraceCompressor,
 			EmitterBatchProcessor:          s.EnvConfig.TelemetryEmitterBatchProcessor,
 			EmitterExportTimeout:           s.EnvConfig.TelemetryEmitterExportTimeout,
 			EmitterExportInterval:          s.EnvConfig.TelemetryEmitterExportInterval,
@@ -168,11 +168,11 @@ func (s *Server) start(opts ...ServerOpt) error {
 			LogExportMaxBatchSize:          s.EnvConfig.TelemetryLogExportMaxBatchSize,
 			LogExportInterval:              s.EnvConfig.TelemetryLogExportInterval,
 			LogMaxQueueSize:                s.EnvConfig.TelemetryLogMaxQueueSize,
-			LogCompressionDisabled:         s.EnvConfig.TelemetryLogCompressionDisabled,
+			LogCompressor:                  s.EnvConfig.TelemetryLogCompressor,
 			ChipIngressEmitterEnabled:      s.EnvConfig.ChipIngressEndpoint != "",
 			ChipIngressEmitterGRPCEndpoint: s.EnvConfig.ChipIngressEndpoint,
 			ChipIngressInsecureConnection:  s.EnvConfig.ChipIngressInsecureConnection,
-			MetricCompressionDisabled:      s.EnvConfig.TelemetryMetricCompressionDisabled,
+			MetricCompressor:               s.EnvConfig.TelemetryMetricCompressor,
 		}
 
 		// Configure beholder auth - the client will determine rotating vs static mode