fix(otelhttp): make Body handling in Transport consistent with stdlib

CHANGELOG.md

@@ -8,6 +8,10 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ## [Unreleased]
 
+### Fixed
+
+- Make `Body` handling in `Transport` consistent with stdlib in `go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp`. (#8618)
+
 ### Removed
 
 - Drop support for [Go 1.24]. (#8628)

instrumentation/net/http/otelhttp/transport.go

@@ -5,7 +5,6 @@ package otelhttp // import "go.opentelemetry.io/contrib/instrumentation/net/http
 
 import (
 	"context"
-	"fmt"
 	"io"
 	"net/http"
 	"net/http/httptrace"
@@ -85,8 +84,6 @@ func defaultTransportFormatter(_ string, r *http.Request) string {
 // RoundTrip creates a Span and propagates its context via the provided request's headers
 // before handing the request to the configured base RoundTripper. The created span will
 // end when the response body is closed or when a read from the body returns io.EOF.
-// If GetBody returns an error, the error is reported via otel.Handle and the request
-// continues with the original Body.
 func (t *Transport) RoundTrip(r *http.Request) (*http.Response, error) {
 	requestStartTime := time.Now()
 	for _, f := range t.filters {
@@ -119,23 +116,26 @@ func (t *Transport) RoundTrip(r *http.Request) (*http.Response, error) {
 
 	r = r.Clone(ctx) // According to RoundTripper spec, we shouldn't modify the origin request.
 
-	// GetBody is preferred over direct access to Body if the function is set.
-	// If the resulting body is nil or is NoBody, we don't want to mutate the body as it
-	// will affect the identity of it in an unforeseeable way because we assert
-	// ReadCloser fulfills a certain interface and it is indeed nil or NoBody.
-	body := r.Body
-	if r.GetBody != nil {
-		b, err := r.GetBody()
-		if err != nil {
-			otel.Handle(fmt.Errorf("http.Request GetBody returned an error: %w", err))
-		} else {
-			body = b
+	var lastBW *request.BodyWrapper // Records the last body wrapper. Can be nil.
+	maybeWrapBody := func(body io.ReadCloser) io.ReadCloser {
+		if body == nil || body == http.NoBody {
+			return body
 		}
+		bw := request.NewBodyWrapper(body, func(int64) {})
+		lastBW = bw
+		return bw
 	}
-
-	bw := request.NewBodyWrapper(body, func(int64) {})
-	if body != nil && body != http.NoBody {
-		r.Body = bw
+	r.Body = maybeWrapBody(r.Body)
+	if r.GetBody != nil {
+		originalGetBody := r.GetBody
+		r.GetBody = func() (io.ReadCloser, error) {
+			b, err := originalGetBody()
+			if err != nil {
+				lastBW = nil // The underlying transport will fail to make a retry request, hence, record no data.
+				return nil, err
+			}
+			return maybeWrapBody(b), nil
+		}
 	}
 
 	span.SetAttributes(t.semconv.RequestTraceAttrs(r)...)
@@ -148,10 +148,14 @@ func (t *Transport) RoundTrip(r *http.Request) (*http.Response, error) {
 	if err == nil {
 		statusCode = res.StatusCode
 	}
+	var requestSize int64
+	if lastBW != nil {
+		requestSize = lastBW.BytesRead()
+	}
 	t.semconv.RecordMetrics(
 		ctx,
 		semconv.MetricData{
-			RequestSize:     bw.BytesRead(),
+			RequestSize:     requestSize,
 			RequestDuration: time.Since(requestStartTime),
 		},
 		t.semconv.MetricOptions(semconv.MetricAttributes{

instrumentation/net/http/otelhttp/transport_test.go

@@ -20,7 +20,6 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/codes"
 	"go.opentelemetry.io/otel/propagation"
@@ -1118,88 +1117,80 @@ func BenchmarkTransportRoundTrip(b *testing.B) {
 }
 
 func TestRequestBodyReuse(t *testing.T) {
-	var bodies []string
+	var (
+		receivedBody1, receivedBody2 string
+		requests                     int
+	)
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		body, err := io.ReadAll(r.Body)
+		b, err := io.ReadAll(r.Body)
 		assert.NoError(t, err)
 
-		defer r.Body.Close()
-		bodies = append(bodies, string(body))
-
-		w.WriteHeader(http.StatusOK)
+		requests++
+		if requests == 1 {
+			receivedBody1 = string(b)
+			http.Redirect(w, r, "/", http.StatusPermanentRedirect) // body is not sent again on 302, but it is on 308.
+		} else {
+			receivedBody2 = string(b)
+		}
 	}))
 	defer server.Close()
 
-	transport := NewTransport(http.DefaultTransport)
-	client := http.Client{Transport: transport}
-
-	body := bytes.NewBuffer([]byte("hello"))
+	body1 := "plain text body1"
+	body2 := "plain text body2"
 
-	req, err := http.NewRequestWithContext(t.Context(), http.MethodPost, server.URL, body)
+	r, err := http.NewRequestWithContext(
+		t.Context(),
+		http.MethodPost,
+		server.URL,
+		io.NopCloser(strings.NewReader(body1)), // hide *strings.Reader from NewRequestWithContext()
+	)
 	require.NoError(t, err)
-
-	for range 2 {
-		resp, err := client.Do(req)
-		assert.NoError(t, err, "failed to perform http req")
-
-		_, err = io.Copy(io.Discard, resp.Body)
-		require.NoError(t, err)
-
-		err = resp.Body.Close()
-		require.NoError(t, err)
+	r.GetBody = func() (io.ReadCloser, error) {
+		return io.NopCloser(strings.NewReader(body2)), nil
 	}
 
-	require.Len(t, bodies, 2, "expected exactly 2 req bodies")
-	assert.Equal(t, bodies[0], bodies[1], "req bodies do not match")
-}
+	client := http.Client{Transport: NewTransport(http.DefaultTransport)}
 
-type testErrHandler struct {
-	err error
-}
+	resp, err := client.Do(r)
+	require.NoError(t, err)
+	require.NoError(t, resp.Body.Close())
 
-func (h *testErrHandler) Handle(err error) { h.err = err }
+	assert.Equal(t, http.StatusOK, resp.StatusCode)
+	assert.Equal(t, body1, receivedBody1)
+	assert.Equal(t, body2, receivedBody2)
+	assert.Equal(t, 2, requests)
+}
 
 func TestHandleErrorOnGetBodyError(t *testing.T) {
-	errHandler := otel.GetErrorHandler()
-
-	testErrHandler := &testErrHandler{}
-
-	otel.SetErrorHandler(testErrHandler)
-	defer func() {
-		otel.SetErrorHandler(errHandler)
-	}()
-
 	var receivedBody string
-
+	var requests int
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requests++
 		b, err := io.ReadAll(r.Body)
 		assert.NoError(t, err)
-
 		receivedBody = string(b)
-
-		w.WriteHeader(http.StatusOK)
+		http.Redirect(w, r, "/", http.StatusPermanentRedirect) // body is not sent again on 302, but it is on 308.
 	}))
 	defer server.Close()
 
 	body := "plain text body"
 
-	r, err := http.NewRequestWithContext(t.Context(), http.MethodPost, server.URL, strings.NewReader(body))
+	r, err := http.NewRequestWithContext(
+		t.Context(),
+		http.MethodPost,
+		server.URL,
+		io.NopCloser(strings.NewReader(body)), // hide *strings.Reader from NewRequestWithContext()
+	)
 	require.NoError(t, err)
 
 	getBodyErr := errors.New("GetBody error")
 	r.GetBody = func() (io.ReadCloser, error) {
 		return nil, getBodyErr
 	}
 
-	transport := NewTransport(http.DefaultTransport)
-	client := http.Client{Transport: transport}
-
-	resp, err := client.Do(r)
-	require.NotNil(t, resp)
-	require.NoError(t, err)
-	require.NoError(t, resp.Body.Close())
-
-	assert.Equal(t, http.StatusOK, resp.StatusCode)
-	assert.Equal(t, body, receivedBody, "should fallback to http.Request body on http.Request GetBody error")
-	assert.ErrorContains(t, testErrHandler.err, "http.Request GetBody returned an error: GetBody error")
+	client := http.Client{Transport: NewTransport(http.DefaultTransport)}
+	_, err = client.Do(r) //nolint:bodyclose // no response is returned here
+	assert.ErrorContains(t, err, getBodyErr.Error())
+	assert.Equal(t, body, receivedBody) // got it one time
+	assert.Equal(t, 1, requests)
 }