[1/2] chipingress: add ErrorCodeFor for partial delivery error classification

pkg/chipingress/batch/client.go

@@ -14,11 +14,18 @@ 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"
 )
 
+var (
+	ErrMessageBufferFull = errors.New("message buffer is full")
+	ErrClientShutdown    = errors.New("client is shutdown")
+)
+
 type messageWithCallback struct {
 	event    *chipingress.CloudEventPb
 	callback func(error)
@@ -51,6 +58,40 @@ 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) {
+		if pubErr.Code == ErrCodeResultsMismatch {
+			return "results_mismatch"
+		}
+		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
@@ -242,7 +283,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 +317,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
 	}
 }
 

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: "results_mismatch",
+		},
+		{
+			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))
+}