Propagate non-retryable error messages to client

PR open-telemetry#5541 (and issue open-telemetry#5536) enhance error handling, returning body text as part of the error. However, this is only done for retryable errors; if non-retryable, error text still does not propagate to clients. This PR adds handling of non-retryable errors, ensuring any body text is part of the message returned to the user's code. There is no change to the circumstances under which errors are reported, just an enhancement of the content of such an error.
mark-pictor-csec · Nov 12, 2024 · e3332de · e3332de
1 parent 0bf9572
commit e3332de
Show file tree

Hide file tree

Showing 6 changed files with 179 additions and 80 deletions.
diff --git a/exporters/otlp/otlplog/otlploghttp/client.go b/exporters/otlp/otlplog/otlploghttp/client.go
@@ -157,9 +157,7 @@ func (c *httpClient) uploadLogs(ctx context.Context, data []*logpb.ResourceLogs)
 			}()
 		}
 
-		var rErr error
-		switch sc := resp.StatusCode; {
-		case sc >= 200 && sc <= 299:
+		if sc := resp.StatusCode; sc >= 200 && sc <= 299 {
 			// Success, do not retry.
 
 			// Read the partial success message, if any.
@@ -187,34 +185,41 @@ func (c *httpClient) uploadLogs(ctx context.Context, data []*logpb.ResourceLogs)
 				}
 			}
 			return nil
-		case sc == http.StatusTooManyRequests,
-			sc == http.StatusBadGateway,
-			sc == http.StatusServiceUnavailable,
-			sc == http.StatusGatewayTimeout:
-			// Retry-able failure.
-			rErr = newResponseError(resp.Header, 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 {
-				return err
-			}
-
-			// overwrite the error message with the response body
-			// if it is not empty
-			if respStr := strings.TrimSpace(respData.String()); respStr != "" {
-				// Include response for context.
-				e := errors.New(respStr)
-				rErr = newResponseError(resp.Header, e)
+		}
+		// Error cases.
+
+		// 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 {
+			return err
+		}
+		respStr := strings.TrimSpace(respData.String())
+
+		switch resp.StatusCode {
+		case http.StatusTooManyRequests,
+			http.StatusBadGateway,
+			http.StatusServiceUnavailable,
+			http.StatusGatewayTimeout:
+			// Retryable failure.
+
+			var err error
+			if len(respStr) > 0 {
+				// include response body for context
+				err = errors.New(respStr)
 			}
+			return newResponseError(resp.Header, err)
 		default:
-			rErr = fmt.Errorf("failed to send logs to %s: %s", request.URL, resp.Status)
+			// Non-retryable failure.
+			if len(respStr) > 0 {
+				// include response body for context
+				err = errors.New(respStr)
+				return fmt.Errorf("failed to send logs to %s: %s (%w)", request.URL, resp.Status, err)
+			}
+			return fmt.Errorf("failed to send logs to %s: %s", request.URL, resp.Status)
 		}
-
-		return rErr
 	})
 }
 

diff --git a/exporters/otlp/otlplog/otlploghttp/client_test.go b/exporters/otlp/otlplog/otlploghttp/client_test.go
@@ -779,3 +779,39 @@ func TestConfig(t *testing.T) {
 		assert.Equal(t, []string{headerValueSetInProxy}, got[headerKeySetInProxy])
 	})
 }
+
+// borrows from TestConfig
+func TestNonRetryable(t *testing.T) {
+	factoryFunc := func(ePt string, rCh <-chan exportResult, o ...Option) (log.Exporter, *httpCollector) {
+		coll, err := newHTTPCollector(ePt, rCh)
+		require.NoError(t, err)
+
+		opts := []Option{WithEndpoint(coll.Addr().String())}
+		if !strings.HasPrefix(strings.ToLower(ePt), "https") {
+			opts = append(opts, WithInsecure())
+		}
+		opts = append(opts, o...)
+
+		ctx := context.Background()
+		exp, err := New(ctx, opts...)
+		require.NoError(t, err)
+		return exp, coll
+	}
+	exporterErr := errors.New("missing required attribute aaaa")
+	rCh := make(chan exportResult, 1)
+	rCh <- exportResult{Err: &httpResponseError{
+		Status: http.StatusBadRequest,
+		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, make([]log.Record, 1))
+	assert.ErrorContains(t, err, exporterErr.Error())
+}
diff --git a/exporters/otlp/otlpmetric/otlpmetrichttp/client.go b/exporters/otlp/otlpmetric/otlpmetrichttp/client.go
@@ -160,9 +160,7 @@ func (c *client) UploadMetrics(ctx context.Context, protoMetrics *metricpb.Resou
 			}()
 		}
 
