beholder: handle partial-delivery errors via metric and log suppression

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

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"),

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"])
 	})