Parse errormsgs in retryable status codes

CHANGELOG.md

@@ -42,6 +42,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - Fix panic in baggage creation when a member contains 0x80 char in key or value. (#5494)
 - Correct comments for the priority of the `WithEndpoint` and `WithEndpointURL` options and their coresponding environment variables in  in `go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc`. (#5508)
 - Fix stale timestamps reported by the lastvalue aggregation. (#5517)
+- Pass the underlying error rather than a generic retry-able failure in `go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp`. (#5541)
 
 ## [1.27.0/0.49.0/0.3.0] 2024-05-21
 

exporters/otlp/otlplog/otlploghttp/client.go

@@ -14,9 +14,11 @@ import (
 	"net/http"
 	"net/url"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 
+	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/proto"
 
 	"go.opentelemetry.io/otel"
@@ -143,7 +145,7 @@ func (c *httpClient) uploadLogs(ctx context.Context, data []*logpb.ResourceLogs)
 		resp, err := c.client.Do(request.Request)
 		var urlErr *url.Error
 		if errors.As(err, &urlErr) && urlErr.Temporary() {
-			return newResponseError(http.Header{})
+			return newResponseError(http.Header{}, err.Error())
 		}
 		if err != nil {
 			return err
@@ -184,13 +186,26 @@ func (c *httpClient) uploadLogs(ctx context.Context, data []*logpb.ResourceLogs)
 			sc == http.StatusServiceUnavailable,
 			sc == http.StatusGatewayTimeout:
 			// Retry-able failure.
-			rErr = newResponseError(resp.Header)
+			rErr = newResponseError(resp.Header, "")
 
-			// Going to retry, drain the body to reuse the connection.
-			if _, err := io.Copy(io.Discard, resp.Body); err != nil {
+			// server may return a message with the response
+			// body, so we read it to include in the error
+			// message to be returned. It will help in
+			// debugging the actual issue.
+			var respData bytes.Buffer
+			if _, err := io.Copy(&respData, resp.Body); err != nil {
 				_ = resp.Body.Close()
 				return err
 			}
+
+			// overwrite the error message with the response body
+			// if it is not empty
+			respStr := strings.TrimSpace(respData.String())
+			if respStr != "" {
+				// pass the error message along with retry-able error,
+				// so that it can be retried and also passes the message
+				rErr = newResponseError(resp.Header, respStr)
+			}
 		default:
 			rErr = fmt.Errorf("failed to send logs to %s: %s", request.URL, resp.Status)
 		}
@@ -266,21 +281,38 @@ func (r *request) reset(ctx context.Context) {
 // retryableError represents a request failure that can be retried.
 type retryableError struct {
 	throttle int64
+	errMsg   string
 }
 
 // newResponseError returns a retryableError and will extract any explicit
-// throttle delay contained in headers.
-func newResponseError(header http.Header) error {
+// throttle delay contained in headers and if there is message in the response
+// body, it will be used as the error message. If errMsg is not empty, it will
+// be used as the error message instead of the standard "retry-able" failure.
+func newResponseError(header http.Header, body string) error {
 	var rErr retryableError
 	if v := header.Get("Retry-After"); v != "" {
 		if t, err := strconv.ParseInt(v, 10, 64); err == nil {
 			rErr.throttle = t
 		}
 	}
+
+	// Extract the error message from the response body.
+	st, ok := status.FromError(errors.New(body))
+	rErr.errMsg = fmt.Sprintf("rpc error: code = %s desc = %s", st.Code(), st.Message())
+
+	// if response body is not in expected format
+	if !ok {
+		rErr.errMsg = body
+	}
+
 	return rErr
 }
 
 func (e retryableError) Error() string {
+	if e.errMsg != "" {
+		return e.errMsg
+	}
+
 	return "retry-able request failure"
 }
 

exporters/otlp/otlplog/otlploghttp/go.mod

@@ -12,6 +12,7 @@ require (
 	go.opentelemetry.io/otel/sdk/log v0.3.0
 	go.opentelemetry.io/otel/trace v1.27.0
 	go.opentelemetry.io/proto/otlp v1.3.1
+	google.golang.org/grpc v1.64.0
 	google.golang.org/protobuf v1.34.2
 )
 
@@ -29,7 +30,6 @@ require (
 	golang.org/x/text v0.16.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240617180043-68d350f18fd4 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240617180043-68d350f18fd4 // indirect
-	google.golang.org/grpc v1.64.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
 

exporters/otlp/otlpmetric/otlpmetrichttp/client.go

@@ -14,9 +14,11 @@ import (
 	"net/http"
 	"net/url"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 
+	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/proto"
 
 	"go.opentelemetry.io/otel"
@@ -146,7 +148,7 @@ func (c *client) UploadMetrics(ctx context.Context, protoMetrics *metricpb.Resou
 		resp, err := c.httpClient.Do(request.Request)
 		var urlErr *url.Error
 		if errors.As(err, &urlErr) && urlErr.Temporary() {
-			return newResponseError(http.Header{})
+			return newResponseError(http.Header{}, err.Error())
 		}
 		if err != nil {
 			return err
@@ -187,13 +189,27 @@ func (c *client) UploadMetrics(ctx context.Context, protoMetrics *metricpb.Resou
 			sc == http.StatusServiceUnavailable,
 			sc == http.StatusGatewayTimeout:
 			// Retry-able failure.
-			rErr = newResponseError(resp.Header)
+			rErr = newResponseError(resp.Header, "")
 
-			// Going to retry, drain the body to reuse the connection.
-			if _, err := io.Copy(io.Discard, resp.Body); err != nil {
+			// server may return a message with the response
+			// body, so we read it to include in the error
+			// message to be returned. It will help in
+			// debugging the actual issue.
+			var respData bytes.Buffer
+			if _, err := io.Copy(&respData, resp.Body); err != nil {
 				_ = resp.Body.Close()
 				return err
 			}
+
+			// overwrite the error message with the response body
+			// if it is not empty
+			respStr := strings.TrimSpace(respData.String())
+			if respStr != "" {
+				// pass the error message along with retry-able error,
+				// so that it can be retried and also passes the message
+				rErr = newResponseError(resp.Header, respStr)
+			}
+
 		default:
 			rErr = fmt.Errorf("failed to send metrics to %s: %s", request.URL, resp.Status)
 		}
@@ -269,21 +285,38 @@ func (r *request) reset(ctx context.Context) {
 // retryableError represents a request failure that can be retried.
 type retryableError struct {
 	throttle int64
+	errMsg   string
 }
 
 // newResponseError returns a retryableError and will extract any explicit
-// throttle delay contained in headers.
-func newResponseError(header http.Header) error {
+// throttle delay contained in headers and if there is message in the response
+// body, it will be used as the error message. If errMsg is not empty, it will
+// be used as the error message instead of the standard "retry-able" failure.
+func newResponseError(header http.Header, body string) error {
 	var rErr retryableError
 	if v := header.Get("Retry-After"); v != "" {
 		if t, err := strconv.ParseInt(v, 10, 64); err == nil {
 			rErr.throttle = t
 		}
 	}
+
+	// Extract the error message from the response body.
+	st, ok := status.FromError(errors.New(body))
+	rErr.errMsg = fmt.Sprintf("rpc error: code = %s desc = %s", st.Code(), st.Message())
+
+	// if response body is not in expected format
+	if !ok {
+		rErr.errMsg = body
+	}
+
 	return rErr
 }
 
 func (e retryableError) Error() string {
+	if e.errMsg != "" {
+		return e.errMsg
+	}
+
 	return "retry-able request failure"
 }
 

exporters/otlp/otlpmetric/otlpmetrichttp/client_test.go

@@ -192,6 +192,25 @@ func TestConfig(t *testing.T) {
 		assert.Len(t, rCh, 0, "failed HTTP responses did not occur")
 	})
 
+	t.Run("WithRetryAndExporterErr", func(t *testing.T) {
+		exporterErr := errors.New("rpc error: code = Unavailable desc = service.name not found in resource attributes")
+		rCh := make(chan otest.ExportResult, 1)
+		rCh <- otest.ExportResult{Err: &otest.HTTPResponseError{
+			Status: http.StatusTooManyRequests,
+			Err:    exporterErr,
+		}}
+		exp, coll := factoryFunc("", rCh, WithRetry(RetryConfig{
+			Enabled: false,
+		}))
+		ctx := context.Background()
+		t.Cleanup(func() { require.NoError(t, coll.Shutdown(ctx)) })
+		// Push this after Shutdown so the HTTP server doesn't hang.
+		t.Cleanup(func() { close(rCh) })
+		t.Cleanup(func() { require.NoError(t, exp.Shutdown(ctx)) })
+		err := exp.Export(ctx, &metricdata.ResourceMetrics{})
+		assert.EqualError(t, err, fmt.Sprintf("failed to upload metrics: %d: %v", http.StatusTooManyRequests, exporterErr))
+	})
+
 	t.Run("WithURLPath", func(t *testing.T) {
 		path := "/prefix/v2/metrics"
 		ePt := fmt.Sprintf("http://localhost:0%s", path)

exporters/otlp/otlptrace/otlptracehttp/client.go

@@ -14,9 +14,11 @@ import (
 	"net/http"
 	"net/url"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 
+	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/proto"
 
 	"go.opentelemetry.io/otel"
@@ -151,7 +153,7 @@ func (d *client) UploadTraces(ctx context.Context, protoSpans []*tracepb.Resourc
 		resp, err := d.client.Do(request.Request)
 		var urlErr *url.Error
 		if errors.As(err, &urlErr) && urlErr.Temporary() {
-			return newResponseError(http.Header{})
+			return newResponseError(http.Header{}, err.Error())
 		}
 		if err != nil {
 			return err
@@ -198,11 +200,28 @@ func (d *client) UploadTraces(ctx context.Context, protoSpans []*tracepb.Resourc
 			sc == http.StatusBadGateway,
 			sc == http.StatusServiceUnavailable,
 			sc == http.StatusGatewayTimeout:
-			// Retry-able failures.  Drain the body to reuse the connection.
-			if _, err := io.Copy(io.Discard, resp.Body); err != nil {
-				otel.Handle(err)
+			// Retry-able failures.
+			rErr := newResponseError(resp.Header, "")
+
+			// server may return a message with the response
+			// body, so we read it to include in the error
+			// message to be returned. It will help in
+			// debugging the actual issue.
+			var respData bytes.Buffer
+			if _, err := io.Copy(&respData, resp.Body); err != nil {
+				_ = resp.Body.Close()
+				return err
 			}
-			return newResponseError(resp.Header)
+
+			// overwrite the error message with the response body
+			// if it is not empty
+			respStr := strings.TrimSpace(respData.String())
+			if respStr != "" {
+				// pass the error message along with retry-able error,
+				// so that it can be retried and also passes the message
+				rErr = newResponseError(resp.Header, respStr)
+			}
+			return rErr
 		default:
 			return fmt.Errorf("failed to send to %s: %s", request.URL, resp.Status)
 		}
@@ -291,21 +310,38 @@ func (r *request) reset(ctx context.Context) {
 // retryableError represents a request failure that can be retried.
 type retryableError struct {
 	throttle int64
+	errMsg   string
 }
 
 // newResponseError returns a retryableError and will extract any explicit
-// throttle delay contained in headers.
-func newResponseError(header http.Header) error {
+// throttle delay contained in headers and if there is message in the response
+// body, it will be used as the error message. If errMsg is not empty, it will
+// be used as the error message instead of the standard "retry-able" failure.
+func newResponseError(header http.Header, body string) error {
 	var rErr retryableError
 	if s, ok := header["Retry-After"]; ok {
 		if t, err := strconv.ParseInt(s[0], 10, 64); err == nil {
 			rErr.throttle = t
 		}
 	}
+
+	// Extract the error message from the response body.
+	st, ok := status.FromError(errors.New(body))
+	rErr.errMsg = fmt.Sprintf("rpc error: code = %s desc = %s", st.Code(), st.Message())
+
+	// if response body is not in expected format
+	if !ok {
+		rErr.errMsg = body
+	}
+
 	return rErr
 }
 
 func (e retryableError) Error() string {
+	if e.errMsg != "" {
+		return e.errMsg
+	}
+
 	return "retry-able request failure"
 }
 

exporters/otlp/otlptrace/otlptracehttp/client_test.go

@@ -238,7 +238,7 @@ func TestTimeout(t *testing.T) {
 		assert.NoError(t, exporter.Shutdown(ctx))
 	}()
 	err = exporter.ExportSpans(ctx, otlptracetest.SingleReadOnlySpan())
-	assert.ErrorContains(t, err, "retry-able request failure")
+	assert.ErrorContains(t, err, "context deadline exceeded")
 }
 
 func TestNoRetry(t *testing.T) {