-		var rErr error
-		switch sc := resp.StatusCode; {
-		case sc >= 200 && sc <= 299:
+		if sc := resp.StatusCode; sc >= 200 && sc <= 299 {
 			// Success, do not retry.
 
 			// Read the partial success message, if any.
@@ -190,34 +188,42 @@ func (c *client) UploadMetrics(ctx context.Context, protoMetrics *metricpb.Resou
 				}
 			}
 			return nil
-		case sc == http.StatusTooManyRequests,
-			sc == http.StatusBadGateway,
-			sc == http.StatusServiceUnavailable,
-			sc == http.StatusGatewayTimeout:
-			// Retry-able failure.
-			rErr = newResponseError(resp.Header, 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 {
-				return err
+		}
+		// Error cases.
+
+		// 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 {
+			return err
+		}
+		respStr := strings.TrimSpace(respData.String())
+
+		switch resp.StatusCode {
+		case http.StatusTooManyRequests,
+			http.StatusBadGateway,
+			http.StatusServiceUnavailable,
+			http.StatusGatewayTimeout:
+			// Retryable failure.
+
+			var err error
+			if len(respStr) > 0 {
+				// include response body for context
+				err = errors.New(respStr)
 			}
+			return newResponseError(resp.Header, err)
+		default:
+			// Non-retryable failure.
 
-			// overwrite the error message with the response body
-			// if it is not empty
-			if respStr := strings.TrimSpace(respData.String()); respStr != "" {
-				// Include response for context.
+			if len(respStr) > 0 {
+				// include response body for context
 				e := errors.New(respStr)
-				rErr = newResponseError(resp.Header, e)
+				return fmt.Errorf("failed to send metrics to %s: %s (%w)", request.URL, resp.Status, e)
 			}
-		default:
-			rErr = fmt.Errorf("failed to send metrics to %s: %s", request.URL, resp.Status)
+			return fmt.Errorf("failed to send metrics to %s: %s", request.URL, resp.Status)
 		}
-
-		return rErr
 	})
 }
 

diff --git a/exporters/otlp/otlpmetric/otlpmetrichttp/client_test.go b/exporters/otlp/otlpmetric/otlpmetrichttp/client_test.go
@@ -271,3 +271,40 @@ func TestConfig(t *testing.T) {
 		assert.Equal(t, []string{headerValueSetInProxy}, got[headerKeySetInProxy])
 	})
 }
+
+// borrows from TestConfig
+func TestNonRetryable(t *testing.T) {
+	factoryFunc := func(ePt string, rCh <-chan otest.ExportResult, o ...Option) (metric.Exporter, *otest.HTTPCollector) {
+		coll, err := otest.NewHTTPCollector(ePt, rCh)
+		require.NoError(t, err)
+
+		opts := []Option{WithEndpoint(coll.Addr().String())}
+		if !strings.HasPrefix(strings.ToLower(ePt), "https") {
+			opts = append(opts, WithInsecure())
+		}
+		opts = append(opts, o...)
+
+		ctx := context.Background()
+		exp, err := New(ctx, opts...)
+		require.NoError(t, err)
+		return exp, coll
+	}
+	exporterErr := errors.New("missing required attribute aaa")
+	rCh := make(chan otest.ExportResult, 1)
+	rCh <- otest.ExportResult{Err: &otest.HTTPResponseError{
+		Status: http.StatusBadRequest,
+		Err:    exporterErr,
+	}}
+	exp, coll := factoryFunc("", rCh)
+	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)) })
+	exCtx, cancel := context.WithTimeout(ctx, time.Second)
+	defer cancel()
+	err := exp.Export(exCtx, &metricdata.ResourceMetrics{})
+	assert.ErrorContains(t, err, exporterErr.Error())
+
+	assert.NoError(t, exCtx.Err())
+}
diff --git a/exporters/otlp/otlptrace/otlptracehttp/client.go b/exporters/otlp/otlptrace/otlptracehttp/client.go
@@ -166,8 +166,7 @@ func (d *client) UploadTraces(ctx context.Context, protoSpans []*tracepb.Resourc
 			}()
 		}
 
