diff --git a/pkg/chipingress/batch/client.go b/pkg/chipingress/batch/client.go index ca66cb944e..b07d9c7999 100644 --- a/pkg/chipingress/batch/client.go +++ b/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 } } diff --git a/pkg/chipingress/batch/client_test.go b/pkg/chipingress/batch/client_test.go index 6896a3a1fd..c642371d9f 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: "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)) +}