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

pkg/loop/config.go

@@ -61,6 +61,7 @@ const (
 	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"
 
 	envChipIngressEndpoint = "CL_CHIP_INGRESS_ENDPOINT"
 )
@@ -115,6 +116,7 @@ type EnvConfig struct {
 	TelemetryEmitterExportInterval     time.Duration
 	TelemetryEmitterExportMaxBatchSize int
 	TelemetryEmitterMaxQueueSize       int
+	TelemetryLogStreamingEnabled       bool
 
 	ChipIngressEndpoint string
 }
@@ -182,6 +184,7 @@ func (e *EnvConfig) AsCmdEnv() (env []string) {
 	add(envTelemetryEmitterExportInterval, e.TelemetryEmitterExportInterval.String())
 	add(envTelemetryEmitterExportMaxBatchSize, strconv.Itoa(e.TelemetryEmitterExportMaxBatchSize))
 	add(envTelemetryEmitterMaxQueueSize, strconv.Itoa(e.TelemetryEmitterMaxQueueSize))
+	add(envTelemetryLogStreamingEnabled, strconv.FormatBool(e.TelemetryLogStreamingEnabled))
 
 	add(envChipIngressEndpoint, e.ChipIngressEndpoint)
 
@@ -341,6 +344,10 @@ func (e *EnvConfig) parse() error {
 		if err != nil {
 			return fmt.Errorf("failed to parse %s: %w", envTelemetryEmitterMaxQueueSize, err)
 		}
+		e.TelemetryLogStreamingEnabled, err = getBool(envTelemetryLogStreamingEnabled)
+		if err != nil {
+			return fmt.Errorf("failed to parse %s: %w", envTelemetryLogStreamingEnabled, err)
+		}
 		// Optional
 		e.ChipIngressEndpoint = os.Getenv(envChipIngressEndpoint)
 	}

pkg/loop/config_test.go

@@ -74,6 +74,7 @@ func TestEnvConfig_parse(t *testing.T) {
 				envTelemetryEmitterExportInterval:     "2s",
 				envTelemetryEmitterExportMaxBatchSize: "100",
 				envTelemetryEmitterMaxQueueSize:       "1000",
+				envTelemetryLogStreamingEnabled:       "false",
 
 				envChipIngressEndpoint: "http://chip-ingress.example.com",
 			},
@@ -171,6 +172,7 @@ var envCfgFull = EnvConfig{
 	TelemetryEmitterExportInterval:     2 * time.Second,
 	TelemetryEmitterExportMaxBatchSize: 100,
 	TelemetryEmitterMaxQueueSize:       1000,
+	TelemetryLogStreamingEnabled:       false,
 
 	ChipIngressEndpoint: "http://chip-ingress.example.com",
 }
@@ -219,6 +221,7 @@ func TestEnvConfig_AsCmdEnv(t *testing.T) {
 	assert.Equal(t, "2s", got[envTelemetryEmitterExportInterval])
 	assert.Equal(t, "100", got[envTelemetryEmitterExportMaxBatchSize])
 	assert.Equal(t, "1000", got[envTelemetryEmitterMaxQueueSize])
+	assert.Equal(t, "false", got[envTelemetryLogStreamingEnabled])
 
 	// Assert ChipIngress environment variables
 	assert.Equal(t, "http://chip-ingress.example.com", got[envChipIngressEndpoint])

pkg/loop/server.go

@@ -123,6 +123,7 @@ func (s *Server) start() error {
 			EmitterExportInterval:          s.EnvConfig.TelemetryEmitterExportInterval,
 			EmitterExportMaxBatchSize:      s.EnvConfig.TelemetryEmitterExportMaxBatchSize,
 			EmitterMaxQueueSize:            s.EnvConfig.TelemetryEmitterMaxQueueSize,
+			LogStreamingEnabled:            s.EnvConfig.TelemetryLogStreamingEnabled,
 			ChipIngressEmitterEnabled:      s.EnvConfig.ChipIngressEndpoint != "",
 			ChipIngressEmitterGRPCEndpoint: s.EnvConfig.ChipIngressEndpoint,
 			ChipIngressInsecureConnection:  s.EnvConfig.TelemetryInsecureConnection,
@@ -166,7 +167,7 @@ func (s *Server) start() error {
 			LockTimeout:            s.EnvConfig.DatabaseLockTimeout,
 			MaxOpenConns:           s.EnvConfig.DatabaseMaxOpenConns,
 			MaxIdleConns:           s.EnvConfig.DatabaseMaxIdleConns,
-			EnableTracing:         	s.EnvConfig.DatabaseTracingEnabled,
+			EnableTracing:          s.EnvConfig.DatabaseTracingEnabled,
 		}.New(ctx, dbURL, pg.DriverPostgres)
 		if err != nil {
 			return fmt.Errorf("error connecting to DataBase: %w", err)