INFOPLAT-1468 Enable batching for beholder emitter

core/cmd/shell.go

@@ -103,6 +103,8 @@ func initGlobals(cfgProm config.Prometheus, cfgTracing config.Tracing, cfgTeleme
 				OtelExporterGRPCEndpoint: cfgTelemetry.OtelExporterGRPCEndpoint(),
 				ResourceAttributes:       attributes,
 				TraceSampleRatio:         cfgTelemetry.TraceSampleRatio(),
+				EmitterBatchProcessor:    cfgTelemetry.EmitterBatchProcessor(),
+				EmitterExportTimeout:     cfgTelemetry.EmitterExportTimeout(),
 			}
 			if tracingCfg.Enabled {
 				clientCfg.TraceSpanExporter, err = tracingCfg.NewSpanExporter()

core/config/docs/core.toml

@@ -706,6 +706,10 @@ CACertFile = 'cert-file' # Example
 InsecureConnection = false # Default
 # TraceSampleRatio is the rate at which to sample traces. Must be between 0 and 1.
 TraceSampleRatio = 0.01 # Default
+# Enables batching for telemetry events
+EmitterBatchProcessor = true # Default
+# Sets export timeout for telemetry events
+EmitterExportTimeout = "1s" # Default
 
 # ResourceAttributes are global metadata to include with all telemetry.
 [Telemetry.ResourceAttributes]

core/config/telemetry_config.go

@@ -1,10 +1,14 @@
 package config
 
+import "time"
+
 type Telemetry interface {
 	Enabled() bool
 	InsecureConnection() bool
 	CACertFile() string
 	OtelExporterGRPCEndpoint() string
 	ResourceAttributes() map[string]string
 	TraceSampleRatio() float64
+	EmitterBatchProcessor() bool
+	EmitterExportTimeout() time.Duration
 }

core/config/toml/types.go

@@ -1657,12 +1657,14 @@ func (t *Tracing) ValidateConfig() (err error) {
 }
 
 type Telemetry struct {
-	Enabled            *bool
-	CACertFile         *string
-	Endpoint           *string
-	InsecureConnection *bool
-	ResourceAttributes map[string]string `toml:",omitempty"`
-	TraceSampleRatio   *float64
+	Enabled               *bool
+	CACertFile            *string
+	Endpoint              *string
+	InsecureConnection    *bool
+	ResourceAttributes    map[string]string `toml:",omitempty"`
+	TraceSampleRatio      *float64
+	EmitterBatchProcessor *bool
+	EmitterExportTimeout  *commonconfig.Duration
 }
 
 func (b *Telemetry) setFrom(f *Telemetry) {
@@ -1684,6 +1686,12 @@ func (b *Telemetry) setFrom(f *Telemetry) {
 	if v := f.TraceSampleRatio; v != nil {
 		b.TraceSampleRatio = v
 	}
+	if v := f.EmitterBatchProcessor; v != nil {
+		b.EmitterBatchProcessor = v
+	}
+	if v := f.EmitterExportTimeout; v != nil {
+		b.EmitterExportTimeout = v
+	}
 }
 
 func (b *Telemetry) ValidateConfig() (err error) {

core/services/chainlink/config_telemetry.go

@@ -1,6 +1,8 @@
 package chainlink
 
 import (
+	"time"
+
 	"github.com/smartcontractkit/chainlink/v2/core/config/toml"
 )
 
@@ -41,3 +43,17 @@ func (b *telemetryConfig) TraceSampleRatio() float64 {
 	}
 	return *b.s.TraceSampleRatio
 }
+
+func (b *telemetryConfig) EmitterBatchProcessor() bool {
+	if b.s.EmitterBatchProcessor == nil {
+		return false
+	}
+	return *b.s.EmitterBatchProcessor
+}
+
+func (b *telemetryConfig) EmitterExportTimeout() time.Duration {
+	if b.s.EmitterExportTimeout == nil {
+		return 0
+	}
+	return b.s.EmitterExportTimeout.Duration()
+}

plugins/loop_registry_test.go

@@ -2,6 +2,7 @@ package plugins
 
 import (
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 
@@ -55,6 +56,10 @@ func (m mockCfgTelemetry) ResourceAttributes() map[string]string {
 
 func (m mockCfgTelemetry) TraceSampleRatio() float64 { return 0.42 }
 
+func (m mockCfgTelemetry) EmitterBatchProcessor() bool { return false }
+
+func (m mockCfgTelemetry) EmitterExportTimeout() time.Duration { return 0 }
+
 func TestLoopRegistry_Register(t *testing.T) {
 	mockCfgTracing := &mockCfgTracing{}
 	mockCfgTelemetry := &mockCfgTelemetry{}