fix(otlploghttp): replay gzipped bodies on redirect (#8152)

When gzip compression is enabled, the OTLP log HTTP client compressed
the request body for the initial send but left GetBody wired to the
original uncompressed protobuf payload.

Any redirect or retry path that rebuilt the request from GetBody could
then resend an uncompressed body while still advertising
Content-Encoding: gzip.

Update GetBody to return the gzipped buffer and add a redirect
regression test that verifies the replayed body matches the original
compressed request and can be decompressed successfully.

CHANGELOG.md

@@ -19,6 +19,10 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - `ErrorType` in `go.opentelemetry.io/otel/semconv` now unwraps errors created with `fmt.Errorf` when deriving the `error.type` attribute. (#8133)
 - `go.opentelemetry.io/otel/sdk/log` now unwraps error chains created with `fmt.Errorf` when deriving the `error.type` attribute from errors on log records. (#8133)
 
+### Fixed
+
+- Fix gzipped request body replay on redirect in `go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp`. (#8152)
+
 <!-- Released section -->
 <!-- Don't change this section unless doing release -->
 

exporters/otlp/otlplog/otlploghttp/client.go

@@ -285,7 +285,10 @@ func (c *httpClient) newRequest(ctx context.Context, body []byte) (request, erro
 		r.Header.Set("Content-Encoding", "gzip")
 
 		gz := gzPool.Get().(*gzip.Writer)
-		defer gzPool.Put(gz)
+		defer func() {
+			gz.Reset(io.Discard)
+			gzPool.Put(gz)
+		}()
 
 		var b bytes.Buffer
 		gz.Reset(&b)
@@ -299,7 +302,7 @@ func (c *httpClient) newRequest(ctx context.Context, body []byte) (request, erro
 		}
 
 		req.bodyReader = bodyReader(b.Bytes())
-		req.GetBody = bodyReaderErr(body)
+		req.GetBody = bodyReaderErr(b.Bytes())
 	}
 
 	return req, nil

exporters/otlp/otlplog/otlploghttp/client_test.go

@@ -886,6 +886,75 @@ func TestGetBodyCalledOnRedirect(t *testing.T) {
 	assert.Equal(t, requestBodies[0], requestBodies[1], "redirect body should match original")
 }
 
+func TestGetBodyCalledOnRedirectWithGzip(t *testing.T) {
+	// Test that req.GetBody replays the gzipped request body on redirects.
+	var mu sync.Mutex
+	var requestBodies [][]byte
+
+	handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, err := io.ReadAll(r.Body)
+		if err != nil {
+			w.WriteHeader(http.StatusInternalServerError)
+			return
+		}
+
+		assert.Equal(t, "gzip", r.Header.Get("Content-Encoding"))
+
+		mu.Lock()
+		requestBodies = append(requestBodies, body)
+		isFirstRequest := len(requestBodies) == 1
+		mu.Unlock()
+
+		if isFirstRequest {
+			w.Header().Set("Location", "/v1/logs/final")
+			w.WriteHeader(http.StatusTemporaryRedirect)
+			return
+		}
+
+		w.Header().Set("Content-Type", "application/x-protobuf")
+		w.WriteHeader(http.StatusOK)
+	})
+
+	server := httptest.NewServer(handler)
+	defer server.Close()
+
+	opts := []Option{
+		WithEndpoint(server.Listener.Addr().String()),
+		WithInsecure(),
+		WithCompression(GzipCompression),
+	}
+	cfg := newConfig(opts)
+	client, err := newHTTPClient(t.Context(), cfg)
+	require.NoError(t, err)
+
+	exporter, err := newExporter(client, cfg)
+	require.NoError(t, err)
+	ctx := t.Context()
+	defer func() { _ = exporter.Shutdown(ctx) }()
+
+	err = exporter.Export(ctx, make([]log.Record, 1))
+	require.NoError(t, err)
+
+	mu.Lock()
+	defer mu.Unlock()
+
+	require.Len(t, requestBodies, 2, "expected 2 requests (original + redirect)")
+	assert.NotEmpty(t, requestBodies[0], "original request body should not be empty")
+	assert.Equal(t, requestBodies[0], requestBodies[1], "redirect body should match original")
+
+	for _, body := range requestBodies {
+		reader, err := gzip.NewReader(bytes.NewReader(body))
+		require.NoError(t, err)
+		func() {
+			defer func() { assert.NoError(t, reader.Close()) }()
+
+			decoded, err := io.ReadAll(reader)
+			require.NoError(t, err)
+			assert.NotEmpty(t, decoded)
+		}()
+	}
+}
+
 // SetExporterID sets the exporter ID counter to v and returns the previous
 // value.
 //