diff --git a/go.mod b/go.mod index 40479c3317..d454543566 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.20260716165322-7f2edff6e954 + github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260722161838-36e9940d44a4 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 e4b11c5902..b9cbf6e6a4 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.20260716165322-7f2edff6e954 h1:QhTMiEn3s+AB4xBoScuQglsqHGJYxheYrgpxdIdqNAI= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260716165322-7f2edff6e954/go.mod h1:UYcRMb4dZcoaIPgZJ3hckCySTqtJc9K4Q+tOKErwTq0= +github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260722161838-36e9940d44a4 h1:S99tTVhpw/ndmt42XAAxY5IjPgwlQme8X/tkWuP7ljA= +github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260722161838-36e9940d44a4/go.mod h1:J5n1H3YFMfrZAPTc6eA5ByP/7hlNeZnPOvMf+m2L6xQ= 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 700d90bf13..f18418e407 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 @@ -90,8 +91,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{ @@ -142,7 +144,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 de56839237..ef931fa92a 100644 --- a/pkg/beholder/batch_emitter_service_test.go +++ b/pkg/beholder/batch_emitter_service_test.go @@ -122,6 +122,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 1bd792a8f9..5c0eedc06f 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 @@ -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)) +}