Merge branch 'main' into cache-settings-store-flag

Author: jinhoonbang

.github/workflows/api-diff.yml

@@ -1,5 +1,4 @@
 name: Analyze API Changes
-description: Runs apidiff-go action to check for breaking changes to modules' public APIs
 
 on:
   pull_request:
@@ -23,7 +22,7 @@ jobs:
           go-version-file: "go.mod"
           cache: false
 
-      - uses: smartcontractkit/.github/actions/apidiff-go@apidiff-go/0.1.0
+      - uses: smartcontractkit/.github/actions/apidiff-go@apidiff-go/0.2.0
         env:
           GITHUB_TOKEN: ${{ github.token }}
         with:

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)

pkg/settings/cresettings/defaults.json

@@ -2,6 +2,7 @@
 	"WorkflowLimit": "200",
 	"WorkflowRegistrationQueueLimit": "20",
 	"WorkflowExecutionConcurrencyLimit": "50",
+	"WorkflowTriggerRateLimit": "200rps:200",
 	"GatewayUnauthenticatedRequestRateLimit": "100rps:-1",
 	"GatewayUnauthenticatedRequestRateLimitPerIP": "1rps:-1",
 	"GatewayIncomingPayloadSizeLimit": "10kb",
@@ -10,7 +11,8 @@
 		"ZeroBalancePruningTimeout": "24h0m0s"
 	},
 	"PerOwner": {
-		"WorkflowExecutionConcurrencyLimit": "50"
+		"WorkflowExecutionConcurrencyLimit": "50",
+		"WorkflowTriggerRateLimit": "200rps:200"
 	},
 	"PerWorkflow": {
 		"TriggerLimit": "10",
@@ -40,7 +42,6 @@
 			"RateLimit": "every30s:3"
 		},
 		"LogTrigger": {
-			"RateLimit": "every10s:-1",
 			"Limit": "5",
 			"EventRateLimit": "every6s:10",
 			"FilterAddressLimit": "5",

pkg/settings/cresettings/defaults.toml

@@ -1,6 +1,7 @@
 WorkflowLimit = '200'
 WorkflowRegistrationQueueLimit = '20'
 WorkflowExecutionConcurrencyLimit = '50'
+WorkflowTriggerRateLimit = '200rps:200'
 GatewayUnauthenticatedRequestRateLimit = '100rps:-1'
 GatewayUnauthenticatedRequestRateLimitPerIP = '1rps:-1'
 GatewayIncomingPayloadSizeLimit = '10kb'
@@ -11,6 +12,7 @@ ZeroBalancePruningTimeout = '24h0m0s'
 
 [PerOwner]
 WorkflowExecutionConcurrencyLimit = '50'
+WorkflowTriggerRateLimit = '200rps:200'
 
 [PerWorkflow]
 TriggerLimit = '10'
@@ -41,7 +43,6 @@ RateLimit = 'every30s:1'
 RateLimit = 'every30s:3'
 
 [PerWorkflow.LogTrigger]
-RateLimit = 'every10s:-1'
 Limit = '5'
 EventRateLimit = 'every6s:10'
 FilterAddressLimit = '5'

pkg/settings/cresettings/settings.go

@@ -34,6 +34,7 @@ var Default = Schema{
 	WorkflowLimit:                               Int(200),
 	WorkflowRegistrationQueueLimit:              Int(20),
 	WorkflowExecutionConcurrencyLimit:           Int(50),
+	WorkflowTriggerRateLimit:                    Rate(200, 200),
 	GatewayUnauthenticatedRequestRateLimit:      Rate(rate.Every(time.Second/100), -1),
 	GatewayUnauthenticatedRequestRateLimitPerIP: Rate(rate.Every(time.Second), -1),
 	GatewayIncomingPayloadSizeLimit:             Size(10 * config.KByte),
@@ -44,6 +45,7 @@ var Default = Schema{
 	},
 	PerOwner: Owners{
 		WorkflowExecutionConcurrencyLimit: Int(50),
+		WorkflowTriggerRateLimit:          Rate(200, 200),
 	},
 	PerWorkflow: Workflows{
 		TriggerLimit:                  Int(10),
@@ -73,9 +75,8 @@ var Default = Schema{
 			RateLimit: Rate(rate.Every(30*time.Second), 3),
 		},
 		LogTrigger: logTrigger{
-			RateLimit:                Rate(rate.Every(10*time.Second), -1), //TODO
 			Limit:                    Int(5),
-			EventRateLimit:           Rate(rate.Every(time.Minute/10), 10), // TODO
+			EventRateLimit:           Rate(rate.Every(time.Minute/10), 10),
 			FilterAddressLimit:       Int(5),
 			FilterTopicsPerSlotLimit: Int(10),
 		},
@@ -91,7 +92,7 @@ var Default = Schema{
 			TargetsLimit:    Int(3),
 			ReportSizeLimit: Size(config.KByte),
 			EVM: evmChainWrite{
-				TransactionGasLimit: Uint64(500_000), //TODO
+				TransactionGasLimit: Uint64(500_000),
 			},
 		},
 		ChainRead: chainRead{
@@ -106,6 +107,7 @@ type Schema struct {
 	WorkflowLimit                               Setting[int] `unit:"{workflow}"`
 	WorkflowRegistrationQueueLimit              Setting[int] `unit:"{workflow}"`
 	WorkflowExecutionConcurrencyLimit           Setting[int] `unit:"{workflow}"`
+	WorkflowTriggerRateLimit                    Setting[config.Rate]
 	GatewayUnauthenticatedRequestRateLimit      Setting[config.Rate]
 	GatewayUnauthenticatedRequestRateLimitPerIP Setting[config.Rate]
 	GatewayIncomingPayloadSizeLimit             Setting[config.Size]
@@ -121,6 +123,7 @@ type Orgs struct {
 
 type Owners struct {
 	WorkflowExecutionConcurrencyLimit Setting[int] `unit:"{workflow}"`
+	WorkflowTriggerRateLimit          Setting[config.Rate]
 }
 
 type Workflows struct {
@@ -166,7 +169,6 @@ type httpTrigger struct {
 	RateLimit Setting[config.Rate]
 }
 type logTrigger struct {
-	RateLimit                Setting[config.Rate]
 	Limit                    Setting[int] `unit:"{trigger}"`
 	EventRateLimit           Setting[config.Rate]
 	FilterAddressLimit       Setting[int] `unit:"{address}"`

pkg/workflows/wasm/host/execution.go

@@ -27,6 +27,8 @@ type execution[T any] struct {
 	mode                 sdkpb.Mode
 	donSeed              int64
 	nodeSeed             int64
+	donLogCount          uint32
+	nodeLogCount         uint32
 }
 
 // callCapAsync async calls a capability by placing execution results onto a
@@ -139,6 +141,35 @@ func (e *execution[T]) awaitSecrets(ctx context.Context, acr *sdkpb.AwaitSecrets
 }
 
 func (e *execution[T]) log(caller *wasmtime.Caller, ptr int32, ptrlen int32) {
+	switch e.mode {
+	case sdkpb.Mode_MODE_DON:
+		e.donLogCount++
+		if e.donLogCount == e.module.cfg.MaxLogCountDONMode {
+			e.module.cfg.Logger.Warnf("max log count for don mode reached: %d - all subsequent logs will be dropped", e.donLogCount)
+		}
+		if e.donLogCount > e.module.cfg.MaxLogCountDONMode {
+			// silently drop to avoid spamming logs
+			return
+		}
+	case sdkpb.Mode_MODE_NODE:
+		e.nodeLogCount++
+		if e.nodeLogCount == e.module.cfg.MaxLogCountNodeMode {
+			e.module.cfg.Logger.Warnf("max log count for node mode reached: %d - all subsequent logs will be dropped", e.nodeLogCount)
+		}
+		if e.nodeLogCount > e.module.cfg.MaxLogCountNodeMode {
+			// silently drop to avoid spamming logs
+			return
+		}
+	default:
+		// unexpected / malicious
+		return
+	}
+
+	if ptrlen > int32(e.module.cfg.MaxLogLenBytes) {
+		e.module.cfg.Logger.Warnf("log message too long: %d - dropping", ptrlen)
+		return
+	}
+
 	b, innerErr := wasmRead(caller, ptr, ptrlen)
 	if innerErr != nil {
 		e.module.cfg.Logger.Errorf("error calling log: %s", innerErr)