INFOPLAT-2672: beholder: config option to enable streaming logs to OTel in core

.changeset/short-parks-retire.md

@@ -0,0 +1,5 @@
+---
+"chainlink": patch
+---
+
+#added add LogStreamingEnabled config that enables streaming logs to the OTel exporter

core/config/docs/core.toml

@@ -815,6 +815,8 @@ EmitterExportTimeout = '1s' # Default
 ChipIngressEndpoint = '' # Default
 # HeartbeatInterval is the interval at which a the application heartbeat is sent to telemetry backends.
 HeartbeatInterval = '1s' # Default
+# LogStreamingEnabled enables log streaming to the OTel log exporter
+LogStreamingEnabled = false # Default
 
 # ResourceAttributes are global metadata to include with all telemetry.
 [Telemetry.ResourceAttributes]

core/config/telemetry_config.go

@@ -13,4 +13,5 @@ type Telemetry interface {
 	EmitterExportTimeout() time.Duration
 	ChipIngressEndpoint() string
 	HeartbeatInterval() time.Duration
+	LogStreamingEnabled() bool
 }

core/config/toml/types.go

@@ -2275,6 +2275,7 @@ type Telemetry struct {
 	EmitterExportTimeout  *commonconfig.Duration
 	ChipIngressEndpoint   *string
 	HeartbeatInterval     *commonconfig.Duration
+	LogStreamingEnabled   *bool
 }
 
 func (b *Telemetry) setFrom(f *Telemetry) {
@@ -2308,6 +2309,9 @@ func (b *Telemetry) setFrom(f *Telemetry) {
 	if v := f.HeartbeatInterval; v != nil {
 		b.HeartbeatInterval = v
 	}
+	if v := f.LogStreamingEnabled; v != nil {
+		b.LogStreamingEnabled = v
+	}
 }
 
 func (b *Telemetry) ValidateConfig() (err error) {

core/services/chainlink/config_telemetry.go

@@ -95,3 +95,10 @@ func (b *telemetryConfig) HeartbeatInterval() time.Duration {
 	}
 	return b.s.HeartbeatInterval.Duration()
 }
+
+func (b *telemetryConfig) LogStreamingEnabled() bool {
+	if b.s.LogStreamingEnabled == nil {
+		return false
+	}
+	return *b.s.LogStreamingEnabled
+}

core/services/chainlink/config_telemetry_test.go

@@ -225,3 +225,21 @@ func TestTelemetryConfig_HeartbeatInterval(t *testing.T) {
 		})
 	}
 }
+
+func TestTelemetryConfig_LogStreamingEnabled(t *testing.T) {
+	tests := []struct {
+		name      string
+		telemetry toml.Telemetry
+		expected  bool
+	}{
+		{"LogStreamingEnabledTrue", toml.Telemetry{LogStreamingEnabled: ptr(true)}, true},
+		{"LogStreamingEnabledFalse", toml.Telemetry{LogStreamingEnabled: ptr(false)}, false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tc := telemetryConfig{s: tt.telemetry}
+			assert.Equal(t, tt.expected, tc.LogStreamingEnabled())
+		})
+	}
+}

core/services/chainlink/config_test.go

