beholder: send resource attributes to chip ingress as gRPC metadata

go.mod

@@ -43,7 +43,7 @@ require (
 	github.com/scylladb/go-reflectx v1.0.1
 	github.com/shopspring/decimal v1.4.0
 	github.com/smartcontractkit/chain-selectors v1.0.100
-	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260724142814-45996a1bcb72
+	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260727152657-992a2cd2ec36
 	github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20251024234028-0988426d98f4
 	github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260622152157-c8e129347b8b
 	github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20251002192024-d2ad9222409b

pkg/beholder/batch_emitter_service_test.go

@@ -620,7 +620,6 @@ func TestChipIngressBatchEmitterService_RPCError(t *testing.T) {
 	})
 }
 
-
 func TestChipIngressBatchEmitterService_Metrics(t *testing.T) {
 	t.Run("records events_sent on successful publish", func(t *testing.T) {
 		reader, restore := useEmitterTestMeterProvider(t)

pkg/beholder/chip_ingress_emitter.go

@@ -14,6 +14,10 @@ import (
 
 // ChipIngressEmitter wraps a synchronous chipingress.Client.Publish call
 // in a fire-and-forget goroutine so callers are never blocked.
+//
+// Resource attributes are not stamped on events here. They describe the producer rather than any
+// individual event, so they travel once per request as gRPC metadata configured on the client (see
+// chipingress.WithResourceAttributeHeaders) rather than being repeated on every event.
 type ChipIngressEmitter struct {
 	client chipingress.Client
 	lggr   logger.Logger
@@ -43,7 +47,7 @@ func (c ChipIngressEmitterConfig) New(client chipingress.Client) (Emitter, error
 
 	return &ChipIngressEmitter{
 		client: client,
-		lggr:    lggr,
+		lggr:   lggr,
 		stopCh: make(services.StopChan),
 	}, nil
 }

pkg/beholder/client.go

@@ -191,6 +191,7 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (*Client, erro
 	// eventually we will remove the dual source emitter and just use chip ingress
 	if cfg.ChipIngressEmitterEnabled || cfg.ChipIngressEmitterGRPCEndpoint != "" {
 		var opts []chipingress.Opt
+		resourceAttrs := resourceAttributesToStringMap(cfg.ResourceAttributes)
 
 		if cfg.ChipIngressInsecureConnection {
 			opts = append(opts, chipingress.WithInsecureConnection())
@@ -215,6 +216,10 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (*Client, erro
 		opts = append(opts, chipingress.WithMeterProvider(meterProvider))
 		opts = append(opts, chipingress.WithTracerProvider(tracerProvider))
 
+		if len(resourceAttrs) > 0 {
+			opts = append(opts, chipingress.WithResourceAttributeHeaders(resourceAttrs))
+		}
+
 		chipIngressClient, err = chipingress.NewClient(cfg.ChipIngressEmitterGRPCEndpoint, opts...)
 		if err != nil {
 			return nil, err

pkg/beholder/client_test.go

@@ -6,18 +6,24 @@ import (
 	"encoding/hex"
 	"errors"
 	"fmt"
+	"net"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
+	cepb "github.com/cloudevents/sdk-go/binding/format/protobuf/v2/pb"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc"
 	"go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp"
 	otellog "go.opentelemetry.io/otel/log"
 	sdklog "go.opentelemetry.io/otel/sdk/log"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/metadata"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/beholder"
 	"github.com/smartcontractkit/chainlink-common/pkg/beholder/internal/mocks"
@@ -486,6 +492,93 @@ func TestNewGRPCClient_ChipIngressEmitter(t *testing.T) {
 	})
 }
 
+// capturingChipServer records the gRPC metadata of the last Publish it handles.
+type capturingChipServer struct {
+	pb.UnimplementedChipIngressServer
+
+	mu     sync.Mutex
+	lastMD metadata.MD
+}
+
+func (s *capturingChipServer) Publish(ctx context.Context, _ *cepb.CloudEvent) (*pb.PublishResponse, error) {
+	md, _ := metadata.FromIncomingContext(ctx)
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.lastMD = md
+	return &pb.PublishResponse{}, nil
+}
+
+func (s *capturingChipServer) metadata() metadata.MD {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.lastMD
+}
+
+// TestNewGRPCClient_AuthHeaderCoexistsWithResourceAttributes is the beholder-level counterpart to
+// chipingress' TestClient_AuthHeaderCoexistsWithResourceAttributes. Wiring resource attributes
+// added a unary header interceptor to a connection that previously carried no context metadata at
+// all, while the CSA node auth token travels separately as per-RPC credentials. This asserts on a
+// real connection that configuring both leaves the auth token intact and delivers the resource
+// attributes alongside it.
+func TestNewGRPCClient_AuthHeaderCoexistsWithResourceAttributes(t *testing.T) {
+	const authHeaderKey = "X-Beholder-Node-Auth-Token"
+	const authToken = "1:abc:2:def"
+
+	lis, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	defer lis.Close()
+
+	srv := grpc.NewServer()
+	capture := &capturingChipServer{}
+	pb.RegisterChipIngressServer(srv, capture)
+	go func() { _ = srv.Serve(lis) }()
+	defer srv.Stop()
+
+	cfg := beholder.Config{
+		OtelExporterGRPCEndpoint:       "localhost:4317",
+		ChipIngressEmitterEnabled:      true,
+		ChipIngressEmitterGRPCEndpoint: lis.Addr().String(),
+		ChipIngressInsecureConnection:  true,
+		AuthHeaders:                    map[string]string{authHeaderKey: authToken},
+		ResourceAttributes: []attribute.KeyValue{
+			attribute.String("csa_public_key", "abc123"),
+			attribute.String("service.name", "chainlink"),
+		},
+	}
+
+	otlploggrpcNew := func(options ...otlploggrpc.Option) (sdklog.Exporter, error) {
+		return &mockLogExporter{}, nil
+	}
+
+	client, err := beholder.NewGRPCClient(cfg, otlploggrpcNew)
+	require.NoError(t, err)
+	require.NotNil(t, client)
+
+	require.NoError(t, client.Emitter.Emit(t.Context(), []byte("payload"),
+		beholder.AttrKeyDomain, "my-domain",
+		beholder.AttrKeyEntity, "my-entity",
+		beholder.AttrKeyDataSchema, "/schemas/ids/1001",
+	))
+
+	// ChipIngressEmitter.Emit publishes fire-and-forget in a goroutine.
+	require.Eventually(t, func() bool { return capture.metadata() != nil }, 5*time.Second, 10*time.Millisecond)
+
+	md := capture.metadata()
+	assert.Equal(t, []string{authToken}, md.Get(authHeaderKey),
+		"the CSA auth token must arrive exactly once, unmodified")
+
+	// Assert against the sanitizer rather than hardcoding key spellings: the property under test
+	// is that auth and resource attributes coexist, not how chipingress normalizes a key.
+	want := chipingress.SanitizeMetadataHeaders(map[string]string{
+		"csa_public_key": "abc123",
+		"service.name":   "chainlink",
+	})
+	require.Len(t, want, 2, "both attributes must survive sanitization for this test to mean anything")
+	for key, val := range want {
+		assert.Equal(t, []string{val}, md.Get(key), "resource attribute %q missing from metadata", key)
+	}
+}
+
 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{

pkg/beholder/resource_attributes.go

@@ -0,0 +1,15 @@
+package beholder
+
+import "go.opentelemetry.io/otel/attribute"
+
+// resourceAttributesToStringMap converts OTel resource attributes into a plain string map,
+// using attribute.Value.Emit for canonical stringification of any value type. This is the
+// single source of truth used to derive both the gRPC metadata headers and the CloudEvent
+// extension keys/values sent to ChipIngress, so both mechanisms stay consistent.
+func resourceAttributesToStringMap(attrs []attribute.KeyValue) map[string]string {
+	m := make(map[string]string, len(attrs))
+	for _, kv := range attrs {
+		m[string(kv.Key)] = kv.Value.Emit()
+	}
+	return m
+}

pkg/beholder/resource_attributes_test.go

@@ -0,0 +1,28 @@
+package beholder
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"go.opentelemetry.io/otel/attribute"
+)
+
+func TestResourceAttributesToStringMap(t *testing.T) {
+	attrs := []attribute.KeyValue{
+		attribute.String("chain_id", "1"),
+		attribute.Bool("is_bootstrap", true),
+		attribute.Int64("node_index", 42),
+	}
+
+	got := resourceAttributesToStringMap(attrs)
+
+	assert.Equal(t, map[string]string{
+		"chain_id":     "1",
+		"is_bootstrap": "true",
+		"node_index":   "42",
+	}, got)
+}
+
+func TestResourceAttributesToStringMap_Empty(t *testing.T) {
+	assert.Empty(t, resourceAttributesToStringMap(nil))
+}