Beholder OTel logs streaming for LOOPP's

pkg/logger/logger.go

@@ -96,6 +96,17 @@ func NewWith(cfgFn func(*zap.Config)) (Logger, error) {
 	return &logger{core.Sugar()}, nil
 }
 
+// NewCore returns a new Logger core from a modified [zap.Config].
+func NewCore(cfgFn func(*zap.Config)) (zapcore.Core, error) {
+	cfg := zap.NewProductionConfig()
+	cfgFn(&cfg)
+	logger, err := cfg.Build()
+	if err != nil {
+		return nil, err
+	}
+	return logger.Core(), nil
+}
+
 // NewWithSync returns a new Logger with a given SyncWriter.
 func NewWithSync(w io.Writer) Logger {
 	core := zapcore.NewCore(zapcore.NewJSONEncoder(zap.NewProductionEncoderConfig()), zapcore.AddSync(w), zapcore.InfoLevel)

pkg/logger/logger_test.go

@@ -324,3 +324,25 @@ type differentLogger interface {
 
 	Sync() error
 }
+
+func TestNewCore(t *testing.T) {
+	// First core at Info (would drop Debug), second core at Debug
+	obsCore, obsLogs := observer.New(zap.DebugLevel)
+
+	primaryCore, err := NewCore(func(cfg *zap.Config) {
+		cfg.Level = zap.NewAtomicLevelAt(zap.InfoLevel)
+	})
+	if err != nil {
+		t.Fatalf("NewCore error: %v", err)
+	}
+
+	lggr := NewWithCores(primaryCore, obsCore)
+
+	lggr.Debug("debug message should reach observer core")
+	if got := obsLogs.Len(); got != 1 {
+		t.Fatalf("expected 1 log in observer core, got %d", got)
+	}
+	if msg := obsLogs.All()[0].Message; msg != "debug message should reach observer core" {
+		t.Fatalf("unexpected message: %s", msg)
+	}
+}

pkg/loop/logger.go

@@ -13,7 +13,10 @@ import (
 	"go.uber.org/zap/zapcore"
 	"golang.org/x/exp/slices"
 
+	otellog "go.opentelemetry.io/otel/log"
+
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/chainlink-common/pkg/logger/otelzap"
 )
 
 // HCLogLogger returns an [hclog.Logger] backed by the given [logger.Logger].
@@ -162,13 +165,38 @@ func (h *hclSinkAdapter) Accept(_ string, level hclog.Level, msg string, args ..
 
 // NewLogger returns a new [logger.Logger] configured to encode [hclog] compatible JSON.
 func NewLogger() (logger.Logger, error) {
-	return logger.NewWith(func(cfg *zap.Config) {
-		cfg.Level.SetLevel(zap.DebugLevel)
-		cfg.EncoderConfig.LevelKey = "@level"
-		cfg.EncoderConfig.MessageKey = "@message"
-		cfg.EncoderConfig.TimeKey = "@timestamp"
-		cfg.EncoderConfig.EncodeTime = zapcore.TimeEncoderOfLayout("2006-01-02T15:04:05.000000Z07:00")
-	})
+	return logger.NewWith(configureHCLogEncoder)
+}
+
+// configureHCLogEncoder mutates cfg to use hclog-compatible field names and timestamp format.
+// NOTE: It also sets the log level to Debug to preserve prior behavior where each caller
+// manually set Debug before applying identical encoder tweaks. Centralizing avoids drift.
+// If a different level is desired, callers should override cfg.Level AFTER calling this helper.
+func configureHCLogEncoder(cfg *zap.Config) {
+	cfg.Level.SetLevel(zap.DebugLevel)
+	cfg.EncoderConfig.LevelKey = "@level"
+	cfg.EncoderConfig.MessageKey = "@message"
+	cfg.EncoderConfig.TimeKey = "@timestamp"
+	cfg.EncoderConfig.EncodeTime = zapcore.TimeEncoderOfLayout("2006-01-02T15:04:05.000000Z07:00")
+}
+
+// NewOtelLogger returns a logger with two cores:
+//  1. The primary JSON core configured via cfgFn (encoder keys changed to @level, @message, @timestamp).
+//  2. The otel core (otelzap.NewCore) which receives the raw zap.Entry and fields.
+//
+// Important:
+// The cfgFn only mutates the encoder config used to build the first core.
+// otelzap.NewCore implements zapcore.Core and does NOT use that encoder; it derives attributes from the zap.Entry
+// (Message, Level, Time, etc.) and zap.Fields directly. Therefore changing encoder keys here does NOT affect how
+// the otel core extracts data, and only the first core's JSON output format is altered.
+// This preserves backward compatibility for OTEL export while allowing hclog-compatible key names in the primary output.
+func NewOtelLogger(otelLogger otellog.Logger) (logger.Logger, error) {
+	primaryCore, err := logger.NewCore(configureHCLogEncoder)
+	if err != nil {
+		return nil, err
+	}
+	// set debug level from primaryCore to match otelzap.NewCore
+	return logger.NewWithCores(primaryCore, otelzap.NewCore(otelLogger, otelzap.WithLevel(zapcore.DebugLevel))), nil
 }
 
 // onceValue returns a function that invokes f only once and returns the value

pkg/loop/logger_test.go

@@ -1,9 +1,15 @@
 package loop
 
 import (
+	"context"
+	"sync"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	sdklog "go.opentelemetry.io/otel/sdk/log"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/chainlink-common/pkg/logger/otelzap"
 )
 
 func Test_removeArg(t *testing.T) {
@@ -38,3 +44,74 @@ func Test_removeArg(t *testing.T) {
 		})
 	}
 }