@@ -591,6 +591,7 @@ func TestConfig_Marshal(t *testing.T) {
 		EmitterExportTimeout:  commoncfg.MustNewDuration(1 * time.Second),
 		ChipIngressEndpoint:   ptr("example.com/chip-ingress"),
 		HeartbeatInterval:     commoncfg.MustNewDuration(1 * time.Second),
+		LogStreamingEnabled:   ptr(false),
 	}
 	full.CRE = toml.CreConfig{
 		Streams: &toml.StreamsConfig{

core/services/chainlink/testdata/config-empty-effective.toml

@@ -335,6 +335,7 @@ EmitterBatchProcessor = true
 EmitterExportTimeout = '1s'
 ChipIngressEndpoint = ''
 HeartbeatInterval = '1s'
+LogStreamingEnabled = false
 
 [Workflows]
 [Workflows.Limits]

core/services/chainlink/testdata/config-full.toml

@@ -345,6 +345,7 @@ EmitterBatchProcessor = true
 EmitterExportTimeout = '1s'
 ChipIngressEndpoint = 'example.com/chip-ingress'
 HeartbeatInterval = '1s'
+LogStreamingEnabled = false
 
 [Telemetry.ResourceAttributes]
 Baz = 'test'

core/services/chainlink/testdata/config-multi-chain-effective.toml

@@ -335,6 +335,7 @@ EmitterBatchProcessor = true
 EmitterExportTimeout = '1s'
 ChipIngressEndpoint = ''
 HeartbeatInterval = '1s'
+LogStreamingEnabled = false
 
 [Workflows]
 [Workflows.Limits]

core/web/resolver/testdata/config-empty-effective.toml

@@ -335,6 +335,7 @@ EmitterBatchProcessor = true
 EmitterExportTimeout = '1s'
 ChipIngressEndpoint = ''
 HeartbeatInterval = '1s'
+LogStreamingEnabled = false
 
 [Workflows]
 [Workflows.Limits]

core/web/resolver/testdata/config-full.toml

@@ -345,6 +345,7 @@ EmitterBatchProcessor = true
 EmitterExportTimeout = '1s'
 ChipIngressEndpoint = 'example.com/chip-ingress'
 HeartbeatInterval = '1s'
+LogStreamingEnabled = false
 
 [Telemetry.ResourceAttributes]
 Baz = 'test'

core/web/resolver/testdata/config-multi-chain-effective.toml

@@ -335,6 +335,7 @@ EmitterBatchProcessor = true
 EmitterExportTimeout = '1s'
 ChipIngressEndpoint = ''
 HeartbeatInterval = '1s'
+LogStreamingEnabled = false
 
 [Workflows]
 [Workflows.Limits]

docs/CONFIG.md

@@ -2222,6 +2222,7 @@ EmitterBatchProcessor = true # Default
 EmitterExportTimeout = '1s' # Default
 ChipIngressEndpoint = '' # Default
 HeartbeatInterval = '1s' # Default
+LogStreamingEnabled = false # Default
 ```
 Telemetry holds OTEL settings.
 This data includes open telemetry metrics, traces, & logs.
@@ -2283,6 +2284,12 @@ HeartbeatInterval = '1s' # Default
 ```
 HeartbeatInterval is the interval at which a the application heartbeat is sent to telemetry backends.
 
+### LogStreamingEnabled
+```toml
+LogStreamingEnabled = false # Default
+```
+LogStreamingEnabled enables log streaming to the OTel log exporter
+
 ## Telemetry.ResourceAttributes
 ```toml
 [Telemetry.ResourceAttributes]

plugins/loop_registry.go

@@ -127,6 +127,7 @@ func (m *LoopRegistry) Register(id string) (*RegisteredLoop, error) {
 		envCfg.TelemetryEmitterExportTimeout = m.cfgTelemetry.EmitterExportTimeout()
 		envCfg.TelemetryAuthPubKeyHex = m.telemetryAuthPubKeyHex
 		envCfg.ChipIngressEndpoint = m.cfgTelemetry.ChipIngressEndpoint()
+		envCfg.TelemetryLogStreamingEnabled = m.cfgTelemetry.LogStreamingEnabled()
 	}
 	m.lggr.Debugf("Registered loopp %q with port %d", id, envCfg.PrometheusPort)
 

plugins/loop_registry_test.go

@@ -70,6 +70,8 @@ func (m mockCfgTelemetry) HeartbeatInterval() time.Duration {
 	return 5 * time.Second
 }
 
+func (m mockCfgTelemetry) LogStreamingEnabled() bool { return false }
+
 type mockCfgDatabase struct{}
 
 func (m mockCfgDatabase) Backup() config.Backup { panic("unimplemented") }
@@ -215,6 +217,7 @@ func TestLoopRegistry_Register(t *testing.T) {
 	require.Equal(t, 0.42, envCfg.TelemetryTraceSampleRatio)
 	require.True(t, envCfg.TelemetryEmitterBatchProcessor)
 	require.Equal(t, 1*time.Second, envCfg.TelemetryEmitterExportTimeout)
+	require.False(t, envCfg.TelemetryLogStreamingEnabled)
 
 	require.Equal(t, "example.com/chip-ingress", envCfg.ChipIngressEndpoint)
 }

testdata/scripts/config/merge_raw_configs.txtar

@@ -482,6 +482,7 @@ EmitterBatchProcessor = true
 EmitterExportTimeout = '1s'
 ChipIngressEndpoint = ''
 HeartbeatInterval = '1s'
+LogStreamingEnabled = false
 
 [Workflows]
 [Workflows.Limits]

testdata/scripts/node/validate/default.txtar

@@ -347,6 +347,7 @@ EmitterBatchProcessor = true
 EmitterExportTimeout = '1s'
 ChipIngressEndpoint = ''
 HeartbeatInterval = '1s'
+LogStreamingEnabled = false
 
 [Workflows]
 [Workflows.Limits]

testdata/scripts/node/validate/defaults-override.txtar

@@ -408,6 +408,7 @@ EmitterBatchProcessor = true
 EmitterExportTimeout = '1s'
 ChipIngressEndpoint = ''
 HeartbeatInterval = '1s'
+LogStreamingEnabled = false
 
 [Workflows]
 [Workflows.Limits]

testdata/scripts/node/validate/disk-based-logging-disabled.txtar

@@ -391,6 +391,7 @@ EmitterBatchProcessor = true
 EmitterExportTimeout = '1s'
 ChipIngressEndpoint = ''
 HeartbeatInterval = '1s'
+LogStreamingEnabled = false
 
 [Workflows]
 [Workflows.Limits]

testdata/scripts/node/validate/disk-based-logging-no-dir.txtar

@@ -391,6 +391,7 @@ EmitterBatchProcessor = true
 EmitterExportTimeout = '1s'
 ChipIngressEndpoint = ''
 HeartbeatInterval = '1s'
+LogStreamingEnabled = false
 
 [Workflows]
 [Workflows.Limits]

testdata/scripts/node/validate/disk-based-logging.txtar

@@ -391,6 +391,7 @@ EmitterBatchProcessor = true
 EmitterExportTimeout = '1s'
 ChipIngressEndpoint = ''
 HeartbeatInterval = '1s'
+LogStreamingEnabled = false
 
 [Workflows]
 [Workflows.Limits]

testdata/scripts/node/validate/fallback-override.txtar

@@ -489,6 +489,7 @@ EmitterBatchProcessor = true
 EmitterExportTimeout = '1s'
 ChipIngressEndpoint = ''
 HeartbeatInterval = '1s'
+LogStreamingEnabled = false
 
 [Workflows]
 [Workflows.Limits]

testdata/scripts/node/validate/invalid-ocr-p2p.txtar

@@ -376,6 +376,7 @@ EmitterBatchProcessor = true
 EmitterExportTimeout = '1s'
 ChipIngressEndpoint = ''
 HeartbeatInterval = '1s'
+LogStreamingEnabled = false
 
 [Workflows]
 [Workflows.Limits]

testdata/scripts/node/validate/invalid.txtar

@@ -387,6 +387,7 @@ EmitterBatchProcessor = true
 EmitterExportTimeout = '1s'
 ChipIngressEndpoint = ''
 HeartbeatInterval = '1s'
+LogStreamingEnabled = false
 
 [Workflows]
 [Workflows.Limits]

testdata/scripts/node/validate/valid.txtar

@@ -388,6 +388,7 @@ EmitterBatchProcessor = true
 EmitterExportTimeout = '1s'
 ChipIngressEndpoint = ''
 HeartbeatInterval = '1s'
+LogStreamingEnabled = false
 
 [Workflows]
 [Workflows.Limits]

testdata/scripts/node/validate/warnings.txtar

@@ -370,6 +370,7 @@ EmitterBatchProcessor = true
 EmitterExportTimeout = '1s'
 ChipIngressEndpoint = ''
 HeartbeatInterval = '1s'
+LogStreamingEnabled = false
 
 [Workflows]
 [Workflows.Limits]