Return proper http response code based on retryable error

receiver/otlpreceiver/internal/errors/errors.go

@@ -0,0 +1,46 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package errors // import "go.opentelemetry.io/collector/receiver/otlpreceiver/internal/errors"
+
+import (
+	"net/http"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	"go.opentelemetry.io/collector/consumer/consumererror"
+)
+
+func GetStatusFromError(err error) *status.Status {
+	s, ok := status.FromError(err)
+	if !ok {
+		// Default to a retryable error
+		// https://github.com/open-telemetry/opentelemetry-proto/blob/main/docs/specification.md#failures
+		code := codes.Unavailable
+		if consumererror.IsPermanent(err) {
+			code = codes.InvalidArgument
+		}
+		s = status.New(code, err.Error())
+	}
+	return s
+}
+
+func GetHTTPStatusCodeFromStatus(err error) int {
+	s, ok := status.FromError(err)
+	if !ok {
+		return http.StatusInternalServerError
+	}
+	// See https://github.com/open-telemetry/opentelemetry-proto/blob/main/docs/specification.md#failures
+	// to see if a code is retryable.
+	// See https://github.com/open-telemetry/opentelemetry-proto/blob/main/docs/specification.md#failures-1
+	// to see a list of retryable http status codes.
+	switch s.Code() {
+	// Retryable
+	case codes.Canceled, codes.DeadlineExceeded, codes.Aborted, codes.OutOfRange, codes.Unavailable, codes.DataLoss:
+		return http.StatusServiceUnavailable
+	// Not Retryable
+	default:
+		return http.StatusInternalServerError
+	}
+}

receiver/otlpreceiver/internal/errors/errors_test.go

@@ -0,0 +1,76 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package errors // import "go.opentelemetry.io/collector/receiver/otlpreceiver/internal/util"
+
+import (
+	"fmt"
+	"net/http"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	"go.opentelemetry.io/collector/consumer/consumererror"
+)
+
+func Test_GetStatusFromError(t *testing.T) {
+	tests := []struct {
+		name     string
+		input    error
+		expected *status.Status
+	}{
+		{
+			name:     "Status",
+			input:    status.Error(codes.Aborted, "test"),
+			expected: status.New(codes.Aborted, "test"),
+		},
+		{
+			name:     "Permanent Error",
+			input:    consumererror.NewPermanent(fmt.Errorf("test")),
+			expected: status.New(codes.InvalidArgument, "Permanent error: test"),
+		},
+		{
+			name:     "Non-Permanent Error",
+			input:    fmt.Errorf("test"),
+			expected: status.New(codes.Unavailable, "test"),
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := GetStatusFromError(tt.input)
+			assert.Equal(t, tt.expected, result)
+		})
+	}
+}
+
+func Test_GetHTTPStatusCodeFromStatus(t *testing.T) {
+	tests := []struct {
+		name     string
+		input    error
+		expected int
+	}{
+		{
+			name:     "Not a Status",
+			input:    fmt.Errorf("not a status error"),
+			expected: http.StatusInternalServerError,
+		},
+		{
+			name:     "Retryable Status",
+			input:    status.New(codes.Unavailable, "test").Err(),
+			expected: http.StatusServiceUnavailable,
+		},
+		{
+			name:     "Non-retryable Status",
+			input:    status.New(codes.InvalidArgument, "test").Err(),
+			expected: http.StatusInternalServerError,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := GetHTTPStatusCodeFromStatus(tt.input)
+			assert.Equal(t, tt.expected, result)
+		})
+	}
+}

receiver/otlpreceiver/internal/logs/otlp.go

@@ -6,12 +6,9 @@ package logs // import "go.opentelemetry.io/collector/receiver/otlpreceiver/inte
 import (
 	"context"
 
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
-
 	"go.opentelemetry.io/collector/consumer"
-	"go.opentelemetry.io/collector/consumer/consumererror"
 	"go.opentelemetry.io/collector/pdata/plog/plogotlp"
+	"go.opentelemetry.io/collector/receiver/otlpreceiver/internal/errors"
 	"go.opentelemetry.io/collector/receiver/receiverhelper"
 )
 
