diff --git a/go.mod b/go.mod index 1299440695..3ca141bb2b 100644 --- a/go.mod +++ b/go.mod @@ -158,3 +158,5 @@ require ( google.golang.org/genproto/googleapis/api v0.0.0-20260401024825-9d38bb4040a9 // indirect gopkg.in/yaml.v2 v2.4.0 // indirect ) + +replace github.com/smartcontractkit/chainlink-common/pkg/chipingress => ./pkg/chipingress diff --git a/go.sum b/go.sum index 3441393f7f..4c0ec03690 100644 --- a/go.sum +++ b/go.sum @@ -258,8 +258,6 @@ 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.20260626151909-052e55e62e62 h1:o7vfwNQjQbMKQ9YsZFQOxvU7RMXD/wKnZsX5N9sDS3w= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260626151909-052e55e62e62/go.mod h1:HmUyH2oD9m+GRpKq7q3vuRnm1F2Uczf/Nd1v3ipMSK8= 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.go b/pkg/beholder/batch_emitter_service.go index a81fd82bc9..4a01d8d2b8 100644 --- a/pkg/beholder/batch_emitter_service.go +++ b/pkg/beholder/batch_emitter_service.go @@ -21,7 +21,8 @@ type ChipIngressBatchEmitterService struct { services.Service eng *services.Engine - batchClient *batch.Client + batchClient *batch.Client + resourceAttrs map[string]string metricAttrsCache sync.Map // map[string]otelmetric.MeasurementOption metrics batchEmitterMetrics @@ -84,8 +85,9 @@ func NewChipIngressBatchEmitterService(client chipingress.Client, cfg Config, lg } e := &ChipIngressBatchEmitterService{ - batchClient: batchClient, - metrics: metrics, + batchClient: batchClient, + resourceAttrs: resourceAttributesToStringMap(cfg.ResourceAttributes), + metrics: metrics, } e.Service, e.eng = services.Config{ @@ -136,7 +138,7 @@ func (e *ChipIngressBatchEmitterService) emitInternal(ctx context.Context, body attributes := newAttributes(attrKVs...) - event, err := chipingress.NewEvent(domain, entity, body, attributes) + event, err := chipingress.NewEventWithOpts(domain, entity, body, attributes, chipingress.WithResourceAttributeExtensions(e.resourceAttrs)) if err != nil { return fmt.Errorf("failed to create CloudEvent: %w", err) } diff --git a/pkg/beholder/batch_emitter_service_test.go b/pkg/beholder/batch_emitter_service_test.go index 4137d450d5..7d8863b87b 100644 --- a/pkg/beholder/batch_emitter_service_test.go +++ b/pkg/beholder/batch_emitter_service_test.go @@ -121,6 +121,50 @@ func TestChipIngressBatchEmitterService_Emit(t *testing.T) { }) } +func TestChipIngressBatchEmitterService_ResourceAttributes(t *testing.T) { + clientMock := mocks.NewClient(t) + clientMock.EXPECT().Close().Return(nil).Maybe() + + var mu sync.Mutex + var receivedBatch *chipingress.CloudEventBatch + clientMock. + On("PublishBatch", mock.Anything, mock.Anything). + Run(func(args mock.Arguments) { + mu.Lock() + defer mu.Unlock() + receivedBatch = args.Get(1).(*chipingress.CloudEventBatch) + }). + Return(nil, nil) + + cfg := newTestConfig() + cfg.ChipIngressSendInterval = 50 * time.Millisecond + cfg.ResourceAttributes = []attribute.KeyValue{attribute.String("chain_id", "1")} + + emitter, err := beholder.NewChipIngressBatchEmitterService(clientMock, cfg, newTestLogger(t)) + require.NoError(t, err) + require.NoError(t, emitter.Start(t.Context())) + + err = emitter.Emit(t.Context(), []byte("test-payload"), + beholder.AttrKeyDomain, "my-domain", + beholder.AttrKeyEntity, "my-entity", + ) + require.NoError(t, err) + + assert.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return receivedBatch != nil + }, 2*time.Second, 10*time.Millisecond) + + require.NoError(t, emitter.Close()) + + mu.Lock() + defer mu.Unlock() + require.Len(t, receivedBatch.Events, 1) + require.NotNil(t, receivedBatch.Events[0].Attributes["chainid"]) + assert.Equal(t, "1", receivedBatch.Events[0].Attributes["chainid"].GetCeString()) +} + func TestChipIngressBatchEmitterService_CloudEventFormat(t *testing.T) { clientMock := mocks.NewClient(t) clientMock.EXPECT().Close().Return(nil).Maybe() diff --git a/pkg/beholder/chip_ingress_emitter.go b/pkg/beholder/chip_ingress_emitter.go index 4bca08b47f..f8525b4eeb 100644 --- a/pkg/beholder/chip_ingress_emitter.go +++ b/pkg/beholder/chip_ingress_emitter.go @@ -12,14 +12,24 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/services" ) +// resourceAttrExtensions holds resource attributes to stamp as CloudEvent extensions on every +// emitted event. It is stored behind a pointer on ChipIngressEmitter (rather than as a bare map +// field) so the struct itself stays a comparable type — a map field would make it incomparable, +// which is an exported-API-breaking change per apidiff. A nil *resourceAttrExtensions means no +// resource attributes are configured. +type resourceAttrExtensions struct { + attrs map[string]string +} + // ChipIngressEmitter wraps a synchronous chipingress.Client.Publish call // in a fire-and-forget goroutine so callers are never blocked. type ChipIngressEmitter struct { - client chipingress.Client - lggr logger.Logger - stopCh services.StopChan - wg services.WaitGroup - closed atomic.Bool + client chipingress.Client + lggr logger.Logger + resourceAttrs *resourceAttrExtensions + stopCh services.StopChan + wg services.WaitGroup + closed atomic.Bool } func NewChipIngressEmitter(client chipingress.Client) (Emitter, error) { @@ -31,8 +41,15 @@ type ChipIngressEmitterConfig struct { Lggr logger.Logger } -// New creates a ChipIngressEmitter from the config. +// New creates a ChipIngressEmitter from the config, with no resource attributes configured. func (c ChipIngressEmitterConfig) New(client chipingress.Client) (Emitter, error) { + return c.NewWithResourceAttributes(client, nil) +} + +// NewWithResourceAttributes creates a ChipIngressEmitter from the config, additionally stamping +// attrs as CloudEvent extensions (via chipingress.WithResourceAttributeExtensions) on every +// emitted event. +func (c ChipIngressEmitterConfig) NewWithResourceAttributes(client chipingress.Client, attrs map[string]string) (Emitter, error) { if client == nil { return nil, errors.New("chip ingress client is nil") } @@ -41,10 +58,16 @@ func (c ChipIngressEmitterConfig) New(client chipingress.Client) (Emitter, error lggr = logger.Nop() } + var resourceAttrs *resourceAttrExtensions + if len(attrs) > 0 { + resourceAttrs = &resourceAttrExtensions{attrs: attrs} + } + return &ChipIngressEmitter{ - client: client, - lggr: lggr, - stopCh: make(services.StopChan), + client: client, + lggr: lggr, + resourceAttrs: resourceAttrs, + stopCh: make(services.StopChan), }, nil } @@ -65,7 +88,12 @@ func (c *ChipIngressEmitter) Emit(ctx context.Context, body []byte, attrKVs ...a return err } - event, err := chipingress.NewEvent(sourceDomain, entityType, body, newAttributes(attrKVs...)) + var event chipingress.CloudEvent + if c.resourceAttrs != nil { + event, err = chipingress.NewEventWithOpts(sourceDomain, entityType, body, newAttributes(attrKVs...), chipingress.WithResourceAttributeExtensions(c.resourceAttrs.attrs)) + } else { + event, err = chipingress.NewEvent(sourceDomain, entityType, body, newAttributes(attrKVs...)) + } if err != nil { return err } diff --git a/pkg/beholder/chip_ingress_emitter_test.go b/pkg/beholder/chip_ingress_emitter_test.go index 11349d335b..0590befd5d 100644 --- a/pkg/beholder/chip_ingress_emitter_test.go +++ b/pkg/beholder/chip_ingress_emitter_test.go @@ -9,6 +9,7 @@ import ( "github.com/stretchr/testify/require" "github.com/smartcontractkit/chainlink-common/pkg/beholder" + "github.com/smartcontractkit/chainlink-common/pkg/chipingress" "github.com/smartcontractkit/chainlink-common/pkg/chipingress/mocks" "github.com/smartcontractkit/chainlink-common/pkg/logger" ) @@ -73,6 +74,59 @@ func TestChipIngressEmit(t *testing.T) { assert.Error(t, err) }) + t.Run("resource attributes are stamped as sanitized CE extensions", func(t *testing.T) { + clientMock := mocks.NewClient(t) + + var published *chipingress.CloudEventPb + clientMock. + On("Publish", mock.Anything, mock.Anything). + Run(func(args mock.Arguments) { + published = args.Get(1).(*chipingress.CloudEventPb) + }). + Return(nil, nil) + clientMock.On("Close").Return(nil) + + emitter, err := beholder.ChipIngressEmitterConfig{ + Lggr: logger.Test(t), + }.NewWithResourceAttributes(clientMock, map[string]string{"chain_id": "1"}) + require.NoError(t, err) + + err = emitter.Emit(t.Context(), body, beholder.AttrKeyDomain, domain, beholder.AttrKeyEntity, entity) + require.NoError(t, err) + + require.NoError(t, emitter.Close()) + clientMock.AssertExpectations(t) + + require.NotNil(t, published) + require.NotNil(t, published.Attributes["chainid"]) + assert.Equal(t, "1", published.Attributes["chainid"].GetCeString()) + }) + + t.Run("New(client) with no resource attributes stamps no extra extensions (backward-compat)", func(t *testing.T) { + clientMock := mocks.NewClient(t) + + var published *chipingress.CloudEventPb + clientMock. + On("Publish", mock.Anything, mock.Anything). + Run(func(args mock.Arguments) { + published = args.Get(1).(*chipingress.CloudEventPb) + }). + Return(nil, nil) + clientMock.On("Close").Return(nil) + + emitter, err := beholder.ChipIngressEmitterConfig{Lggr: logger.Test(t)}.New(clientMock) + require.NoError(t, err) + + err = emitter.Emit(t.Context(), body, beholder.AttrKeyDomain, domain, beholder.AttrKeyEntity, entity) + require.NoError(t, err) + + require.NoError(t, emitter.Close()) + clientMock.AssertExpectations(t) + + require.NotNil(t, published) + assert.Nil(t, published.Attributes["chainid"]) + }) + t.Run("logs error when Publish fails", func(t *testing.T) { clientMock := mocks.NewClient(t) diff --git a/pkg/beholder/client.go b/pkg/beholder/client.go index cf66c39306..9eac29302b 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.WithHeaderProvider(chipingress.NewStaticHeaderProvider(chipingress.SanitizeMetadataHeaders(resourceAttrs)))) + } + chipIngressClient, err = chipingress.NewClient(cfg.ChipIngressEmitterGRPCEndpoint, opts...) if err != nil { return nil, err @@ -235,7 +240,7 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (*Client, erro // teardown after parent close hook completes. chipIngressEmitter = noCloseEmitter{Emitter: batchEmitterService} } else { - chipIngressEmitter, err = ChipIngressEmitterConfig{Lggr: lggr}.New(chipIngressClient) + chipIngressEmitter, err = ChipIngressEmitterConfig{Lggr: lggr}.NewWithResourceAttributes(chipIngressClient, resourceAttrs) if err != nil { return nil, fmt.Errorf("failed to create chip ingress emitter: %w", err) } 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)) +} diff --git a/pkg/chipingress/client.go b/pkg/chipingress/client.go index ff244bde6c..5c7e421c9e 100644 --- a/pkg/chipingress/client.go +++ b/pkg/chipingress/client.go @@ -5,6 +5,8 @@ import ( "crypto/tls" "fmt" "net" + "sort" + "strings" "time" "github.com/google/uuid" @@ -112,14 +114,22 @@ func NewClient(address string, opts ...Opt) (Client, error) { if cfg.perRPCCredentials != nil { grpcOpts = append(grpcOpts, grpc.WithPerRPCCredentials(cfg.perRPCCredentials)) } - // Add headers as a unary interceptor, use for non-auth headers + // Add headers as unary interceptors, use for non-auth headers. + // WithChainUnaryInterceptor is used (rather than WithUnaryInterceptor) so that + // headerProvider and nopInfoHeaderProvider compose instead of the second call + // silently overriding the first (grpc.WithUnaryInterceptor is last-one-wins). + var unaryInterceptors []grpc.UnaryClientInterceptor if cfg.headerProvider != nil { - grpcOpts = append(grpcOpts, grpc.WithUnaryInterceptor(newHeaderInterceptor(cfg.headerProvider))) + unaryInterceptors = append(unaryInterceptors, newHeaderInterceptor(cfg.headerProvider)) // NOTE: not supporting streaming interceptors } if cfg.nopInfoHeaderProvider != nil { - grpcOpts = append(grpcOpts, grpc.WithUnaryInterceptor(newHeaderInterceptor(cfg.nopInfoHeaderProvider))) + unaryInterceptors = append(unaryInterceptors, newHeaderInterceptor(cfg.nopInfoHeaderProvider)) + } + + if len(unaryInterceptors) > 0 { + grpcOpts = append(grpcOpts, grpc.WithChainUnaryInterceptor(unaryInterceptors...)) } conn, err := grpc.NewClient(address, grpcOpts...) @@ -267,9 +277,60 @@ func newHeaderInterceptor(provider HeaderProvider) grpc.UnaryClientInterceptor { } } +// EventOpt configures a CloudEvent after its well-known attributes have been set by NewEvent. +type EventOpt func(*ce.Event) + +// SanitizeExtensionName lower-cases name and strips every rune outside [a-z0-9], the character +// set the CloudEvents spec requires for extension attribute names. +func SanitizeExtensionName(name string) string { + var b strings.Builder + for _, r := range strings.ToLower(name) { + if (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') { + b.WriteRune(r) + } + } + return b.String() +} + +// WithResourceAttributeExtensions returns an EventOpt that sets a CloudEvent extension for each +// entry in attrs, sanitizing keys via SanitizeExtensionName so they satisfy the CloudEvents +// extension-name character set. Entries that sanitize to an empty string, or that collide with a +// reserved extension name (see reservedExtensionNames), are skipped. Keys are applied in sorted +// order so that if two distinct keys sanitize to the same name, the result is deterministic. +func WithResourceAttributeExtensions(attrs map[string]string) EventOpt { + return func(event *ce.Event) { + keys := make([]string, 0, len(attrs)) + for k := range attrs { + keys = append(keys, k) + } + sort.Strings(keys) + + set := make(map[string]struct{}, len(attrs)) + for _, k := range keys { + name := SanitizeExtensionName(k) + if name == "" { + continue + } + if _, reserved := reservedExtensionNames[name]; reserved { + continue + } + if _, already := set[name]; already { + continue + } + set[name] = struct{}{} + event.SetExtension(name, attrs[k]) + } + } +} + // NewEvent creates a new CloudEvent with the specified domain, entity, payload, and optional attributes. func NewEvent(domain, entity string, payload []byte, attributes map[string]any) (CloudEvent, error) { + return NewEventWithOpts(domain, entity, payload, attributes) +} +// NewEventWithOpts creates a new CloudEvent like NewEvent, additionally applying opts (e.g. +// WithResourceAttributeExtensions) to the event before its data is set. +func NewEventWithOpts(domain, entity string, payload []byte, attributes map[string]any, opts ...EventOpt) (CloudEvent, error) { event := ce.NewEvent() event.SetSource(domain) event.SetType(entity) @@ -303,6 +364,10 @@ func NewEvent(domain, entity string, payload []byte, attributes map[string]any) event.SetExtension(IdempotencyKeyAttr, val) } + for _, opt := range opts { + opt(&event) + } + err := event.SetData(ceformat.ContentTypeProtobuf, payload) if err != nil { return ce.Event{}, fmt.Errorf("could not set data on event: %w", err) diff --git a/pkg/chipingress/client_test.go b/pkg/chipingress/client_test.go index bfd61e5ca1..9cb04f622d 100644 --- a/pkg/chipingress/client_test.go +++ b/pkg/chipingress/client_test.go @@ -3,6 +3,7 @@ package chipingress import ( "context" "fmt" + "net" "testing" "time" @@ -167,6 +168,89 @@ func TestNewEvent_IdempotencyKey(t *testing.T) { }) } +func TestSanitizeExtensionName(t *testing.T) { + tests := []struct { + name string + in string + want string + }{ + {name: "snake_case", in: "chain_id", want: "chainid"}, + {name: "dotted", in: "k8s.pod.name", want: "k8spodname"}, + {name: "already valid", in: "chainid", want: "chainid"}, + {name: "upper case is lowered", in: "ChainID", want: "chainid"}, + {name: "empty", in: "", want: ""}, + {name: "all invalid characters", in: "---...", want: ""}, + {name: "mixed valid and invalid", in: "Service-Name.1", want: "servicename1"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, SanitizeExtensionName(tt.in)) + }) + } +} + +func TestNewEventWithOpts_WithResourceAttributeExtensions(t *testing.T) { + payload := []byte("body") + + t.Run("sanitized keys/values land on the event", func(t *testing.T) { + attrs := map[string]string{"chain_id": "1", "k8s.pod.name": "pod-abc"} + event, err := NewEventWithOpts("domain", "entity", payload, nil, WithResourceAttributeExtensions(attrs)) + require.NoError(t, err) + ext := event.Extensions() + assert.Equal(t, "1", ext["chainid"]) + assert.Equal(t, "pod-abc", ext["k8spodname"]) + }) + + t.Run("empty sanitized name is dropped", func(t *testing.T) { + attrs := map[string]string{"---": "value"} + event, err := NewEventWithOpts("domain", "entity", payload, nil, WithResourceAttributeExtensions(attrs)) + require.NoError(t, err) + assert.Len(t, event.Extensions(), 1) // only the always-set recordedtime extension + }) + + t.Run("reserved name is skipped", func(t *testing.T) { + attrs := map[string]string{IdempotencyKeyAttr: "should-not-override", "subject": "should-not-override"} + event, err := NewEventWithOpts("domain", "entity", payload, map[string]any{IdempotencyKeyAttr: "real-key"}, WithResourceAttributeExtensions(attrs)) + require.NoError(t, err) + ext := event.Extensions() + assert.Equal(t, "real-key", ext[IdempotencyKeyAttr]) + assert.Empty(t, event.Subject()) + }) + + t.Run("duplicate sanitized names resolve deterministically to sorted-first key", func(t *testing.T) { + attrs := map[string]string{"service.name": "from-dotted", "service_name": "from-snake"} + event, err := NewEventWithOpts("domain", "entity", payload, nil, WithResourceAttributeExtensions(attrs)) + require.NoError(t, err) + // sorted order: "service.name" < "service_name" ('.' < '_' in ASCII), so the dotted key wins. + assert.Equal(t, "from-dotted", event.Extensions()["servicename"]) + }) + + t.Run("omitting all opts is a no-op", func(t *testing.T) { + event, err := NewEventWithOpts("domain", "entity", payload, nil) + require.NoError(t, err) + assert.Len(t, event.Extensions(), 1) // only the always-set recordedtime extension + }) +} + +// TestNewEvent_UnchangedSignature is a backward-compatibility guard: NewEvent's exported +// signature must stay exactly as it was before EventOpt/NewEventWithOpts were introduced, and +// must remain equivalent to calling NewEventWithOpts with no opts. +func TestNewEvent_UnchangedSignature(t *testing.T) { + payload := []byte("body") + attributes := map[string]any{"subject": "example-subject"} + + viaNewEvent, err := NewEvent("domain", "entity", payload, attributes) + require.NoError(t, err) + + viaNewEventWithOpts, err := NewEventWithOpts("domain", "entity", payload, attributes) + require.NoError(t, err) + + assert.Equal(t, viaNewEventWithOpts.Subject(), viaNewEvent.Subject()) + assert.Equal(t, viaNewEventWithOpts.Extensions()["recordedtime"].(ce.Timestamp).Time.Truncate(time.Second), + viaNewEvent.Extensions()["recordedtime"].(ce.Timestamp).Time.Truncate(time.Second)) + assert.Equal(t, viaNewEventWithOpts.Data(), viaNewEvent.Data()) +} + func TestEventToProto(t *testing.T) { // Create a test protobuf message testProto := pb.PingResponse{Message: "test message"} @@ -668,6 +752,49 @@ func (m *mockHeaderProvider) Headers(ctx context.Context) (map[string]string, er return m.headers, nil } +// capturingServer is a minimal ChipIngressServer that records the incoming gRPC metadata +// of the last request it handles. +type capturingServer struct { + pb.UnimplementedChipIngressServer + lastMD metadata.MD +} + +func (s *capturingServer) Ping(ctx context.Context, _ *pb.EmptyRequest) (*pb.PingResponse, error) { + s.lastMD, _ = metadata.FromIncomingContext(ctx) + return &pb.PingResponse{}, nil +} + +// TestClient_ChainedHeaderProviders is a regression test for the fix that switched from +// grpc.WithUnaryInterceptor (last-one-wins) to grpc.WithChainUnaryInterceptor: when both +// WithHeaderProvider and WithNOPLookup are configured, both providers' headers must reach +// the server, not just the one registered last. +func TestClient_ChainedHeaderProviders(t *testing.T) { + lis, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", "127.0.0.1:0") + require.NoError(t, err) + defer lis.Close() + + srv := gp.NewServer() + capture := &capturingServer{} + pb.RegisterChipIngressServer(srv, capture) + go func() { _ = srv.Serve(lis) }() + defer srv.Stop() + + client, err := NewClient(lis.Addr().String(), + WithInsecureConnection(), + WithHeaderProvider(&mockHeaderProvider{headers: map[string]string{"x-resource-attr": "chain-1"}}), + WithNOPLookup(), + ) + require.NoError(t, err) + defer client.Close() //nolint:errcheck + + _, err = client.Ping(t.Context(), &EmptyRequest{}) + require.NoError(t, err) + + require.NotNil(t, capture.lastMD) + assert.Equal(t, []string{"chain-1"}, capture.lastMD.Get("x-resource-attr")) + assert.Equal(t, []string{"true"}, capture.lastMD.Get("x-include-nop-info")) +} + func TestWithTLS(t *testing.T) { serverName := "example.com" config := defaultCfg diff --git a/pkg/chipingress/header_provider.go b/pkg/chipingress/header_provider.go index 9a4141f98f..984b815344 100644 --- a/pkg/chipingress/header_provider.go +++ b/pkg/chipingress/header_provider.go @@ -8,6 +8,7 @@ import ( "errors" "fmt" "maps" + "sort" "sync" "sync/atomic" "time" @@ -108,6 +109,74 @@ func newStaticHeaderProvider(headers map[string]string, requireTLS bool) HeaderP return &staticHeaderProvider{headers: headers, requireTLS: requireTLS} } +// NewStaticHeaderProvider returns a HeaderProvider that always returns the given headers, +// for use with WithHeaderProvider to attach fixed, non-auth gRPC metadata (e.g. resource +// attributes) to every request. +func NewStaticHeaderProvider(headers map[string]string) HeaderProvider { + return newStaticHeaderProvider(headers, false) +} + +// SanitizeMetadataValue replaces any byte outside the printable ASCII range [0x20-0x7E] +// with '?'. grpc-go hard-fails the entire RPC when an outgoing metadata value fails this +// check (unlike the CE-extension path, where an invalid entry is simply dropped), so +// values headed for gRPC metadata must be normalized before being sent. +func SanitizeMetadataValue(val string) string { + b := []byte(val) + out := make([]byte, len(b)) + for i, c := range b { + if c >= 0x20 && c <= 0x7E { + out[i] = c + } else { + out[i] = '?' + } + } + return string(out) +} + +// SanitizeMetadataHeaders sanitizes a map of resource-attribute headers for use as outgoing +// gRPC metadata (e.g. via NewStaticHeaderProvider). Keys are sanitized with +// SanitizeExtensionName — the same strict [a-z0-9] charset used for CloudEvent extensions — +// which is a subset of grpc's allowed metadata-key charset, so a sanitized key can never trip +// grpc's key validation or the reserved "-bin" suffix, and produces the same key stem as the +// corresponding CE extension (differing only by the CloudEvents Kafka binding's "ce_" prefix +// once on the wire). Values are sanitized via SanitizeMetadataValue, since grpc-go fails the +// whole RPC on a non-printable value. Entries that sanitize to an empty key, or that collide +// with a reserved extension name (see reservedExtensionNames) or a gRPC-reserved header name +// (see reservedMetadataKeys), are skipped. Keys are applied in sorted order so duplicate +// sanitized keys resolve deterministically (first in sorted order wins), matching +// WithResourceAttributeExtensions' collision handling. +// +// Note: unlike the CloudEvents Kafka binding, gRPC metadata keys are NOT prefixed with "ce_" — +// that prefix is a CloudEvents-binding concept, not a metadata one, and reusing it here would +// collide with the CE binding's own "ce_" Kafka header if the server ever forwards gRPC +// metadata verbatim onto Kafka. +func SanitizeMetadataHeaders(in map[string]string) map[string]string { + keys := make([]string, 0, len(in)) + for k := range in { + keys = append(keys, k) + } + sort.Strings(keys) + + out := make(map[string]string, len(in)) + for _, k := range keys { + name := SanitizeExtensionName(k) + if name == "" { + continue + } + if _, reserved := reservedExtensionNames[name]; reserved { + continue + } + if _, reserved := reservedMetadataKeys[name]; reserved { + continue + } + if _, exists := out[name]; exists { + continue + } + out[name] = SanitizeMetadataValue(in[k]) + } + return out +} + // newRotatingHeaderProvider returns a HeaderProvider that refreshes its // headers every ttl using signer. initialHeaders, if non-empty, are served // until the first rotation occurs. diff --git a/pkg/chipingress/header_provider_test.go b/pkg/chipingress/header_provider_test.go index 0c0f523e90..8069420fd2 100644 --- a/pkg/chipingress/header_provider_test.go +++ b/pkg/chipingress/header_provider_test.go @@ -4,13 +4,18 @@ import ( "context" "crypto/ed25519" "encoding/hex" + "net" "testing" "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" "github.com/smartcontractkit/chainlink-common/pkg/chipingress" + "github.com/smartcontractkit/chainlink-common/pkg/chipingress/pb" ) // fakeSigner is a minimal chipingress.Signer used by tests that need to @@ -260,3 +265,132 @@ func TestNewHeaderProvider(t *testing.T) { assert.Nil(t, provider) }) } + +func TestNewStaticHeaderProvider(t *testing.T) { + headers := map[string]string{"chain_id": "1", "environment": "prod"} + provider := chipingress.NewStaticHeaderProvider(headers) + require.NotNil(t, provider) + + got, err := provider.Headers(t.Context()) + require.NoError(t, err) + assert.Equal(t, headers, got) + + type tsr interface { + RequireTransportSecurity() bool + } + tlsReq, ok := provider.(tsr) + require.True(t, ok) + assert.False(t, tlsReq.RequireTransportSecurity()) +} + +func TestSanitizeMetadataValue(t *testing.T) { + tests := []struct { + name string + in string + want string + }{ + {name: "printable ASCII is unchanged", in: "chain-1_prod.v2", want: "chain-1_prod.v2"}, + {name: "empty", in: "", want: ""}, + {name: "control character replaced", in: "value\nwith\tcontrol", want: "value?with?control"}, + {name: "non-ASCII UTF-8 replaced byte-wise", in: "café", want: "caf??"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, chipingress.SanitizeMetadataValue(tt.in)) + }) + } +} + +func TestSanitizeMetadataHeaders(t *testing.T) { + t.Run("standard OTel-style keys are sanitized to the same stem as CE extensions", func(t *testing.T) { + in := map[string]string{ + "service.name": "beholder", + "chain_id": "1", + "node-operator": "acme", + } + got := chipingress.SanitizeMetadataHeaders(in) + assert.Equal(t, map[string]string{ + "servicename": "beholder", + "chainid": "1", + "nodeoperator": "acme", + }, got) + }) + + t.Run("empty-after-sanitize keys are dropped", func(t *testing.T) { + got := chipingress.SanitizeMetadataHeaders(map[string]string{"---": "value"}) + assert.Empty(t, got) + }) + + t.Run("reserved names are dropped", func(t *testing.T) { + got := chipingress.SanitizeMetadataHeaders(map[string]string{chipingress.IdempotencyKeyAttr: "should-not-appear", "subject": "should-not-appear"}) + assert.Empty(t, got) + }) + + t.Run("gRPC-reserved header 'te' is dropped", func(t *testing.T) { + got := chipingress.SanitizeMetadataHeaders(map[string]string{"te": "trailers"}) + assert.Empty(t, got) + }) + + t.Run("non-printable values are sanitized", func(t *testing.T) { + got := chipingress.SanitizeMetadataHeaders(map[string]string{"chain_id": "1\n2"}) + assert.Equal(t, "1?2", got["chainid"]) + }) + + t.Run("duplicate sanitized keys resolve deterministically to sorted-first key", func(t *testing.T) { + got := chipingress.SanitizeMetadataHeaders(map[string]string{"service.name": "from-dotted", "service_name": "from-snake"}) + // sorted order: "service.name" < "service_name" ('.' < '_' in ASCII), so the dotted key wins. + assert.Equal(t, "from-dotted", got["servicename"]) + }) +} + +// pingServer is a minimal ChipIngressServer that always answers Ping successfully. +type pingServer struct { + pb.UnimplementedChipIngressServer +} + +func (pingServer) Ping(context.Context, *pb.EmptyRequest) (*pb.PingResponse, error) { + return &pb.PingResponse{}, nil +} + +// TestSanitizeMetadataHeaders_AvoidsRPCFailure is a regression/guard test for the core reason +// SanitizeMetadataHeaders exists: grpc-go hard-fails an entire RPC (codes.Internal) when an +// outgoing metadata pair fails its charset validation. An unsanitized resource-attribute key or +// value (dots, non-printable characters) reproduces that failure; running it through +// SanitizeMetadataHeaders first must not. +func TestSanitizeMetadataHeaders_AvoidsRPCFailure(t *testing.T) { + lis, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", "127.0.0.1:0") + require.NoError(t, err) + defer lis.Close() + + srv := grpc.NewServer() + pb.RegisterChipIngressServer(srv, pingServer{}) + go func() { _ = srv.Serve(lis) }() + defer srv.Stop() + + dirty := map[string]string{"k8s.pod.name": "pod-\x01abc"} + + t.Run("unsanitized headers fail the RPC", func(t *testing.T) { + client, err := chipingress.NewClient(lis.Addr().String(), + chipingress.WithInsecureConnection(), + chipingress.WithHeaderProvider(chipingress.NewStaticHeaderProvider(dirty)), + ) + require.NoError(t, err) + defer client.Close() //nolint:errcheck + + _, err = client.Ping(t.Context(), &chipingress.EmptyRequest{}) + require.Error(t, err) + assert.Equal(t, codes.Internal, status.Code(err)) + }) + + t.Run("sanitized headers succeed", func(t *testing.T) { + client, err := chipingress.NewClient(lis.Addr().String(), + chipingress.WithInsecureConnection(), + chipingress.WithHeaderProvider(chipingress.NewStaticHeaderProvider(chipingress.SanitizeMetadataHeaders(dirty))), + ) + require.NoError(t, err) + defer client.Close() //nolint:errcheck + + _, err = client.Ping(t.Context(), &chipingress.EmptyRequest{}) + require.NoError(t, err) + }) +} diff --git a/pkg/chipingress/types.go b/pkg/chipingress/types.go index e27c18d194..54cb844477 100644 --- a/pkg/chipingress/types.go +++ b/pkg/chipingress/types.go @@ -13,6 +13,35 @@ import ( // Kafka headers named "ce_" (e.g., ce_idempotencykey), enabling downstream deduplication. const IdempotencyKeyAttr = "idempotencykey" +// reservedExtensionNames holds every CloudEvent extension name that NewEvent sets internally, +// plus the CloudEvents core context attribute names (id, source, type, specversion, time, +// subject, dataschema, datacontenttype) and the spec-forbidden "data" name. WithResourceAttributeExtensions +// consults this set so that a resource attribute can never silently overwrite event-lifecycle +// metadata or collide with a CloudEvents core attribute. +var reservedExtensionNames = map[string]struct{}{ + IdempotencyKeyAttr: {}, + "recordedtime": {}, + "id": {}, + "source": {}, + "type": {}, + "specversion": {}, + "time": {}, + "subject": {}, + "dataschema": {}, + "datacontenttype": {}, + "data": {}, +} + +// reservedMetadataKeys holds gRPC-reserved header names that could otherwise be reached by +// SanitizeExtensionName's [a-z0-9] sanitization. Verified against grpc-go v1.79.1's +// isReservedHeader: every other reserved header (pseudo-headers, "content-type", "grpc-*") +// contains a ':' or '-' that sanitization strips, so "te" is the only one actually reachable. +// SanitizeMetadataHeaders consults this set so that edge case is handled deterministically +// rather than relying on grpc's own (silent) handling of a reserved header. +var reservedMetadataKeys = map[string]struct{}{ + "te": {}, +} + type ( // Cloudevents types CloudEvent = ce.Event