diff --git a/go.mod b/go.mod index 40479c3317..1c02b8ecaf 100644 --- a/go.mod +++ b/go.mod @@ -43,7 +43,7 @@ require ( github.com/scylladb/go-reflectx v1.0.1 github.com/shopspring/decimal v1.4.0 github.com/smartcontractkit/chain-selectors v1.0.100 - github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260716165322-7f2edff6e954 + github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260724142814-45996a1bcb72 github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20251024234028-0988426d98f4 github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260622152157-c8e129347b8b github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20251002192024-d2ad9222409b diff --git a/go.sum b/go.sum index e4b11c5902..072c8556ab 100644 --- a/go.sum +++ b/go.sum @@ -258,8 +258,8 @@ github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME= github.com/smartcontractkit/chain-selectors v1.0.100 h1:wpiSpmI/eFjY+wx/nPr5VuNF4hki0prIBMKEaQWn3g4= github.com/smartcontractkit/chain-selectors v1.0.100/go.mod h1:qy7whtgG5g+7z0jt0nRyii9bLND9m15NZTzuQPkMZ5w= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260716165322-7f2edff6e954 h1:QhTMiEn3s+AB4xBoScuQglsqHGJYxheYrgpxdIdqNAI= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260716165322-7f2edff6e954/go.mod h1:UYcRMb4dZcoaIPgZJ3hckCySTqtJc9K4Q+tOKErwTq0= +github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260724142814-45996a1bcb72 h1:uWEwl7i2ryuRVoV4DmIKm6mqYevf1lH/8cQYhw/JXko= +github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260724142814-45996a1bcb72/go.mod h1:UYcRMb4dZcoaIPgZJ3hckCySTqtJc9K4Q+tOKErwTq0= github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20251024234028-0988426d98f4 h1:GCzrxDWn3b7jFfEA+WiYRi8CKoegsayiDoJBCjYkneE= github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20251024234028-0988426d98f4/go.mod h1:HHGeDUpAsPa0pmOx7wrByCitjQ0mbUxf0R9v+g67uCA= github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260622152157-c8e129347b8b h1:VDgJWDipihV9f7M5+d21d1RzSsg5rEv+iI12oN1VQbo= diff --git a/pkg/beholder/batch_emitter_service.go b/pkg/beholder/batch_emitter_service.go index 700d90bf13..d2560a981e 100644 --- a/pkg/beholder/batch_emitter_service.go +++ b/pkg/beholder/batch_emitter_service.go @@ -2,6 +2,7 @@ package beholder import ( "context" + "errors" "fmt" "sync" @@ -163,8 +164,23 @@ func (e *ChipIngressBatchEmitterService) emitInternal(ctx context.Context, body // for metric recording — OTel Add is non-blocking and tolerates // cancelled contexts. if sendErr != nil { - e.metrics.eventsDropped.Add(ctx, 1, metricAttrs) - e.eng.Errorw("failed to emit to chip ingress", "error", sendErr, "domain", domain, "entity", entity) + 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. + var pubErr *batch.PublishError + if !errors.As(sendErr, &pubErr) { + 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) } @@ -173,8 +189,15 @@ func (e *ChipIngressBatchEmitterService) emitInternal(ctx context.Context, body } }) if queueErr != nil { - e.metrics.eventsDropped.Add(ctx, 1, metricAttrs) - e.eng.Errorw("failed to queue message for chip ingress", "error", queueErr, "domain", domain, "entity", entity) + 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) } @@ -198,6 +221,26 @@ func (e *ChipIngressBatchEmitterService) metricAttrsFor(domain, entity string) o return v.(otelmetric.MeasurementOption) } +// dropMetricAttrsFor returns a measurement option for the eventsDropped counter. +// Not cached — drop paths are not on the hot path. +// +// error_reason is deliberately excluded: it is free-form text from the server/gRPC +// stack (e.g. validation messages, status details) and is not a bounded value, so using +// it as a metric attribute would create unbounded cardinality. domain/entity/client_name/error_code +// are all closed, bounded sets. error_reason is still available on the corresponding +// log line. +func (e *ChipIngressBatchEmitterService) dropMetricAttrsFor(domain, entity, errorCode string) otelmetric.MeasurementOption { + attrs := []attribute.KeyValue{ + attribute.String("domain", domain), + attribute.String("entity", entity), + attribute.String("client_name", batch.ClientNameBeholder), + } + if errorCode != "" { + attrs = append(attrs, attribute.String("error_code", errorCode)) + } + return otelmetric.WithAttributeSet(attribute.NewSet(attrs...)) +} + func newBatchEmitterMetrics(meter otelmetric.Meter) (batchEmitterMetrics, error) { eventsSent, err := meter.Int64Counter("chip_ingress.events_sent", otelmetric.WithDescription("Total events successfully sent via PublishBatch"), diff --git a/pkg/beholder/batch_emitter_service_test.go b/pkg/beholder/batch_emitter_service_test.go index de56839237..5a405a6b9b 100644 --- a/pkg/beholder/batch_emitter_service_test.go +++ b/pkg/beholder/batch_emitter_service_test.go @@ -16,6 +16,8 @@ import ( "go.opentelemetry.io/otel/sdk/metric/metricdata" "go.uber.org/zap" "go.uber.org/zap/zaptest/observer" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" "github.com/smartcontractkit/chainlink-common/pkg/beholder" "github.com/smartcontractkit/chainlink-common/pkg/chipingress" @@ -459,6 +461,166 @@ func TestChipIngressBatchEmitterService_EmitWithCallback(t *testing.T) { }) } +func TestChipIngressBatchEmitterService_PartialDeliveryError(t *testing.T) { + t.Run("does not log on per-event PublishError, only records the metric", func(t *testing.T) { + lggr, observed := logger.TestObserved(t, zap.InfoLevel) + + clientMock := mocks.NewClient(t) + clientMock.EXPECT().Close().Return(nil).Maybe() + + partialResp := &chipingress.PublishResponse{ + Results: []*chipingress.PublishResult{ + { + Error: &chipingress.PublishError{ + ErrorCode: chipingress.PublishErrorCode(1), // VALIDATION_FAILED + Reason: "schema not found", + }, + }, + }, + } + done := make(chan struct{}) + clientMock. + On("PublishBatch", mock.Anything, mock.Anything). + Return(partialResp, nil). + Run(func(_ mock.Arguments) { close(done) }). + Once() + + cfg := newTestConfig() + cfg.ChipIngressMaxBatchSize = 1 + cfg.ChipIngressSendInterval = time.Second + + emitter, err := beholder.NewChipIngressBatchEmitterService(clientMock, cfg, lggr) + require.NoError(t, err) + require.NoError(t, emitter.Start(t.Context())) + + err = emitter.Emit(t.Context(), []byte("body"), + beholder.AttrKeyDomain, "platform", + beholder.AttrKeyEntity, "TestEvent", + ) + require.NoError(t, err) + + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("timeout waiting for publish") + } + require.NoError(t, emitter.Close()) + + // Partial delivery is intentionally not logged (see dropMetricAttrsFor / emitInternal): + // it's a per-event, often persistent condition that would otherwise spam logs at + // fleet-wide volume. The events_dropped metric (error_code) is the intended signal; + // see the "records events_dropped" subtest below. + logs := observed.FilterMessage("failed to emit to chip ingress") + assert.Zero(t, logs.Len(), "partial delivery drops must not be logged") + }) + + t.Run("records events_dropped with the PublishError code", func(t *testing.T) { + reader, restore := useEmitterTestMeterProvider(t) + defer restore() + + clientMock := mocks.NewClient(t) + clientMock.EXPECT().Close().Return(nil).Maybe() + + partialResp := &chipingress.PublishResponse{ + Results: []*chipingress.PublishResult{ + { + Error: &chipingress.PublishError{ + ErrorCode: chipingress.PublishErrorCode(1), + Reason: "encode error", + }, + }, + }, + } + done := make(chan struct{}) + clientMock. + On("PublishBatch", mock.Anything, mock.Anything). + Return(partialResp, nil). + Run(func(_ mock.Arguments) { close(done) }). + Once() + + cfg := newTestConfig() + cfg.ChipIngressMaxBatchSize = 1 + cfg.ChipIngressSendInterval = time.Second + + emitter, err := beholder.NewChipIngressBatchEmitterService(clientMock, cfg, newTestLogger(t)) + require.NoError(t, err) + require.NoError(t, emitter.Start(t.Context())) + + err = emitter.Emit(t.Context(), []byte("body"), + beholder.AttrKeyDomain, "platform", + beholder.AttrKeyEntity, "PartialEvent", + ) + require.NoError(t, err) + + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("timeout waiting for publish") + } + require.NoError(t, emitter.Close()) + + rm := collectEmitterMetrics(t, reader) + metric := mustEmitterMetric(t, rm, "chip_ingress.events_dropped") + sum, ok := metric.Data.(metricdata.Sum[int64]) + require.True(t, ok) + dp := mustEmitterInt64SumPoint(t, sum, "domain", "platform", "entity", "PartialEvent") + assert.True(t, hasEmitterStringAttr(dp.Attributes, "error_code", "PUBLISH_ERROR_CODE_VALIDATION_FAILED")) + assert.True(t, hasEmitterStringAttr(dp.Attributes, "client_name", batch.ClientNameBeholder)) + assert.GreaterOrEqual(t, dp.Value, int64(1)) + }) +} + +func TestChipIngressBatchEmitterService_RPCError(t *testing.T) { + t.Run("records events_dropped with error_code on RPC failure", func(t *testing.T) { + reader, restore := useEmitterTestMeterProvider(t) + defer restore() + + clientMock := mocks.NewClient(t) + clientMock.EXPECT().Close().Return(nil).Maybe() + + done := make(chan struct{}) + clientMock. + On("PublishBatch", mock.Anything, mock.Anything). + Return(nil, status.Error(codes.Internal, "failed to publish events")). + Run(func(_ mock.Arguments) { close(done) }). + Once() + + cfg := newTestConfig() + cfg.ChipIngressMaxBatchSize = 1 + cfg.ChipIngressSendInterval = time.Second + + emitter, err := beholder.NewChipIngressBatchEmitterService(clientMock, cfg, newTestLogger(t)) + require.NoError(t, err) + require.NoError(t, emitter.Start(t.Context())) + + err = emitter.Emit(t.Context(), []byte("body"), + beholder.AttrKeyDomain, "platform", + beholder.AttrKeyEntity, "RPCDropEvent", + ) + require.NoError(t, err) + + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("timeout waiting for publish") + } + require.NoError(t, emitter.Close()) + + rm := collectEmitterMetrics(t, reader) + metric := mustEmitterMetric(t, rm, "chip_ingress.events_dropped") + sum, ok := metric.Data.(metricdata.Sum[int64]) + require.True(t, ok) + dp := mustEmitterInt64SumPoint(t, sum, "domain", "platform", "entity", "RPCDropEvent") + assert.True(t, hasEmitterStringAttr(dp.Attributes, "error_code", "Internal")) + assert.True(t, hasEmitterStringAttr(dp.Attributes, "client_name", batch.ClientNameBeholder)) + assert.GreaterOrEqual(t, dp.Value, int64(1)) + // error_reason is free-form server/gRPC text and must never be a metric attribute - + // it would create unbounded cardinality. It's still available on the log line. + assert.False(t, hasEmitterStringAttr(dp.Attributes, "error_reason", "failed to publish events")) + }) +} + + func TestChipIngressBatchEmitterService_Metrics(t *testing.T) { t.Run("records events_sent on successful publish", func(t *testing.T) { reader, restore := useEmitterTestMeterProvider(t) @@ -513,7 +675,7 @@ func TestChipIngressBatchEmitterService_Metrics(t *testing.T) { done := make(chan struct{}) clientMock. On("PublishBatch", mock.Anything, mock.Anything). - Return(nil, assert.AnError). + Return(nil, status.Error(codes.DeadlineExceeded, "context deadline exceeded")). Run(func(_ mock.Arguments) { close(done) }). Once() @@ -542,6 +704,7 @@ func TestChipIngressBatchEmitterService_Metrics(t *testing.T) { sum, ok := metric.Data.(metricdata.Sum[int64]) require.True(t, ok) dp := mustEmitterInt64SumPoint(t, sum, "domain", "platform", "entity", "MetricDropEvent") + assert.True(t, hasEmitterStringAttr(dp.Attributes, "error_code", "DeadlineExceeded")) assert.True(t, hasEmitterStringAttr(dp.Attributes, "client_name", batch.ClientNameBeholder)) assert.GreaterOrEqual(t, dp.Value, int64(1)) @@ -551,6 +714,7 @@ func TestChipIngressBatchEmitterService_Metrics(t *testing.T) { assert.Equal(t, zap.ErrorLevel, entry.Level) fieldMap := logFieldMap(entry) assert.Contains(t, fieldMap, "error") + assert.Equal(t, "DeadlineExceeded", fieldMap["error_code"]) assert.Equal(t, "platform", fieldMap["domain"]) assert.Equal(t, "MetricDropEvent", fieldMap["entity"]) })