OTLP HTTP Exporter: Propagate HTTP 429s

.chloggen/fix-429-logic.yaml

@@ -0,0 +1,25 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: otlphttpexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Fixes a bug that was preventing the otlp http exporter from propagating status.
+
+# One or more tracking issues or pull requests related to the change
+issues: [9892]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

exporter/otlphttpexporter/otlp.go

@@ -16,13 +16,14 @@ import (
 	"time"
 
 	"go.uber.org/zap"
-	"google.golang.org/genproto/googleapis/rpc/status"
+	spb "google.golang.org/genproto/googleapis/rpc/status"
 	"google.golang.org/protobuf/proto"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer/consumererror"
 	"go.opentelemetry.io/collector/exporter"
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
+	"go.opentelemetry.io/collector/internal/httphelper"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/plog/plogotlp"
 	"go.opentelemetry.io/collector/pdata/pmetric"
@@ -184,16 +185,18 @@ func (e *baseExporter) export(ctx context.Context, url string, request []byte, p
 	respStatus := readResponseStatus(resp)
 
 	// Format the error message. Use the status if it is present in the response.
+	var errString string
 	var formattedErr error
 	if respStatus != nil {
-		formattedErr = fmt.Errorf(
+		errString = fmt.Sprintf(
 			"error exporting items, request to %s responded with HTTP Status Code %d, Message=%s, Details=%v",
 			url, resp.StatusCode, respStatus.Message, respStatus.Details)
 	} else {
-		formattedErr = fmt.Errorf(
+		errString = fmt.Sprintf(
 			"error exporting items, request to %s responded with HTTP Status Code %d",
 			url, resp.StatusCode)
 	}
+	formattedErr = httphelper.NewStatusFromMsgAndHTTPCode(errString, resp.StatusCode).Err()
 
 	if isRetryableStatusCode(resp.StatusCode) {
 		// A retry duration of 0 seconds will trigger the default backoff policy
@@ -267,8 +270,8 @@ func readResponseBody(resp *http.Response) ([]byte, error) {
 
 // Read the response and decode the status.Status from the body.
 // Returns nil if the response is empty or cannot be decoded.
-func readResponseStatus(resp *http.Response) *status.Status {
-	var respStatus *status.Status
+func readResponseStatus(resp *http.Response) *spb.Status {
+	var respStatus *spb.Status
 	if resp.StatusCode >= 400 && resp.StatusCode <= 599 {
 		// Request failed. Read the body. OTLP spec says:
 		// "Response body for all HTTP 4xx and HTTP 5xx responses MUST be a
@@ -279,7 +282,7 @@ func readResponseStatus(resp *http.Response) *status.Status {
 		}
 
 		// Decode it as Status struct. See https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/protocol/otlp.md#failures
-		respStatus = &status.Status{}
+		respStatus = &spb.Status{}
 		err = proto.Unmarshal(respBytes, respStatus)
 		if err != nil {
 			return nil

exporter/otlphttpexporter/otlp_test.go

@@ -128,12 +128,12 @@ func TestErrorResponses(t *testing.T) {
 			isPermErr:      true,
 		},
 		{
-			name:           "419",
+			name:           "429",
 			responseStatus: http.StatusTooManyRequests,
-			responseBody:   status.New(codes.InvalidArgument, "Quota exceeded"),
+			responseBody:   status.New(codes.ResourceExhausted, "Quota exceeded"),
 			err: func(srv *httptest.Server) error {
 				return exporterhelper.NewThrottleRetry(
-					errors.New(errMsgPrefix(srv)+"429, Message=Quota exceeded, Details=[]"),
+					status.New(codes.ResourceExhausted, errMsgPrefix(srv)+"429, Message=Quota exceeded, Details=[]").Err(),
 					time.Duration(0)*time.Second)
 			},
 		},
@@ -149,7 +149,7 @@ func TestErrorResponses(t *testing.T) {
 			responseBody:   status.New(codes.InvalidArgument, "Bad gateway"),
 			err: func(srv *httptest.Server) error {
 				return exporterhelper.NewThrottleRetry(
-					errors.New(errMsgPrefix(srv)+"502, Message=Bad gateway, Details=[]"),
+					status.New(codes.Unavailable, errMsgPrefix(srv)+"502, Message=Bad gateway, Details=[]").Err(),
 					time.Duration(0)*time.Second)
 			},
 		},
@@ -159,7 +159,7 @@ func TestErrorResponses(t *testing.T) {
 			responseBody:   status.New(codes.InvalidArgument, "Server overloaded"),
 			err: func(srv *httptest.Server) error {
 				return exporterhelper.NewThrottleRetry(
-					errors.New(errMsgPrefix(srv)+"503, Message=Server overloaded, Details=[]"),
+					status.New(codes.Unavailable, errMsgPrefix(srv)+"503, Message=Server overloaded, Details=[]").Err(),
 					time.Duration(0)*time.Second)
 			},
 		},
@@ -170,7 +170,7 @@ func TestErrorResponses(t *testing.T) {
 			headers:        map[string]string{"Retry-After": "30"},
 			err: func(srv *httptest.Server) error {
 				return exporterhelper.NewThrottleRetry(
-					errors.New(errMsgPrefix(srv)+"503, Message=Server overloaded, Details=[]"),
+					status.New(codes.Unavailable, errMsgPrefix(srv)+"503, Message=Server overloaded, Details=[]").Err(),
 					time.Duration(30)*time.Second)
 			},
 		},
@@ -180,7 +180,7 @@ func TestErrorResponses(t *testing.T) {
 			responseBody:   status.New(codes.InvalidArgument, "Gateway timeout"),
 			err: func(srv *httptest.Server) error {
 				return exporterhelper.NewThrottleRetry(
-					errors.New(errMsgPrefix(srv)+"504, Message=Gateway timeout, Details=[]"),
+					status.New(codes.Unavailable, errMsgPrefix(srv)+"504, Message=Gateway timeout, Details=[]").Err(),
 					time.Duration(0)*time.Second)
 			},
 		},
@@ -190,7 +190,7 @@ func TestErrorResponses(t *testing.T) {
 			responseBody:   status.New(codes.InvalidArgument, strings.Repeat("a", maxHTTPResponseReadBytes+1)),
 			err: func(srv *httptest.Server) error {
 				return exporterhelper.NewThrottleRetry(
-					errors.New(errMsgPrefix(srv)+"503"),
+					status.New(codes.Unavailable, errMsgPrefix(srv)+"503").Err(),
 					time.Duration(0)*time.Second)
 			},
 		},

go.mod

@@ -14,6 +14,7 @@ require (
 	go.uber.org/goleak v1.3.0
 	go.uber.org/multierr v1.11.0
 	go.uber.org/zap v1.27.0
+	google.golang.org/grpc v1.62.1
 )
 
 require (
@@ -64,7 +65,6 @@ require (
 	golang.org/x/text v0.14.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240123012728-ef4313101c80 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240123012728-ef4313101c80 // indirect
-	google.golang.org/grpc v1.62.1 // indirect
 	google.golang.org/protobuf v1.33.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

internal/httphelper/helper.go

@@ -0,0 +1,36 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package httphelper // import "go.opentelemetry.io/collector/internal/httphelper"
+
+import (
+	"net/http"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// NewStatusFromMsgAndHTTPCode returns a gRPC status based on an error message string and a http status code.
+// This function is shared between the http receiver and http exporter for error propagation.
+func NewStatusFromMsgAndHTTPCode(errMsg string, statusCode int) *status.Status {
+	var c codes.Code
+	// Mapping based on https://github.com/grpc/grpc/blob/master/doc/http-grpc-status-mapping.md
+	// 429 mapping to ResourceExhausted and 400 mapping to StatusBadRequest are exceptions.
+	switch statusCode {
+	case http.StatusBadRequest:
+		c = codes.InvalidArgument
+	case http.StatusUnauthorized:
+		c = codes.Unauthenticated
+	case http.StatusForbidden:
+		c = codes.PermissionDenied
+	case http.StatusNotFound:
+		c = codes.Unimplemented
+	case http.StatusTooManyRequests:
+		c = codes.ResourceExhausted
+	case http.StatusBadGateway, http.StatusServiceUnavailable, http.StatusGatewayTimeout:
+		c = codes.Unavailable
+	default:
+		c = codes.Unknown
+	}
+	return status.New(c, errMsg)
+}

internal/httphelper/helper_test.go

@@ -0,0 +1,83 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package httphelper
+
+import (
+	"net/http"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+func Test_ErrorMsgAndHTTPCodeToStatus(t *testing.T) {
+	tests := []struct {
+		name       string
+		errMsg     string
+		statusCode int
+		expected   *status.Status
+	}{
+		{
+			name:       "Bad Request",
+			errMsg:     "test",
+			statusCode: http.StatusBadRequest,
+			expected:   status.New(codes.InvalidArgument, "test"),
+		},
+		{
+			name:       "Unauthorized",
+			errMsg:     "test",
+			statusCode: http.StatusUnauthorized,
+			expected:   status.New(codes.Unauthenticated, "test"),
+		},
+		{
+			name:       "Forbidden",
+			errMsg:     "test",
+			statusCode: http.StatusForbidden,
+			expected:   status.New(codes.PermissionDenied, "test"),
+		},
+		{
+			name:       "Not Found",
+			errMsg:     "test",
+			statusCode: http.StatusNotFound,
+			expected:   status.New(codes.Unimplemented, "test"),
+		},
+		{
+			name:       "Too Many Requests",
+			errMsg:     "test",
+			statusCode: http.StatusTooManyRequests,
+			expected:   status.New(codes.ResourceExhausted, "test"),
+		},
+		{
+			name:       "Bad Gateway",
+			errMsg:     "test",
+			statusCode: http.StatusBadGateway,
+			expected:   status.New(codes.Unavailable, "test"),
+		},
+		{
+			name:       "Service Unavailable",
+			errMsg:     "test",
+			statusCode: http.StatusServiceUnavailable,
+			expected:   status.New(codes.Unavailable, "test"),
+		},
+		{
+			name:       "Gateway Timeout",
+			errMsg:     "test",
+			statusCode: http.StatusGatewayTimeout,
+			expected:   status.New(codes.Unavailable, "test"),
+		},
+		{
+			name:       "Unsupported Media Type",
+			errMsg:     "test",
+			statusCode: http.StatusUnsupportedMediaType,
+			expected:   status.New(codes.Unknown, "test"),
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := NewStatusFromMsgAndHTTPCode(tt.errMsg, tt.statusCode)
+			assert.Equal(t, tt.expected, result)
+		})
+	}
+}

receiver/otlpreceiver/internal/errors/errors.go

@@ -44,26 +44,3 @@ func GetHTTPStatusCodeFromStatus(s *status.Status) int {
 		return http.StatusInternalServerError
 	}
 }
-
-func NewStatusFromMsgAndHTTPCode(errMsg string, statusCode int) *status.Status {
-	var c codes.Code
-	// Mapping based on https://github.com/grpc/grpc/blob/master/doc/http-grpc-status-mapping.md
-	// 429 mapping to ResourceExhausted and 400 mapping to StatusBadRequest are exceptions.
-	switch statusCode {
-	case http.StatusBadRequest:
-		c = codes.InvalidArgument
-	case http.StatusUnauthorized:
-		c = codes.Unauthenticated
-	case http.StatusForbidden:
-		c = codes.PermissionDenied
-	case http.StatusNotFound:
-		c = codes.Unimplemented
-	case http.StatusTooManyRequests:
-		c = codes.ResourceExhausted
-	case http.StatusBadGateway, http.StatusServiceUnavailable, http.StatusGatewayTimeout:
-		c = codes.Unavailable
-	default:
-		c = codes.Unknown
-	}
-	return status.New(c, errMsg)
-}

receiver/otlpreceiver/internal/errors/errors_test.go

@@ -74,73 +74,3 @@ func Test_GetHTTPStatusCodeFromStatus(t *testing.T) {
 		})
 	}
 }
-
-func Test_ErrorMsgAndHTTPCodeToStatus(t *testing.T) {
-	tests := []struct {
-		name       string
-		errMsg     string
-		statusCode int
-		expected   *status.Status
-	}{
-		{
-			name:       "Bad Request",
-			errMsg:     "test",
-			statusCode: http.StatusBadRequest,
-			expected:   status.New(codes.InvalidArgument, "test"),
-		},
-		{
-			name:       "Unauthorized",
-			errMsg:     "test",
-			statusCode: http.StatusUnauthorized,
-			expected:   status.New(codes.Unauthenticated, "test"),
-		},
-		{
-			name:       "Forbidden",
-			errMsg:     "test",
-			statusCode: http.StatusForbidden,
-			expected:   status.New(codes.PermissionDenied, "test"),
-		},
-		{
-			name:       "Not Found",
-			errMsg:     "test",
-			statusCode: http.StatusNotFound,
-			expected:   status.New(codes.Unimplemented, "test"),
-		},
-		{
-			name:       "Too Many Requests",
-			errMsg:     "test",
-			statusCode: http.StatusTooManyRequests,
-			expected:   status.New(codes.ResourceExhausted, "test"),
-		},
-		{
-			name:       "Bad Gateway",
-			errMsg:     "test",
-			statusCode: http.StatusBadGateway,
-			expected:   status.New(codes.Unavailable, "test"),
-		},
-		{
-			name:       "Service Unavailable",
-			errMsg:     "test",
-			statusCode: http.StatusServiceUnavailable,
-			expected:   status.New(codes.Unavailable, "test"),
-		},
-		{
-			name:       "Gateway Timeout",
-			errMsg:     "test",
-			statusCode: http.StatusGatewayTimeout,
-			expected:   status.New(codes.Unavailable, "test"),
-		},
-		{
-			name:       "Unsupported Media Type",
-			errMsg:     "test",
-			statusCode: http.StatusUnsupportedMediaType,
-			expected:   status.New(codes.Unknown, "test"),
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			result := NewStatusFromMsgAndHTTPCode(tt.errMsg, tt.statusCode)
-			assert.Equal(t, tt.expected, result)
-		})
-	}
-}