Merge branch 'main' into update-ocr3-metadata

Author: silaslenihan

go.mod

@@ -35,7 +35,7 @@ require (
 	github.com/santhosh-tekuri/jsonschema/v5 v5.2.0
 	github.com/scylladb/go-reflectx v1.0.1
 	github.com/shopspring/decimal v1.4.0
-	github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20250604191313-8f161cf29ad5
+	github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20250612182447-1c32d2efe48f
 	github.com/smartcontractkit/freeport v0.1.1
 	github.com/smartcontractkit/grpc-proxy v0.0.0-20240830132753-a7e17fec5ab7
 	github.com/smartcontractkit/libocr v0.0.0-20250220133800-f3b940c4f298

go.sum

@@ -296,8 +296,8 @@ github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp
 github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME=
 github.com/sirupsen/logrus v1.4.1/go.mod h1:ni0Sbl8bgC9z8RoU9G6nDWqqs/fq4eDPysMBDgk/93Q=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
-github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20250604191313-8f161cf29ad5 h1:Qrz1cG/myK+EUEz4aLfz1pS4cODit7v61BuXq3MDifk=
-github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20250604191313-8f161cf29ad5/go.mod h1:HHGeDUpAsPa0pmOx7wrByCitjQ0mbUxf0R9v+g67uCA=
+github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20250612182447-1c32d2efe48f h1:OYOifWMKVL54uM+sxrY7oPVTZptTXK45tA8dXN5nZSE=
+github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20250612182447-1c32d2efe48f/go.mod h1:HHGeDUpAsPa0pmOx7wrByCitjQ0mbUxf0R9v+g67uCA=
 github.com/smartcontractkit/chainlink-protos/workflows/go v0.0.0-20250430163438-97d324ef9061 h1:5BKk6j2QWmb5TFoWYVLuL8U2XUIzTUUo5HBkCHTX2kM=
 github.com/smartcontractkit/chainlink-protos/workflows/go v0.0.0-20250430163438-97d324ef9061/go.mod h1:HIpGvF6nKCdtZ30xhdkKWGM9+4Z4CVqJH8ZBL1FTEiY=
 github.com/smartcontractkit/freeport v0.1.1 h1:B5fhEtmgomdIhw03uPVbVTP6oPv27fBhZsoZZMSIS8I=

pkg/beholder/beholdertest/beholder.go

@@ -0,0 +1,192 @@
+package beholdertest
+
+import (
+	"context"
+	"fmt"
+	"reflect"
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	otellognoop "go.opentelemetry.io/otel/log/noop"
+	otelmetricnoop "go.opentelemetry.io/otel/metric/noop"
+	oteltracenoop "go.opentelemetry.io/otel/trace/noop"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/beholder"
+	"github.com/smartcontractkit/chainlink-common/pkg/beholder/pb"
+)
+
+const (
+	packageNameBeholder = "test_beholder"
+)
+
+// Observer is a test helper that provides assertion methods on received messages within the beholder client.
+type Observer struct {
+	emitter *assertMessageEmitter
+}
+
+// Len returns the total number of messages received that match the provided attribute key/value pairs.
+func (o Observer) Len(t *testing.T, attrKVs ...any) int {
+	t.Helper()
+
+	found := o.msgsForKVs(t, attrKVs...)
+
+	return len(found)
+}
+
+// Messages returns messages matching the provided keys and values.
+func (o Observer) Messages(t *testing.T, attrKVs ...any) []beholder.Message {
+	t.Helper()
+
+	if attrKVs == nil {
+		return o.emitter.msgs
+	}
+
+	return o.msgsForKVs(t, attrKVs...)
+}
+func (o Observer) msgsForKVs(t *testing.T, attrKVs ...any) []beholder.Message {
+	t.Helper()
+
+	o.emitter.mu.RLock()
+	defer o.emitter.mu.RUnlock()
+
+	found := []beholder.Message{}
+
+	for _, eMsg := range o.emitter.msgs {
+		var i, j int
+
+		for i < len(attrKVs)-1 {
+			j = i + 1
+
+			key, ok := attrKVs[i].(string)
+			require.True(t, ok)
+
+			value := attrKVs[j]
+			val, ok := eMsg.Attrs[key]
+
+			if ok && reflect.DeepEqual(value, val) {
+				found = append(found, eMsg)
+			}
+
+			i = i + 2
+		}
+	}
+
+	return found
+}
+
+func (o Observer) BaseMessagesForLabels(t *testing.T, labels map[string]string) ([]*pb.BaseMessage, error) {
+	t.Helper()
+
+	o.emitter.mu.RLock()
+	defer o.emitter.mu.RUnlock()
+
+	var found []*pb.BaseMessage
+
+messageLoop:
+	for _, eMsg := range o.emitter.msgs {
+		dataSchema, ok := eMsg.Attrs["beholder_entity"].(string)
+		if !ok {
+			continue
+		}
+
+		if dataSchema != "BaseMessage" {
+			continue
+		}
+
+		payload := pb.BaseMessage{}
+		err := proto.Unmarshal(eMsg.Body, &payload)
+		if err != nil {
+			return nil, fmt.Errorf("error unmarshalling base message: %v", err)
+		}
+
+		for k, v := range labels {
+			if payload.Labels[k] != v {
+				continue messageLoop
+			}
+		}
+
+		found = append(found, &payload)
+	}
+
+	return found, nil
+}
+
+// NewObserver sets the global beholder client as a message collector and returns an Observer that provides helper assertion
+// functions on received messages.
+//
+// NewObserver affects the whole process, it cannot be used in parallel tests or tests with parallel ancestors.
+func NewObserver(t *testing.T) Observer {
+	t.Helper()
+
+	cfg := beholder.DefaultConfig()
+
+	// Logger
+	loggerProvider := otellognoop.NewLoggerProvider()
+	logger := loggerProvider.Logger(packageNameBeholder)
+
+	// Tracer
+	tracerProvider := oteltracenoop.NewTracerProvider()
+	tracer := tracerProvider.Tracer(packageNameBeholder)
+
+	// Meter
+	meterProvider := otelmetricnoop.NewMeterProvider()
+	meter := meterProvider.Meter(packageNameBeholder)
+
+	// MessageEmitter
+	messageEmitter := &assertMessageEmitter{t: t}
+
+	client := &beholder.Client{
+		Config:                cfg,
+		Logger:                logger,
+		Tracer:                tracer,
+		Meter:                 meter,
+		Emitter:               messageEmitter,
+		LoggerProvider:        loggerProvider,
+		TracerProvider:        tracerProvider,
+		MeterProvider:         meterProvider,
+		MessageLoggerProvider: loggerProvider,
+		OnClose:               func() error { return nil },
+	}
+
+	//reset NewObserver state after the test
+	prevClient := beholder.GetClient()
+	t.Cleanup(func() {
+		beholder.SetClient(prevClient)
+		t.Setenv(packageNameBeholder, packageNameBeholder)
+	})
+	beholder.SetClient(client)
+
+	return Observer{emitter: messageEmitter}
+}
+
+// assertMessageEmitter is implemented with the same interface as the noopMessageEmitter in pkg/beholder/noop.go
+// it is unknown at this time whether EmitMessage is needed, but it exists in the case that it is needed
+type assertMessageEmitter struct {
+	t    *testing.T
+	mu   sync.RWMutex
+	msgs []beholder.Message
+}
+
+func (e *assertMessageEmitter) Emit(_ context.Context, msg []byte, attrKVs ...any) error {
+	e.t.Helper()
+
+	e.mu.Lock()
+	defer e.mu.Unlock()
+
+	e.msgs = append(e.msgs, beholder.NewMessage(msg, attrKVs...))
+
+	return nil
+}
+
+func (e *assertMessageEmitter) EmitMessage(_ context.Context, msg beholder.Message) error {
+	e.t.Helper()
+
+	e.mu.Lock()
+	defer e.mu.Unlock()
+
+	e.msgs = append(e.msgs, msg)
+
+	return nil
+}

pkg/beholder/client.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/chipingress"
 	"go.opentelemetry.io/otel/attribute"
@@ -132,6 +133,7 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (*Client, erro
 	} else {
 		loggerProcessor = sdklog.NewSimpleProcessor(sharedLogExporter)
 	}
+
 	loggerAttributes := []attribute.KeyValue{
 		attribute.String(AttrKeyDataType, "zap_log_message"),
 	}
@@ -146,6 +148,12 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (*Client, erro
 		sdklog.WithResource(loggerResource),
 		sdklog.WithProcessor(loggerProcessor),
 	)
+
+	// If log streaming is disabled, use a noop logger provider
+	if !cfg.LogStreamingEnabled {
+		loggerProvider = BeholderNoopLoggerProvider()
+	}
+
 	logger := loggerProvider.Logger(defaultPackageName)
 
 	// Tracer
@@ -211,21 +219,29 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (*Client, erro
 	// eventually we will remove the dual source emitter and just use chip ingress
 
 	if cfg.ChipIngressEmitterEnabled {
-
-		// Create a header provider that implements the chipingress.HeaderProvider interface
-
-		chipIngressOpts := []chipingress.Opt{
-			chipingress.WithTransportCredentials(creds),
+		chipIngressOpts := make([]chipingress.Opt, 0, 2)
+		chipIngressEndpoint := cfg.ChipIngressEmitterGRPCEndpoint
+		chipIngressHost, err := getHost(chipIngressEndpoint)
+		if err != nil {
+			return nil, fmt.Errorf("failed to extract host from address '%s': %w", chipIngressEndpoint, err)
+		}
+		// Set Authority
+		chipIngressOpts = append(chipIngressOpts, chipingress.WithAuthority(chipIngressHost))
+
+		if cfg.ChipIngressInsecureConnection {
+			// Use insecure credentials when TLS is not required
+			chipIngressOpts = append(chipIngressOpts, chipingress.WithInsecureConnection())
+		} else {
+			chipIngressOpts = append(chipIngressOpts, chipingress.WithTLSAndHTTP2(chipIngressHost))
 		}
-
 		// Only add headers if they exist
 		if len(cfg.AuthHeaders) > 0 {
 			headerProvider := NewStaticAuthHeaderProvider(cfg.AuthHeaders)
 			chipIngressOpts = append(chipIngressOpts, chipingress.WithHeaderProvider(headerProvider))
 		}
 
 		chipIngressClient, err := chipingress.NewChipIngressClient(
-			cfg.ChipIngressEmitterGRPCEndpoint,
+			chipIngressEndpoint,
 			chipIngressOpts...,
 		)
 
@@ -409,3 +425,11 @@ func newMeterProvider(config Config, resource *sdkresource.Resource, creds crede
 	)
 	return mp, nil
 }
+
+func getHost(address string) (string, error) {
+	host, _, err := net.SplitHostPort(address)
+	if err != nil {
+		return "", err
+	}
+	return host, nil
+}