From f49aea8c8f1de1363b7f3995522748e52c9b1b0f Mon Sep 17 00:00:00 2001 From: 4of9 <177086174+4of9@users.noreply.github.com> Date: Tue, 27 Oct 2026 15:22:39 -0400 Subject: [PATCH 01/15] beholder: add per-domain chip ingress emitters (ForDomain, EmitEvent) Co-Authored-By: Claude Opus 5.5 --- go.mod | 2 +- go.sum | 4 +- pkg/beholder/batch_emitter_service.go | 142 +++++++---- pkg/beholder/client.go | 38 ++- pkg/beholder/config.go | 6 + pkg/beholder/domain_emitter.go | 97 ++++++++ pkg/beholder/domain_emitter_test.go | 284 ++++++++++++++++++++++ pkg/beholder/testdata/config-example.json | 1 + 8 files changed, 518 insertions(+), 56 deletions(-) create mode 100644 pkg/beholder/domain_emitter.go create mode 100644 pkg/beholder/domain_emitter_test.go diff --git a/go.mod b/go.mod index 34adbda213..fbc0bfd450 100644 --- a/go.mod +++ b/go.mod @@ -41,7 +41,7 @@ require ( github.com/scylladb/go-reflectx v1.0.1 github.com/shopspring/decimal v1.4.0 github.com/smartcontractkit/chain-selectors v1.0.104 - github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20261005132539-b7d63d5fd549 + github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20261026195813-8ed7b172ab8c github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20251024234028-0988426d98f4 github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20261013212015-e622e8b83b6b github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20251002192024-d2ad9222409b diff --git a/go.sum b/go.sum index 12dc36d338..a8a39b2635 100644 --- a/go.sum +++ b/go.sum @@ -251,8 +251,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.104 h1:/n9pPGM5W/+r1eHoWZv4VwX9LNS1af4+ICyhM8zKRNM= github.com/smartcontractkit/chain-selectors v1.0.104/go.mod h1:qy7whtgG5g+7z0jt0nRyii9bLND9m15NZTzuQPkMZ5w= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20261005132539-b7d63d5fd549 h1:J4vYRJxkBO8Yf//17wScKjYWmS45VZaFWx0zt+4ETEU= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20261005132539-b7d63d5fd549/go.mod h1:Scyk7O80fde5rJF3Ty3fEE3xiS6j84eWbu8/2gl3pIY= +github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20261026195813-8ed7b172ab8c h1:lgshnYtVJKzYnh8ejHpE7MAoR2Jx3O/kezw13PmifZo= +github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20261026195813-8ed7b172ab8c/go.mod h1:Scyk7O80fde5rJF3Ty3fEE3xiS6j84eWbu8/2gl3pIY= 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-20261013212015-e622e8b83b6b h1:tlxPkVQ6ZpJIElqUWc4lU5VH4BB6OzhckG/gHwpaaa0= diff --git a/pkg/beholder/batch_emitter_service.go b/pkg/beholder/batch_emitter_service.go index 7ca1a82ea2..f17123b858 100644 --- a/pkg/beholder/batch_emitter_service.go +++ b/pkg/beholder/batch_emitter_service.go @@ -23,6 +23,10 @@ type ChipIngressBatchEmitterService struct { eng *services.Engine batchClient *batch.Client + clientName string + // ownedClient, when set, is closed on stop after the batch client drains. Domain emitters + // own their chip ingress connection; the default emitter shares the Client's. + ownedClient chipingress.Client metricAttrsCache sync.Map // map[string]otelmetric.MeasurementOption metrics batchEmitterMetrics @@ -35,6 +39,12 @@ type batchEmitterMetrics struct { // NewChipIngressBatchEmitterService creates a batch emitter service backed by the given chipingress client. func NewChipIngressBatchEmitterService(client chipingress.Client, cfg Config, lggr logger.Logger) (*ChipIngressBatchEmitterService, error) { + return newChipIngressBatchEmitterService(client, cfg, lggr, batch.ClientNameBeholder) +} + +// newChipIngressBatchEmitterService is NewChipIngressBatchEmitterService with a client name and +// extra batch options (applied last), used to build domain emitters. +func newChipIngressBatchEmitterService(client chipingress.Client, cfg Config, lggr logger.Logger, clientName string, extraOpts ...batch.Opt) (*ChipIngressBatchEmitterService, error) { if client == nil { return nil, errors.New("chip ingress client is nil") } @@ -75,7 +85,7 @@ func NewChipIngressBatchEmitterService(client chipingress.Client, cfg Config, lg return nil, fmt.Errorf("failed to create batch emitter metrics: %w", err) } - batchClient, err := batch.NewBatchClient(client, + opts := append([]batch.Opt{ batch.WithBatchSize(maxBatchSize), batch.WithMessageBuffer(bufferSize), batch.WithBatchInterval(sendInterval), @@ -84,14 +94,16 @@ func NewChipIngressBatchEmitterService(client chipingress.Client, cfg Config, lg batch.WithMaxConcurrentSends(maxConcurrentSends), batch.WithMaxGRPCRequestSize(maxGRPCRequestSize), batch.WithEventClone(false), - batch.WithClientName(batch.ClientNameBeholder), - ) + batch.WithClientName(clientName), + }, extraOpts...) + batchClient, err := batch.NewBatchClient(client, opts...) if err != nil { return nil, fmt.Errorf("failed to create batch client: %w", err) } e := &ChipIngressBatchEmitterService{ batchClient: batchClient, + clientName: clientName, metrics: metrics, } @@ -115,6 +127,9 @@ func (e *ChipIngressBatchEmitterService) start(_ context.Context) error { func (e *ChipIngressBatchEmitterService) stop() error { e.batchClient.Stop() + if e.ownedClient != nil { + return e.ownedClient.Close() + } return nil } @@ -156,56 +171,85 @@ func (e *ChipIngressBatchEmitterService) emitInternal(ctx context.Context, body return err } - metricAttrs := e.metricAttrsFor(domain, entity) - - queueErr := e.batchClient.QueueMessage(eventPb, func(sendErr error) { - // The callback fires asynchronously after the batch is sent, - // so the caller's ctx may already be cancelled. Use ctx directly - // for metric recording — OTel Add is non-blocking and tolerates - // cancelled contexts. - if sendErr != nil { - errorCode := batch.ErrorCodeFor(sendErr) - e.metrics.eventsDropped.Add(ctx, 1, e.dropMetricAttrsFor(domain, entity, errorCode)) - // Partial delivery is not logged: it is a per-event, often persistent - // server-side condition (e.g. missing schema) that would otherwise log on - // every dropped event. chip_ingress.events_dropped (error_code) already - // captures that it's happening and roughly why; the full reason isn't - // needed at fleet-wide log volume. - if _, ok := errors.AsType[*batch.PublishError](sendErr); !ok { - e.eng.Errorw("failed to emit to chip ingress", - "error", sendErr, - "error_code", errorCode, - "error_reason", sendErr.Error(), - "domain", domain, - "entity", entity, - ) - } - } else { - e.metrics.eventsSent.Add(ctx, 1, metricAttrs) - } - if callback != nil { - callback(sendErr) - } - }) - if queueErr != nil { - errorCode := batch.ErrorCodeFor(queueErr) - e.metrics.eventsDropped.Add(ctx, 1, e.dropMetricAttrsFor(domain, entity, errorCode)) - e.eng.Errorw("failed to queue message for chip ingress", - "error", queueErr, - "error_code", errorCode, - "error_reason", queueErr.Error(), - "domain", domain, - "entity", entity, - ) - if callback != nil { - callback(queueErr) - } + e.queue(ctx, eventPb, domain, entity, callback) + return nil + }) +} + +// EmitEvent queues a fully built CloudEvent for batched delivery without blocking. Unlike +// Emit, which builds the event from attributes, it keeps every extension the caller set. +// +// If EmitEvent returns a non-nil error, the callback will NOT be invoked. +// If it returns nil, a non-nil callback is guaranteed to fire exactly once. +func (e *ChipIngressBatchEmitterService) EmitEvent(ctx context.Context, event chipingress.CloudEvent, callback func(error)) error { + return e.eng.IfStarted(func() error { + if err := event.Validate(); err != nil { + return fmt.Errorf("invalid CloudEvent: %w", err) + } + eventPb, err := chipingress.EventToProto(event) + if err != nil { + return fmt.Errorf("failed to convert to proto: %w", err) + } + + if err := ctx.Err(); err != nil { + return err } + e.queue(ctx, eventPb, event.Source(), event.Type(), callback) return nil }) } +// queue hands eventPb to the batch client and records its fate. domain and entity are the +// event's source and type, used for metrics and logs. +func (e *ChipIngressBatchEmitterService) queue(ctx context.Context, eventPb *chipingress.CloudEventPb, domain, entity string, callback func(error)) { + metricAttrs := e.metricAttrsFor(domain, entity) + + queueErr := e.batchClient.QueueMessage(eventPb, func(sendErr error) { + // The callback fires asynchronously after the batch is sent, + // so the caller's ctx may already be cancelled. Use ctx directly + // for metric recording — OTel Add is non-blocking and tolerates + // cancelled contexts. + if sendErr != nil { + errorCode := batch.ErrorCodeFor(sendErr) + e.metrics.eventsDropped.Add(ctx, 1, e.dropMetricAttrsFor(domain, entity, errorCode)) + // Partial delivery is not logged: it is a per-event, often persistent + // server-side condition (e.g. missing schema) that would otherwise log on + // every dropped event. chip_ingress.events_dropped (error_code) already + // captures that it's happening and roughly why; the full reason isn't + // needed at fleet-wide log volume. + if _, ok := errors.AsType[*batch.PublishError](sendErr); !ok { + e.eng.Errorw("failed to emit to chip ingress", + "error", sendErr, + "error_code", errorCode, + "error_reason", sendErr.Error(), + "domain", domain, + "entity", entity, + ) + } + } else { + e.metrics.eventsSent.Add(ctx, 1, metricAttrs) + } + if callback != nil { + callback(sendErr) + } + }) + if queueErr != nil { + errorCode := batch.ErrorCodeFor(queueErr) + e.metrics.eventsDropped.Add(ctx, 1, e.dropMetricAttrsFor(domain, entity, errorCode)) + e.eng.Errorw("failed to queue message for chip ingress", + "error", queueErr, + "error_code", errorCode, + "error_reason", queueErr.Error(), + "domain", domain, + "entity", entity, + ) + if callback != nil { + callback(queueErr) + } + } +} + func (e *ChipIngressBatchEmitterService) metricAttrsFor(domain, entity string) otelmetric.MeasurementOption { key := domain + "\x00" + entity if v, ok := e.metricAttrsCache.Load(key); ok { @@ -214,7 +258,7 @@ func (e *ChipIngressBatchEmitterService) metricAttrsFor(domain, entity string) o attrs := otelmetric.WithAttributeSet(attribute.NewSet( attribute.String("domain", domain), attribute.String("entity", entity), - attribute.String("client_name", batch.ClientNameBeholder), + attribute.String("client_name", e.clientName), )) v, _ := e.metricAttrsCache.LoadOrStore(key, attrs) return v.(otelmetric.MeasurementOption) @@ -232,7 +276,7 @@ func (e *ChipIngressBatchEmitterService) dropMetricAttrsFor(domain, entity, erro attrs := []attribute.KeyValue{ attribute.String("domain", domain), attribute.String("entity", entity), - attribute.String("client_name", batch.ClientNameBeholder), + attribute.String("client_name", e.clientName), } if errorCode != "" { attrs = append(attrs, attribute.String("error_code", errorCode)) diff --git a/pkg/beholder/client.go b/pkg/beholder/client.go index bb7727d676..1f0f07b2e2 100644 --- a/pkg/beholder/client.go +++ b/pkg/beholder/client.go @@ -51,6 +51,9 @@ type Client struct { // Chip Chip chipingress.Client + // domainEmitters are built when Config.ChipIngressDomainEmittersEnabled is set and retrieved with ForDomain. + domainEmitters map[Domain]*domainEmitter + // Providers LoggerProvider otellog.LoggerProvider TracerProvider oteltrace.TracerProvider @@ -71,10 +74,16 @@ func (c *Client) initService(lggr pkglogger.Logger, batchSvc *ChipIngressBatchEm Name: "BeholderClient", Close: c.close, NewSubServices: func(l pkglogger.Logger) []services.Service { - if batchSvc == nil { - return nil + var subs []services.Service + if batchSvc != nil { + subs = append(subs, batchSvc) + } + for _, domain := range domains { + if em, ok := c.domainEmitters[domain]; ok { + subs = append(subs, em.ChipIngressBatchEmitterService) + } } - return []services.Service{batchSvc} + return subs }, }.NewServiceEngine(lggr) } @@ -210,14 +219,19 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (_ *Client, er // and logs will be sent via OTLP using the regular Logger instead of calling Emit emitter := NewMessageEmitter(messageLogger) var batchEmitterService *ChipIngressBatchEmitterService + var domainEmitters map[Domain]*domainEmitter var chipIngressClient chipingress.Client = &chipingress.NoopClient{} lggr := cfg.ChipIngressLogger if lggr == nil { lggr = pkglogger.Nop() } + chipIngressEnabled := cfg.ChipIngressEmitterEnabled || cfg.ChipIngressEmitterGRPCEndpoint != "" + if cfg.ChipIngressDomainEmittersEnabled && !chipIngressEnabled { + return nil, errors.New("ChipIngressDomainEmittersEnabled requires chip ingress (ChipIngressEmitterGRPCEndpoint)") + } // if chip ingress is enabled, create dual source emitter that sends to both otel collector and chip ingress // eventually we will remove the dual source emitter and just use chip ingress - if cfg.ChipIngressEmitterEnabled || cfg.ChipIngressEmitterGRPCEndpoint != "" { + if chipIngressEnabled { var opts []chipingress.Opt resourceAttrs := resourceAttributesToStringMap(cfg.ResourceAttributes) @@ -244,6 +258,21 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (_ *Client, er opts = append(opts, chipingress.WithMeterProvider(meterProvider)) opts = append(opts, chipingress.WithTracerProvider(tracerProvider)) + // Domain emitters share everything above and set their own header provider. + if cfg.ChipIngressDomainEmittersEnabled { + domainEmitters, err = newDomainEmitters(cfg, opts, resourceAttrs, lggr) + if err != nil { + return nil, err + } + defer func() { + if err != nil { + for _, em := range domainEmitters { + _ = em.ownedClient.Close() + } + } + }() + } + if len(resourceAttrs) > 0 { opts = append(opts, chipingress.WithResourceAttributeHeaders(resourceAttrs)) } @@ -293,6 +322,7 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (_ *Client, er Meter: meter, Emitter: emitter, Chip: chipIngressClient, + domainEmitters: domainEmitters, LoggerProvider: loggerProvider, TracerProvider: tracerProvider, MeterProvider: meterProvider, diff --git a/pkg/beholder/config.go b/pkg/beholder/config.go index c0bb0c2398..d824f5cdcf 100644 --- a/pkg/beholder/config.go +++ b/pkg/beholder/config.go @@ -79,6 +79,12 @@ type Config struct { ChipIngressMaxGRPCRequestSize int // Max serialized PublishBatch request size in bytes (default 10 MiB) ChipIngressLogger logger.Logger // Required when ChipIngressBatchEmitterEnabled is true + // ChipIngressDomainEmittersEnabled builds one chip ingress emitter per Domain, retrieved with + // Client.ForDomain. Each has its own connection and buffer (sized by the batch settings + // above), sends the DomainHeaderKey routing header, sends to chip ingress only, and publishes + // with PublishBatchNoAck. Requires chip ingress. Off by default. + ChipIngressDomainEmittersEnabled bool + // OTel Log LogExportTimeout time.Duration LogExportInterval time.Duration diff --git a/pkg/beholder/domain_emitter.go b/pkg/beholder/domain_emitter.go new file mode 100644 index 0000000000..ddcfb259dd --- /dev/null +++ b/pkg/beholder/domain_emitter.go @@ -0,0 +1,97 @@ +package beholder + +import ( + "context" + "fmt" + + "github.com/smartcontractkit/chainlink-common/pkg/chipingress" + "github.com/smartcontractkit/chainlink-common/pkg/chipingress/batch" + pkglogger "github.com/smartcontractkit/chainlink-common/pkg/logger" +) + +// Domain is a top-level product domain with its own chip ingress deployment. +type Domain string + +// Domains are fixed in code, not configured: each one needs a matching load balancer rule and +// chip ingress deployment, so adding one is a reviewed change here plus infra. +const ( + DomainDataFeeds Domain = "data-feeds" + DomainDataStreams Domain = "data-streams" + DomainCCIP Domain = "ccip" +) + +// domains lists every Domain; Config.ChipIngressDomainEmittersEnabled builds an emitter for each. +var domains = []Domain{DomainDataFeeds, DomainDataStreams, DomainCCIP} + +// DomainHeaderKey is the gRPC metadata key carrying a domain emitter's Domain. Load balancers route +// on it to the chip ingress deployment that serves the domain; chip ingress itself ignores it. +const DomainHeaderKey = "chainlink-domain" + +// DomainEmitter sends one domain's events to chip ingress only, with PublishBatchNoAck, on its own +// connection and buffer. Obtain one with Client.ForDomain. +type DomainEmitter interface { + // Emitter builds the event from attributes, like the Client's Emitter. Close is a no-op: the + // Client starts and closes its domain emitters. + Emitter + // EmitEvent queues a fully built CloudEvent, keeping every extension the caller set. If it + // returns a non-nil error the callback is not invoked; otherwise a non-nil callback fires + // exactly once with the event's fate. + EmitEvent(ctx context.Context, event chipingress.CloudEvent, callback func(error)) error +} + +// ForDomain returns the Client's emitter for domain. It creates nothing, so repeated calls return +// the same emitter. It is an error unless Config.ChipIngressDomainEmittersEnabled is set and domain +// is a known Domain. +func (c *Client) ForDomain(domain Domain) (DomainEmitter, error) { + em, ok := c.domainEmitters[domain] + if !ok { + return nil, fmt.Errorf("beholder: no chip ingress emitter for domain %q (ChipIngressDomainEmittersEnabled=%t)", + domain, c.Config.ChipIngressDomainEmittersEnabled) + } + return em, nil +} + +// domainEmitter is a domain's batch emitter as handed out by ForDomain. Close is a no-op so holders +// cannot stop an emitter the Client owns. +type domainEmitter struct { + *ChipIngressBatchEmitterService +} + +func (*domainEmitter) Close() error { return nil } + +// newDomainEmitters builds one emitter per Domain. opts are the Client's chip ingress options +// (transport, auth, OTel providers) without its header provider, which each domain emitter +// replaces with one carrying the resource attribute headers plus its DomainHeaderKey. +func newDomainEmitters(cfg Config, opts []chipingress.Opt, resourceAttrs map[string]string, lggr pkglogger.Logger) (_ map[Domain]*domainEmitter, err error) { + emitters := make(map[Domain]*domainEmitter, len(domains)) + defer func() { + if err != nil { + for _, em := range emitters { + _ = em.ownedClient.Close() + } + } + }() + for _, domain := range domains { + headers := chipingress.SanitizeMetadataHeaders(resourceAttrs) + headers[DomainHeaderKey] = string(domain) + clientName := batch.ClientNameBeholder + "_" + string(domain) + + domainOpts := append(append([]chipingress.Opt{}, opts...), + chipingress.WithHeaderProvider(chipingress.NewStaticHeaderProvider(headers)), + chipingress.WithNOPLookup(), + chipingress.WithClientName(clientName), + ) + chipClient, err := chipingress.NewClient(cfg.ChipIngressEmitterGRPCEndpoint, domainOpts...) + if err != nil { + return nil, fmt.Errorf("chip ingress client for domain %q: %w", domain, err) + } + svc, err := newChipIngressBatchEmitterService(chipClient, cfg, pkglogger.Named(lggr, string(domain)), clientName, batch.WithNoAck()) + if err != nil { + _ = chipClient.Close() + return nil, fmt.Errorf("chip ingress emitter for domain %q: %w", domain, err) + } + svc.ownedClient = chipClient + emitters[domain] = &domainEmitter{svc} + } + return emitters, nil +} diff --git a/pkg/beholder/domain_emitter_test.go b/pkg/beholder/domain_emitter_test.go new file mode 100644 index 0000000000..7fcba811a2 --- /dev/null +++ b/pkg/beholder/domain_emitter_test.go @@ -0,0 +1,284 @@ +package beholder_test + +import ( + "context" + "net" + "sync" + "testing" + "time" + + cepb "github.com/cloudevents/sdk-go/binding/format/protobuf/v2/pb" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc" + 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/chipingress" + "github.com/smartcontractkit/chainlink-common/pkg/chipingress/pb" + "github.com/smartcontractkit/chainlink-common/pkg/logger" +) + +// publishCall is one batch RPC received by batchCapturingChipServer. +type publishCall struct { + method string + md metadata.MD + events []*cepb.CloudEvent +} + +// batchCapturingChipServer records every PublishBatch / PublishBatchNoAck call and answers +// with one positional success result per event, as the real server does. +type batchCapturingChipServer struct { + pb.UnimplementedChipIngressServer + + mu sync.Mutex + calls []publishCall +} + +func (s *batchCapturingChipServer) record(ctx context.Context, method string, in *pb.CloudEventBatch) (*pb.PublishResponse, error) { + md, _ := metadata.FromIncomingContext(ctx) + s.mu.Lock() + defer s.mu.Unlock() + s.calls = append(s.calls, publishCall{method: method, md: md, events: in.Events}) + return &pb.PublishResponse{Results: make([]*pb.PublishResult, len(in.Events))}, nil +} + +func (s *batchCapturingChipServer) PublishBatch(ctx context.Context, in *pb.CloudEventBatch) (*pb.PublishResponse, error) { + return s.record(ctx, "PublishBatch", in) +} + +func (s *batchCapturingChipServer) PublishBatchNoAck(ctx context.Context, in *pb.CloudEventBatch) (*pb.PublishResponse, error) { + return s.record(ctx, "PublishBatchNoAck", in) +} + +// callsWithEventSource returns the calls that carried an event with the given CloudEvent source. +func (s *batchCapturingChipServer) callsWithEventSource(source string) []publishCall { + s.mu.Lock() + defer s.mu.Unlock() + var out []publishCall + for _, c := range s.calls { + for _, ev := range c.events { + if ev.Source == source { + out = append(out, c) + break + } + } + } + return out +} + +func startBatchCapturingChipServer(t *testing.T) (*batchCapturingChipServer, string) { + t.Helper() + lis, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", "127.0.0.1:0") + require.NoError(t, err) + srv := grpc.NewServer() + capture := &batchCapturingChipServer{} + pb.RegisterChipIngressServer(srv, capture) + go func() { _ = srv.Serve(lis) }() + t.Cleanup(srv.Stop) + return capture, lis.Addr().String() +} + +const ( + testAuthHeaderKey = "X-Beholder-Node-Auth-Token" + testAuthToken = "1:abc:2:def" +) + +func domainTestConfig(t *testing.T, chipAddr string) beholder.Config { + return beholder.Config{ + OtelExporterGRPCEndpoint: "localhost:4317", + ChipIngressEmitterEnabled: true, + ChipIngressEmitterGRPCEndpoint: chipAddr, + ChipIngressInsecureConnection: true, + ChipIngressBatchEmitterEnabled: true, + ChipIngressLogger: logger.Test(t), + ChipIngressSendInterval: 10 * time.Millisecond, + AuthHeaders: map[string]string{testAuthHeaderKey: testAuthToken}, + ResourceAttributes: []attribute.KeyValue{ + attribute.String("csa_public_key", "abc123"), + }, + ChipIngressDomainEmittersEnabled: true, + } +} + +func newDomainTestClient(t *testing.T, cfg beholder.Config) *beholder.Client { + t.Helper() + client, err := beholder.NewGRPCClient(cfg, func(...otlploggrpc.Option) (sdklog.Exporter, error) { + return &mockLogExporter{}, nil + }) + require.NoError(t, err) + require.NoError(t, client.Start(t.Context())) + t.Cleanup(func() { _ = client.Close() }) + return client +} + +// legacyStyleEvent builds an event the way the node's telemetry manager will: a normal +// CloudEvent carrying its own extensions, which beholder's attribute-based Emit cannot express. +func legacyStyleEvent(t *testing.T, source string) chipingress.CloudEvent { + t.Helper() + ev, err := chipingress.NewEvent(source, "pb.TelemetryWrapper", []byte("payload"), nil) + require.NoError(t, err) + ev.SetExtension("telemetrytype", "ocr2-median") + ev.SetExtension("partitionkey", "0xcontract-abc123") + return ev +} + +func TestForDomain_EmitEventSendsNoAckWithDomainHeader(t *testing.T) { + capture, addr := startBatchCapturingChipServer(t) + client := newDomainTestClient(t, domainTestConfig(t, addr)) + + em, err := client.ForDomain(beholder.DomainDataFeeds) + require.NoError(t, err) + + const source = "data-feeds.telemetry.ocr2-median" + done := make(chan error, 1) + require.NoError(t, em.EmitEvent(t.Context(), legacyStyleEvent(t, source), func(err error) { done <- err })) + + select { + case err := <-done: + require.NoError(t, err, "callback reports the per-event result") + case <-time.After(5 * time.Second): + t.Fatal("callback not invoked") + } + + calls := capture.callsWithEventSource(source) + require.Len(t, calls, 1) + call := calls[0] + + assert.Equal(t, "PublishBatchNoAck", call.method, "domain emitters are always no-ack") + assert.Equal(t, []string{"data-feeds"}, call.md.Get(beholder.DomainHeaderKey), "routing header carries the domain") + assert.Equal(t, []string{testAuthToken}, call.md.Get(testAuthHeaderKey), "inherits the client's auth") + for key, val := range chipingress.SanitizeMetadataHeaders(map[string]string{"csa_public_key": "abc123"}) { + assert.Equal(t, []string{val}, call.md.Get(key), "inherits resource attribute header %q", key) + } + assert.Equal(t, []string{"true"}, call.md.Get("x-include-nop-info"), "domain emitters request NOP info") + + require.Len(t, call.events, 1) + attrs := call.events[0].Attributes + require.Contains(t, attrs, "telemetrytype", "caller-set extensions survive") + assert.Equal(t, "ocr2-median", attrs["telemetrytype"].GetCeString()) + require.Contains(t, attrs, "partitionkey") + assert.Equal(t, "0xcontract-abc123", attrs["partitionkey"].GetCeString()) +} + +func TestForDomain_EachDomainUsesItsOwnHeader(t *testing.T) { + capture, addr := startBatchCapturingChipServer(t) + client := newDomainTestClient(t, domainTestConfig(t, addr)) + + ccip, err := client.ForDomain(beholder.DomainCCIP) + require.NoError(t, err) + require.NoError(t, ccip.EmitEvent(t.Context(), legacyStyleEvent(t, "ccip.telemetry.ocr3"), nil)) + + require.Eventually(t, func() bool { return len(capture.callsWithEventSource("ccip.telemetry.ocr3")) == 1 }, + 5*time.Second, 10*time.Millisecond) + assert.Equal(t, []string{"ccip"}, capture.callsWithEventSource("ccip.telemetry.ocr3")[0].md.Get(beholder.DomainHeaderKey)) +} + +func TestForDomain_DefaultEmitterIsUnaffected(t *testing.T) { + capture, addr := startBatchCapturingChipServer(t) + client := newDomainTestClient(t, domainTestConfig(t, addr)) + + require.NoError(t, client.Emitter.Emit(t.Context(), []byte("payload"), + beholder.AttrKeyDomain, "platform-custom", + beholder.AttrKeyEntity, "SomeEntity", + beholder.AttrKeyDataSchema, "/schemas/ids/1001", + )) + + require.Eventually(t, func() bool { return len(capture.callsWithEventSource("platform-custom")) == 1 }, + 5*time.Second, 10*time.Millisecond) + call := capture.callsWithEventSource("platform-custom")[0] + assert.Equal(t, "PublishBatch", call.method, "the default emitter keeps acked publishing") + assert.Empty(t, call.md.Get(beholder.DomainHeaderKey), "the default emitter sends no routing header") +} + +func TestForDomain_BuildsEveryDomainAndRejectsUnknownOnes(t *testing.T) { + _, addr := startBatchCapturingChipServer(t) + client := newDomainTestClient(t, domainTestConfig(t, addr)) + + for _, d := range []beholder.Domain{beholder.DomainDataFeeds, beholder.DomainDataStreams, beholder.DomainCCIP} { + _, err := client.ForDomain(d) + require.NoError(t, err, "domain %q is built", d) + } + + first, err := client.ForDomain(beholder.DomainDataFeeds) + require.NoError(t, err) + second, err := client.ForDomain(beholder.DomainDataFeeds) + require.NoError(t, err) + assert.Same(t, first, second, "ForDomain is a lookup, not a constructor") + + _, err = client.ForDomain(beholder.Domain("not-a-domain")) + require.Error(t, err, "an unknown domain is an error, never a silent fallback") +} + +func TestForDomain_CloseOnReturnedEmitterDoesNotStopIt(t *testing.T) { + capture, addr := startBatchCapturingChipServer(t) + client := newDomainTestClient(t, domainTestConfig(t, addr)) + + em, err := client.ForDomain(beholder.DomainDataFeeds) + require.NoError(t, err) + require.NoError(t, em.Close(), "the client owns the emitter's lifecycle") + + require.NoError(t, em.EmitEvent(t.Context(), legacyStyleEvent(t, "data-feeds.telemetry.ocr"), nil)) + require.Eventually(t, func() bool { return len(capture.callsWithEventSource("data-feeds.telemetry.ocr")) == 1 }, + 5*time.Second, 10*time.Millisecond) +} + +func TestForDomain_ClientCloseDrainsQueuedEvents(t *testing.T) { + capture, addr := startBatchCapturingChipServer(t) + cfg := domainTestConfig(t, addr) + cfg.ChipIngressSendInterval = time.Hour // nothing flushes on its own + + client, err := beholder.NewGRPCClient(cfg, func(...otlploggrpc.Option) (sdklog.Exporter, error) { + return &mockLogExporter{}, nil + }) + require.NoError(t, err) + require.NoError(t, client.Start(t.Context())) + + em, err := client.ForDomain(beholder.DomainDataFeeds) + require.NoError(t, err) + require.NoError(t, em.EmitEvent(t.Context(), legacyStyleEvent(t, "data-feeds.telemetry.drain"), nil)) + + // Close also shuts down the OTel providers, which fail to flush to the absent test + // collector; only the drain matters here. + _ = client.Close() + assert.Len(t, capture.callsWithEventSource("data-feeds.telemetry.drain"), 1, "queued events are published on close") +} + +func TestForDomain_EmitEventRejectsInvalidEvent(t *testing.T) { + _, addr := startBatchCapturingChipServer(t) + client := newDomainTestClient(t, domainTestConfig(t, addr)) + + em, err := client.ForDomain(beholder.DomainDataFeeds) + require.NoError(t, err) + + called := false + err = em.EmitEvent(t.Context(), chipingress.CloudEvent{}, func(error) { called = true }) + require.Error(t, err, "an event without id/source/type is rejected up front") + assert.False(t, called, "no callback when EmitEvent returns an error") +} + +func TestNewGRPCClient_DomainEmittersRequireChipIngress(t *testing.T) { + cfg := beholder.Config{OtelExporterGRPCEndpoint: "localhost:4317", ChipIngressDomainEmittersEnabled: true} + _, err := beholder.NewGRPCClient(cfg, func(...otlploggrpc.Option) (sdklog.Exporter, error) { + return &mockLogExporter{}, nil + }) + require.Error(t, err) +} + +func TestForDomain_DisabledByDefault(t *testing.T) { + _, addr := startBatchCapturingChipServer(t) + cfg := domainTestConfig(t, addr) + cfg.ChipIngressDomainEmittersEnabled = false + client := newDomainTestClient(t, cfg) + + _, err := client.ForDomain(beholder.DomainDataFeeds) + require.Error(t, err, "chip ingress alone does not build domain emitters") +} + +func TestNoopClient_ForDomain(t *testing.T) { + _, err := beholder.NewNoopClient().ForDomain(beholder.DomainDataFeeds) + require.Error(t, err) +} diff --git a/pkg/beholder/testdata/config-example.json b/pkg/beholder/testdata/config-example.json index 6a82336c4b..f9b212a2b3 100644 --- a/pkg/beholder/testdata/config-example.json +++ b/pkg/beholder/testdata/config-example.json @@ -49,6 +49,7 @@ "ChipIngressMaxConcurrentSends": 0, "ChipIngressMaxGRPCRequestSize": 0, "ChipIngressLogger": null, + "ChipIngressDomainEmittersEnabled": false, "LogExportTimeout": 1000000000, "LogExportInterval": 1000000000, "LogExportMaxBatchSize": 512, From c2f5a8a887300e5d6ddae0bafceb17a5fc6f076f Mon Sep 17 00:00:00 2001 From: 4of9 <177086174+4of9@users.noreply.github.com> Date: Tue, 27 Oct 2026 16:31:33 -0400 Subject: [PATCH 02/15] beholder: reject domain emitter events from other domains Co-Authored-By: Claude Opus 5.5 --- pkg/beholder/batch_emitter_service.go | 19 +++++++++++ pkg/beholder/domain_emitter.go | 5 +++ pkg/beholder/domain_emitter_test.go | 45 +++++++++++++++++++++++++++ 3 files changed, 69 insertions(+) diff --git a/pkg/beholder/batch_emitter_service.go b/pkg/beholder/batch_emitter_service.go index f17123b858..360b3b415f 100644 --- a/pkg/beholder/batch_emitter_service.go +++ b/pkg/beholder/batch_emitter_service.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "strings" "sync" "go.opentelemetry.io/otel" @@ -27,6 +28,9 @@ type ChipIngressBatchEmitterService struct { // ownedClient, when set, is closed on stop after the batch client drains. Domain emitters // own their chip ingress connection; the default emitter shares the Client's. ownedClient chipingress.Client + // domain, when set, restricts events to this Domain or its subdomains, so an event cannot + // ride another domain's route. Unset for the default emitter. + domain Domain metricAttrsCache sync.Map // map[string]otelmetric.MeasurementOption metrics batchEmitterMetrics @@ -155,6 +159,9 @@ func (e *ChipIngressBatchEmitterService) emitInternal(ctx context.Context, body if err != nil { return err } + if err := e.checkDomain(domain); err != nil { + return err + } attributes := newAttributes(attrKVs...) @@ -186,6 +193,9 @@ func (e *ChipIngressBatchEmitterService) EmitEvent(ctx context.Context, event ch if err := event.Validate(); err != nil { return fmt.Errorf("invalid CloudEvent: %w", err) } + if err := e.checkDomain(event.Source()); err != nil { + return err + } eventPb, err := chipingress.EventToProto(event) if err != nil { return fmt.Errorf("failed to convert to proto: %w", err) @@ -200,6 +210,15 @@ func (e *ChipIngressBatchEmitterService) EmitEvent(ctx context.Context, event ch }) } +// checkDomain rejects a source outside e.domain: it must be the domain itself or a subdomain of +// it (e.g. "data-feeds" or "data-feeds.telemetry.ocr2-median" for DomainDataFeeds). +func (e *ChipIngressBatchEmitterService) checkDomain(source string) error { + if e.domain == "" || source == string(e.domain) || strings.HasPrefix(source, string(e.domain)+".") { + return nil + } + return fmt.Errorf("beholder: event source %q is not in domain %q", source, e.domain) +} + // queue hands eventPb to the batch client and records its fate. domain and entity are the // event's source and type, used for metrics and logs. func (e *ChipIngressBatchEmitterService) queue(ctx context.Context, eventPb *chipingress.CloudEventPb, domain, entity string, callback func(error)) { diff --git a/pkg/beholder/domain_emitter.go b/pkg/beholder/domain_emitter.go index ddcfb259dd..afab2fe3ef 100644 --- a/pkg/beholder/domain_emitter.go +++ b/pkg/beholder/domain_emitter.go @@ -29,6 +29,10 @@ const DomainHeaderKey = "chainlink-domain" // DomainEmitter sends one domain's events to chip ingress only, with PublishBatchNoAck, on its own // connection and buffer. Obtain one with Client.ForDomain. +// +// Both methods reject an event whose source (the CloudEvent source, or the beholder_domain +// attribute for Emit) is not the emitter's Domain or a subdomain of it: for DomainDataFeeds, +// "data-feeds" and "data-feeds.telemetry.ocr2-median" are accepted, "ccip.telemetry.ocr3" is not. type DomainEmitter interface { // Emitter builds the event from attributes, like the Client's Emitter. Close is a no-op: the // Client starts and closes its domain emitters. @@ -91,6 +95,7 @@ func newDomainEmitters(cfg Config, opts []chipingress.Opt, resourceAttrs map[str return nil, fmt.Errorf("chip ingress emitter for domain %q: %w", domain, err) } svc.ownedClient = chipClient + svc.domain = domain emitters[domain] = &domainEmitter{svc} } return emitters, nil diff --git a/pkg/beholder/domain_emitter_test.go b/pkg/beholder/domain_emitter_test.go index 7fcba811a2..7fc84e169f 100644 --- a/pkg/beholder/domain_emitter_test.go +++ b/pkg/beholder/domain_emitter_test.go @@ -260,6 +260,51 @@ func TestForDomain_EmitEventRejectsInvalidEvent(t *testing.T) { assert.False(t, called, "no callback when EmitEvent returns an error") } +func TestForDomain_RejectsEventsFromOtherDomains(t *testing.T) { + capture, addr := startBatchCapturingChipServer(t) + client := newDomainTestClient(t, domainTestConfig(t, addr)) + + em, err := client.ForDomain(beholder.DomainDataFeeds) + require.NoError(t, err) + + for _, tc := range []struct { + source string + ok bool + }{ + {"data-feeds", true}, // the top-level domain itself + {"data-feeds.telemetry.ocr2-median", true}, // a subdomain of it + {"ccip.telemetry.ocr3", false}, // another domain + {"data-feeds-extra.telemetry", false}, // shares a prefix, not a subdomain + {"telemetry.data-feeds", false}, // contains it, not under it + } { + t.Run(tc.source, func(t *testing.T) { + called := false + err := em.EmitEvent(t.Context(), legacyStyleEvent(t, tc.source), func(error) { called = true }) + if tc.ok { + require.NoError(t, err) + require.Eventually(t, func() bool { return len(capture.callsWithEventSource(tc.source)) == 1 }, + 5*time.Second, 10*time.Millisecond) + return + } + require.Error(t, err, "event from another domain must not ride this domain's route") + assert.False(t, called, "no callback when EmitEvent returns an error") + }) + } + + t.Run("Emit is checked too", func(t *testing.T) { + err := em.Emit(t.Context(), []byte("payload"), + beholder.AttrKeyDomain, "ccip.telemetry.ocr3", + beholder.AttrKeyEntity, "SomeEntity", + ) + require.Error(t, err) + }) + + // Nothing from the rejected sources reached the server. + for _, source := range []string{"ccip.telemetry.ocr3", "data-feeds-extra.telemetry", "telemetry.data-feeds"} { + assert.Empty(t, capture.callsWithEventSource(source), "source %q", source) + } +} + func TestNewGRPCClient_DomainEmittersRequireChipIngress(t *testing.T) { cfg := beholder.Config{OtelExporterGRPCEndpoint: "localhost:4317", ChipIngressDomainEmittersEnabled: true} _, err := beholder.NewGRPCClient(cfg, func(...otlploggrpc.Option) (sdklog.Exporter, error) { From 0d120a651c028b3b0348517660f06a70ced93493 Mon Sep 17 00:00:00 2001 From: 4of9 <177086174+4of9@users.noreply.github.com> Date: Wed, 28 Oct 2026 13:45:56 -0400 Subject: [PATCH 03/15] beholder: close domain emitter connections once batch.Client.Stop already closes the chip ingress client, so stop() closing it again made every Client.Close return one Canceled error per domain. The connection is now only closed directly when construction fails. Co-Authored-By: Claude Opus 5.5 --- pkg/beholder/batch_emitter_service.go | 17 +++++------- pkg/beholder/client.go | 2 +- pkg/beholder/domain_emitter.go | 11 ++++---- pkg/beholder/domain_emitter_test.go | 37 ++++++++++++++++++++++++--- 4 files changed, 48 insertions(+), 19 deletions(-) diff --git a/pkg/beholder/batch_emitter_service.go b/pkg/beholder/batch_emitter_service.go index 360b3b415f..78a45fe1bb 100644 --- a/pkg/beholder/batch_emitter_service.go +++ b/pkg/beholder/batch_emitter_service.go @@ -25,9 +25,6 @@ type ChipIngressBatchEmitterService struct { batchClient *batch.Client clientName string - // ownedClient, when set, is closed on stop after the batch client drains. Domain emitters - // own their chip ingress connection; the default emitter shares the Client's. - ownedClient chipingress.Client // domain, when set, restricts events to this Domain or its subdomains, so an event cannot // ride another domain's route. Unset for the default emitter. domain Domain @@ -43,12 +40,13 @@ type batchEmitterMetrics struct { // NewChipIngressBatchEmitterService creates a batch emitter service backed by the given chipingress client. func NewChipIngressBatchEmitterService(client chipingress.Client, cfg Config, lggr logger.Logger) (*ChipIngressBatchEmitterService, error) { - return newChipIngressBatchEmitterService(client, cfg, lggr, batch.ClientNameBeholder) + return newChipIngressBatchEmitterService(client, cfg, lggr, batch.ClientNameBeholder, "") } -// newChipIngressBatchEmitterService is NewChipIngressBatchEmitterService with a client name and -// extra batch options (applied last), used to build domain emitters. -func newChipIngressBatchEmitterService(client chipingress.Client, cfg Config, lggr logger.Logger, clientName string, extraOpts ...batch.Opt) (*ChipIngressBatchEmitterService, error) { +// newChipIngressBatchEmitterService is NewChipIngressBatchEmitterService with a client name, a +// Domain to restrict events to ("" for none), and extra batch options (applied last), used to +// build domain emitters. +func newChipIngressBatchEmitterService(client chipingress.Client, cfg Config, lggr logger.Logger, clientName string, domain Domain, extraOpts ...batch.Opt) (*ChipIngressBatchEmitterService, error) { if client == nil { return nil, errors.New("chip ingress client is nil") } @@ -108,6 +106,7 @@ func newChipIngressBatchEmitterService(client chipingress.Client, cfg Config, lg e := &ChipIngressBatchEmitterService{ batchClient: batchClient, clientName: clientName, + domain: domain, metrics: metrics, } @@ -129,11 +128,9 @@ func (e *ChipIngressBatchEmitterService) start(_ context.Context) error { return nil } +// stop drains the batch client, which then closes the chip ingress client it was built with. func (e *ChipIngressBatchEmitterService) stop() error { e.batchClient.Stop() - if e.ownedClient != nil { - return e.ownedClient.Close() - } return nil } diff --git a/pkg/beholder/client.go b/pkg/beholder/client.go index 1f0f07b2e2..0c910f5ad1 100644 --- a/pkg/beholder/client.go +++ b/pkg/beholder/client.go @@ -267,7 +267,7 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (_ *Client, er defer func() { if err != nil { for _, em := range domainEmitters { - _ = em.ownedClient.Close() + _ = em.chipClient.Close() } } }() diff --git a/pkg/beholder/domain_emitter.go b/pkg/beholder/domain_emitter.go index afab2fe3ef..13cef7efa1 100644 --- a/pkg/beholder/domain_emitter.go +++ b/pkg/beholder/domain_emitter.go @@ -59,6 +59,9 @@ func (c *Client) ForDomain(domain Domain) (DomainEmitter, error) { // cannot stop an emitter the Client owns. type domainEmitter struct { *ChipIngressBatchEmitterService + // chipClient is the emitter's own connection. Once started, the batch client closes it on + // Stop; it is closed here only when construction fails before the emitter ever starts. + chipClient chipingress.Client } func (*domainEmitter) Close() error { return nil } @@ -71,7 +74,7 @@ func newDomainEmitters(cfg Config, opts []chipingress.Opt, resourceAttrs map[str defer func() { if err != nil { for _, em := range emitters { - _ = em.ownedClient.Close() + _ = em.chipClient.Close() } } }() @@ -89,14 +92,12 @@ func newDomainEmitters(cfg Config, opts []chipingress.Opt, resourceAttrs map[str if err != nil { return nil, fmt.Errorf("chip ingress client for domain %q: %w", domain, err) } - svc, err := newChipIngressBatchEmitterService(chipClient, cfg, pkglogger.Named(lggr, string(domain)), clientName, batch.WithNoAck()) + svc, err := newChipIngressBatchEmitterService(chipClient, cfg, pkglogger.Named(lggr, string(domain)), clientName, domain, batch.WithNoAck()) if err != nil { _ = chipClient.Close() return nil, fmt.Errorf("chip ingress emitter for domain %q: %w", domain, err) } - svc.ownedClient = chipClient - svc.domain = domain - emitters[domain] = &domainEmitter{svc} + emitters[domain] = &domainEmitter{ChipIngressBatchEmitterService: svc, chipClient: chipClient} } return emitters, nil } diff --git a/pkg/beholder/domain_emitter_test.go b/pkg/beholder/domain_emitter_test.go index 7fc84e169f..510146fe87 100644 --- a/pkg/beholder/domain_emitter_test.go +++ b/pkg/beholder/domain_emitter_test.go @@ -241,12 +241,43 @@ func TestForDomain_ClientCloseDrainsQueuedEvents(t *testing.T) { require.NoError(t, err) require.NoError(t, em.EmitEvent(t.Context(), legacyStyleEvent(t, "data-feeds.telemetry.drain"), nil)) - // Close also shuts down the OTel providers, which fail to flush to the absent test - // collector; only the drain matters here. - _ = client.Close() + requireNoDomainEmitterCloseErrors(t, client.Close()) assert.Len(t, capture.callsWithEventSource("data-feeds.telemetry.drain"), 1, "queued events are published on close") } +// requireNoDomainEmitterCloseErrors checks a Client.Close error for anything the domain emitters +// caused. Close also shuts down the OTel providers, which fail to flush to the test's absent +// collector, so those errors are expected and ignored. +func requireNoDomainEmitterCloseErrors(t *testing.T, err error) { + t.Helper() + if err == nil { + return + } + assert.NotContains(t, err.Error(), "client connection is closing", + "domain emitter connections are closed exactly once") +} + +func TestForDomain_ClientCloseClosesDomainConnectionsOnce(t *testing.T) { + capture, addr := startBatchCapturingChipServer(t) + client, err := beholder.NewGRPCClient(domainTestConfig(t, addr), func(...otlploggrpc.Option) (sdklog.Exporter, error) { + return &mockLogExporter{}, nil + }) + require.NoError(t, err) + require.NoError(t, client.Start(t.Context())) + + // Use every domain so each connection is actually dialed before Close. + for _, d := range []beholder.Domain{beholder.DomainDataFeeds, beholder.DomainDataStreams, beholder.DomainCCIP} { + em, err := client.ForDomain(d) + require.NoError(t, err) + source := string(d) + ".telemetry.close" + require.NoError(t, em.EmitEvent(t.Context(), legacyStyleEvent(t, source), nil)) + require.Eventually(t, func() bool { return len(capture.callsWithEventSource(source)) == 1 }, + 5*time.Second, 10*time.Millisecond) + } + + requireNoDomainEmitterCloseErrors(t, client.Close()) +} + func TestForDomain_EmitEventRejectsInvalidEvent(t *testing.T) { _, addr := startBatchCapturingChipServer(t) client := newDomainTestClient(t, domainTestConfig(t, addr)) From 66af14e0b5bcb527af64deb399ad567c49579c06 Mon Sep 17 00:00:00 2001 From: 4of9 <177086174+4of9@users.noreply.github.com> Date: Wed, 28 Oct 2026 14:23:28 -0400 Subject: [PATCH 04/15] beholder: add domain emitter outage and failure tests Covers server down, hung server + Close (each callback fires once), buffer full, a server without PublishBatchNoAck, concurrent EmitEvent and Close, the Emit attribute path, and construction failures. Co-Authored-By: Claude Opus 5.5 --- pkg/beholder/domain_emitter_failure_test.go | 298 ++++++++++++++++++++ 1 file changed, 298 insertions(+) create mode 100644 pkg/beholder/domain_emitter_failure_test.go diff --git a/pkg/beholder/domain_emitter_failure_test.go b/pkg/beholder/domain_emitter_failure_test.go new file mode 100644 index 0000000000..e0d6edd5eb --- /dev/null +++ b/pkg/beholder/domain_emitter_failure_test.go @@ -0,0 +1,298 @@ +package beholder_test + +import ( + "context" + "errors" + "fmt" + "net" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc" + sdklog "go.opentelemetry.io/otel/sdk/log" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + + "github.com/smartcontractkit/chainlink-common/pkg/beholder" + "github.com/smartcontractkit/chainlink-common/pkg/chipingress/batch" + "github.com/smartcontractkit/chainlink-common/pkg/chipingress/pb" +) + +// hungChipServer accepts batch RPCs and never answers until the RPC is cancelled. +type hungChipServer struct { + pb.UnimplementedChipIngressServer + received atomic.Int64 +} + +func (s *hungChipServer) PublishBatchNoAck(ctx context.Context, _ *pb.CloudEventBatch) (*pb.PublishResponse, error) { + s.received.Add(1) + <-ctx.Done() + return nil, ctx.Err() +} + +// ackOnlyChipServer predates PublishBatchNoAck: it implements only PublishBatch. +type ackOnlyChipServer struct { + pb.UnimplementedChipIngressServer +} + +func (s *ackOnlyChipServer) PublishBatch(_ context.Context, in *pb.CloudEventBatch) (*pb.PublishResponse, error) { + return &pb.PublishResponse{Results: make([]*pb.PublishResult, len(in.Events))}, nil +} + +func startChipServer(t *testing.T, impl pb.ChipIngressServer) string { + t.Helper() + lis, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", "127.0.0.1:0") + require.NoError(t, err) + srv := grpc.NewServer() + pb.RegisterChipIngressServer(srv, impl) + go func() { _ = srv.Serve(lis) }() + t.Cleanup(srv.Stop) + return lis.Addr().String() +} + +// closeBound bounds Client.Close in these tests. Close always spends ~10s flushing the OTel +// providers to the absent test collector; the domain emitters add at most their drain timeout +// (under 1s here), so anything near this bound is a real hang. +const closeBound = 20 * time.Second + +// unusedAddr returns an address nothing listens on. +func unusedAddr(t *testing.T) string { + t.Helper() + lis, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", "127.0.0.1:0") + require.NoError(t, err) + addr := lis.Addr().String() + require.NoError(t, lis.Close()) + return addr +} + +func newStartedDomainClient(t *testing.T, cfg beholder.Config) *beholder.Client { + t.Helper() + client, err := beholder.NewGRPCClient(cfg, func(...otlploggrpc.Option) (sdklog.Exporter, error) { + return &mockLogExporter{}, nil + }) + require.NoError(t, err) + require.NoError(t, client.Start(t.Context())) + return client +} + +// callbackCounter records how many times each event's callback fired and with what error. +type callbackCounter struct { + mu sync.Mutex + counts map[int]int + errs map[int]error +} + +func newCallbackCounter() *callbackCounter { + return &callbackCounter{counts: map[int]int{}, errs: map[int]error{}} +} + +func (c *callbackCounter) callback(i int) func(error) { + return func(err error) { + c.mu.Lock() + defer c.mu.Unlock() + c.counts[i]++ + c.errs[i] = err + } +} + +func (c *callbackCounter) snapshot() (map[int]int, map[int]error) { + c.mu.Lock() + defer c.mu.Unlock() + counts := make(map[int]int, len(c.counts)) + errs := make(map[int]error, len(c.errs)) + for k, v := range c.counts { + counts[k] = v + } + for k, v := range c.errs { + errs[k] = v + } + return counts, errs +} + +func TestDomainEmitter_ServerDown(t *testing.T) { + cfg := domainTestConfig(t, unusedAddr(t)) + cfg.ChipIngressSendTimeout = 500 * time.Millisecond + cfg.ChipIngressDrainTimeout = time.Second + client := newStartedDomainClient(t, cfg) + + em, err := client.ForDomain(beholder.DomainDataFeeds) + require.NoError(t, err) + + done := make(chan error, 1) + require.NoError(t, em.EmitEvent(t.Context(), legacyStyleEvent(t, "data-feeds.telemetry.down"), func(err error) { done <- err })) + select { + case err := <-done: + require.Error(t, err, "an unreachable server fails the event") + case <-time.After(5 * time.Second): + t.Fatal("callback not invoked while the server is down") + } + + closed := make(chan error, 1) + go func() { closed <- client.Close() }() + select { + case err := <-closed: + requireNoDomainEmitterCloseErrors(t, err) + case <-time.After(closeBound): + t.Fatal("Close hung while the server is down") + } +} + +func TestDomainEmitter_HungServerCloseFiresEachCallbackOnce(t *testing.T) { + hung := &hungChipServer{} + cfg := domainTestConfig(t, startChipServer(t, hung)) + cfg.ChipIngressSendTimeout = 30 * time.Second // the send outlives the drain + cfg.ChipIngressDrainTimeout = 500 * time.Millisecond + client := newStartedDomainClient(t, cfg) + + em, err := client.ForDomain(beholder.DomainDataFeeds) + require.NoError(t, err) + + const n = 5 + counter := newCallbackCounter() + for i := range n { + require.NoError(t, em.EmitEvent(t.Context(), legacyStyleEvent(t, fmt.Sprintf("data-feeds.telemetry.hung-%d", i)), counter.callback(i))) + } + require.Eventually(t, func() bool { return hung.received.Load() > 0 }, 5*time.Second, 10*time.Millisecond, + "a batch is in flight before Close") + + closed := make(chan struct{}) + go func() { _ = client.Close(); close(closed) }() + select { + case <-closed: + case <-time.After(closeBound): + t.Fatal("Close did not return against a hung server") + } + + // In-flight callbacks can fire just after Close returns (the connection is force-closed). + require.Eventually(t, func() bool { + counts, _ := counter.snapshot() + return len(counts) == n + }, 5*time.Second, 10*time.Millisecond, "every callback fires") + time.Sleep(200 * time.Millisecond) // give any duplicate a chance to show up + counts, errs := counter.snapshot() + for i := range n { + assert.Equal(t, 1, counts[i], "callback %d fires exactly once", i) + assert.Error(t, errs[i], "event %d was never acknowledged", i) + } +} + +func TestDomainEmitter_BufferFull(t *testing.T) { + cfg := domainTestConfig(t, startChipServer(t, &hungChipServer{})) + cfg.ChipIngressBufferSize = 1 + cfg.ChipIngressMaxBatchSize = 1 + cfg.ChipIngressMaxConcurrentSends = 1 + cfg.ChipIngressSendTimeout = 30 * time.Second + cfg.ChipIngressDrainTimeout = 200 * time.Millisecond + client := newStartedDomainClient(t, cfg) + + em, err := client.ForDomain(beholder.DomainDataFeeds) + require.NoError(t, err) + + // With the only send slot hung, the buffer fills; the next event is dropped via its callback. + var ( + pending sync.WaitGroup + dropped atomic.Bool + ) + for i := 0; i < 100 && !dropped.Load(); i++ { + pending.Add(1) + require.NoError(t, em.EmitEvent(t.Context(), legacyStyleEvent(t, fmt.Sprintf("data-feeds.telemetry.full-%d", i)), func(err error) { + defer pending.Done() + if errors.Is(err, batch.ErrMessageBufferFull) { + dropped.Store(true) + } + })) + time.Sleep(5 * time.Millisecond) + } + require.True(t, dropped.Load(), "an event is dropped with ErrMessageBufferFull") + + // The hung sends outlive the drain timeout, so their callbacks fire after Close returns. Wait + // for every one before the test ends, so nothing logs through the finished test's logger. + requireNoDomainEmitterCloseErrors(t, client.Close()) + waited := make(chan struct{}) + go func() { pending.Wait(); close(waited) }() + select { + case <-waited: + case <-time.After(closeBound): + t.Fatal("not every accepted event's callback fired") + } +} + +func TestDomainEmitter_ServerWithoutNoAck(t *testing.T) { + client := newDomainTestClient(t, domainTestConfig(t, startChipServer(t, &ackOnlyChipServer{}))) + + em, err := client.ForDomain(beholder.DomainDataFeeds) + require.NoError(t, err) + + done := make(chan error, 1) + require.NoError(t, em.EmitEvent(t.Context(), legacyStyleEvent(t, "data-feeds.telemetry.old-server"), func(err error) { done <- err })) + select { + case err := <-done: + require.Error(t, err, "no fallback to PublishBatch") + assert.Equal(t, codes.Unimplemented.String(), batch.ErrorCodeFor(err)) + case <-time.After(5 * time.Second): + t.Fatal("callback not invoked") + } +} + +func TestDomainEmitter_ConcurrentEmitAndClose(t *testing.T) { + _, addr := startBatchCapturingChipServer(t) + client := newStartedDomainClient(t, domainTestConfig(t, addr)) + + em, err := client.ForDomain(beholder.DomainDataFeeds) + require.NoError(t, err) + + var wg sync.WaitGroup + for g := range 8 { + wg.Go(func() { + for i := range 200 { + // Errors are expected once Close has stopped the emitter. + _ = em.EmitEvent(context.Background(), legacyStyleEvent(t, fmt.Sprintf("data-feeds.telemetry.race-%d-%d", g, i)), nil) + } + }) + } + time.Sleep(20 * time.Millisecond) + requireNoDomainEmitterCloseErrors(t, client.Close()) + wg.Wait() +} + +func TestDomainEmitter_EmitAttributePath(t *testing.T) { + capture, addr := startBatchCapturingChipServer(t) + client := newDomainTestClient(t, domainTestConfig(t, addr)) + + em, err := client.ForDomain(beholder.DomainDataFeeds) + require.NoError(t, err) + require.NoError(t, em.Emit(t.Context(), []byte("payload"), + beholder.AttrKeyDomain, "data-feeds.telemetry.emit", + beholder.AttrKeyEntity, "SomeEntity", + )) + + require.Eventually(t, func() bool { return len(capture.callsWithEventSource("data-feeds.telemetry.emit")) == 1 }, + 5*time.Second, 10*time.Millisecond) + call := capture.callsWithEventSource("data-feeds.telemetry.emit")[0] + assert.Equal(t, "PublishBatchNoAck", call.method) + assert.Equal(t, []string{"data-feeds"}, call.md.Get(beholder.DomainHeaderKey)) +} + +func TestNewGRPCClient_DomainEmitterConstructionFailures(t *testing.T) { + newClient := func(cfg beholder.Config) error { + _, err := beholder.NewGRPCClient(cfg, func(...otlploggrpc.Option) (sdklog.Exporter, error) { + return &mockLogExporter{}, nil + }) + return err + } + + t.Run("bad endpoint fails while building domain emitters", func(t *testing.T) { + err := newClient(domainTestConfig(t, "missing-port")) + require.ErrorContains(t, err, "domain") + }) + t.Run("failure after domain emitters are built", func(t *testing.T) { + // Domain emitters are built before this check, so their connections must be cleaned up. + cfg := domainTestConfig(t, "localhost:9999") + cfg.ChipIngressLogger = nil + require.ErrorContains(t, newClient(cfg), "ChipIngressLogger is required") + }) +} From 5770d394ea10d02b3400d89500ee87d769c06b78 Mon Sep 17 00:00:00 2001 From: 4of9 <177086174+4of9@users.noreply.github.com> Date: Wed, 28 Oct 2026 14:30:14 -0400 Subject: [PATCH 05/15] beholder: rate-limit per-event chip ingress failure logs During an outage the batch emitter logged one Error line per dropped event. It now logs the first failure per error code, then at most one per minute with a suppressed count. chip_ingress.events_dropped still counts every event. Co-Authored-By: Claude Opus 5.5 --- pkg/beholder/batch_emitter_service.go | 38 ++++++++++------ pkg/beholder/domain_emitter_failure_test.go | 28 ++++++++++++ pkg/beholder/error_log_limiter.go | 48 +++++++++++++++++++++ pkg/beholder/error_log_limiter_test.go | 33 ++++++++++++++ 4 files changed, 133 insertions(+), 14 deletions(-) create mode 100644 pkg/beholder/error_log_limiter.go create mode 100644 pkg/beholder/error_log_limiter_test.go diff --git a/pkg/beholder/batch_emitter_service.go b/pkg/beholder/batch_emitter_service.go index 78a45fe1bb..d95210315c 100644 --- a/pkg/beholder/batch_emitter_service.go +++ b/pkg/beholder/batch_emitter_service.go @@ -6,6 +6,7 @@ import ( "fmt" "strings" "sync" + "time" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" @@ -28,6 +29,8 @@ type ChipIngressBatchEmitterService struct { // domain, when set, restricts events to this Domain or its subdomains, so an event cannot // ride another domain's route. Unset for the default emitter. domain Domain + // errorLogs limits per-event failure logs to one per error code per errorLogInterval. + errorLogs *errorLogLimiter metricAttrsCache sync.Map // map[string]otelmetric.MeasurementOption metrics batchEmitterMetrics @@ -107,6 +110,7 @@ func newChipIngressBatchEmitterService(client chipingress.Client, cfg Config, lg batchClient: batchClient, clientName: clientName, domain: domain, + errorLogs: newErrorLogLimiter(errorLogInterval), metrics: metrics, } @@ -235,13 +239,16 @@ func (e *ChipIngressBatchEmitterService) queue(ctx context.Context, eventPb *chi // captures that it's happening and roughly why; the full reason isn't // needed at fleet-wide log volume. if _, ok := errors.AsType[*batch.PublishError](sendErr); !ok { - e.eng.Errorw("failed to emit to chip ingress", - "error", sendErr, - "error_code", errorCode, - "error_reason", sendErr.Error(), - "domain", domain, - "entity", entity, - ) + if log, suppressed := e.errorLogs.allow("emit:"+errorCode, time.Now()); log { + e.eng.Errorw("failed to emit to chip ingress", + "error", sendErr, + "error_code", errorCode, + "error_reason", sendErr.Error(), + "domain", domain, + "entity", entity, + "suppressed_since_last_log", suppressed, + ) + } } } else { e.metrics.eventsSent.Add(ctx, 1, metricAttrs) @@ -253,13 +260,16 @@ func (e *ChipIngressBatchEmitterService) queue(ctx context.Context, eventPb *chi if queueErr != nil { errorCode := batch.ErrorCodeFor(queueErr) e.metrics.eventsDropped.Add(ctx, 1, e.dropMetricAttrsFor(domain, entity, errorCode)) - e.eng.Errorw("failed to queue message for chip ingress", - "error", queueErr, - "error_code", errorCode, - "error_reason", queueErr.Error(), - "domain", domain, - "entity", entity, - ) + if log, suppressed := e.errorLogs.allow("queue:"+errorCode, time.Now()); log { + e.eng.Errorw("failed to queue message for chip ingress", + "error", queueErr, + "error_code", errorCode, + "error_reason", queueErr.Error(), + "domain", domain, + "entity", entity, + "suppressed_since_last_log", suppressed, + ) + } if callback != nil { callback(queueErr) } diff --git a/pkg/beholder/domain_emitter_failure_test.go b/pkg/beholder/domain_emitter_failure_test.go index e0d6edd5eb..cf3966f40e 100644 --- a/pkg/beholder/domain_emitter_failure_test.go +++ b/pkg/beholder/domain_emitter_failure_test.go @@ -14,12 +14,14 @@ import ( "github.com/stretchr/testify/require" "go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc" sdklog "go.opentelemetry.io/otel/sdk/log" + "go.uber.org/zap/zapcore" "google.golang.org/grpc" "google.golang.org/grpc/codes" "github.com/smartcontractkit/chainlink-common/pkg/beholder" "github.com/smartcontractkit/chainlink-common/pkg/chipingress/batch" "github.com/smartcontractkit/chainlink-common/pkg/chipingress/pb" + "github.com/smartcontractkit/chainlink-common/pkg/logger" ) // hungChipServer accepts batch RPCs and never answers until the RPC is cancelled. @@ -238,6 +240,32 @@ func TestDomainEmitter_ServerWithoutNoAck(t *testing.T) { } } +func TestDomainEmitter_OutageLogsAreRateLimited(t *testing.T) { + lggr, logs := logger.TestObserved(t, zapcore.ErrorLevel) + cfg := domainTestConfig(t, startChipServer(t, &ackOnlyChipServer{})) + cfg.ChipIngressLogger = lggr + cfg.ChipIngressMaxBatchSize = 1 // one RPC, and so one failure, per event + client := newDomainTestClient(t, cfg) + + em, err := client.ForDomain(beholder.DomainDataFeeds) + require.NoError(t, err) + + const n = 20 + var pending sync.WaitGroup + for i := range n { + pending.Add(1) + require.NoError(t, em.EmitEvent(t.Context(), legacyStyleEvent(t, fmt.Sprintf("data-feeds.telemetry.outage-%d", i)), + func(err error) { + defer pending.Done() + assert.Error(t, err) + })) + } + pending.Wait() + + assert.Equal(t, 1, logs.FilterMessage("failed to emit to chip ingress").Len(), + "every event failed, but only the first failure per error code is logged") +} + func TestDomainEmitter_ConcurrentEmitAndClose(t *testing.T) { _, addr := startBatchCapturingChipServer(t) client := newStartedDomainClient(t, domainTestConfig(t, addr)) diff --git a/pkg/beholder/error_log_limiter.go b/pkg/beholder/error_log_limiter.go new file mode 100644 index 0000000000..8f4b181dd4 --- /dev/null +++ b/pkg/beholder/error_log_limiter.go @@ -0,0 +1,48 @@ +package beholder + +import ( + "sync" + "time" +) + +// errorLogInterval is how often a batch emitter logs a repeating failure for one error code. +const errorLogInterval = time.Minute + +// errorLogLimiter rate-limits per-event failure logs: the first failure per key is logged, then +// at most one per interval, reporting how many were suppressed in between. During an outage this +// turns one log line per dropped event into one per error code per interval; the +// chip_ingress.events_dropped metric still counts every event. Keys are bounded (error codes). +type errorLogLimiter struct { + interval time.Duration + + mu sync.Mutex + state map[string]*errorLogState +} + +type errorLogState struct { + lastLogged time.Time + suppressed int +} + +func newErrorLogLimiter(interval time.Duration) *errorLogLimiter { + return &errorLogLimiter{interval: interval, state: make(map[string]*errorLogState)} +} + +// allow reports whether a failure for key should be logged now and, if so, how many were +// suppressed since the last log for that key. +func (l *errorLogLimiter) allow(key string, now time.Time) (ok bool, suppressed int) { + l.mu.Lock() + defer l.mu.Unlock() + s, seen := l.state[key] + if !seen { + l.state[key] = &errorLogState{lastLogged: now} + return true, 0 + } + if now.Sub(s.lastLogged) < l.interval { + s.suppressed++ + return false, 0 + } + suppressed = s.suppressed + s.lastLogged, s.suppressed = now, 0 + return true, suppressed +} diff --git a/pkg/beholder/error_log_limiter_test.go b/pkg/beholder/error_log_limiter_test.go new file mode 100644 index 0000000000..2751d585ce --- /dev/null +++ b/pkg/beholder/error_log_limiter_test.go @@ -0,0 +1,33 @@ +package beholder + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +func TestErrorLogLimiter(t *testing.T) { + start := time.Unix(1000, 0) + l := newErrorLogLimiter(time.Minute) + + ok, suppressed := l.allow("Unavailable", start) + assert.True(t, ok, "the first failure for a code is logged") + assert.Zero(t, suppressed) + + for i := range 3 { + ok, _ = l.allow("Unavailable", start.Add(time.Duration(i+1)*time.Second)) + assert.False(t, ok, "repeats within the interval are suppressed") + } + + ok, suppressed = l.allow("DeadlineExceeded", start.Add(5*time.Second)) + assert.True(t, ok, "each code is limited independently") + assert.Zero(t, suppressed) + + ok, suppressed = l.allow("Unavailable", start.Add(time.Minute)) + assert.True(t, ok, "the next failure after the interval is logged") + assert.Equal(t, 3, suppressed, "and reports how many were suppressed since the last log") + + ok, _ = l.allow("Unavailable", start.Add(time.Minute+time.Second)) + assert.False(t, ok, "the interval restarts from the last log") +} From b3de7aa7469ffd016b8981aafd18f247cb2f9531 Mon Sep 17 00:00:00 2001 From: 4of9 <177086174+4of9@users.noreply.github.com> Date: Wed, 28 Oct 2026 14:35:52 -0400 Subject: [PATCH 06/15] beholder: address domain emitter review nits Domain list as a func so it cannot be mutated, slices.Concat for the domain emitter options, and docs for the source check (case-sensitive, "." separator) and for ChipIngressDomainEmittersEnabled (always batch emitters, per-node cost). Co-Authored-By: Claude Opus 5.5 --- pkg/beholder/client.go | 2 +- pkg/beholder/config.go | 4 ++++ pkg/beholder/domain_emitter.go | 19 +++++++++++++------ pkg/beholder/domain_emitter_test.go | 3 ++- 4 files changed, 20 insertions(+), 8 deletions(-) diff --git a/pkg/beholder/client.go b/pkg/beholder/client.go index 0c910f5ad1..d7776249a7 100644 --- a/pkg/beholder/client.go +++ b/pkg/beholder/client.go @@ -78,7 +78,7 @@ func (c *Client) initService(lggr pkglogger.Logger, batchSvc *ChipIngressBatchEm if batchSvc != nil { subs = append(subs, batchSvc) } - for _, domain := range domains { + for _, domain := range allDomains() { if em, ok := c.domainEmitters[domain]; ok { subs = append(subs, em.ChipIngressBatchEmitterService) } diff --git a/pkg/beholder/config.go b/pkg/beholder/config.go index d824f5cdcf..04004b026b 100644 --- a/pkg/beholder/config.go +++ b/pkg/beholder/config.go @@ -83,6 +83,10 @@ type Config struct { // Client.ForDomain. Each has its own connection and buffer (sized by the batch settings // above), sends the DomainHeaderKey routing header, sends to chip ingress only, and publishes // with PublishBatchNoAck. Requires chip ingress. Off by default. + // + // Domain emitters are always batch emitters, whatever ChipIngressBatchEmitterEnabled says. + // Per-node cost: one ChipIngressBufferSize buffer per Domain; a connection only for domains + // actually used (gRPC dials lazily). ChipIngressDomainEmittersEnabled bool // OTel Log diff --git a/pkg/beholder/domain_emitter.go b/pkg/beholder/domain_emitter.go index 13cef7efa1..33130ebae4 100644 --- a/pkg/beholder/domain_emitter.go +++ b/pkg/beholder/domain_emitter.go @@ -3,6 +3,7 @@ package beholder import ( "context" "fmt" + "slices" "github.com/smartcontractkit/chainlink-common/pkg/chipingress" "github.com/smartcontractkit/chainlink-common/pkg/chipingress/batch" @@ -20,8 +21,11 @@ const ( DomainCCIP Domain = "ccip" ) -// domains lists every Domain; Config.ChipIngressDomainEmittersEnabled builds an emitter for each. -var domains = []Domain{DomainDataFeeds, DomainDataStreams, DomainCCIP} +// allDomains lists every Domain; Config.ChipIngressDomainEmittersEnabled builds an emitter for +// each. A func, not a var, so the list cannot be modified. +func allDomains() []Domain { + return []Domain{DomainDataFeeds, DomainDataStreams, DomainCCIP} +} // DomainHeaderKey is the gRPC metadata key carrying a domain emitter's Domain. Load balancers route // on it to the chip ingress deployment that serves the domain; chip ingress itself ignores it. @@ -31,8 +35,10 @@ const DomainHeaderKey = "chainlink-domain" // connection and buffer. Obtain one with Client.ForDomain. // // Both methods reject an event whose source (the CloudEvent source, or the beholder_domain -// attribute for Emit) is not the emitter's Domain or a subdomain of it: for DomainDataFeeds, -// "data-feeds" and "data-feeds.telemetry.ocr2-median" are accepted, "ccip.telemetry.ocr3" is not. +// attribute for Emit) is not the emitter's Domain or a subdomain of it. The match is +// case-sensitive and a subdomain must follow a "." separator: for DomainDataFeeds, +// "data-feeds" and "data-feeds.telemetry.ocr2-median" are accepted; "ccip.telemetry.ocr3", +// "data-feeds-extra" and "Data-Feeds" are not. type DomainEmitter interface { // Emitter builds the event from attributes, like the Client's Emitter. Close is a no-op: the // Client starts and closes its domain emitters. @@ -70,6 +76,7 @@ func (*domainEmitter) Close() error { return nil } // (transport, auth, OTel providers) without its header provider, which each domain emitter // replaces with one carrying the resource attribute headers plus its DomainHeaderKey. func newDomainEmitters(cfg Config, opts []chipingress.Opt, resourceAttrs map[string]string, lggr pkglogger.Logger) (_ map[Domain]*domainEmitter, err error) { + domains := allDomains() emitters := make(map[Domain]*domainEmitter, len(domains)) defer func() { if err != nil { @@ -83,11 +90,11 @@ func newDomainEmitters(cfg Config, opts []chipingress.Opt, resourceAttrs map[str headers[DomainHeaderKey] = string(domain) clientName := batch.ClientNameBeholder + "_" + string(domain) - domainOpts := append(append([]chipingress.Opt{}, opts...), + domainOpts := slices.Concat(opts, []chipingress.Opt{ chipingress.WithHeaderProvider(chipingress.NewStaticHeaderProvider(headers)), chipingress.WithNOPLookup(), chipingress.WithClientName(clientName), - ) + }) chipClient, err := chipingress.NewClient(cfg.ChipIngressEmitterGRPCEndpoint, domainOpts...) if err != nil { return nil, fmt.Errorf("chip ingress client for domain %q: %w", domain, err) diff --git a/pkg/beholder/domain_emitter_test.go b/pkg/beholder/domain_emitter_test.go index 510146fe87..b536cdaa61 100644 --- a/pkg/beholder/domain_emitter_test.go +++ b/pkg/beholder/domain_emitter_test.go @@ -307,6 +307,7 @@ func TestForDomain_RejectsEventsFromOtherDomains(t *testing.T) { {"ccip.telemetry.ocr3", false}, // another domain {"data-feeds-extra.telemetry", false}, // shares a prefix, not a subdomain {"telemetry.data-feeds", false}, // contains it, not under it + {"Data-Feeds.telemetry", false}, // the match is case-sensitive } { t.Run(tc.source, func(t *testing.T) { called := false @@ -331,7 +332,7 @@ func TestForDomain_RejectsEventsFromOtherDomains(t *testing.T) { }) // Nothing from the rejected sources reached the server. - for _, source := range []string{"ccip.telemetry.ocr3", "data-feeds-extra.telemetry", "telemetry.data-feeds"} { + for _, source := range []string{"ccip.telemetry.ocr3", "data-feeds-extra.telemetry", "telemetry.data-feeds", "Data-Feeds.telemetry"} { assert.Empty(t, capture.callsWithEventSource(source), "source %q", source) } } From 79eb8d9ac766a7d372155a52a90ca19c0c9c54d3 Mon Sep 17 00:00:00 2001 From: 4of9 <177086174+4of9@users.noreply.github.com> Date: Thu, 29 Oct 2026 12:42:35 -0400 Subject: [PATCH 07/15] beholder: harden domain emitters after review - Report queueing failures to the callback after the service lock is released, so a callback that re-enters the emitter cannot deadlock with Close (also fixes EmitWithCallback). - Require ChipIngressLogger when domain emitters are enabled. - ForDomain errors wrap ErrDomainEmittersDisabled or ErrUnknownDomain. - Reject ChipIngressDomainEmittersEnabled in the HTTP client. - Close the chip ingress client when NewGRPCClient fails after creating it. Co-Authored-By: Claude Opus 5.5 --- pkg/beholder/batch_emitter_service.go | 40 +++++++--- .../batch_emitter_service_reentrant_test.go | 79 +++++++++++++++++++ pkg/beholder/client.go | 9 +++ pkg/beholder/domain_emitter.go | 19 ++++- pkg/beholder/domain_emitter_failure_test.go | 32 ++++++-- pkg/beholder/domain_emitter_test.go | 36 ++++++++- pkg/beholder/httpclient.go | 3 + 7 files changed, 195 insertions(+), 23 deletions(-) create mode 100644 pkg/beholder/batch_emitter_service_reentrant_test.go diff --git a/pkg/beholder/batch_emitter_service.go b/pkg/beholder/batch_emitter_service.go index d95210315c..039e5d6596 100644 --- a/pkg/beholder/batch_emitter_service.go +++ b/pkg/beholder/batch_emitter_service.go @@ -155,7 +155,8 @@ func (e *ChipIngressBatchEmitterService) EmitWithCallback(ctx context.Context, b } func (e *ChipIngressBatchEmitterService) emitInternal(ctx context.Context, body []byte, callback func(error), attrKVs ...any) error { - return e.eng.IfStarted(func() error { + var queueErr error + if err := e.eng.IfStarted(func() error { domain, entity, err := ExtractSourceAndType(attrKVs...) if err != nil { return err @@ -179,9 +180,13 @@ func (e *ChipIngressBatchEmitterService) emitInternal(ctx context.Context, body return err } - e.queue(ctx, eventPb, domain, entity, callback) + queueErr = e.queue(ctx, eventPb, domain, entity, callback) return nil - }) + }); err != nil { + return err + } + e.failQueued(callback, queueErr) + return nil } // EmitEvent queues a fully built CloudEvent for batched delivery without blocking. Unlike @@ -190,7 +195,8 @@ func (e *ChipIngressBatchEmitterService) emitInternal(ctx context.Context, body // If EmitEvent returns a non-nil error, the callback will NOT be invoked. // If it returns nil, a non-nil callback is guaranteed to fire exactly once. func (e *ChipIngressBatchEmitterService) EmitEvent(ctx context.Context, event chipingress.CloudEvent, callback func(error)) error { - return e.eng.IfStarted(func() error { + var queueErr error + if err := e.eng.IfStarted(func() error { if err := event.Validate(); err != nil { return fmt.Errorf("invalid CloudEvent: %w", err) } @@ -206,9 +212,13 @@ func (e *ChipIngressBatchEmitterService) EmitEvent(ctx context.Context, event ch return err } - e.queue(ctx, eventPb, event.Source(), event.Type(), callback) + queueErr = e.queue(ctx, eventPb, event.Source(), event.Type(), callback) return nil - }) + }); err != nil { + return err + } + e.failQueued(callback, queueErr) + return nil } // checkDomain rejects a source outside e.domain: it must be the domain itself or a subdomain of @@ -220,9 +230,19 @@ func (e *ChipIngressBatchEmitterService) checkDomain(source string) error { return fmt.Errorf("beholder: event source %q is not in domain %q", source, e.domain) } +// failQueued reports a queueing failure to the caller's callback. It runs after the service's +// lock is released, so a callback may safely call back into the emitter (e.g. to retry) even +// while Close is waiting to stop the service. +func (e *ChipIngressBatchEmitterService) failQueued(callback func(error), queueErr error) { + if queueErr != nil && callback != nil { + callback(queueErr) + } +} + // queue hands eventPb to the batch client and records its fate. domain and entity are the -// event's source and type, used for metrics and logs. -func (e *ChipIngressBatchEmitterService) queue(ctx context.Context, eventPb *chipingress.CloudEventPb, domain, entity string, callback func(error)) { +// event's source and type, used for metrics and logs. A queueing failure is recorded and +// returned, not passed to callback: the caller does that once the service's lock is released. +func (e *ChipIngressBatchEmitterService) queue(ctx context.Context, eventPb *chipingress.CloudEventPb, domain, entity string, callback func(error)) error { metricAttrs := e.metricAttrsFor(domain, entity) queueErr := e.batchClient.QueueMessage(eventPb, func(sendErr error) { @@ -270,10 +290,8 @@ func (e *ChipIngressBatchEmitterService) queue(ctx context.Context, eventPb *chi "suppressed_since_last_log", suppressed, ) } - if callback != nil { - callback(queueErr) - } } + return queueErr } func (e *ChipIngressBatchEmitterService) metricAttrsFor(domain, entity string) otelmetric.MeasurementOption { diff --git a/pkg/beholder/batch_emitter_service_reentrant_test.go b/pkg/beholder/batch_emitter_service_reentrant_test.go new file mode 100644 index 0000000000..0718598fd7 --- /dev/null +++ b/pkg/beholder/batch_emitter_service_reentrant_test.go @@ -0,0 +1,79 @@ +package beholder + +import ( + "context" + "errors" + "fmt" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + + "github.com/smartcontractkit/chainlink-common/pkg/chipingress" + "github.com/smartcontractkit/chainlink-common/pkg/chipingress/batch" + "github.com/smartcontractkit/chainlink-common/pkg/logger" +) + +// hangingNoAckClient never answers PublishBatchNoAck until released or the call's context ends. +type hangingNoAckClient struct { + chipingress.NoopClient + release chan struct{} +} + +func (c *hangingNoAckClient) PublishBatchNoAck(ctx context.Context, _ *chipingress.CloudEventBatch, _ ...grpc.CallOption) (*chipingress.PublishResponse, error) { + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-c.release: + return &chipingress.PublishResponse{}, nil + } +} + +// A callback that re-enters EmitEvent (e.g. to retry a buffer-full drop) while Close is waiting +// to stop the service must not deadlock: the callback must not run under the service's lock. +func TestChipIngressBatchEmitterService_ReentrantCallbackDuringClose(t *testing.T) { + client := &hangingNoAckClient{release: make(chan struct{})} + t.Cleanup(func() { close(client.release) }) + + svc, err := newChipIngressBatchEmitterService(client, Config{ + ChipIngressBufferSize: 1, + ChipIngressMaxBatchSize: 1, + ChipIngressMaxConcurrentSends: 1, + ChipIngressSendInterval: 10 * time.Millisecond, + ChipIngressSendTimeout: 30 * time.Second, + ChipIngressDrainTimeout: 200 * time.Millisecond, + }, logger.Test(t), "test", "", batch.WithNoAck()) + require.NoError(t, err) + require.NoError(t, svc.Start(t.Context())) + + event := func(i int) chipingress.CloudEvent { + ev, err := chipingress.NewEvent("test-domain", "Entity", []byte(fmt.Sprint(i)), nil) + require.NoError(t, err) + return ev + } + + done := make(chan struct{}) + go func() { + defer close(done) + var retried atomic.Bool + for i := 0; i < 100 && !retried.Load(); i++ { + _ = svc.EmitEvent(context.Background(), event(i), func(err error) { + if !errors.Is(err, batch.ErrMessageBufferFull) || !retried.CompareAndSwap(false, true) { + return + } + go func() { _ = svc.Close() }() + time.Sleep(100 * time.Millisecond) // let Close start waiting to stop the service + _ = svc.EmitEvent(context.Background(), event(-1), nil) + }) + time.Sleep(5 * time.Millisecond) + } + }() + + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("deadlock: a callback re-entering EmitEvent during Close never returned") + } +} diff --git a/pkg/beholder/client.go b/pkg/beholder/client.go index d7776249a7..e80f1b4e3d 100644 --- a/pkg/beholder/client.go +++ b/pkg/beholder/client.go @@ -229,6 +229,10 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (_ *Client, er if cfg.ChipIngressDomainEmittersEnabled && !chipIngressEnabled { return nil, errors.New("ChipIngressDomainEmittersEnabled requires chip ingress (ChipIngressEmitterGRPCEndpoint)") } + // Domain emitters are always batch emitters; like the batch emitter, they need a real logger. + if cfg.ChipIngressDomainEmittersEnabled && cfg.ChipIngressLogger == nil { + return nil, errors.New("ChipIngressLogger is required when ChipIngressDomainEmittersEnabled is true") + } // if chip ingress is enabled, create dual source emitter that sends to both otel collector and chip ingress // eventually we will remove the dual source emitter and just use chip ingress if chipIngressEnabled { @@ -281,6 +285,11 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (_ *Client, er if err != nil { return nil, err } + defer func() { + if err != nil { + _ = chipIngressClient.Close() + } + }() var chipIngressEmitter Emitter if cfg.ChipIngressBatchEmitterEnabled { diff --git a/pkg/beholder/domain_emitter.go b/pkg/beholder/domain_emitter.go index 33130ebae4..a4043526fa 100644 --- a/pkg/beholder/domain_emitter.go +++ b/pkg/beholder/domain_emitter.go @@ -2,6 +2,7 @@ package beholder import ( "context" + "errors" "fmt" "slices" @@ -27,6 +28,14 @@ func allDomains() []Domain { return []Domain{DomainDataFeeds, DomainDataStreams, DomainCCIP} } +var ( + // ErrDomainEmittersDisabled is returned by ForDomain when the Client has no domain emitters: + // Config.ChipIngressDomainEmittersEnabled is off, or the Client is a noop or HTTP client. + ErrDomainEmittersDisabled = errors.New("beholder: chip ingress domain emitters are disabled") + // ErrUnknownDomain is returned by ForDomain for a Domain the Client has no emitter for. + ErrUnknownDomain = errors.New("beholder: unknown chip ingress domain") +) + // DomainHeaderKey is the gRPC metadata key carrying a domain emitter's Domain. Load balancers route // on it to the chip ingress deployment that serves the domain; chip ingress itself ignores it. const DomainHeaderKey = "chainlink-domain" @@ -50,13 +59,15 @@ type DomainEmitter interface { } // ForDomain returns the Client's emitter for domain. It creates nothing, so repeated calls return -// the same emitter. It is an error unless Config.ChipIngressDomainEmittersEnabled is set and domain -// is a known Domain. +// the same emitter. The error wraps ErrDomainEmittersDisabled if the Client has no domain emitters, +// or ErrUnknownDomain if domain is not a known Domain. func (c *Client) ForDomain(domain Domain) (DomainEmitter, error) { + if c.domainEmitters == nil { + return nil, fmt.Errorf("%w: no emitter for domain %q", ErrDomainEmittersDisabled, domain) + } em, ok := c.domainEmitters[domain] if !ok { - return nil, fmt.Errorf("beholder: no chip ingress emitter for domain %q (ChipIngressDomainEmittersEnabled=%t)", - domain, c.Config.ChipIngressDomainEmittersEnabled) + return nil, fmt.Errorf("%w: %q", ErrUnknownDomain, domain) } return em, nil } diff --git a/pkg/beholder/domain_emitter_failure_test.go b/pkg/beholder/domain_emitter_failure_test.go index cf3966f40e..73891067ca 100644 --- a/pkg/beholder/domain_emitter_failure_test.go +++ b/pkg/beholder/domain_emitter_failure_test.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "net" + "runtime" "sync" "sync/atomic" "testing" @@ -315,12 +316,29 @@ func TestNewGRPCClient_DomainEmitterConstructionFailures(t *testing.T) { t.Run("bad endpoint fails while building domain emitters", func(t *testing.T) { err := newClient(domainTestConfig(t, "missing-port")) - require.ErrorContains(t, err, "domain") - }) - t.Run("failure after domain emitters are built", func(t *testing.T) { - // Domain emitters are built before this check, so their connections must be cleaned up. - cfg := domainTestConfig(t, "localhost:9999") - cfg.ChipIngressLogger = nil - require.ErrorContains(t, newClient(cfg), "ChipIngressLogger is required") + require.ErrorContains(t, err, `chip ingress client for domain "data-feeds"`) }) } + +// A NewGRPCClient that fails after creating its chip ingress client must close it. The +// ChipIngressLogger check fails after the default chip client exists (domain emitters off, since +// they require the logger up front). Each leaked client keeps gRPC goroutines alive. +func TestNewGRPCClient_ConstructionFailureClosesChipClient(t *testing.T) { + cfg := beholder.Config{ + OtelExporterGRPCEndpoint: "localhost:4317", + ChipIngressEmitterEnabled: true, + ChipIngressEmitterGRPCEndpoint: "localhost:9999", + ChipIngressInsecureConnection: true, + ChipIngressBatchEmitterEnabled: true, // with a nil ChipIngressLogger + } + before := runtime.NumGoroutine() + for range 3 { + _, err := beholder.NewGRPCClient(cfg, func(...otlploggrpc.Option) (sdklog.Exporter, error) { + return &mockLogExporter{}, nil + }) + require.ErrorContains(t, err, "ChipIngressLogger is required") + } + // Leaking the chip client costs 3 goroutines per failed call; allow some unrelated noise. + require.Eventually(t, func() bool { return runtime.NumGoroutine()-before < 5 }, 5*time.Second, 50*time.Millisecond, + "failed constructions leaked goroutines: before=%d now=%d", before, runtime.NumGoroutine()) +} diff --git a/pkg/beholder/domain_emitter_test.go b/pkg/beholder/domain_emitter_test.go index b536cdaa61..b43ba38380 100644 --- a/pkg/beholder/domain_emitter_test.go +++ b/pkg/beholder/domain_emitter_test.go @@ -357,5 +357,39 @@ func TestForDomain_DisabledByDefault(t *testing.T) { func TestNoopClient_ForDomain(t *testing.T) { _, err := beholder.NewNoopClient().ForDomain(beholder.DomainDataFeeds) - require.Error(t, err) + require.ErrorIs(t, err, beholder.ErrDomainEmittersDisabled) +} + +func TestForDomain_ErrorsAreDistinguishable(t *testing.T) { + _, addr := startBatchCapturingChipServer(t) + + enabled := newDomainTestClient(t, domainTestConfig(t, addr)) + _, err := enabled.ForDomain(beholder.Domain("not-a-domain")) + require.ErrorIs(t, err, beholder.ErrUnknownDomain) + require.NotErrorIs(t, err, beholder.ErrDomainEmittersDisabled) + + cfg := domainTestConfig(t, addr) + cfg.ChipIngressDomainEmittersEnabled = false + disabled := newDomainTestClient(t, cfg) + _, err = disabled.ForDomain(beholder.DomainDataFeeds) + require.ErrorIs(t, err, beholder.ErrDomainEmittersDisabled) + require.NotErrorIs(t, err, beholder.ErrUnknownDomain) +} + +func TestNewGRPCClient_DomainEmittersRequireLogger(t *testing.T) { + cfg := domainTestConfig(t, "localhost:9999") + cfg.ChipIngressLogger = nil + cfg.ChipIngressBatchEmitterEnabled = false // the batch emitter's own logger check must not be what fails + _, err := beholder.NewGRPCClient(cfg, func(...otlploggrpc.Option) (sdklog.Exporter, error) { + return &mockLogExporter{}, nil + }) + require.ErrorContains(t, err, "ChipIngressLogger is required when ChipIngressDomainEmittersEnabled is true") +} + +func TestNewHTTPClient_RejectsDomainEmitters(t *testing.T) { + _, err := beholder.NewClient(beholder.Config{ + OtelExporterHTTPEndpoint: "localhost:4318", + ChipIngressDomainEmittersEnabled: true, + }) + require.ErrorContains(t, err, "ChipIngressDomainEmittersEnabled is not supported by the HTTP client") } diff --git a/pkg/beholder/httpclient.go b/pkg/beholder/httpclient.go index adbc2f9f9c..8e5dac4865 100644 --- a/pkg/beholder/httpclient.go +++ b/pkg/beholder/httpclient.go @@ -37,6 +37,9 @@ func newCertFromFile(certFile string) (*x509.CertPool, error) { // NewHTTPClient creates a HTTP based beholder Client. Use NewClient to create a client from a Config which will pick // the best client type from the Config. func NewHTTPClient(cfg Config, otlploghttpNew otlploghttpFactory) (*Client, error) { + if cfg.ChipIngressDomainEmittersEnabled { + return nil, errors.New("ChipIngressDomainEmittersEnabled is not supported by the HTTP client") + } baseResource, err := newOtelResource(cfg) if err != nil { return nil, err From 6ca644e74c79316381174cfd88ff7f6ecf3963d4 Mon Sep 17 00:00:00 2001 From: 4of9 <177086174+4of9@users.noreply.github.com> Date: Thu, 29 Oct 2026 12:53:21 -0400 Subject: [PATCH 08/15] beholder: close domain emitter test gaps The invalid-event test now isolates CloudEvent validation (it passed on the domain check before). New coverage: a cancelled context, domain emitters with the batch emitter disabled, the per-domain client name, the logged event source, and rate-limited buffer-full logs. The concurrency test no longer calls require off the test goroutine and checks that emits overlap Close. Co-Authored-By: Claude Opus 5.5 --- pkg/beholder/domain_emitter_failure_test.go | 69 +++++++++++++++++++-- pkg/beholder/domain_emitter_test.go | 41 +++++++++++- 2 files changed, 103 insertions(+), 7 deletions(-) diff --git a/pkg/beholder/domain_emitter_failure_test.go b/pkg/beholder/domain_emitter_failure_test.go index 73891067ca..e70338da37 100644 --- a/pkg/beholder/domain_emitter_failure_test.go +++ b/pkg/beholder/domain_emitter_failure_test.go @@ -20,6 +20,7 @@ import ( "google.golang.org/grpc/codes" "github.com/smartcontractkit/chainlink-common/pkg/beholder" + "github.com/smartcontractkit/chainlink-common/pkg/chipingress" "github.com/smartcontractkit/chainlink-common/pkg/chipingress/batch" "github.com/smartcontractkit/chainlink-common/pkg/chipingress/pb" "github.com/smartcontractkit/chainlink-common/pkg/logger" @@ -263,8 +264,54 @@ func TestDomainEmitter_OutageLogsAreRateLimited(t *testing.T) { } pending.Wait() - assert.Equal(t, 1, logs.FilterMessage("failed to emit to chip ingress").Len(), - "every event failed, but only the first failure per error code is logged") + emitLogs := logs.FilterMessage("failed to emit to chip ingress") + if assert.Equal(t, 1, emitLogs.Len(), + "every event failed, but only the first failure per error code is logged") { + assert.Contains(t, emitLogs.All()[0].ContextMap()["domain"], "data-feeds.telemetry.outage-", + "the log carries the event's source") + } +} + +func TestDomainEmitter_BufferFullLogsAreRateLimited(t *testing.T) { + lggr, logs := logger.TestObserved(t, zapcore.ErrorLevel) + cfg := domainTestConfig(t, startChipServer(t, &hungChipServer{})) + cfg.ChipIngressLogger = lggr + cfg.ChipIngressBufferSize = 1 + cfg.ChipIngressMaxBatchSize = 1 + cfg.ChipIngressMaxConcurrentSends = 1 + cfg.ChipIngressSendTimeout = 30 * time.Second + cfg.ChipIngressDrainTimeout = 200 * time.Millisecond + client := newStartedDomainClient(t, cfg) + + em, err := client.ForDomain(beholder.DomainDataFeeds) + require.NoError(t, err) + + var ( + pending sync.WaitGroup + drops atomic.Int64 + ) + for i := 0; i < 200 && drops.Load() < 3; i++ { + pending.Add(1) + require.NoError(t, em.EmitEvent(t.Context(), legacyStyleEvent(t, fmt.Sprintf("data-feeds.telemetry.fulllog-%d", i)), func(err error) { + defer pending.Done() + if errors.Is(err, batch.ErrMessageBufferFull) { + drops.Add(1) + } + })) + time.Sleep(5 * time.Millisecond) + } + require.GreaterOrEqual(t, drops.Load(), int64(3), "several events are dropped with ErrMessageBufferFull") + assert.Equal(t, 1, logs.FilterMessage("failed to queue message for chip ingress").Len(), + "only the first buffer-full drop per interval is logged") + + requireNoDomainEmitterCloseErrors(t, client.Close()) + waited := make(chan struct{}) + go func() { pending.Wait(); close(waited) }() + select { + case <-waited: + case <-time.After(closeBound): + t.Fatal("not every accepted event's callback fired") + } } func TestDomainEmitter_ConcurrentEmitAndClose(t *testing.T) { @@ -274,18 +321,30 @@ func TestDomainEmitter_ConcurrentEmitAndClose(t *testing.T) { em, err := client.ForDomain(beholder.DomainDataFeeds) require.NoError(t, err) - var wg sync.WaitGroup + var ( + wg sync.WaitGroup + accepted atomic.Int64 + ) for g := range 8 { wg.Go(func() { for i := range 200 { + // Built without require: FailNow must not be called off the test goroutine. + ev, err := chipingress.NewEvent(fmt.Sprintf("data-feeds.telemetry.race-%d-%d", g, i), "pb.TelemetryWrapper", []byte("payload"), nil) + if err != nil { + continue + } // Errors are expected once Close has stopped the emitter. - _ = em.EmitEvent(context.Background(), legacyStyleEvent(t, fmt.Sprintf("data-feeds.telemetry.race-%d-%d", g, i)), nil) + if em.EmitEvent(context.Background(), ev, nil) == nil { + accepted.Add(1) + } } }) } - time.Sleep(20 * time.Millisecond) + // Close only once emits are flowing, so the two genuinely overlap. + require.Eventually(t, func() bool { return accepted.Load() > 0 }, 5*time.Second, time.Millisecond) requireNoDomainEmitterCloseErrors(t, client.Close()) wg.Wait() + assert.Positive(t, accepted.Load()) } func TestDomainEmitter_EmitAttributePath(t *testing.T) { diff --git a/pkg/beholder/domain_emitter_test.go b/pkg/beholder/domain_emitter_test.go index b43ba38380..1feedcbf2f 100644 --- a/pkg/beholder/domain_emitter_test.go +++ b/pkg/beholder/domain_emitter_test.go @@ -155,6 +155,8 @@ func TestForDomain_EmitEventSendsNoAckWithDomainHeader(t *testing.T) { assert.Equal(t, []string{val}, call.md.Get(key), "inherits resource attribute header %q", key) } assert.Equal(t, []string{"true"}, call.md.Get("x-include-nop-info"), "domain emitters request NOP info") + require.Len(t, call.md.Get("user-agent"), 1) + assert.Contains(t, call.md.Get("user-agent")[0], "beholder_data-feeds", "per-domain client name") require.Len(t, call.events, 1) attrs := call.events[0].Attributes @@ -285,12 +287,47 @@ func TestForDomain_EmitEventRejectsInvalidEvent(t *testing.T) { em, err := client.ForDomain(beholder.DomainDataFeeds) require.NoError(t, err) + // In-domain source, so only CloudEvent validation (a missing id) can reject it. + ev := legacyStyleEvent(t, "data-feeds.telemetry.invalid") + ev.SetID("") + + called := false + err = em.EmitEvent(t.Context(), ev, func(error) { called = true }) + require.ErrorContains(t, err, "invalid CloudEvent") + assert.False(t, called, "no callback when EmitEvent returns an error") +} + +func TestForDomain_EmitEventRejectsCancelledContext(t *testing.T) { + _, addr := startBatchCapturingChipServer(t) + client := newDomainTestClient(t, domainTestConfig(t, addr)) + + em, err := client.ForDomain(beholder.DomainDataFeeds) + require.NoError(t, err) + + ctx, cancel := context.WithCancel(t.Context()) + cancel() called := false - err = em.EmitEvent(t.Context(), chipingress.CloudEvent{}, func(error) { called = true }) - require.Error(t, err, "an event without id/source/type is rejected up front") + err = em.EmitEvent(ctx, legacyStyleEvent(t, "data-feeds.telemetry.cancelled"), func(error) { called = true }) + require.ErrorIs(t, err, context.Canceled) assert.False(t, called, "no callback when EmitEvent returns an error") } +// Domain emitters are always batch emitters, whatever ChipIngressBatchEmitterEnabled says. +func TestForDomain_WorksWithBatchEmitterDisabled(t *testing.T) { + capture, addr := startBatchCapturingChipServer(t) + cfg := domainTestConfig(t, addr) + cfg.ChipIngressBatchEmitterEnabled = false + client := newDomainTestClient(t, cfg) + + em, err := client.ForDomain(beholder.DomainDataFeeds) + require.NoError(t, err) + require.NoError(t, em.EmitEvent(t.Context(), legacyStyleEvent(t, "data-feeds.telemetry.unbatched"), nil)) + + require.Eventually(t, func() bool { return len(capture.callsWithEventSource("data-feeds.telemetry.unbatched")) == 1 }, + 5*time.Second, 10*time.Millisecond) + assert.Equal(t, "PublishBatchNoAck", capture.callsWithEventSource("data-feeds.telemetry.unbatched")[0].method) +} + func TestForDomain_RejectsEventsFromOtherDomains(t *testing.T) { capture, addr := startBatchCapturingChipServer(t) client := newDomainTestClient(t, domainTestConfig(t, addr)) From 7ff0b8b607e494df6ea30a2912a018a2f7d2248d Mon Sep 17 00:00:00 2001 From: 4of9 <177086174+4of9@users.noreply.github.com> Date: Thu, 29 Oct 2026 12:54:22 -0400 Subject: [PATCH 09/15] beholder: document the domain emitter contract Spell out what callers need: source and type must be low-cardinality, EmitEvent sends the event as given (seqnum is overwritten, nothing is stamped, the event must not be modified afterwards), when callbacks fire and what a nil no-ack result means, that the source check is a client-side guard, and that the Client must be started before Close to release connections. Co-Authored-By: Claude Opus 5.5 --- pkg/beholder/batch_emitter_service.go | 22 +++++++++++++-------- pkg/beholder/domain_emitter.go | 28 +++++++++++++++++++-------- 2 files changed, 34 insertions(+), 16 deletions(-) diff --git a/pkg/beholder/batch_emitter_service.go b/pkg/beholder/batch_emitter_service.go index 039e5d6596..10be6a24fd 100644 --- a/pkg/beholder/batch_emitter_service.go +++ b/pkg/beholder/batch_emitter_service.go @@ -18,16 +18,18 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/services" ) -// ChipIngressBatchEmitterService batches events and sends them via chipingress.Client.PublishBatch. -// It implements the Emitter interface. +// ChipIngressBatchEmitterService batches events and sends them to chip ingress with +// chipingress.Client.PublishBatch (or PublishBatchNoAck when built with batch.WithNoAck, as domain +// emitters are). It implements the Emitter interface. A domain emitter also restricts events to +// its Domain; see DomainEmitter. type ChipIngressBatchEmitterService struct { services.Service eng *services.Engine batchClient *batch.Client clientName string - // domain, when set, restricts events to this Domain or its subdomains, so an event cannot - // ride another domain's route. Unset for the default emitter. + // domain, when set, restricts events to this Domain or its subdomains: a client-side guard so + // an event is not sent down another domain's route. Unset for the default emitter. domain Domain // errorLogs limits per-event failure logs to one per error code per errorLogInterval. errorLogs *errorLogLimiter @@ -189,11 +191,14 @@ func (e *ChipIngressBatchEmitterService) emitInternal(ctx context.Context, body return nil } -// EmitEvent queues a fully built CloudEvent for batched delivery without blocking. Unlike -// Emit, which builds the event from attributes, it keeps every extension the caller set. +// EmitEvent queues a fully built CloudEvent for batched delivery without blocking. Unlike Emit, +// which builds the event from attributes, it sends the event as given (see DomainEmitter.EmitEvent +// for what that means, including the "seqnum" overwrite and not reusing the event afterwards). // // If EmitEvent returns a non-nil error, the callback will NOT be invoked. -// If it returns nil, a non-nil callback is guaranteed to fire exactly once. +// If it returns nil, a non-nil callback is guaranteed to fire exactly once: synchronously before +// EmitEvent returns if the event cannot be queued, otherwise after the batch is sent (possibly +// after Close returns). func (e *ChipIngressBatchEmitterService) EmitEvent(ctx context.Context, event chipingress.CloudEvent, callback func(error)) error { var queueErr error if err := e.eng.IfStarted(func() error { @@ -222,7 +227,8 @@ func (e *ChipIngressBatchEmitterService) EmitEvent(ctx context.Context, event ch } // checkDomain rejects a source outside e.domain: it must be the domain itself or a subdomain of -// it (e.g. "data-feeds" or "data-feeds.telemetry.ocr2-median" for DomainDataFeeds). +// it (e.g. "data-feeds" or "data-feeds.telemetry.ocr2-median" for DomainDataFeeds). It is a +// client-side guard against misrouting, not server-side enforcement. func (e *ChipIngressBatchEmitterService) checkDomain(source string) error { if e.domain == "" || source == string(e.domain) || strings.HasPrefix(source, string(e.domain)+".") { return nil diff --git a/pkg/beholder/domain_emitter.go b/pkg/beholder/domain_emitter.go index a4043526fa..52cb46c20b 100644 --- a/pkg/beholder/domain_emitter.go +++ b/pkg/beholder/domain_emitter.go @@ -37,24 +37,36 @@ var ( ) // DomainHeaderKey is the gRPC metadata key carrying a domain emitter's Domain. Load balancers route -// on it to the chip ingress deployment that serves the domain; chip ingress itself ignores it. +// on it to the chip ingress deployment that serves the domain; chip ingress currently ignores it. +// It is not authenticated (the CSA token does not cover it), so it selects a route, not a right. const DomainHeaderKey = "chainlink-domain" // DomainEmitter sends one domain's events to chip ingress only, with PublishBatchNoAck, on its own -// connection and buffer. Obtain one with Client.ForDomain. +// connection and buffer. Obtain one with Client.ForDomain. The Client starts and closes it, and +// releases its connection only if the Client was started: call Start before Close. // // Both methods reject an event whose source (the CloudEvent source, or the beholder_domain // attribute for Emit) is not the emitter's Domain or a subdomain of it. The match is // case-sensitive and a subdomain must follow a "." separator: for DomainDataFeeds, // "data-feeds" and "data-feeds.telemetry.ocr2-median" are accepted; "ccip.telemetry.ocr3", -// "data-feeds-extra" and "Data-Feeds" are not. +// "data-feeds-extra" and "Data-Feeds" are not. This is a client-side guard against sending an +// event down another domain's route, not server-side enforcement. +// +// An event's source and type become metric attributes (and per-pair counters), so keep both +// low-cardinality: never put job, feed, or contract IDs in them. +// +// Callbacks: if a method returns a non-nil error the callback is not invoked; otherwise a non-nil +// callback fires exactly once with the event's fate. It may fire synchronously, before the method +// returns, when the event cannot be queued (e.g. the buffer is full), and it may fire after the +// Client's Close returns. Publishing is no-ack: a nil error means chip ingress validated the event +// and handed it to its producer, not that it was persisted. type DomainEmitter interface { - // Emitter builds the event from attributes, like the Client's Emitter. Close is a no-op: the - // Client starts and closes its domain emitters. + // Emitter builds the event from attributes, like the Client's Emitter. Close is a no-op. Emitter - // EmitEvent queues a fully built CloudEvent, keeping every extension the caller set. If it - // returns a non-nil error the callback is not invoked; otherwise a non-nil callback fires - // exactly once with the event's fate. + // EmitEvent queues a fully built CloudEvent as given: it keeps every extension the caller set + // except "seqnum", which the batch client overwrites. Unlike Emit it stamps nothing (no id, + // time or "recordedtime"), so set what the receiver needs. Do not modify the event or its data + // after EmitEvent returns nil: the queued event shares them. EmitEvent(ctx context.Context, event chipingress.CloudEvent, callback func(error)) error } From 2cd0235865710628e9fb8a3cf09342763253a423 Mon Sep 17 00:00:00 2001 From: 4of9 <177086174+4of9@users.noreply.github.com> Date: Thu, 29 Oct 2026 13:02:00 -0400 Subject: [PATCH 10/15] beholder: rename ChipIngressDomainEmittersEnabled to ChipDomainEmittersEnabled Chip ingress is a background service users should not need to know about. Existing ChipIngress* settings are unchanged. Co-Authored-By: Claude Opus 5.5 --- pkg/beholder/client.go | 12 ++++++------ pkg/beholder/config.go | 4 ++-- pkg/beholder/domain_emitter.go | 4 ++-- pkg/beholder/domain_emitter_test.go | 16 ++++++++-------- pkg/beholder/httpclient.go | 4 ++-- pkg/beholder/testdata/config-example.json | 2 +- 6 files changed, 21 insertions(+), 21 deletions(-) diff --git a/pkg/beholder/client.go b/pkg/beholder/client.go index e80f1b4e3d..adc33fd2d2 100644 --- a/pkg/beholder/client.go +++ b/pkg/beholder/client.go @@ -51,7 +51,7 @@ type Client struct { // Chip Chip chipingress.Client - // domainEmitters are built when Config.ChipIngressDomainEmittersEnabled is set and retrieved with ForDomain. + // domainEmitters are built when Config.ChipDomainEmittersEnabled is set and retrieved with ForDomain. domainEmitters map[Domain]*domainEmitter // Providers @@ -226,12 +226,12 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (_ *Client, er lggr = pkglogger.Nop() } chipIngressEnabled := cfg.ChipIngressEmitterEnabled || cfg.ChipIngressEmitterGRPCEndpoint != "" - if cfg.ChipIngressDomainEmittersEnabled && !chipIngressEnabled { - return nil, errors.New("ChipIngressDomainEmittersEnabled requires chip ingress (ChipIngressEmitterGRPCEndpoint)") + if cfg.ChipDomainEmittersEnabled && !chipIngressEnabled { + return nil, errors.New("ChipDomainEmittersEnabled requires chip ingress (ChipIngressEmitterGRPCEndpoint)") } // Domain emitters are always batch emitters; like the batch emitter, they need a real logger. - if cfg.ChipIngressDomainEmittersEnabled && cfg.ChipIngressLogger == nil { - return nil, errors.New("ChipIngressLogger is required when ChipIngressDomainEmittersEnabled is true") + if cfg.ChipDomainEmittersEnabled && cfg.ChipIngressLogger == nil { + return nil, errors.New("ChipIngressLogger is required when ChipDomainEmittersEnabled is true") } // if chip ingress is enabled, create dual source emitter that sends to both otel collector and chip ingress // eventually we will remove the dual source emitter and just use chip ingress @@ -263,7 +263,7 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (_ *Client, er opts = append(opts, chipingress.WithTracerProvider(tracerProvider)) // Domain emitters share everything above and set their own header provider. - if cfg.ChipIngressDomainEmittersEnabled { + if cfg.ChipDomainEmittersEnabled { domainEmitters, err = newDomainEmitters(cfg, opts, resourceAttrs, lggr) if err != nil { return nil, err diff --git a/pkg/beholder/config.go b/pkg/beholder/config.go index 04004b026b..c4e96b0378 100644 --- a/pkg/beholder/config.go +++ b/pkg/beholder/config.go @@ -79,7 +79,7 @@ type Config struct { ChipIngressMaxGRPCRequestSize int // Max serialized PublishBatch request size in bytes (default 10 MiB) ChipIngressLogger logger.Logger // Required when ChipIngressBatchEmitterEnabled is true - // ChipIngressDomainEmittersEnabled builds one chip ingress emitter per Domain, retrieved with + // ChipDomainEmittersEnabled builds one chip ingress emitter per Domain, retrieved with // Client.ForDomain. Each has its own connection and buffer (sized by the batch settings // above), sends the DomainHeaderKey routing header, sends to chip ingress only, and publishes // with PublishBatchNoAck. Requires chip ingress. Off by default. @@ -87,7 +87,7 @@ type Config struct { // Domain emitters are always batch emitters, whatever ChipIngressBatchEmitterEnabled says. // Per-node cost: one ChipIngressBufferSize buffer per Domain; a connection only for domains // actually used (gRPC dials lazily). - ChipIngressDomainEmittersEnabled bool + ChipDomainEmittersEnabled bool // OTel Log LogExportTimeout time.Duration diff --git a/pkg/beholder/domain_emitter.go b/pkg/beholder/domain_emitter.go index 52cb46c20b..fc051d777a 100644 --- a/pkg/beholder/domain_emitter.go +++ b/pkg/beholder/domain_emitter.go @@ -22,7 +22,7 @@ const ( DomainCCIP Domain = "ccip" ) -// allDomains lists every Domain; Config.ChipIngressDomainEmittersEnabled builds an emitter for +// allDomains lists every Domain; Config.ChipDomainEmittersEnabled builds an emitter for // each. A func, not a var, so the list cannot be modified. func allDomains() []Domain { return []Domain{DomainDataFeeds, DomainDataStreams, DomainCCIP} @@ -30,7 +30,7 @@ func allDomains() []Domain { var ( // ErrDomainEmittersDisabled is returned by ForDomain when the Client has no domain emitters: - // Config.ChipIngressDomainEmittersEnabled is off, or the Client is a noop or HTTP client. + // Config.ChipDomainEmittersEnabled is off, or the Client is a noop or HTTP client. ErrDomainEmittersDisabled = errors.New("beholder: chip ingress domain emitters are disabled") // ErrUnknownDomain is returned by ForDomain for a Domain the Client has no emitter for. ErrUnknownDomain = errors.New("beholder: unknown chip ingress domain") diff --git a/pkg/beholder/domain_emitter_test.go b/pkg/beholder/domain_emitter_test.go index 1feedcbf2f..ab0e26d1e3 100644 --- a/pkg/beholder/domain_emitter_test.go +++ b/pkg/beholder/domain_emitter_test.go @@ -100,7 +100,7 @@ func domainTestConfig(t *testing.T, chipAddr string) beholder.Config { ResourceAttributes: []attribute.KeyValue{ attribute.String("csa_public_key", "abc123"), }, - ChipIngressDomainEmittersEnabled: true, + ChipDomainEmittersEnabled: true, } } @@ -375,7 +375,7 @@ func TestForDomain_RejectsEventsFromOtherDomains(t *testing.T) { } func TestNewGRPCClient_DomainEmittersRequireChipIngress(t *testing.T) { - cfg := beholder.Config{OtelExporterGRPCEndpoint: "localhost:4317", ChipIngressDomainEmittersEnabled: true} + cfg := beholder.Config{OtelExporterGRPCEndpoint: "localhost:4317", ChipDomainEmittersEnabled: true} _, err := beholder.NewGRPCClient(cfg, func(...otlploggrpc.Option) (sdklog.Exporter, error) { return &mockLogExporter{}, nil }) @@ -385,7 +385,7 @@ func TestNewGRPCClient_DomainEmittersRequireChipIngress(t *testing.T) { func TestForDomain_DisabledByDefault(t *testing.T) { _, addr := startBatchCapturingChipServer(t) cfg := domainTestConfig(t, addr) - cfg.ChipIngressDomainEmittersEnabled = false + cfg.ChipDomainEmittersEnabled = false client := newDomainTestClient(t, cfg) _, err := client.ForDomain(beholder.DomainDataFeeds) @@ -406,7 +406,7 @@ func TestForDomain_ErrorsAreDistinguishable(t *testing.T) { require.NotErrorIs(t, err, beholder.ErrDomainEmittersDisabled) cfg := domainTestConfig(t, addr) - cfg.ChipIngressDomainEmittersEnabled = false + cfg.ChipDomainEmittersEnabled = false disabled := newDomainTestClient(t, cfg) _, err = disabled.ForDomain(beholder.DomainDataFeeds) require.ErrorIs(t, err, beholder.ErrDomainEmittersDisabled) @@ -420,13 +420,13 @@ func TestNewGRPCClient_DomainEmittersRequireLogger(t *testing.T) { _, err := beholder.NewGRPCClient(cfg, func(...otlploggrpc.Option) (sdklog.Exporter, error) { return &mockLogExporter{}, nil }) - require.ErrorContains(t, err, "ChipIngressLogger is required when ChipIngressDomainEmittersEnabled is true") + require.ErrorContains(t, err, "ChipIngressLogger is required when ChipDomainEmittersEnabled is true") } func TestNewHTTPClient_RejectsDomainEmitters(t *testing.T) { _, err := beholder.NewClient(beholder.Config{ - OtelExporterHTTPEndpoint: "localhost:4318", - ChipIngressDomainEmittersEnabled: true, + OtelExporterHTTPEndpoint: "localhost:4318", + ChipDomainEmittersEnabled: true, }) - require.ErrorContains(t, err, "ChipIngressDomainEmittersEnabled is not supported by the HTTP client") + require.ErrorContains(t, err, "ChipDomainEmittersEnabled is not supported by the HTTP client") } diff --git a/pkg/beholder/httpclient.go b/pkg/beholder/httpclient.go index 8e5dac4865..1f4e03bdc4 100644 --- a/pkg/beholder/httpclient.go +++ b/pkg/beholder/httpclient.go @@ -37,8 +37,8 @@ func newCertFromFile(certFile string) (*x509.CertPool, error) { // NewHTTPClient creates a HTTP based beholder Client. Use NewClient to create a client from a Config which will pick // the best client type from the Config. func NewHTTPClient(cfg Config, otlploghttpNew otlploghttpFactory) (*Client, error) { - if cfg.ChipIngressDomainEmittersEnabled { - return nil, errors.New("ChipIngressDomainEmittersEnabled is not supported by the HTTP client") + if cfg.ChipDomainEmittersEnabled { + return nil, errors.New("ChipDomainEmittersEnabled is not supported by the HTTP client") } baseResource, err := newOtelResource(cfg) if err != nil { diff --git a/pkg/beholder/testdata/config-example.json b/pkg/beholder/testdata/config-example.json index f9b212a2b3..3c768a9aa3 100644 --- a/pkg/beholder/testdata/config-example.json +++ b/pkg/beholder/testdata/config-example.json @@ -49,7 +49,7 @@ "ChipIngressMaxConcurrentSends": 0, "ChipIngressMaxGRPCRequestSize": 0, "ChipIngressLogger": null, - "ChipIngressDomainEmittersEnabled": false, + "ChipDomainEmittersEnabled": false, "LogExportTimeout": 1000000000, "LogExportInterval": 1000000000, "LogExportMaxBatchSize": 512, From 9b4d2d3d8d12cbcb177ac438e3ba988740f9eaa1 Mon Sep 17 00:00:00 2001 From: 4of9 <177086174+4of9@users.noreply.github.com> Date: Tue, 3 Nov 2026 15:15:09 -0500 Subject: [PATCH 11/15] beholder: drain domain emitters before shutting down OTel providers The service engine runs the Client's Close hook, which shuts the OTel providers down, before closing sub-services, so outcomes recorded while the domain emitters drained were not exported. The hook now drains the domain emitters first; a close-once wrapper makes the engine's later sub-service Close a no-op. The default batch emitter is unchanged. Co-Authored-By: Claude Opus 5.5 --- pkg/beholder/client.go | 12 ++++++++- pkg/beholder/domain_emitter.go | 18 +++++++++++++- pkg/beholder/domain_emitter_test.go | 38 +++++++++++++++++++++++++++++ 3 files changed, 66 insertions(+), 2 deletions(-) diff --git a/pkg/beholder/client.go b/pkg/beholder/client.go index adc33fd2d2..9577b81cbe 100644 --- a/pkg/beholder/client.go +++ b/pkg/beholder/client.go @@ -80,7 +80,7 @@ func (c *Client) initService(lggr pkglogger.Logger, batchSvc *ChipIngressBatchEm } for _, domain := range allDomains() { if em, ok := c.domainEmitters[domain]; ok { - subs = append(subs, em.ChipIngressBatchEmitterService) + subs = append(subs, em.closer) } } return subs @@ -345,6 +345,16 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (_ *Client, er // close is the lifecycle Close hook: OTel/message shutdown and CHIP close. func (c *Client) close() (err error) { + // Drain the domain emitters (concurrently) before the OTel providers shut down below, so + // outcomes recorded during the drain are still exported. The engine closes them again as + // sub-services after this hook; closeOnce makes that a no-op. + var domainClosers []*closeOnce + for _, domain := range allDomains() { + if em, ok := c.domainEmitters[domain]; ok { + domainClosers = append(domainClosers, em.closer) + } + } + err = errors.Join(err, services.MultiCloser(domainClosers).Close()) if c.Emitter != nil { err = errors.Join(err, c.Emitter.Close()) } diff --git a/pkg/beholder/domain_emitter.go b/pkg/beholder/domain_emitter.go index fc051d777a..ed2cd67b09 100644 --- a/pkg/beholder/domain_emitter.go +++ b/pkg/beholder/domain_emitter.go @@ -5,10 +5,12 @@ import ( "errors" "fmt" "slices" + "sync" "github.com/smartcontractkit/chainlink-common/pkg/chipingress" "github.com/smartcontractkit/chainlink-common/pkg/chipingress/batch" pkglogger "github.com/smartcontractkit/chainlink-common/pkg/logger" + "github.com/smartcontractkit/chainlink-common/pkg/services" ) // Domain is a top-level product domain with its own chip ingress deployment. @@ -91,10 +93,24 @@ type domainEmitter struct { // chipClient is the emitter's own connection. Once started, the batch client closes it on // Stop; it is closed here only when construction fails before the emitter ever starts. chipClient chipingress.Client + // closer is the emitter's service as registered with the Client: the Client's Close hook + // drains it before the OTel providers shut down, then the engine's sub-service Close is a no-op. + closer *closeOnce } func (*domainEmitter) Close() error { return nil } +// closeOnce wraps a service so that only its first Close does anything; later calls return nil. +type closeOnce struct { + services.Service + once sync.Once +} + +func (c *closeOnce) Close() (err error) { + c.once.Do(func() { err = c.Service.Close() }) + return err +} + // newDomainEmitters builds one emitter per Domain. opts are the Client's chip ingress options // (transport, auth, OTel providers) without its header provider, which each domain emitter // replaces with one carrying the resource attribute headers plus its DomainHeaderKey. @@ -127,7 +143,7 @@ func newDomainEmitters(cfg Config, opts []chipingress.Opt, resourceAttrs map[str _ = chipClient.Close() return nil, fmt.Errorf("chip ingress emitter for domain %q: %w", domain, err) } - emitters[domain] = &domainEmitter{ChipIngressBatchEmitterService: svc, chipClient: chipClient} + emitters[domain] = &domainEmitter{ChipIngressBatchEmitterService: svc, chipClient: chipClient, closer: &closeOnce{Service: svc}} } return emitters, nil } diff --git a/pkg/beholder/domain_emitter_test.go b/pkg/beholder/domain_emitter_test.go index ab0e26d1e3..8f0a31a9f8 100644 --- a/pkg/beholder/domain_emitter_test.go +++ b/pkg/beholder/domain_emitter_test.go @@ -280,6 +280,44 @@ func TestForDomain_ClientCloseClosesDomainConnectionsOnce(t *testing.T) { requireNoDomainEmitterCloseErrors(t, client.Close()) } +// Domain emitters drain before the OTel providers shut down, so outcomes recorded during the +// drain (sent/dropped metrics, RPC metrics and spans) are still exported. OnClose is the hook that +// shuts the providers down; by the time it runs, the domain emitters must already be stopped. +func TestForDomain_DrainsBeforeProvidersShutDown(t *testing.T) { + capture, addr := startBatchCapturingChipServer(t) + cfg := domainTestConfig(t, addr) + cfg.ChipIngressSendInterval = time.Hour // only the drain can publish the queued event + + client, err := beholder.NewGRPCClient(cfg, func(...otlploggrpc.Option) (sdklog.Exporter, error) { + return &mockLogExporter{}, nil + }) + require.NoError(t, err) + require.NoError(t, client.Start(t.Context())) + + em, err := client.ForDomain(beholder.DomainDataFeeds) + require.NoError(t, err) + require.NoError(t, em.EmitEvent(t.Context(), legacyStyleEvent(t, "data-feeds.telemetry.ordering"), nil)) + + var ( + emitDuringShutdown error + drainedBefore int + ) + shutdownProviders := client.OnClose + client.OnClose = func() error { + emitDuringShutdown = em.EmitEvent(context.Background(), legacyStyleEvent(t, "data-feeds.telemetry.late"), nil) + drainedBefore = len(capture.callsWithEventSource("data-feeds.telemetry.ordering")) + return shutdownProviders() + } + + closeErr := client.Close() + require.Error(t, emitDuringShutdown, "the domain emitter is stopped before the providers shut down") + assert.Equal(t, 1, drainedBefore, "the queued event is published before the providers shut down") + requireNoDomainEmitterCloseErrors(t, closeErr) + if closeErr != nil { + assert.NotContains(t, closeErr.Error(), "already been stopped", "each domain emitter is closed once") + } +} + func TestForDomain_EmitEventRejectsInvalidEvent(t *testing.T) { _, addr := startBatchCapturingChipServer(t) client := newDomainTestClient(t, domainTestConfig(t, addr)) From c038d46540226f40d44909a35729a25ca3cd5d78 Mon Sep 17 00:00:00 2001 From: Thomas Kaliakos Date: Thu, 5 Nov 2026 14:10:15 +0200 Subject: [PATCH 12/15] Amend cleanup - copilot comment --- pkg/beholder/client.go | 27 +++++++---- pkg/beholder/client_construction_test.go | 61 ++++++++++++++++++++++++ 2 files changed, 80 insertions(+), 8 deletions(-) create mode 100644 pkg/beholder/client_construction_test.go diff --git a/pkg/beholder/client.go b/pkg/beholder/client.go index b3f6a0d242..5a051b3b08 100644 --- a/pkg/beholder/client.go +++ b/pkg/beholder/client.go @@ -119,6 +119,15 @@ type otlploggrpcFactory func(options ...otlploggrpc.Option) (sdklog.Exporter, er // NewGRPCClient creates a GRPC based beholder Client. Use NewClient to create a client from a Config which will pick // the best client type from the Config. func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (_ *Client, err error) { + chipIngressEnabled := cfg.ChipIngressEmitterEnabled || cfg.ChipIngressEmitterGRPCEndpoint != "" + if cfg.ChipDomainEmittersEnabled && !chipIngressEnabled { + return nil, errors.New("ChipDomainEmittersEnabled requires chip ingress (ChipIngressEmitterGRPCEndpoint)") + } + // Domain emitters are always batch emitters; like the batch emitter, they need a real logger. + if cfg.ChipDomainEmittersEnabled && cfg.ChipIngressLogger == nil { + return nil, errors.New("ChipIngressLogger is required when ChipDomainEmittersEnabled is true") + } + baseResource, err := newOtelResource(cfg) if err != nil { return nil, err @@ -204,6 +213,11 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (_ *Client, er } loggerProvider = sdklog.NewLoggerProvider(loggerOpts...) } + defer func() { + if err != nil { + _ = loggerProvider.Shutdown(context.Background()) + } + }() logger := loggerProvider.Logger(defaultPackageName) // Message emitter @@ -212,6 +226,11 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (_ *Client, er return nil, err } messageLoggerProvider := sdklog.NewLoggerProvider(messageLoggerOpts...) + defer func() { + if err != nil { + _ = messageLoggerProvider.Shutdown(context.Background()) + } + }() messageLogger := messageLoggerProvider.Logger(defaultPackageName) // Use the messageEmitter by default @@ -225,14 +244,6 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (_ *Client, er if lggr == nil { lggr = pkglogger.Nop() } - chipIngressEnabled := cfg.ChipIngressEmitterEnabled || cfg.ChipIngressEmitterGRPCEndpoint != "" - if cfg.ChipDomainEmittersEnabled && !chipIngressEnabled { - return nil, errors.New("ChipDomainEmittersEnabled requires chip ingress (ChipIngressEmitterGRPCEndpoint)") - } - // Domain emitters are always batch emitters; like the batch emitter, they need a real logger. - if cfg.ChipDomainEmittersEnabled && cfg.ChipIngressLogger == nil { - return nil, errors.New("ChipIngressLogger is required when ChipDomainEmittersEnabled is true") - } // if chip ingress is enabled, create dual source emitter that sends to both otel collector and chip ingress // eventually we will remove the dual source emitter and just use chip ingress if chipIngressEnabled { diff --git a/pkg/beholder/client_construction_test.go b/pkg/beholder/client_construction_test.go new file mode 100644 index 0000000000..f1b670a064 --- /dev/null +++ b/pkg/beholder/client_construction_test.go @@ -0,0 +1,61 @@ +package beholder_test + +import ( + "context" + "sync/atomic" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc" + sdklog "go.opentelemetry.io/otel/sdk/log" + + "github.com/smartcontractkit/chainlink-common/pkg/beholder" +) + +type countingLogExporter struct { + sdklog.Exporter + shutdowns *atomic.Int64 +} + +func (e *countingLogExporter) Shutdown(ctx context.Context) error { + e.shutdowns.Add(1) + return e.Exporter.Shutdown(ctx) +} + +func countingExporterFactory(created *atomic.Bool, shutdowns *atomic.Int64) func(...otlploggrpc.Option) (sdklog.Exporter, error) { + return func(...otlploggrpc.Option) (sdklog.Exporter, error) { + created.Store(true) + return &countingLogExporter{Exporter: &mockLogExporter{}, shutdowns: shutdowns}, nil + } +} + +func TestNewGRPCClient_DomainEmitterValidationBuildsNothing(t *testing.T) { + created, shutdowns := &atomic.Bool{}, &atomic.Int64{} + + _, err := beholder.NewGRPCClient(beholder.Config{ + OtelExporterGRPCEndpoint: "localhost:4317", + ChipDomainEmittersEnabled: true, + }, countingExporterFactory(created, shutdowns)) + require.ErrorContains(t, err, "ChipDomainEmittersEnabled requires chip ingress") + + cfg := domainTestConfig(t, "localhost:9999") + cfg.ChipIngressLogger = nil + _, err = beholder.NewGRPCClient(cfg, countingExporterFactory(created, shutdowns)) + require.ErrorContains(t, err, "ChipIngressLogger is required when ChipDomainEmittersEnabled is true") + + assert.False(t, created.Load(), "nothing is constructed for a config error") + assert.Zero(t, shutdowns.Load()) +} + +func TestNewGRPCClient_ConstructionFailureShutsDownLogProviders(t *testing.T) { + created, shutdowns := &atomic.Bool{}, &atomic.Int64{} + + cfg := domainTestConfig(t, "missing-port") + cfg.LogStreamingEnabled = true // both log providers are real, so both shut down + _, err := beholder.NewGRPCClient(cfg, countingExporterFactory(created, shutdowns)) + require.ErrorContains(t, err, `chip ingress client for domain "data-feeds"`) + + assert.True(t, created.Load(), "the log pipeline was built before the failure") + assert.Equal(t, int64(2), shutdowns.Load(), "both log providers are shut down on failure") +} From e141f091bc5187e73c5dcaa6744850314aa1f3da Mon Sep 17 00:00:00 2001 From: Thomas Kaliakos Date: Fri, 6 Nov 2026 15:09:29 +0200 Subject: [PATCH 13/15] Use noop_closer --- pkg/beholder/client.go | 10 +++++----- pkg/beholder/domain_emitter.go | 24 ++++++++++------------- pkg/beholder/noop_closer_test.go | 33 ++++++++++++++++++++++++++++++++ 3 files changed, 48 insertions(+), 19 deletions(-) create mode 100644 pkg/beholder/noop_closer_test.go diff --git a/pkg/beholder/client.go b/pkg/beholder/client.go index 5a051b3b08..d856d55a51 100644 --- a/pkg/beholder/client.go +++ b/pkg/beholder/client.go @@ -357,15 +357,15 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (_ *Client, er // close is the lifecycle Close hook: OTel/message shutdown and CHIP close. func (c *Client) close() (err error) { // Drain the domain emitters (concurrently) before the OTel providers shut down below, so - // outcomes recorded during the drain are still exported. The engine closes them again as - // sub-services after this hook; closeOnce makes that a no-op. - var domainClosers []*closeOnce + // outcomes recorded during the drain are still exported. They are registered with the + // engine as noopClosers, so the engine's later sub-service Close does not close them again. + var emitters []*ChipIngressBatchEmitterService for _, domain := range allDomains() { if em, ok := c.domainEmitters[domain]; ok { - domainClosers = append(domainClosers, em.closer) + emitters = append(emitters, em.ChipIngressBatchEmitterService) } } - err = errors.Join(err, services.MultiCloser(domainClosers).Close()) + err = errors.Join(err, services.MultiCloser(emitters).Close()) if c.Emitter != nil { err = errors.Join(err, c.Emitter.Close()) } diff --git a/pkg/beholder/domain_emitter.go b/pkg/beholder/domain_emitter.go index ed2cd67b09..1623a1de69 100644 --- a/pkg/beholder/domain_emitter.go +++ b/pkg/beholder/domain_emitter.go @@ -5,7 +5,6 @@ import ( "errors" "fmt" "slices" - "sync" "github.com/smartcontractkit/chainlink-common/pkg/chipingress" "github.com/smartcontractkit/chainlink-common/pkg/chipingress/batch" @@ -93,23 +92,20 @@ type domainEmitter struct { // chipClient is the emitter's own connection. Once started, the batch client closes it on // Stop; it is closed here only when construction fails before the emitter ever starts. chipClient chipingress.Client - // closer is the emitter's service as registered with the Client: the Client's Close hook - // drains it before the OTel providers shut down, then the engine's sub-service Close is a no-op. - closer *closeOnce + // closer is the emitter's service as registered with the Client's engine. Close is + // suppressed: the engine starts it and reports its health, but Client.close is the + // sole closer, so the drain happens before the OTel providers shut down. + closer noopCloser } func (*domainEmitter) Close() error { return nil } -// closeOnce wraps a service so that only its first Close does anything; later calls return nil. -type closeOnce struct { - services.Service - once sync.Once -} +// noopCloser wraps a service that something other than the engine closes. +// shortcut: the engine's start-rollback closes subs through this wrapper, so emitter +// start() must stay infallible or that rollback can no longer clean up. +type noopCloser struct{ services.Service } -func (c *closeOnce) Close() (err error) { - c.once.Do(func() { err = c.Service.Close() }) - return err -} +func (noopCloser) Close() error { return nil } // newDomainEmitters builds one emitter per Domain. opts are the Client's chip ingress options // (transport, auth, OTel providers) without its header provider, which each domain emitter @@ -143,7 +139,7 @@ func newDomainEmitters(cfg Config, opts []chipingress.Opt, resourceAttrs map[str _ = chipClient.Close() return nil, fmt.Errorf("chip ingress emitter for domain %q: %w", domain, err) } - emitters[domain] = &domainEmitter{ChipIngressBatchEmitterService: svc, chipClient: chipClient, closer: &closeOnce{Service: svc}} + emitters[domain] = &domainEmitter{ChipIngressBatchEmitterService: svc, chipClient: chipClient, closer: noopCloser{Service: svc}} } return emitters, nil } diff --git a/pkg/beholder/noop_closer_test.go b/pkg/beholder/noop_closer_test.go new file mode 100644 index 0000000000..1688e4113f --- /dev/null +++ b/pkg/beholder/noop_closer_test.go @@ -0,0 +1,33 @@ +package beholder + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/smartcontractkit/chainlink-common/pkg/chipingress" + "github.com/smartcontractkit/chainlink-common/pkg/chipingress/batch" + "github.com/smartcontractkit/chainlink-common/pkg/logger" +) + +// The engine closes sub-services after the parent Close hook; noopCloser suppresses that, so +// Client.close stays the emitter's sole closer. +func TestNoopCloser_SuppressesClose(t *testing.T) { + svc, err := newChipIngressBatchEmitterService(&chipingress.NoopClient{}, Config{ + ChipIngressSendInterval: 10 * time.Millisecond, + ChipIngressSendTimeout: time.Second, + }, logger.Test(t), "test", "", batch.WithNoAck()) + require.NoError(t, err) + require.NoError(t, svc.Start(t.Context())) + + ev, err := chipingress.NewEvent("test-domain", "Entity", []byte("x"), nil) + require.NoError(t, err) + + require.NoError(t, noopCloser{Service: svc}.Close()) + require.NoError(t, svc.EmitEvent(context.Background(), ev, nil), "still started after the suppressed close") + + require.NoError(t, svc.Close(), "the real closer works") + require.Error(t, svc.EmitEvent(context.Background(), ev, nil), "stopped after the real close") +} From b0490ab7d95a440611b4bf79207b8a963afafda6 Mon Sep 17 00:00:00 2001 From: Thomas Kaliakos Date: Fri, 6 Nov 2026 17:04:18 +0200 Subject: [PATCH 14/15] Move comment --- pkg/beholder/batch_emitter_service.go | 2 ++ pkg/beholder/domain_emitter.go | 2 -- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pkg/beholder/batch_emitter_service.go b/pkg/beholder/batch_emitter_service.go index 10be6a24fd..48faad7d98 100644 --- a/pkg/beholder/batch_emitter_service.go +++ b/pkg/beholder/batch_emitter_service.go @@ -126,6 +126,8 @@ func newChipIngressBatchEmitterService(client chipingress.Client, cfg Config, lg } func (e *ChipIngressBatchEmitterService) start(_ context.Context) error { + // Must not fail: emitters registered with the engine as noopClosers cannot be closed + // by the engine's start rollback once started (see noopCloser in domain_emitter.go). // Do not pass the startup ctx — the services contract forbids retaining it // after Start returns. Use the engine's lifecycle context so the batcher // is cancelled when the service shuts down (StopChan closes before stop() runs). diff --git a/pkg/beholder/domain_emitter.go b/pkg/beholder/domain_emitter.go index 1623a1de69..f6f2310690 100644 --- a/pkg/beholder/domain_emitter.go +++ b/pkg/beholder/domain_emitter.go @@ -101,8 +101,6 @@ type domainEmitter struct { func (*domainEmitter) Close() error { return nil } // noopCloser wraps a service that something other than the engine closes. -// shortcut: the engine's start-rollback closes subs through this wrapper, so emitter -// start() must stay infallible or that rollback can no longer clean up. type noopCloser struct{ services.Service } func (noopCloser) Close() error { return nil } From c72d7ec00ee069777f82c9196f0288c53269ba1a Mon Sep 17 00:00:00 2001 From: Thomas Kaliakos Date: Fri, 6 Nov 2026 17:17:43 +0200 Subject: [PATCH 15/15] Revert "Move comment" This reverts commit b0490ab7d95a440611b4bf79207b8a963afafda6. --- pkg/beholder/batch_emitter_service.go | 2 -- pkg/beholder/domain_emitter.go | 2 ++ 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pkg/beholder/batch_emitter_service.go b/pkg/beholder/batch_emitter_service.go index 48faad7d98..10be6a24fd 100644 --- a/pkg/beholder/batch_emitter_service.go +++ b/pkg/beholder/batch_emitter_service.go @@ -126,8 +126,6 @@ func newChipIngressBatchEmitterService(client chipingress.Client, cfg Config, lg } func (e *ChipIngressBatchEmitterService) start(_ context.Context) error { - // Must not fail: emitters registered with the engine as noopClosers cannot be closed - // by the engine's start rollback once started (see noopCloser in domain_emitter.go). // Do not pass the startup ctx — the services contract forbids retaining it // after Start returns. Use the engine's lifecycle context so the batcher // is cancelled when the service shuts down (StopChan closes before stop() runs). diff --git a/pkg/beholder/domain_emitter.go b/pkg/beholder/domain_emitter.go index f6f2310690..1623a1de69 100644 --- a/pkg/beholder/domain_emitter.go +++ b/pkg/beholder/domain_emitter.go @@ -101,6 +101,8 @@ type domainEmitter struct { func (*domainEmitter) Close() error { return nil } // noopCloser wraps a service that something other than the engine closes. +// shortcut: the engine's start-rollback closes subs through this wrapper, so emitter +// start() must stay infallible or that rollback can no longer clean up. type noopCloser struct{ services.Service } func (noopCloser) Close() error { return nil }