Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 8 additions & 1 deletion pkg/beholder/client.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ import (
"context"
"errors"
"fmt"
"github.com/smartcontractkit/chainlink-common/pkg/chipingress"

"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc"
"go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp"
Expand All @@ -19,8 +19,12 @@ import (
oteltrace "go.opentelemetry.io/otel/trace"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/credentials/insecure"

"github.com/smartcontractkit/chainlink-common/pkg/chipingress"
)

const grpcCompressorGzip = "gzip"

type Emitter interface {
// Sends message with bytes and attributes to OTel Collector
Emit(ctx context.Context, body []byte, attrKVs ...any) error
Expand Down Expand Up @@ -391,6 +395,9 @@ func newMeterProvider(config Config, resource *sdkresource.Resource, creds crede
otlpmetricgrpc.WithEndpoint(config.OtelExporterGRPCEndpoint),
otlpmetricgrpc.WithHeaders(config.AuthHeaders),
}
if config.MetricCompressorEnabled {
opts = append(opts, otlpmetricgrpc.WithCompressor(grpcCompressorGzip))
}
if config.MetricRetryConfig != nil {
// NOTE: By default, the retry is enabled in the OTel SDK
opts = append(opts, otlpmetricgrpc.WithRetry(otlpmetricgrpc.RetryConfig{
Expand Down
2 changes: 2 additions & 0 deletions pkg/beholder/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,8 @@ type Config struct {
MetricReaderInterval time.Duration
MetricRetryConfig *RetryConfig
MetricViews []metric.View
// MetricCompressorEnabled enables gRPC compression for metrics (uses "gzip").
MetricCompressorEnabled bool

// Custom Events via Chip Ingress Emitter
ChipIngressEmitterEnabled bool
Expand Down
7 changes: 7 additions & 0 deletions pkg/loop/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,7 @@ const (
envTelemetryEmitterExportMaxBatchSize = "CL_TELEMETRY_EMITTER_EXPORT_MAX_BATCH_SIZE"
envTelemetryEmitterMaxQueueSize = "CL_TELEMETRY_EMITTER_MAX_QUEUE_SIZE"
envTelemetryLogStreamingEnabled = "CL_TELEMETRY_LOG_STREAMING_ENABLED"
envTelemetryMetricCompressorEnabled = "CL_TELEMETRY_METRIC_COMPRESSOR_ENABLED"

envChipIngressEndpoint = "CL_CHIP_INGRESS_ENDPOINT"
envChipIngressInsecureConnection = "CL_CHIP_INGRESS_INSECURE_CONNECTION"
Expand Down Expand Up @@ -118,6 +119,7 @@ type EnvConfig struct {
TelemetryEmitterExportMaxBatchSize int
TelemetryEmitterMaxQueueSize int
TelemetryLogStreamingEnabled bool
TelemetryMetricCompressorEnabled bool

ChipIngressEndpoint string
ChipIngressInsecureConnection bool
Expand Down Expand Up @@ -187,6 +189,7 @@ func (e *EnvConfig) AsCmdEnv() (env []string) {
add(envTelemetryEmitterExportMaxBatchSize, strconv.Itoa(e.TelemetryEmitterExportMaxBatchSize))
add(envTelemetryEmitterMaxQueueSize, strconv.Itoa(e.TelemetryEmitterMaxQueueSize))
add(envTelemetryLogStreamingEnabled, strconv.FormatBool(e.TelemetryLogStreamingEnabled))
add(envTelemetryMetricCompressorEnabled, strconv.FormatBool(e.TelemetryMetricCompressorEnabled))

add(envChipIngressEndpoint, e.ChipIngressEndpoint)
add(envChipIngressInsecureConnection, strconv.FormatBool(e.ChipIngressInsecureConnection))
Expand Down Expand Up @@ -351,6 +354,10 @@ func (e *EnvConfig) parse() error {
if err != nil {
return fmt.Errorf("failed to parse %s: %w", envTelemetryLogStreamingEnabled, err)
}
e.TelemetryMetricCompressorEnabled, err = getBool(envTelemetryMetricCompressorEnabled)
if err != nil {
return fmt.Errorf("failed to parse %s: %w", envTelemetryMetricCompressorEnabled, err)
}
// Optional
e.ChipIngressEndpoint = os.Getenv(envChipIngressEndpoint)
e.ChipIngressInsecureConnection, err = getBool(envChipIngressInsecureConnection)
Expand Down
1 change: 1 addition & 0 deletions pkg/loop/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,7 @@ func (s *Server) start() error {
ChipIngressEmitterEnabled: s.EnvConfig.ChipIngressEndpoint != "",
ChipIngressEmitterGRPCEndpoint: s.EnvConfig.ChipIngressEndpoint,
ChipIngressInsecureConnection: s.EnvConfig.ChipIngressInsecureConnection,
MetricCompressorEnabled: s.EnvConfig.TelemetryMetricCompressorEnabled,
}

if tracingConfig.Enabled {
Expand Down
Loading