diff --git a/go.mod b/go.mod index b03f35e4fd..e12bf614df 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 773442329f..a811d57488 100644 --- a/go.sum +++ b/go.sum @@ -258,8 +258,8 @@ github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME= github.com/smartcontractkit/chain-selectors v1.0.100 h1:wpiSpmI/eFjY+wx/nPr5VuNF4hki0prIBMKEaQWn3g4= github.com/smartcontractkit/chain-selectors v1.0.100/go.mod h1:qy7whtgG5g+7z0jt0nRyii9bLND9m15NZTzuQPkMZ5w= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260724142814-45996a1bcb72 h1:uWEwl7i2ryuRVoV4DmIKm6mqYevf1lH/8cQYhw/JXko= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260724142814-45996a1bcb72/go.mod h1:UYcRMb4dZcoaIPgZJ3hckCySTqtJc9K4Q+tOKErwTq0= +github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260727152657-992a2cd2ec36 h1:p+jHKdHfmlyA7HLcSF25IvAN4DuEKVpuxWCLVIHW4q4= +github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260727152657-992a2cd2ec36/go.mod h1:n7wTMqh5BGLDcL5XFLeWaAvf3PRqfRu7G1Uig07YrVc= github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20251024234028-0988426d98f4 h1:GCzrxDWn3b7jFfEA+WiYRi8CKoegsayiDoJBCjYkneE= github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20251024234028-0988426d98f4/go.mod h1:HHGeDUpAsPa0pmOx7wrByCitjQ0mbUxf0R9v+g67uCA= github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260622152157-c8e129347b8b h1:VDgJWDipihV9f7M5+d21d1RzSsg5rEv+iI12oN1VQbo= diff --git a/pkg/beholder/batch_emitter_service_test.go b/pkg/beholder/batch_emitter_service_test.go index 5a405a6b9b..085ac94aa7 100644 --- a/pkg/beholder/batch_emitter_service_test.go +++ b/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) diff --git a/pkg/beholder/chip_ingress_emitter.go b/pkg/beholder/chip_ingress_emitter.go index 4bca08b47f..fa52953dd8 100644 --- a/pkg/beholder/chip_ingress_emitter.go +++ b/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 } diff --git a/pkg/beholder/client.go b/pkg/beholder/client.go index 1bd792a8f9..a4fe75ae05 100644 --- a/pkg/beholder/client.go +++ b/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 diff --git a/pkg/beholder/client_test.go b/pkg/beholder/client_test.go index 48dca5efff..35e7c843d5 100644 --- a/pkg/beholder/client_test.go +++ b/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{ diff --git a/pkg/beholder/resource_attributes.go b/pkg/beholder/resource_attributes.go new file mode 100644 index 0000000000..7888fd12ab --- /dev/null +++ b/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 +} diff --git a/pkg/beholder/resource_attributes_test.go b/pkg/beholder/resource_attributes_test.go new file mode 100644 index 0000000000..35dceccdd8 --- /dev/null +++ b/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)) +}