Merge branch 'INFOPLAT-2938-logs-streaming-loopp' into INFOPLAT-2938-logs-streaming-loopp-heartbeat

Author: pkcll

go.mod

@@ -37,7 +37,7 @@ require (
 	github.com/scylladb/go-reflectx v1.0.1
 	github.com/shopspring/decimal v1.4.0
 	github.com/smartcontractkit/chain-selectors v1.0.67
-	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.1
+	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.4
 	github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20250722225531-876fd6b94976
 	github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20250911124514-5874cc6d62b2
 	github.com/smartcontractkit/chainlink-protos/storage-service v0.3.0

go.sum

@@ -326,8 +326,8 @@ github.com/sirupsen/logrus v1.4.1/go.mod h1:ni0Sbl8bgC9z8RoU9G6nDWqqs/fq4eDPysMB
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
 github.com/smartcontractkit/chain-selectors v1.0.67 h1:gxTqP/JC40KDe3DE1SIsIKSTKTZEPyEU1YufO1admnw=
 github.com/smartcontractkit/chain-selectors v1.0.67/go.mod h1:xsKM0aN3YGcQKTPRPDDtPx2l4mlTN1Djmg0VVXV40b8=
-github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.1 h1:ca2z5OXgnbBPQRxpwXwBLJsUA1+cAp5ncfW4Ssvd6eY=
-github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.1/go.mod h1:NZv/qKYGFRnkjOYBouajnDfFoZ+WDa6H2KNmSf1dnKc=
+github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.4 h1:hvqATtrZ0iMRTI80cpBot/3JFbjz2j+2tvpfooVhRHw=
+github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.4/go.mod h1:eKGyfTKzr0/PeR7qKN4l2FcW9p+HzyKUwAfGhm/5YZc=
 github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20250722225531-876fd6b94976 h1:mF3FiDUoV0QbJcks9R2y7ydqntNL1Z0VCPBJgx/Ms+0=
 github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20250722225531-876fd6b94976/go.mod h1:HHGeDUpAsPa0pmOx7wrByCitjQ0mbUxf0R9v+g67uCA=
 github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20250911124514-5874cc6d62b2 h1:1/KdO5AbUr3CmpLjMPuJXPo2wHMbfB8mldKLsg7D4M8=

pkg/beholder/chip_client.go

@@ -0,0 +1,45 @@
+package beholder
+
+import (
+	"context"
+	"fmt"
+	"github.com/smartcontractkit/chainlink-common/pkg/chipingress"
+	"github.com/smartcontractkit/chainlink-common/pkg/chipingress/pb"
+)
+
+type ChipIngressClient interface {
+	RegisterSchema(ctx context.Context, schemas ...*pb.Schema) (map[string]int, error)
+}
+
+type chipIngressClient struct {
+	client chipingress.Client
+}
+
+func NewChipIngressClient(client chipingress.Client) (ChipIngressClient, error) {
+	if client == nil {
+		return nil, fmt.Errorf("chip ingress client is nil")
+	}
+
+	return &chipIngressClient{
+		client: client,
+	}, nil
+}
+
+// RegisterSchema registers one or more schemas with the Chip Ingress service. Returns a map of subject to version for each registered schema.
+func (sr *chipIngressClient) RegisterSchema(ctx context.Context, schemas ...*pb.Schema) (map[string]int, error) {
+	request := &pb.RegisterSchemaRequest{
+		Schemas: schemas,
+	}
+
+	resp, err := sr.client.RegisterSchema(ctx, request)
+	if err != nil {
+		return nil, fmt.Errorf("failed to register schema: %w", err)
+	}
+
+	registeredMap := make(map[string]int)
+	for _, schema := range resp.Registered {
+		registeredMap[schema.Subject] = int(schema.Version)
+	}
+
+	return registeredMap, err
+}

pkg/beholder/chip_client_test.go

@@ -0,0 +1,72 @@
+package beholder_test
+
+import (
+	"fmt"
+	"github.com/smartcontractkit/chainlink-common/pkg/beholder"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/chipingress/mocks"
+	"github.com/smartcontractkit/chainlink-common/pkg/chipingress/pb"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+	"testing"
+)
+
+func TestNewChipClient(t *testing.T) {
+	t.Run("returns error when client is nil", func(t *testing.T) {
+		registry, err := beholder.NewChipIngressClient(nil)
+		assert.Nil(t, registry)
+		assert.EqualError(t, err, "chip ingress client is nil")
+	})
+
+	t.Run("returns schema registry when client is valid", func(t *testing.T) {
+		mockClient := mocks.NewClient(t)
+		registry, err := beholder.NewChipIngressClient(mockClient)
+		require.NoError(t, err)
+		assert.NotNil(t, registry)
+	})
+}
+
+func TestRegisterSchema(t *testing.T) {
+	t.Run("successfully registers schemas", func(t *testing.T) {
+		mockClient := mocks.NewClient(t)
+		mockClient.
+			On("RegisterSchema", mock.Anything, mock.Anything).
+			Return(&pb.RegisterSchemaResponse{
+				Registered: []*pb.RegisteredSchema{
+					{Subject: "schema1", Version: 1},
+					{Subject: "schema2", Version: 2},
+				},
+			}, nil)
+
+		registry, err := beholder.NewChipIngressClient(mockClient)
+		require.NoError(t, err)
+
+		schemas := []*pb.Schema{
+			{Subject: "schema1", Schema: `{"type":"record","name":"Test","fields":[{"name":"field1"}]}`, Format: 1},
+			{Subject: "schema2", Schema: `{"type":"record","name":"Test2","fields":[{"name":"field2"}]}`, Format: 2},
+		}
+
+		result, err := registry.RegisterSchema(t.Context(), schemas...)
+		require.NoError(t, err)
+		assert.Equal(t, map[string]int{"schema1": 1, "schema2": 2}, result)
+	})
+
+	t.Run("returns error when registration fails", func(t *testing.T) {
+		mockClient := mocks.NewClient(t)
+		mockClient.
+			On("RegisterSchema", mock.Anything, mock.Anything).
+			Return(nil, fmt.Errorf("registration failed"))
+
+		registry, err := beholder.NewChipIngressClient(mockClient)
+		require.NoError(t, err)
+
+		schemas := []*pb.Schema{
+			{Subject: "schema1", Schema: `{"type":"record","name":"Test","fields":[{"name":"field1"}]}`, Format: 1},
+		}
+
+		result, err := registry.RegisterSchema(t.Context(), schemas...)
+		assert.Nil(t, result)
+		assert.EqualError(t, err, "failed to register schema: registration failed")
+	})
+}

pkg/beholder/client.go

@@ -36,6 +36,8 @@ type Client struct {
 	Meter otelmetric.Meter
 	// Message Emitter
 	Emitter Emitter
+	// Chip
+	Chip ChipIngressClient
 
 	// Providers
 	LoggerProvider        otellog.LoggerProvider
@@ -213,10 +215,11 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (*Client, erro
 	// This will eventually be removed in favor of chip-ingress emitter
 	// and logs will be sent via OTLP using the regular Logger instead of calling Emit
 	emitter := NewMessageEmitter(messageLogger)
+	var chipIngressClient chipingress.Client
+
 	// if chip ingress is enabled, create dual source emitter that sends to both otel collector and chip ingress
 	// eventually we will remove the dual source emitter and just use chip ingress
-
-	if cfg.ChipIngressEmitterEnabled {
+	if cfg.ChipIngressEmitterEnabled || cfg.ChipIngressEmitterGRPCEndpoint != "" {
 		chipIngressOpts := make([]chipingress.Opt, 0, 2)
 
 		if cfg.ChipIngressInsecureConnection {
@@ -231,7 +234,7 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (*Client, erro
 			chipIngressOpts = append(chipIngressOpts, chipingress.WithTokenAuth(headerProvider))
 		}
 
-		chipIngressClient, err := chipingress.NewClient(
+		chipIngressClient, err = chipingress.NewClient(
 			cfg.ChipIngressEmitterGRPCEndpoint,
 			chipIngressOpts...,
 		)
@@ -251,13 +254,22 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (*Client, erro
 		}
 	}
 
+	// Create interface/wrapper to chip-ingress for schema registry
+	var chip ChipIngressClient
+	if chipIngressClient != nil {
+		chip, err = NewChipIngressClient(chipIngressClient)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create interface to chip ingress: %w", err)
+		}
+	}
+
 	onClose := func() (err error) {
 		for _, provider := range []shutdowner{messageLoggerProvider, loggerProvider, tracerProvider, meterProvider, messageLoggerProvider} {
 			err = errors.Join(err, provider.Shutdown(context.Background()))
 		}
 		return
 	}
-	return &Client{cfg, logger, tracer, meter, emitter, loggerProvider, tracerProvider, meterProvider, messageLoggerProvider, onClose}, nil
+	return &Client{cfg, logger, tracer, meter, emitter, chip, loggerProvider, tracerProvider, meterProvider, messageLoggerProvider, onClose}, nil
 }
 
 // Closes all providers, flushes all data and stops all background processes

pkg/beholder/client_test.go

@@ -480,6 +480,42 @@ func TestNewGRPCClient_ChipIngressEmitter(t *testing.T) {
 	})
 }
 
+func TestNewClient_Chip(t *testing.T) {
+	t.Run("chip interface available with chip-ingress endpoint provided", func(t *testing.T) {
+		client, err := beholder.NewClient(beholder.Config{
+			OtelExporterGRPCEndpoint:       "grpc-endpoint",
+			ChipIngressEmitterEnabled:      true,
+			ChipIngressEmitterGRPCEndpoint: "chip-ingress.example.com:9090",
+			ChipIngressInsecureConnection:  false,
+		})
+		require.NoError(t, err)
+		assert.NotNil(t, client)
+		assert.NotNil(t, client.Chip)
+	})
+
+	t.Run("chip interface can be enabled when chip ingress dual emitter is not enabled ", func(t *testing.T) {
+		client, err := beholder.NewClient(beholder.Config{
+			OtelExporterGRPCEndpoint:       "grpc-endpoint",
+			ChipIngressEmitterEnabled:      false,
+			ChipIngressEmitterGRPCEndpoint: "chip-ingress.example.com:9090",
+			ChipIngressInsecureConnection:  false,
+		})
+		require.NoError(t, err)
+		assert.NotNil(t, client)
+		assert.NotNil(t, client.Chip)
+		assert.NotNil(t, client.Emitter)
+	})
+
+	t.Run("chip interface is nil when chip ingress config is missing", func(t *testing.T) {
+		client, err := beholder.NewClient(beholder.Config{
+			OtelExporterGRPCEndpoint:  "grpc-endpoint",
+			ChipIngressEmitterEnabled: true,
+		})
+		require.Error(t, err)
+		assert.Nil(t, client)
+	})
+}
+
 // mockLogExporter is a no-op exporter for testing purposes.
 type mockLogExporter struct{}
 

pkg/beholder/httpclient.go

@@ -186,7 +186,7 @@ func NewHTTPClient(cfg Config, otlploghttpNew otlploghttpFactory) (*Client, erro
 		}
 		return
 	}
-	return &Client{cfg, logger, tracer, meter, emitter, loggerProvider, tracerProvider, meterProvider, messageLoggerProvider, onClose}, nil
+	return &Client{cfg, logger, tracer, meter, emitter, nil, loggerProvider, tracerProvider, meterProvider, messageLoggerProvider, onClose}, nil
 }
 
 func newHTTPTracerProvider(config Config, resource *sdkresource.Resource, tlsConfig *tls.Config) (*sdktrace.TracerProvider, error) {

pkg/beholder/noop.go

@@ -35,7 +35,7 @@ func NewNoopClient() *Client {
 	// MessageEmitter
 	messageEmitter := noopMessageEmitter{}
 
-	return &Client{cfg, logger, tracer, meter, messageEmitter, loggerProvider, tracerProvider, meterProvider, loggerProvider, noopOnClose}
+	return &Client{cfg, logger, tracer, meter, messageEmitter, nil, loggerProvider, tracerProvider, meterProvider, loggerProvider, noopOnClose}
 }
 
 // NewStdoutClient creates a new Client with exporters which send telemetry data to standard output

pkg/capabilities/capabilities.go

@@ -11,6 +11,7 @@ import (
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/anypb"
 
+	"github.com/smartcontractkit/chainlink-common/pkg/contexts"
 	"github.com/smartcontractkit/chainlink-protos/cre/go/values"
 )
 
@@ -108,13 +109,27 @@ type RequestMetadata struct {
 	WorkflowTag string
 }
 
+func (m *RequestMetadata) ContextWithCRE(ctx context.Context) context.Context {
+	return contexts.WithCRE(ctx, contexts.CRE{
+		Owner:    m.WorkflowOwner,
+		Workflow: m.WorkflowID,
+	})
+}
+
 type RegistrationMetadata struct {
 	WorkflowID    string
 	WorkflowOwner string
 	// The step reference ID of the workflow
 	ReferenceID string
 }
 
+func (m *RegistrationMetadata) ContextWithCRE(ctx context.Context) context.Context {
+	return contexts.WithCRE(ctx, contexts.CRE{
+		Owner:    m.WorkflowOwner,
+		Workflow: m.WorkflowID,
+	})
+}
+
 // CapabilityRequest is a struct for the Execute request of a capability.
 type CapabilityRequest struct {
 	Metadata RequestMetadata

pkg/capabilities/consensus/ocr3/aggregators/reduce_aggregator.go

@@ -306,24 +306,33 @@ func (a *reduceAggregator) extractValues(lggr logger.Logger, observations map[oc
 		// values are then re-wrapped here to handle aggregating against Value types
 		// which is used for mode aggregation
 		switch val := val.(type) {
-		case map[string]interface{}:
+		case map[string]any:
 			_, ok := val[aggregationKey]
 			if !ok {
 				continue
 			}
+			if val[aggregationKey] == nil {
+				lggr.Warnf("node %d contributed with a nil value under key %s", nodeID, aggregationKey)
+				continue
+			}
 
 			rewrapped, err := values.Wrap(val[aggregationKey])
 			if err != nil {
 				lggr.Warnf("unable to wrap value %s", val[aggregationKey])
 				continue
 			}
 			vals = append(vals, rewrapped)
-		case []interface{}:
+		case []any:
 			i, err := strconv.Atoi(aggregationKey)
 			if err != nil {
 				lggr.Warnf("aggregation key %s could not be used to index a list type", aggregationKey)
 				continue
 			}
+			if i >= len(val) {
+				lggr.Warnf("node %d contributed with an array shorter than index %s", nodeID, aggregationKey)
+				continue
+			}
+
 			rewrapped, err := values.Wrap(val[i])
 			if err != nil {
 				lggr.Warnf("unable to wrap value %s", val[i])