diff --git a/go.mod b/go.mod index 1299440695..0254d2ef1b 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.20260626151909-052e55e62e62 + github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260713200205-3ec922174fe1 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 3441393f7f..699a274548 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.20260626151909-052e55e62e62 h1:o7vfwNQjQbMKQ9YsZFQOxvU7RMXD/wKnZsX5N9sDS3w= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260626151909-052e55e62e62/go.mod h1:HmUyH2oD9m+GRpKq7q3vuRnm1F2Uczf/Nd1v3ipMSK8= +github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260713200205-3ec922174fe1 h1:pBlv8M7iERb44h6cDyLWIzmh1MgLo4Z/0cHpfPmPs38= +github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260713200205-3ec922174fe1/go.mod h1:HmUyH2oD9m+GRpKq7q3vuRnm1F2Uczf/Nd1v3ipMSK8= github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20251024234028-0988426d98f4 h1:GCzrxDWn3b7jFfEA+WiYRi8CKoegsayiDoJBCjYkneE= github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20251024234028-0988426d98f4/go.mod h1:HHGeDUpAsPa0pmOx7wrByCitjQ0mbUxf0R9v+g67uCA= github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260622152157-c8e129347b8b h1:VDgJWDipihV9f7M5+d21d1RzSsg5rEv+iI12oN1VQbo= diff --git a/pkg/beholder/batch_emitter_service.go b/pkg/beholder/batch_emitter_service.go index a81fd82bc9..c26d5ad148 100644 --- a/pkg/beholder/batch_emitter_service.go +++ b/pkg/beholder/batch_emitter_service.go @@ -157,8 +157,26 @@ 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) + drop := batch.ClassifyDropFailure(sendErr) + e.metrics.eventsDropped.Add(ctx, 1, e.dropMetricAttrsFor(domain, entity, drop)) + if drop.ErrorType == batch.ErrorTypePartialDelivery { + e.eng.Warnw("failed to emit to chip ingress (partial delivery)", + "error_type", drop.ErrorType, + "error_code", drop.ErrorCode, + "error_reason", drop.ErrorReason, + "domain", domain, + "entity", entity, + ) + } else { + e.eng.Errorw("failed to emit to chip ingress", + "error", sendErr, + "error_type", drop.ErrorType, + "error_code", drop.ErrorCode, + "error_reason", drop.ErrorReason, + "domain", domain, + "entity", entity, + ) + } } else { e.metrics.eventsSent.Add(ctx, 1, metricAttrs) } @@ -167,8 +185,16 @@ 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) + drop := batch.ClassifyDropFailure(queueErr) + e.metrics.eventsDropped.Add(ctx, 1, e.dropMetricAttrsFor(domain, entity, drop)) + e.eng.Errorw("failed to queue message for chip ingress", + "error", queueErr, + "error_type", drop.ErrorType, + "error_code", drop.ErrorCode, + "error_reason", drop.ErrorReason, + "domain", domain, + "entity", entity, + ) if callback != nil { callback(queueErr) } @@ -191,6 +217,23 @@ 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. +func (e *ChipIngressBatchEmitterService) dropMetricAttrsFor(domain, entity string, drop batch.DropFailure) otelmetric.MeasurementOption { + attrs := []attribute.KeyValue{ + attribute.String("domain", domain), + attribute.String("entity", entity), + attribute.String("error_type", drop.ErrorType), + } + if drop.ErrorCode != "" { + attrs = append(attrs, attribute.String("error_code", drop.ErrorCode)) + } + if drop.ErrorReason != "" { + attrs = append(attrs, attribute.String("error_reason", drop.ErrorReason)) + } + 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 4137d450d5..5ba1817268 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" @@ -458,6 +460,169 @@ func TestChipIngressBatchEmitterService_EmitWithCallback(t *testing.T) { }) } +func TestChipIngressBatchEmitterService_PartialDeliveryError(t *testing.T) { + t.Run("logs error_code and reason on per-event PublishError", 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()) + + logs := observed.FilterMessage("failed to emit to chip ingress (partial delivery)") + require.GreaterOrEqual(t, logs.Len(), 1, "expected partial delivery error log") + entry := logs.All()[0] + assert.Equal(t, zap.WarnLevel, entry.Level) + fieldMap := logFieldMap(entry) + assert.Equal(t, "platform", fieldMap["domain"]) + assert.Equal(t, "TestEvent", fieldMap["entity"]) + assert.Equal(t, "partial_delivery", fieldMap["error_type"]) + assert.Contains(t, fieldMap["error_code"], "PUBLISH_ERROR_CODE") + assert.Equal(t, "schema not found", fieldMap["error_reason"]) + }) + + t.Run("records events_dropped with partial_delivery error_type", 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, "error_type", "partial_delivery", "entity", "PartialEvent") + assert.GreaterOrEqual(t, dp.Value, int64(1)) + assert.True(t, + hasEmitterStringAttr(dp.Attributes, "error_code", "PUBLISH_ERROR_CODE_VALIDATION_FAILED"), + "expected error_code label on partial_delivery drop metric datapoint", + ) + }) +} + +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, "error_type", "rpc_error", "entity", "RPCDropEvent") + assert.GreaterOrEqual(t, dp.Value, int64(1)) + assert.True(t, hasEmitterStringAttr(dp.Attributes, "error_code", "Internal")) + assert.True(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) @@ -511,7 +676,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() @@ -539,8 +704,9 @@ func TestChipIngressBatchEmitterService_Metrics(t *testing.T) { 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", "MetricDropEvent") + dp := mustEmitterInt64SumPoint(t, sum, "error_type", "rpc_error", "entity", "MetricDropEvent") assert.GreaterOrEqual(t, dp.Value, int64(1)) + assert.True(t, hasEmitterStringAttr(dp.Attributes, "error_code", "DeadlineExceeded")) logs := observed.FilterMessage("failed to emit to chip ingress") require.GreaterOrEqual(t, logs.Len(), 1, "expected error log for publish failure") @@ -548,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"]) }) diff --git a/pkg/chipingress/batch/client.go b/pkg/chipingress/batch/client.go index 90c3bca980..22c249bf9e 100644 --- a/pkg/chipingress/batch/client.go +++ b/pkg/chipingress/batch/client.go @@ -19,6 +19,11 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/chipingress" ) +var ( + ErrMessageBufferFull = errors.New("message buffer is full") + ErrClientShutdown = errors.New("client is shutdown") +) + type messageWithCallback struct { event *chipingress.CloudEventPb callback func(error) @@ -240,7 +245,7 @@ func (b *Client) QueueMessage(event *chipingress.CloudEventPb, callback func(err // Check shutdown first to avoid race with buffer send select { case <-b.stopCh: - return errors.New("client is shutdown") + return ErrClientShutdown default: } @@ -274,7 +279,7 @@ func (b *Client) QueueMessage(event *chipingress.CloudEventPb, callback func(err case b.messageBuffer <- msg: return nil default: - return errors.New("message buffer is full") + return ErrMessageBufferFull } } diff --git a/pkg/chipingress/batch/drop_failure.go b/pkg/chipingress/batch/drop_failure.go new file mode 100644 index 0000000000..07586ac94d --- /dev/null +++ b/pkg/chipingress/batch/drop_failure.go @@ -0,0 +1,79 @@ +package batch + +import ( + "context" + "errors" + + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +const ( + ErrorTypePartialDelivery = "partial_delivery" + ErrorTypeRPCError = "rpc_error" + ErrorTypeBufferFull = "buffer_full" + ErrorTypeClientError = "client_error" +) + +// DropFailure describes a dropped event for metrics and logging. +type DropFailure struct { + ErrorType string // partial_delivery | rpc_error | buffer_full | client_error + ErrorCode string // PUBLISH_ERROR_CODE_* or gRPC code name (DeadlineExceeded, …) + ErrorReason string // server reason, status message, or client error text +} + +// ClassifyDropFailure maps batch send/queue errors to metric dimensions. +func ClassifyDropFailure(err error) DropFailure { + if err == nil { + return DropFailure{} + } + + var pubErr *PublishError + if errors.As(err, &pubErr) { + return DropFailure{ + ErrorType: ErrorTypePartialDelivery, + ErrorCode: pubErr.Code.String(), + ErrorReason: pubErr.Reason, + } + } + + if errors.Is(err, ErrMessageBufferFull) { + return DropFailure{ + ErrorType: ErrorTypeBufferFull, + ErrorReason: ErrMessageBufferFull.Error(), + } + } + if errors.Is(err, ErrClientShutdown) { + return DropFailure{ + ErrorType: ErrorTypeClientError, + ErrorReason: ErrClientShutdown.Error(), + } + } + if errors.Is(err, context.DeadlineExceeded) { + return DropFailure{ + ErrorType: ErrorTypeRPCError, + ErrorCode: codes.DeadlineExceeded.String(), + ErrorReason: context.DeadlineExceeded.Error(), + } + } + if errors.Is(err, context.Canceled) { + return DropFailure{ + ErrorType: ErrorTypeRPCError, + ErrorCode: codes.Canceled.String(), + ErrorReason: context.Canceled.Error(), + } + } + if st, ok := status.FromError(err); ok { + return DropFailure{ + ErrorType: ErrorTypeRPCError, + ErrorCode: st.Code().String(), + ErrorReason: st.Message(), + } + } + + return DropFailure{ + ErrorType: ErrorTypeRPCError, + ErrorCode: codes.Unknown.String(), + ErrorReason: err.Error(), + } +} diff --git a/pkg/chipingress/batch/drop_failure_test.go b/pkg/chipingress/batch/drop_failure_test.go new file mode 100644 index 0000000000..70dfb75ad9 --- /dev/null +++ b/pkg/chipingress/batch/drop_failure_test.go @@ -0,0 +1,98 @@ +package batch_test + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + "github.com/smartcontractkit/chainlink-common/pkg/chipingress" + "github.com/smartcontractkit/chainlink-common/pkg/chipingress/batch" +) + +func TestClassifyDropFailure(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + err error + wantType string + wantCode string + wantReason string + }{ + { + name: "partial delivery publish error", + err: &batch.PublishError{Code: chipingress.PublishErrorCode(1), Reason: "schema not found"}, + wantType: batch.ErrorTypePartialDelivery, + wantCode: chipingress.PublishErrorCode(1).String(), + wantReason: "schema not found", + }, + { + name: "deadline exceeded status", + err: status.Error(codes.DeadlineExceeded, "context deadline exceeded"), + wantType: batch.ErrorTypeRPCError, + wantCode: codes.DeadlineExceeded.String(), + wantReason: "context deadline exceeded", + }, + { + name: "deadline exceeded context", + err: context.DeadlineExceeded, + wantType: batch.ErrorTypeRPCError, + wantCode: codes.DeadlineExceeded.String(), + wantReason: context.DeadlineExceeded.Error(), + }, + { + name: "unavailable gateway 502", + err: status.Error(codes.Unavailable, `unexpected HTTP status code received from server: 502 (Bad Gateway)`), + wantType: batch.ErrorTypeRPCError, + wantCode: codes.Unavailable.String(), + wantReason: `unexpected HTTP status code received from server: 502 (Bad Gateway)`, + }, + { + name: "internal publish failure", + err: status.Error(codes.Internal, "failed to publish events"), + wantType: batch.ErrorTypeRPCError, + wantCode: codes.Internal.String(), + wantReason: "failed to publish events", + }, + { + name: "buffer full", + err: batch.ErrMessageBufferFull, + wantType: batch.ErrorTypeBufferFull, + wantReason: "message buffer is full", + }, + { + name: "client shutdown", + err: batch.ErrClientShutdown, + wantType: batch.ErrorTypeClientError, + wantReason: "client is shutdown", + }, + { + name: "unknown error", + err: errors.New("something else"), + wantType: batch.ErrorTypeRPCError, + wantCode: codes.Unknown.String(), + wantReason: "something else", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + got := batch.ClassifyDropFailure(tt.err) + assert.Equal(t, tt.wantType, got.ErrorType) + assert.Equal(t, tt.wantCode, got.ErrorCode) + assert.Equal(t, tt.wantReason, got.ErrorReason) + }) + } +} + +func TestClassifyDropFailure_nil(t *testing.T) { + t.Parallel() + got := batch.ClassifyDropFailure(nil) + require.Empty(t, got.ErrorType) +}