@@ -51,15 +48,7 @@ func (r *Receiver) Export(ctx context.Context, req plogotlp.ExportRequest) (plog
 	// NonPermanent errors will be converted to codes.Unavailable (equivalent to HTTP 503)
 	// Permanent errors will be converted to codes.InvalidArgument (equivalent to HTTP 400)
 	if err != nil {
-		s, ok := status.FromError(err)
-		if !ok {
-			code := codes.Unavailable
-			if consumererror.IsPermanent(err) {
-				code = codes.InvalidArgument
-			}
-			s = status.New(code, err.Error())
-		}
-		return plogotlp.NewExportResponse(), s.Err()
+		return plogotlp.NewExportResponse(), errors.GetStatusFromError(err).Err()
 	}
 
 	return plogotlp.NewExportResponse(), nil

receiver/otlpreceiver/internal/metrics/otlp.go

@@ -6,12 +6,9 @@ package metrics // import "go.opentelemetry.io/collector/receiver/otlpreceiver/i
 import (
 	"context"
 
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
-
 	"go.opentelemetry.io/collector/consumer"
-	"go.opentelemetry.io/collector/consumer/consumererror"
 	"go.opentelemetry.io/collector/pdata/pmetric/pmetricotlp"
+	"go.opentelemetry.io/collector/receiver/otlpreceiver/internal/errors"
 	"go.opentelemetry.io/collector/receiver/receiverhelper"
 )
 
@@ -51,15 +48,7 @@ func (r *Receiver) Export(ctx context.Context, req pmetricotlp.ExportRequest) (p
 	// NonPermanent errors will be converted to codes.Unavailable (equivalent to HTTP 503)
 	// Permanent errors will be converted to codes.InvalidArgument (equivalent to HTTP 400)
 	if err != nil {
-		s, ok := status.FromError(err)
-		if !ok {
-			code := codes.Unavailable
-			if consumererror.IsPermanent(err) {
-				code = codes.InvalidArgument
-			}
-			s = status.New(code, err.Error())
-		}
-		return pmetricotlp.NewExportResponse(), s.Err()
+		return pmetricotlp.NewExportResponse(), errors.GetStatusFromError(err).Err()
 	}
 
 	return pmetricotlp.NewExportResponse(), nil

receiver/otlpreceiver/internal/trace/otlp.go

@@ -6,12 +6,9 @@ package trace // import "go.opentelemetry.io/collector/receiver/otlpreceiver/int
 import (
 	"context"
 
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
-
 	"go.opentelemetry.io/collector/consumer"
-	"go.opentelemetry.io/collector/consumer/consumererror"
 	"go.opentelemetry.io/collector/pdata/ptrace/ptraceotlp"
+	"go.opentelemetry.io/collector/receiver/otlpreceiver/internal/errors"
 	"go.opentelemetry.io/collector/receiver/receiverhelper"
 )
 
@@ -52,15 +49,7 @@ func (r *Receiver) Export(ctx context.Context, req ptraceotlp.ExportRequest) (pt
 	// NonPermanent errors will be converted to codes.Unavailable (equivalent to HTTP 503)
 	// Permanent errors will be converted to codes.InvalidArgument (equivalent to HTTP 400)
 	if err != nil {
-		s, ok := status.FromError(err)
-		if !ok {
-			code := codes.Unavailable
-			if consumererror.IsPermanent(err) {
-				code = codes.InvalidArgument
-			}
-			s = status.New(code, err.Error())
-		}
-		return ptraceotlp.NewExportResponse(), s.Err()
+		return ptraceotlp.NewExportResponse(), errors.GetStatusFromError(err).Err()
 	}
 
 	return ptraceotlp.NewExportResponse(), nil