+
+func TestNewOtelLogger(t *testing.T) {
+	tests := []struct {
+		name    string
+		logFn   func(l logger.Logger)
+		wantMsg string
+	}{
+		{
+			name: "debug",
+			logFn: func(l logger.Logger) {
+				l.Debugw("hello world", "k", "v")
+			},
+			wantMsg: "hello world",
+		},
+		{
+			name: "info",
+			logFn: func(l logger.Logger) {
+				l.Infow("info msg", "a", 1)
+			},
+			wantMsg: "info msg",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			exp := &recordingExporter{}
+			lp := sdklog.NewLoggerProvider(
+				sdklog.WithProcessor(sdklog.NewSimpleProcessor(exp)),
+			)
+			otelLggr := lp.Logger("test-" + tt.name)
+
+			lggr, err := NewOtelLogger(otelLggr)
+			if err != nil {
+				t.Fatalf("NewOtelLogger error: %v", err)
+			}
+
+			tt.logFn(lggr)
+
+			// Force flush the logger provider to ensure records are exported
+			if err := lp.ForceFlush(context.Background()); err != nil {
+				t.Fatalf("ForceFlush error: %v", err)
+			}
+
+			if len(exp.records) != 1 {
+				t.Fatalf("expected 1 exported record, got %d", len(exp.records))
+			}
+			if got := exp.records[0].Body().AsString(); got != tt.wantMsg {
+				t.Fatalf("unexpected body: got %q want %q", got, tt.wantMsg)
+			}
+		})
+	}
+}
+
+// recordingExporter captures exported log records (current sdk/log Export signature).
+type recordingExporter struct {
+	mu      sync.Mutex
+	records []sdklog.Record
+}
+
+func (r *recordingExporter) Export(_ context.Context, recs []sdklog.Record) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.records = append(r.records, recs...)
+	return nil
+}
+func (r *recordingExporter) ForceFlush(context.Context) error { return nil }
+func (r *recordingExporter) Shutdown(context.Context) error   { return nil }
+
+// Compile-time assertion that otelzap.NewCore still satisfies zapcore.Core usage pattern.
+// (Guards against accidental API break causing this test file to silently compile with stubs.)
+var _ = otelzap.NewCore
+var _ logger.Logger // silence unused import of logger in case future refactors remove usage

pkg/loop/server.go

@@ -167,6 +167,14 @@ func (s *Server) start() error {
 		}
 		beholder.SetClient(beholderClient)
 		beholder.SetGlobalOtelProviders()
+
+		if beholderCfg.LogStreamingEnabled {
+			otelLogger, err := NewOtelLogger(beholderClient.Logger)
+			if err != nil {
+				return fmt.Errorf("failed to enable log streaming: %w", err)
+			}
+			s.Logger = logger.Sugared(logger.Named(otelLogger, s.Logger.Name()))
+		}
 	}
 
 	s.promServer = NewPromServer(s.EnvConfig.PrometheusPort, s.Logger)