-		switch sc := resp.StatusCode; {
-		case sc >= 200 && sc <= 299:
+		if sc := resp.StatusCode; sc >= 200 && sc <= 299 {
 			// Success, do not retry.
 			// Read the partial success message, if any.
 			var respData bytes.Buffer
@@ -194,32 +193,40 @@ func (d *client) UploadTraces(ctx context.Context, protoSpans []*tracepb.Resourc
 				}
 			}
 			return nil
-
-		case sc == http.StatusTooManyRequests,
-			sc == http.StatusBadGateway,
-			sc == http.StatusServiceUnavailable,
-			sc == http.StatusGatewayTimeout:
-			// Retry-able failures.
-			rErr := newResponseError(resp.Header, 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 {
-				return err
+		}
+		// Error cases.
+
+		// 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 {
+			return err
+		}
+		respStr := strings.TrimSpace(respData.String())
+
+		switch resp.StatusCode {
+		case http.StatusTooManyRequests,
+			http.StatusBadGateway,
+			http.StatusServiceUnavailable,
+			http.StatusGatewayTimeout:
+			// Retryable failure.
+
+			var err error
+			if len(respStr) > 0 {
+				// include response body for context
+				err = errors.New(respStr)
 			}
+			return newResponseError(resp.Header, err)
+		default:
+			// Non-retryable failure.
 
-			// overwrite the error message with the response body
-			// if it is not empty
-			if respStr := strings.TrimSpace(respData.String()); respStr != "" {
-				// Include response for context.
+			if len(respStr) > 0 {
+				// include response body for context
 				e := errors.New(respStr)
-				rErr = newResponseError(resp.Header, e)
+				return fmt.Errorf("failed to send to %s: %s (%w)", request.URL, resp.Status, e)
 			}
-			return rErr
-		default:
 			return fmt.Errorf("failed to send to %s: %s", request.URL, resp.Status)
 		}
 	})

diff --git a/exporters/otlp/otlptrace/otlptracehttp/client_test.go b/exporters/otlp/otlptrace/otlptracehttp/client_test.go
@@ -244,6 +244,9 @@ func TestTimeout(t *testing.T) {
 func TestNoRetry(t *testing.T) {
 	mc := runMockCollector(t, mockCollectorConfig{
 		InjectHTTPStatus: []int{http.StatusBadRequest},
+		Partial: &coltracepb.ExportTracePartialSuccess{
+			ErrorMessage: "missing required attribute aaa",
+		},
 	})
 	defer mc.MustStop(t)
 	driver := otlptracehttp.NewClient(
@@ -265,9 +268,14 @@ func TestNoRetry(t *testing.T) {
 	}()
 	err = exporter.ExportSpans(ctx, otlptracetest.SingleReadOnlySpan())
 	assert.Error(t, err)
-	unwrapped := errors.Unwrap(err)
-	assert.Equal(t, fmt.Sprintf("failed to send to http://%s/v1/traces: 400 Bad Request", mc.endpoint), unwrapped.Error())
 	assert.True(t, strings.HasPrefix(err.Error(), "traces export: "))
+
+	unwrapped := errors.Unwrap(err)
+	assert.Contains(t, unwrapped.Error(), fmt.Sprintf("failed to send to http://%s/v1/traces: 400 Bad Request", mc.endpoint))
+
+	unwrapped2 := errors.Unwrap(unwrapped)
+	assert.Contains(t, unwrapped2.Error(), "missing required attribute aaa")
+
 	assert.Empty(t, mc.GetSpans())
 }