From 74cdc15f829d628b6b6d1002b6b3aba60f7fbbaf Mon Sep 17 00:00:00 2001 From: Pavel <177363085+pkcll@users.noreply.github.com> Date: Tue, 21 Jul 2026 01:58:07 -0400 Subject: [PATCH 1/9] beholder: remove partial delivery WARN log, keep metric only --- go.mod | 2 +- go.sum | 4 +- pkg/beholder/batch_emitter_service.go | 52 ++++++- pkg/beholder/batch_emitter_service_test.go | 169 ++++++++++++++++++++- 4 files changed, 218 insertions(+), 9 deletions(-) diff --git a/go.mod b/go.mod index 50ecc3555a..19aba1c61f 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.20260721060739-5b8513fe4094 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 28bc26fe4c..2101502dc3 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.20260721060739-5b8513fe4094 h1:l0K4bj62Rpdlmz5qfji9hpiYuE8DfkK0k152x7vTepc= +github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260721060739-5b8513fe4094/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 a81fd82bc9..effe780a3d 100644 --- a/pkg/beholder/batch_emitter_service.go +++ b/pkg/beholder/batch_emitter_service.go @@ -157,8 +157,24 @@ 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)) + // 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_type, error_code) + // already captures that it's happening and roughly why; the free-text + // error_reason is not on the metric, but a bounded error_code is enough to + // alert on, and the full reason isn't needed at fleet-wide log volume. + if drop.ErrorType != batch.ErrorTypePartialDelivery { + 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 +183,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 +215,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/error_type/ +// error_code are all closed, bounded sets. error_reason is still available on the +// corresponding log line. +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)) + } + 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 35a321cdac..bc19ff243e 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,167 @@ 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_type, 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 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")) + // 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) @@ -511,7 +674,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 +702,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 +712,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"]) }) From 0a5c31d271e6cb35ff2646abcd299df6da838dfe Mon Sep 17 00:00:00 2001 From: Pavel <177363085+pkcll@users.noreply.github.com> Date: Tue, 21 Jul 2026 16:58:27 -0400 Subject: [PATCH 2/9] beholder: use ErrorCodeFor instead of ClassifyDropFailure --- go.mod | 2 +- go.sum | 4 +-- pkg/beholder/batch_emitter_service.go | 50 +++++++++++++++++++-------- 3 files changed, 38 insertions(+), 18 deletions(-) diff --git a/go.mod b/go.mod index 19aba1c61f..5ad9cab85e 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.20260721060739-5b8513fe4094 + github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260721212710-e8d7286bd7f3 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 2101502dc3..123c106430 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.20260721060739-5b8513fe4094 h1:l0K4bj62Rpdlmz5qfji9hpiYuE8DfkK0k152x7vTepc= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260721060739-5b8513fe4094/go.mod h1:UYcRMb4dZcoaIPgZJ3hckCySTqtJc9K4Q+tOKErwTq0= +github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260721212710-e8d7286bd7f3 h1:qe1E8HK0EFQjzPrmUMpVtZViByfksVpF3LeEXBgW53Y= +github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260721212710-e8d7286bd7f3/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 effe780a3d..3cb25c55f4 100644 --- a/pkg/beholder/batch_emitter_service.go +++ b/pkg/beholder/batch_emitter_service.go @@ -3,6 +3,7 @@ package beholder import ( "context" "fmt" + "strings" "sync" "go.opentelemetry.io/otel" @@ -157,20 +158,21 @@ func (e *ChipIngressBatchEmitterService) emitInternal(ctx context.Context, body // for metric recording — OTel Add is non-blocking and tolerates // cancelled contexts. if sendErr != nil { - drop := batch.ClassifyDropFailure(sendErr) - e.metrics.eventsDropped.Add(ctx, 1, e.dropMetricAttrsFor(domain, entity, drop)) + errorCode := batch.ErrorCodeFor(sendErr) + errorType := errorTypeFromCode(errorCode) + e.metrics.eventsDropped.Add(ctx, 1, e.dropMetricAttrsFor(domain, entity, errorType, 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_type, error_code) // already captures that it's happening and roughly why; the free-text // error_reason is not on the metric, but a bounded error_code is enough to // alert on, and the full reason isn't needed at fleet-wide log volume. - if drop.ErrorType != batch.ErrorTypePartialDelivery { + if errorType != batch.ErrorTypePartialDelivery { e.eng.Errorw("failed to emit to chip ingress", "error", sendErr, - "error_type", drop.ErrorType, - "error_code", drop.ErrorCode, - "error_reason", drop.ErrorReason, + "error_type", errorType, + "error_code", errorCode, + "error_reason", sendErr.Error(), "domain", domain, "entity", entity, ) @@ -183,13 +185,14 @@ func (e *ChipIngressBatchEmitterService) emitInternal(ctx context.Context, body } }) if queueErr != nil { - drop := batch.ClassifyDropFailure(queueErr) - e.metrics.eventsDropped.Add(ctx, 1, e.dropMetricAttrsFor(domain, entity, drop)) + errorCode := batch.ErrorCodeFor(queueErr) + errorType := errorTypeFromCode(errorCode) + e.metrics.eventsDropped.Add(ctx, 1, e.dropMetricAttrsFor(domain, entity, errorType, errorCode)) e.eng.Errorw("failed to queue message for chip ingress", "error", queueErr, - "error_type", drop.ErrorType, - "error_code", drop.ErrorCode, - "error_reason", drop.ErrorReason, + "error_type", errorType, + "error_code", errorCode, + "error_reason", queueErr.Error(), "domain", domain, "entity", entity, ) @@ -215,6 +218,23 @@ func (e *ChipIngressBatchEmitterService) metricAttrsFor(domain, entity string) o return v.(otelmetric.MeasurementOption) } +// errorTypeFromCode infers the error category from the bounded error code +// produced by batch.ErrorCodeFor. +func errorTypeFromCode(code string) string { + switch { + case code == "": + return "" + case strings.HasPrefix(code, chipingress.PublishErrorCode(0).String()) || strings.HasPrefix(code, "PUBLISH_ERROR_CODE_"): + return batch.ErrorTypePartialDelivery + case code == batch.ErrMessageBufferFull.Error(): + return batch.ErrorTypeBufferFull + case code == batch.ErrClientShutdown.Error() || code == batch.ErrorTypeClientError: + return batch.ErrorTypeClientError + default: + return batch.ErrorTypeRPCError + } +} + // dropMetricAttrsFor returns a measurement option for the eventsDropped counter. // Not cached — drop paths are not on the hot path. // @@ -223,14 +243,14 @@ func (e *ChipIngressBatchEmitterService) metricAttrsFor(domain, entity string) o // it as a metric attribute would create unbounded cardinality. domain/entity/error_type/ // error_code are all closed, bounded sets. error_reason is still available on the // corresponding log line. -func (e *ChipIngressBatchEmitterService) dropMetricAttrsFor(domain, entity string, drop batch.DropFailure) otelmetric.MeasurementOption { +func (e *ChipIngressBatchEmitterService) dropMetricAttrsFor(domain, entity, errorType, errorCode string) otelmetric.MeasurementOption { attrs := []attribute.KeyValue{ attribute.String("domain", domain), attribute.String("entity", entity), - attribute.String("error_type", drop.ErrorType), + attribute.String("error_type", errorType), } - if drop.ErrorCode != "" { - attrs = append(attrs, attribute.String("error_code", drop.ErrorCode)) + if errorCode != "" { + attrs = append(attrs, attribute.String("error_code", errorCode)) } return otelmetric.WithAttributeSet(attribute.NewSet(attrs...)) } From 104555f73c8d80fc963ae47b49a4102d9f7524a4 Mon Sep 17 00:00:00 2001 From: Pavel <177363085+pkcll@users.noreply.github.com> Date: Tue, 21 Jul 2026 17:41:16 -0400 Subject: [PATCH 3/9] chipingress: drop DropFailure, add ErrorCodeFor --- pkg/chipingress/batch/client.go | 9 ++- pkg/chipingress/batch/drop_failure.go | 53 +++++++++++++++ pkg/chipingress/batch/drop_failure_test.go | 78 ++++++++++++++++++++++ 3 files changed, 138 insertions(+), 2 deletions(-) create mode 100644 pkg/chipingress/batch/drop_failure.go create mode 100644 pkg/chipingress/batch/drop_failure_test.go diff --git a/pkg/chipingress/batch/client.go b/pkg/chipingress/batch/client.go index 42e0a88cd6..c1dec4b6ed 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) @@ -242,7 +247,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: } @@ -276,7 +281,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..19e0c93698 --- /dev/null +++ b/pkg/chipingress/batch/drop_failure.go @@ -0,0 +1,53 @@ +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" +) + +// ErrorCodeFor returns a bounded error code string for a batch send/queue error. +// The returned code is intended for metric dimensions; callers can infer the +// error category from the code: +// - PUBLISH_ERROR_CODE_* -> partial_delivery +// - "buffer_full" -> buffer_full +// - "client_shutdown" -> client_error +// - gRPC code name -> rpc_error +// - "client_error" -> client_error (fallback for unrecognized client-side errors) +func ErrorCodeFor(err error) string { + if err == nil { + return "" + } + + var pubErr *PublishError + if errors.As(err, &pubErr) { + return pubErr.Code.String() + } + + if errors.Is(err, ErrMessageBufferFull) { + return ErrMessageBufferFull.Error() + } + if errors.Is(err, ErrClientShutdown) { + return ErrClientShutdown.Error() + } + if st, ok := status.FromError(err); ok { + return st.Code().String() + } + if errors.Is(err, context.DeadlineExceeded) { + return codes.DeadlineExceeded.String() + } + if errors.Is(err, context.Canceled) { + return codes.Canceled.String() + } + + return ErrorTypeClientError +} diff --git a/pkg/chipingress/batch/drop_failure_test.go b/pkg/chipingress/batch/drop_failure_test.go new file mode 100644 index 0000000000..0d1361b70a --- /dev/null +++ b/pkg/chipingress/batch/drop_failure_test.go @@ -0,0 +1,78 @@ +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 TestErrorCodeFor(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + err error + want string + }{ + { + name: "partial delivery publish error", + err: &batch.PublishError{Code: chipingress.PublishErrorCode(1), Reason: "schema not found"}, + want: chipingress.PublishErrorCode(1).String(), + }, + { + name: "deadline exceeded status", + err: status.Error(codes.DeadlineExceeded, "context deadline exceeded"), + want: codes.DeadlineExceeded.String(), + }, + { + name: "deadline exceeded context", + err: context.DeadlineExceeded, + want: codes.DeadlineExceeded.String(), + }, + { + name: "unavailable gateway 502", + err: status.Error(codes.Unavailable, `unexpected HTTP status code received from server: 502 (Bad Gateway)`), + want: codes.Unavailable.String(), + }, + { + name: "internal publish failure", + err: status.Error(codes.Internal, "failed to publish events"), + want: codes.Internal.String(), + }, + { + name: "buffer full", + err: batch.ErrMessageBufferFull, + want: batch.ErrMessageBufferFull.Error(), + }, + { + name: "client shutdown", + err: batch.ErrClientShutdown, + want: batch.ErrClientShutdown.Error(), + }, + { + name: "unknown error", + err: errors.New("something else"), + want: batch.ErrorTypeClientError, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + assert.Equal(t, tt.want, batch.ErrorCodeFor(tt.err)) + }) + } +} + +func TestErrorCodeFor_nil(t *testing.T) { + t.Parallel() + require.Empty(t, batch.ErrorCodeFor(nil)) +} From 20020f32ee2a9be332c390727c2df19663ca4151 Mon Sep 17 00:00:00 2001 From: Pavel <177363085+pkcll@users.noreply.github.com> Date: Wed, 22 Jul 2026 01:27:27 -0400 Subject: [PATCH 4/9] chipingress: return error type alongside code from ErrorCodeFor Avoids beholder having to re-derive the error category from the bounded code string. --- pkg/chipingress/batch/drop_failure.go | 28 ++++---- pkg/chipingress/batch/drop_failure_test.go | 77 ++++++++++++++-------- 2 files changed, 59 insertions(+), 46 deletions(-) diff --git a/pkg/chipingress/batch/drop_failure.go b/pkg/chipingress/batch/drop_failure.go index 19e0c93698..8e1ecf80f6 100644 --- a/pkg/chipingress/batch/drop_failure.go +++ b/pkg/chipingress/batch/drop_failure.go @@ -15,39 +15,33 @@ const ( ErrorTypeClientError = "client_error" ) -// ErrorCodeFor returns a bounded error code string for a batch send/queue error. -// The returned code is intended for metric dimensions; callers can infer the -// error category from the code: -// - PUBLISH_ERROR_CODE_* -> partial_delivery -// - "buffer_full" -> buffer_full -// - "client_shutdown" -> client_error -// - gRPC code name -> rpc_error -// - "client_error" -> client_error (fallback for unrecognized client-side errors) -func ErrorCodeFor(err error) string { +// ErrorCodeFor classifies a batch send/queue error for metric dimensions, +// returning both a coarse-grained error type and a bounded error code. +func ErrorCodeFor(err error) (errorType, errorCode string) { if err == nil { - return "" + return "", "" } var pubErr *PublishError if errors.As(err, &pubErr) { - return pubErr.Code.String() + return ErrorTypePartialDelivery, pubErr.Code.String() } if errors.Is(err, ErrMessageBufferFull) { - return ErrMessageBufferFull.Error() + return ErrorTypeBufferFull, ErrMessageBufferFull.Error() } if errors.Is(err, ErrClientShutdown) { - return ErrClientShutdown.Error() + return ErrorTypeClientError, ErrClientShutdown.Error() } if st, ok := status.FromError(err); ok { - return st.Code().String() + return ErrorTypeRPCError, st.Code().String() } if errors.Is(err, context.DeadlineExceeded) { - return codes.DeadlineExceeded.String() + return ErrorTypeRPCError, codes.DeadlineExceeded.String() } if errors.Is(err, context.Canceled) { - return codes.Canceled.String() + return ErrorTypeRPCError, codes.Canceled.String() } - return ErrorTypeClientError + return ErrorTypeClientError, ErrorTypeClientError } diff --git a/pkg/chipingress/batch/drop_failure_test.go b/pkg/chipingress/batch/drop_failure_test.go index 0d1361b70a..ef36e36881 100644 --- a/pkg/chipingress/batch/drop_failure_test.go +++ b/pkg/chipingress/batch/drop_failure_test.go @@ -18,61 +18,80 @@ func TestErrorCodeFor(t *testing.T) { t.Parallel() tests := []struct { - name string - err error - want string + name string + err error + wantType string + wantCode string }{ { - name: "partial delivery publish error", - err: &batch.PublishError{Code: chipingress.PublishErrorCode(1), Reason: "schema not found"}, - want: chipingress.PublishErrorCode(1).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(), }, { - name: "deadline exceeded status", - err: status.Error(codes.DeadlineExceeded, "context deadline exceeded"), - want: codes.DeadlineExceeded.String(), + name: "results mismatch is classified as partial delivery", + err: &batch.PublishError{Code: batch.ErrCodeResultsMismatch, Reason: "server returned 1 results for 2 events"}, + wantType: batch.ErrorTypePartialDelivery, + wantCode: batch.ErrCodeResultsMismatch.String(), }, { - name: "deadline exceeded context", - err: context.DeadlineExceeded, - want: codes.DeadlineExceeded.String(), + name: "deadline exceeded status", + err: status.Error(codes.DeadlineExceeded, "context deadline exceeded"), + wantType: batch.ErrorTypeRPCError, + wantCode: codes.DeadlineExceeded.String(), }, { - name: "unavailable gateway 502", - err: status.Error(codes.Unavailable, `unexpected HTTP status code received from server: 502 (Bad Gateway)`), - want: codes.Unavailable.String(), + name: "deadline exceeded context", + err: context.DeadlineExceeded, + wantType: batch.ErrorTypeRPCError, + wantCode: codes.DeadlineExceeded.String(), }, { - name: "internal publish failure", - err: status.Error(codes.Internal, "failed to publish events"), - want: codes.Internal.String(), + 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(), }, { - name: "buffer full", - err: batch.ErrMessageBufferFull, - want: batch.ErrMessageBufferFull.Error(), + name: "internal publish failure", + err: status.Error(codes.Internal, "failed to publish events"), + wantType: batch.ErrorTypeRPCError, + wantCode: codes.Internal.String(), }, { - name: "client shutdown", - err: batch.ErrClientShutdown, - want: batch.ErrClientShutdown.Error(), + name: "buffer full", + err: batch.ErrMessageBufferFull, + wantType: batch.ErrorTypeBufferFull, + wantCode: batch.ErrMessageBufferFull.Error(), }, { - name: "unknown error", - err: errors.New("something else"), - want: batch.ErrorTypeClientError, + name: "client shutdown", + err: batch.ErrClientShutdown, + wantType: batch.ErrorTypeClientError, + wantCode: batch.ErrClientShutdown.Error(), + }, + { + name: "unknown error", + err: errors.New("something else"), + wantType: batch.ErrorTypeClientError, + wantCode: batch.ErrorTypeClientError, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { t.Parallel() - assert.Equal(t, tt.want, batch.ErrorCodeFor(tt.err)) + errorType, errorCode := batch.ErrorCodeFor(tt.err) + assert.Equal(t, tt.wantType, errorType) + assert.Equal(t, tt.wantCode, errorCode) }) } } func TestErrorCodeFor_nil(t *testing.T) { t.Parallel() - require.Empty(t, batch.ErrorCodeFor(nil)) + errorType, errorCode := batch.ErrorCodeFor(nil) + require.Empty(t, errorType) + require.Empty(t, errorCode) } From 8fa67f2e20bf2358ee83d62dc32c383bb6f799b6 Mon Sep 17 00:00:00 2001 From: Pavel <177363085+pkcll@users.noreply.github.com> Date: Wed, 22 Jul 2026 01:30:47 -0400 Subject: [PATCH 5/9] beholder: consume simplified ErrorCodeFor, drop errorTypeFromCode ErrorCodeFor now returns the error type directly, so beholder no longer needs to re-derive it by string-matching the bounded error code (which mislabeled the synthetic results-mismatch code as rpc_error). --- go.mod | 2 +- go.sum | 4 ++-- pkg/beholder/batch_emitter_service.go | 24 ++---------------------- 3 files changed, 5 insertions(+), 25 deletions(-) diff --git a/go.mod b/go.mod index 5ad9cab85e..4f36e67b18 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.20260721212710-e8d7286bd7f3 + github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260722052727-20020f32ee2a 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 123c106430..74155b22c2 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.20260721212710-e8d7286bd7f3 h1:qe1E8HK0EFQjzPrmUMpVtZViByfksVpF3LeEXBgW53Y= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260721212710-e8d7286bd7f3/go.mod h1:UYcRMb4dZcoaIPgZJ3hckCySTqtJc9K4Q+tOKErwTq0= +github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260722052727-20020f32ee2a h1:dPMfKLDC6Pa48LHdGp8Pid5riDI3uX8xpAr/j853+KM= +github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260722052727-20020f32ee2a/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 3cb25c55f4..da99c601e7 100644 --- a/pkg/beholder/batch_emitter_service.go +++ b/pkg/beholder/batch_emitter_service.go @@ -3,7 +3,6 @@ package beholder import ( "context" "fmt" - "strings" "sync" "go.opentelemetry.io/otel" @@ -158,8 +157,7 @@ func (e *ChipIngressBatchEmitterService) emitInternal(ctx context.Context, body // for metric recording — OTel Add is non-blocking and tolerates // cancelled contexts. if sendErr != nil { - errorCode := batch.ErrorCodeFor(sendErr) - errorType := errorTypeFromCode(errorCode) + errorType, errorCode := batch.ErrorCodeFor(sendErr) e.metrics.eventsDropped.Add(ctx, 1, e.dropMetricAttrsFor(domain, entity, errorType, 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 @@ -185,8 +183,7 @@ func (e *ChipIngressBatchEmitterService) emitInternal(ctx context.Context, body } }) if queueErr != nil { - errorCode := batch.ErrorCodeFor(queueErr) - errorType := errorTypeFromCode(errorCode) + errorType, errorCode := batch.ErrorCodeFor(queueErr) e.metrics.eventsDropped.Add(ctx, 1, e.dropMetricAttrsFor(domain, entity, errorType, errorCode)) e.eng.Errorw("failed to queue message for chip ingress", "error", queueErr, @@ -218,23 +215,6 @@ func (e *ChipIngressBatchEmitterService) metricAttrsFor(domain, entity string) o return v.(otelmetric.MeasurementOption) } -// errorTypeFromCode infers the error category from the bounded error code -// produced by batch.ErrorCodeFor. -func errorTypeFromCode(code string) string { - switch { - case code == "": - return "" - case strings.HasPrefix(code, chipingress.PublishErrorCode(0).String()) || strings.HasPrefix(code, "PUBLISH_ERROR_CODE_"): - return batch.ErrorTypePartialDelivery - case code == batch.ErrMessageBufferFull.Error(): - return batch.ErrorTypeBufferFull - case code == batch.ErrClientShutdown.Error() || code == batch.ErrorTypeClientError: - return batch.ErrorTypeClientError - default: - return batch.ErrorTypeRPCError - } -} - // dropMetricAttrsFor returns a measurement option for the eventsDropped counter. // Not cached — drop paths are not on the hot path. // From 47eba7e26e3e20e59eeaa755cdd8aff409625a13 Mon Sep 17 00:00:00 2001 From: Pavel <177363085+pkcll@users.noreply.github.com> Date: Wed, 22 Jul 2026 01:40:47 -0400 Subject: [PATCH 6/9] chipingress: merge drop_failure.go into client.go, return only ErrorCode ErrorType was inferable from ErrorCode and added no information the caller couldn't derive itself; drop it so there's a single bounded value to reason about. --- pkg/chipingress/batch/client.go | 33 ++++++++ pkg/chipingress/batch/client_test.go | 70 ++++++++++++++++ pkg/chipingress/batch/drop_failure.go | 47 ----------- pkg/chipingress/batch/drop_failure_test.go | 97 ---------------------- 4 files changed, 103 insertions(+), 144 deletions(-) delete mode 100644 pkg/chipingress/batch/drop_failure.go delete mode 100644 pkg/chipingress/batch/drop_failure_test.go diff --git a/pkg/chipingress/batch/client.go b/pkg/chipingress/batch/client.go index c1dec4b6ed..733891aabc 100644 --- a/pkg/chipingress/batch/client.go +++ b/pkg/chipingress/batch/client.go @@ -14,6 +14,8 @@ import ( "go.opentelemetry.io/otel/attribute" otelmetric "go.opentelemetry.io/otel/metric" "go.uber.org/zap" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" "google.golang.org/protobuf/proto" "github.com/smartcontractkit/chainlink-common/pkg/chipingress" @@ -56,6 +58,37 @@ const ( ErrCodeResultsMismatch = chipingress.PublishErrorCode(-1) // client-side synthetic code ) +// ErrorCodeFor returns a bounded error code string for a batch send/queue +// error, suitable for use as a metric dimension. +func ErrorCodeFor(err error) string { + if err == nil { + return "" + } + + var pubErr *PublishError + if errors.As(err, &pubErr) { + return pubErr.Code.String() + } + + if errors.Is(err, ErrMessageBufferFull) { + return ErrMessageBufferFull.Error() + } + if errors.Is(err, ErrClientShutdown) { + return ErrClientShutdown.Error() + } + if st, ok := status.FromError(err); ok { + return st.Code().String() + } + if errors.Is(err, context.DeadlineExceeded) { + return codes.DeadlineExceeded.String() + } + if errors.Is(err, context.Canceled) { + return codes.Canceled.String() + } + + return "client_error" +} + // Client is a batching client that accumulates messages and sends them in batches. type Client struct { client chipingress.Client diff --git a/pkg/chipingress/batch/client_test.go b/pkg/chipingress/batch/client_test.go index 4f0ffc70eb..00b134d589 100644 --- a/pkg/chipingress/batch/client_test.go +++ b/pkg/chipingress/batch/client_test.go @@ -18,6 +18,8 @@ import ( "go.opentelemetry.io/otel/attribute" sdkmetric "go.opentelemetry.io/otel/sdk/metric" "go.opentelemetry.io/otel/sdk/metric/metricdata" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" "google.golang.org/protobuf/proto" "github.com/smartcontractkit/chainlink-common/pkg/chipingress" @@ -2346,3 +2348,71 @@ func TestTransactionEnabledEdgeCases(t *testing.T) { assert.EqualError(t, err3, "kafka unavailable") }) } + +func TestErrorCodeFor(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + err error + want string + }{ + { + name: "partial delivery publish error", + err: &PublishError{Code: chipingress.PublishErrorCode(1), Reason: "schema not found"}, + want: chipingress.PublishErrorCode(1).String(), + }, + { + name: "results mismatch", + err: &PublishError{Code: ErrCodeResultsMismatch, Reason: "server returned 1 results for 2 events"}, + want: ErrCodeResultsMismatch.String(), + }, + { + name: "deadline exceeded status", + err: status.Error(codes.DeadlineExceeded, "context deadline exceeded"), + want: codes.DeadlineExceeded.String(), + }, + { + name: "deadline exceeded context", + err: context.DeadlineExceeded, + want: codes.DeadlineExceeded.String(), + }, + { + name: "unavailable gateway 502", + err: status.Error(codes.Unavailable, `unexpected HTTP status code received from server: 502 (Bad Gateway)`), + want: codes.Unavailable.String(), + }, + { + name: "internal publish failure", + err: status.Error(codes.Internal, "failed to publish events"), + want: codes.Internal.String(), + }, + { + name: "buffer full", + err: ErrMessageBufferFull, + want: ErrMessageBufferFull.Error(), + }, + { + name: "client shutdown", + err: ErrClientShutdown, + want: ErrClientShutdown.Error(), + }, + { + name: "unknown error", + err: errors.New("something else"), + want: "client_error", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + assert.Equal(t, tt.want, ErrorCodeFor(tt.err)) + }) + } +} + +func TestErrorCodeFor_nil(t *testing.T) { + t.Parallel() + require.Empty(t, ErrorCodeFor(nil)) +} diff --git a/pkg/chipingress/batch/drop_failure.go b/pkg/chipingress/batch/drop_failure.go deleted file mode 100644 index 8e1ecf80f6..0000000000 --- a/pkg/chipingress/batch/drop_failure.go +++ /dev/null @@ -1,47 +0,0 @@ -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" -) - -// ErrorCodeFor classifies a batch send/queue error for metric dimensions, -// returning both a coarse-grained error type and a bounded error code. -func ErrorCodeFor(err error) (errorType, errorCode string) { - if err == nil { - return "", "" - } - - var pubErr *PublishError - if errors.As(err, &pubErr) { - return ErrorTypePartialDelivery, pubErr.Code.String() - } - - if errors.Is(err, ErrMessageBufferFull) { - return ErrorTypeBufferFull, ErrMessageBufferFull.Error() - } - if errors.Is(err, ErrClientShutdown) { - return ErrorTypeClientError, ErrClientShutdown.Error() - } - if st, ok := status.FromError(err); ok { - return ErrorTypeRPCError, st.Code().String() - } - if errors.Is(err, context.DeadlineExceeded) { - return ErrorTypeRPCError, codes.DeadlineExceeded.String() - } - if errors.Is(err, context.Canceled) { - return ErrorTypeRPCError, codes.Canceled.String() - } - - return ErrorTypeClientError, ErrorTypeClientError -} diff --git a/pkg/chipingress/batch/drop_failure_test.go b/pkg/chipingress/batch/drop_failure_test.go deleted file mode 100644 index ef36e36881..0000000000 --- a/pkg/chipingress/batch/drop_failure_test.go +++ /dev/null @@ -1,97 +0,0 @@ -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 TestErrorCodeFor(t *testing.T) { - t.Parallel() - - tests := []struct { - name string - err error - wantType string - wantCode 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(), - }, - { - name: "results mismatch is classified as partial delivery", - err: &batch.PublishError{Code: batch.ErrCodeResultsMismatch, Reason: "server returned 1 results for 2 events"}, - wantType: batch.ErrorTypePartialDelivery, - wantCode: batch.ErrCodeResultsMismatch.String(), - }, - { - name: "deadline exceeded status", - err: status.Error(codes.DeadlineExceeded, "context deadline exceeded"), - wantType: batch.ErrorTypeRPCError, - wantCode: codes.DeadlineExceeded.String(), - }, - { - name: "deadline exceeded context", - err: context.DeadlineExceeded, - wantType: batch.ErrorTypeRPCError, - wantCode: codes.DeadlineExceeded.String(), - }, - { - 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(), - }, - { - name: "internal publish failure", - err: status.Error(codes.Internal, "failed to publish events"), - wantType: batch.ErrorTypeRPCError, - wantCode: codes.Internal.String(), - }, - { - name: "buffer full", - err: batch.ErrMessageBufferFull, - wantType: batch.ErrorTypeBufferFull, - wantCode: batch.ErrMessageBufferFull.Error(), - }, - { - name: "client shutdown", - err: batch.ErrClientShutdown, - wantType: batch.ErrorTypeClientError, - wantCode: batch.ErrClientShutdown.Error(), - }, - { - name: "unknown error", - err: errors.New("something else"), - wantType: batch.ErrorTypeClientError, - wantCode: batch.ErrorTypeClientError, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - t.Parallel() - errorType, errorCode := batch.ErrorCodeFor(tt.err) - assert.Equal(t, tt.wantType, errorType) - assert.Equal(t, tt.wantCode, errorCode) - }) - } -} - -func TestErrorCodeFor_nil(t *testing.T) { - t.Parallel() - errorType, errorCode := batch.ErrorCodeFor(nil) - require.Empty(t, errorType) - require.Empty(t, errorCode) -} From b6d3e94f234e862b2b24f8d7f47cb08dbb9f25f3 Mon Sep 17 00:00:00 2001 From: Pavel <177363085+pkcll@users.noreply.github.com> Date: Wed, 22 Jul 2026 01:43:14 -0400 Subject: [PATCH 7/9] beholder: drop error_type, use errors.As to decide log suppression ErrorCodeFor now returns only the bounded error code. Whether a drop is a per-event PublishError (suppressed from logs) is now decided directly via errors.As on the original error instead of string-matching a derived category. --- go.mod | 2 +- go.sum | 4 +-- pkg/beholder/batch_emitter_service.go | 30 ++++++++++------------ pkg/beholder/batch_emitter_service_test.go | 18 +++++-------- 4 files changed, 23 insertions(+), 31 deletions(-) diff --git a/go.mod b/go.mod index 4f36e67b18..e441163114 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.20260722052727-20020f32ee2a + github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260722054047-47eba7e26e3e 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 74155b22c2..0aebf77319 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.20260722052727-20020f32ee2a h1:dPMfKLDC6Pa48LHdGp8Pid5riDI3uX8xpAr/j853+KM= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260722052727-20020f32ee2a/go.mod h1:UYcRMb4dZcoaIPgZJ3hckCySTqtJc9K4Q+tOKErwTq0= +github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260722054047-47eba7e26e3e h1:FggneW/vLdCRt56lVn79SxNX3mZwNOPQNPyFY3yetz8= +github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260722054047-47eba7e26e3e/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 da99c601e7..207942bd5c 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" @@ -157,18 +158,17 @@ func (e *ChipIngressBatchEmitterService) emitInternal(ctx context.Context, body // for metric recording — OTel Add is non-blocking and tolerates // cancelled contexts. if sendErr != nil { - errorType, errorCode := batch.ErrorCodeFor(sendErr) - e.metrics.eventsDropped.Add(ctx, 1, e.dropMetricAttrsFor(domain, entity, errorType, errorCode)) + 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_type, error_code) - // already captures that it's happening and roughly why; the free-text - // error_reason is not on the metric, but a bounded error_code is enough to - // alert on, and the full reason isn't needed at fleet-wide log volume. - if errorType != batch.ErrorTypePartialDelivery { + // 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_type", errorType, "error_code", errorCode, "error_reason", sendErr.Error(), "domain", domain, @@ -183,11 +183,10 @@ func (e *ChipIngressBatchEmitterService) emitInternal(ctx context.Context, body } }) if queueErr != nil { - errorType, errorCode := batch.ErrorCodeFor(queueErr) - e.metrics.eventsDropped.Add(ctx, 1, e.dropMetricAttrsFor(domain, entity, errorType, errorCode)) + 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_type", errorType, "error_code", errorCode, "error_reason", queueErr.Error(), "domain", domain, @@ -220,14 +219,13 @@ func (e *ChipIngressBatchEmitterService) metricAttrsFor(domain, entity string) o // // 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/error_type/ -// error_code are all closed, bounded sets. error_reason is still available on the -// corresponding log line. -func (e *ChipIngressBatchEmitterService) dropMetricAttrsFor(domain, entity, errorType, errorCode string) otelmetric.MeasurementOption { +// it as a metric attribute would create unbounded cardinality. domain/entity/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("error_type", errorType), } if errorCode != "" { attrs = append(attrs, attribute.String("error_code", errorCode)) diff --git a/pkg/beholder/batch_emitter_service_test.go b/pkg/beholder/batch_emitter_service_test.go index bc19ff243e..67fb2e88b5 100644 --- a/pkg/beholder/batch_emitter_service_test.go +++ b/pkg/beholder/batch_emitter_service_test.go @@ -507,13 +507,13 @@ func TestChipIngressBatchEmitterService_PartialDeliveryError(t *testing.T) { // 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_type, error_code) is the - // intended signal; see the "records events_dropped" subtest below. + // 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 partial_delivery error_type", func(t *testing.T) { + t.Run("records events_dropped with the PublishError code", func(t *testing.T) { reader, restore := useEmitterTestMeterProvider(t) defer restore() @@ -562,12 +562,8 @@ func TestChipIngressBatchEmitterService_PartialDeliveryError(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, "error_type", "partial_delivery", "entity", "PartialEvent") + dp := mustEmitterInt64SumPoint(t, sum, "error_code", "PUBLISH_ERROR_CODE_VALIDATION_FAILED", "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", - ) }) } @@ -611,9 +607,8 @@ func TestChipIngressBatchEmitterService_RPCError(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, "error_type", "rpc_error", "entity", "RPCDropEvent") + dp := mustEmitterInt64SumPoint(t, sum, "error_code", "Internal", "entity", "RPCDropEvent") assert.GreaterOrEqual(t, dp.Value, int64(1)) - assert.True(t, hasEmitterStringAttr(dp.Attributes, "error_code", "Internal")) // 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")) @@ -702,9 +697,8 @@ 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, "error_type", "rpc_error", "entity", "MetricDropEvent") + dp := mustEmitterInt64SumPoint(t, sum, "error_code", "DeadlineExceeded", "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") From 7f99ebd64d948bfcd8cb6f126bb47513fca4f5b3 Mon Sep 17 00:00:00 2001 From: Pavel <177363085+pkcll@users.noreply.github.com> Date: Wed, 22 Jul 2026 11:14:53 -0400 Subject: [PATCH 8/9] beholder: add client_name to events_dropped metric Copilot review feedback: preserve the client_name attribute on chip_ingress.events_dropped when adding error_code, so existing dashboards/alerts keep working. --- pkg/beholder/batch_emitter_service.go | 3 ++- pkg/beholder/batch_emitter_service_test.go | 9 +++++++-- 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/pkg/beholder/batch_emitter_service.go b/pkg/beholder/batch_emitter_service.go index 4ed86a6b5c..d2560a981e 100644 --- a/pkg/beholder/batch_emitter_service.go +++ b/pkg/beholder/batch_emitter_service.go @@ -226,13 +226,14 @@ func (e *ChipIngressBatchEmitterService) metricAttrsFor(domain, entity string) o // // 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/error_code +// 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)) diff --git a/pkg/beholder/batch_emitter_service_test.go b/pkg/beholder/batch_emitter_service_test.go index 4c5fae32ec..5a405a6b9b 100644 --- a/pkg/beholder/batch_emitter_service_test.go +++ b/pkg/beholder/batch_emitter_service_test.go @@ -563,7 +563,9 @@ func TestChipIngressBatchEmitterService_PartialDeliveryError(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, "error_code", "PUBLISH_ERROR_CODE_VALIDATION_FAILED", "entity", "PartialEvent") + 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)) }) } @@ -608,7 +610,9 @@ func TestChipIngressBatchEmitterService_RPCError(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, "error_code", "Internal", "entity", "RPCDropEvent") + 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. @@ -701,6 +705,7 @@ func TestChipIngressBatchEmitterService_Metrics(t *testing.T) { 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)) logs := observed.FilterMessage("failed to emit to chip ingress") From a09f28791b0e81b9a594bc57b4be39a28e797684 Mon Sep 17 00:00:00 2001 From: Pavel <177363085+pkcll@users.noreply.github.com> Date: Fri, 24 Jul 2026 10:43:47 -0400 Subject: [PATCH 9/9] bump pkg/chipingress to main pseudo-version from #2274 Updates the root go.mod/go.sum to use the merged main commit of the chipingress submodule.\nThis resolves the dependency validation failure that required chipingress pseudo-versions to exist on main. --- go.mod | 2 +- go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index 92e8409a64..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.20260722054047-47eba7e26e3e + 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 c253c3cea6..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.20260722054047-47eba7e26e3e h1:FggneW/vLdCRt56lVn79SxNX3mZwNOPQNPyFY3yetz8= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260722054047-47eba7e26e3e/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=