From 28fc5d90407e820b05755a921737fcc20088d22e Mon Sep 17 00:00:00 2001 From: Stephan Behnke Date: Tue, 18 Aug 2026 09:28:16 -0700 Subject: [PATCH] Add OpenTelemetry HTTP instrumentation (#11558) ## What changed? Added reusable OpenTelemetry wrappers for HTTP clients and servers. It is not wired up in this PR yet. ## Why? Nexus uses HTTP in several internal and external paths, but Temporal only had shared tracing support for gRPC. This provides the common HTTP instrumentation used by the rest of the stack. ## How did you test it? - [x] built - [ ] run locally and tested manually - [ ] covered by existing tests - [x] added new unit test(s) - [ ] added new functional test(s) --------- Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> --- common/telemetry/config.go | 8 + common/telemetry/grpc.go | 6 - common/telemetry/http.go | 291 ++++++++++++++++ common/telemetry/http_test.go | 633 ++++++++++++++++++++++++++++++++++ go.mod | 4 +- 5 files changed, 934 insertions(+), 8 deletions(-) create mode 100644 common/telemetry/http.go create mode 100644 common/telemetry/http_test.go diff --git a/common/telemetry/config.go b/common/telemetry/config.go index c4af9bef49..adb0bb3339 100644 --- a/common/telemetry/config.go +++ b/common/telemetry/config.go @@ -46,6 +46,14 @@ var ( NoopTracer = NoopTracerProvider.Tracer("") ) +func isEnabled(tp trace.TracerProvider) bool { + if tp == nil { + return false + } + _, isNoop := tp.(otelnoop.TracerProvider) + return !isNoop +} + type ( metadata struct { Name string diff --git a/common/telemetry/grpc.go b/common/telemetry/grpc.go index 748ffe4f67..d31d9a4612 100644 --- a/common/telemetry/grpc.go +++ b/common/telemetry/grpc.go @@ -8,7 +8,6 @@ import ( "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/propagation" "go.opentelemetry.io/otel/trace" - otelnoop "go.opentelemetry.io/otel/trace/noop" "go.temporal.io/server/common/log" "go.temporal.io/server/common/log/tag" "go.temporal.io/server/common/rpc/interceptor/logtags" @@ -190,8 +189,3 @@ func (c *customServerStatsHandler) TagConn(ctx context.Context, info *stats.Conn func (c *customServerStatsHandler) HandleConn(ctx context.Context, stat stats.ConnStats) { c.wrapped.HandleConn(ctx, stat) } - -func isEnabled(tp trace.TracerProvider) bool { - _, isNoop := tp.(otelnoop.TracerProvider) - return !isNoop -} diff --git a/common/telemetry/http.go b/common/telemetry/http.go new file mode 100644 index 0000000000..1f7b4aab44 --- /dev/null +++ b/common/telemetry/http.go @@ -0,0 +1,291 @@ +package telemetry + +import ( + "bytes" + "context" + "io" + "net/http" + "strings" + + "github.com/felixge/httpsnoop" + "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/propagation" + "go.opentelemetry.io/otel/trace" +) + +// NewHTTPClientTransport instruments outbound HTTP requests with OpenTelemetry client spans +// and injects trace context using propagator. If propagator is nil, it defaults to W3C Trace Context. +func NewHTTPClientTransport( + rt http.RoundTripper, + tracerProvider trace.TracerProvider, + propagator propagation.TextMapPropagator, +) http.RoundTripper { + if !isEnabled(tracerProvider) { + return rt + } + if propagator == nil { + propagator = propagation.TraceContext{} + } + isDebug := DebugMode() + if isDebug { + if rt == nil { + rt = http.DefaultTransport + } + rt = &debugHTTPClientSpanTransport{rt: rt} + } + rt = otelhttp.NewTransport( + rt, + otelhttp.WithTracerProvider(tracerProvider), + otelhttp.WithPropagators(propagator), + ) + if isDebug { + rt = &debugHTTPClientTransport{rt: rt} + } + return rt +} + +// NewHTTPHandler instruments inbound HTTP requests with OpenTelemetry server spans +// and extracts trace context using propagator. If propagator is nil, it defaults to W3C Trace Context. +func NewHTTPHandler( + handler http.Handler, + operation string, + tracerProvider trace.TracerProvider, + propagator propagation.TextMapPropagator, +) http.Handler { + if !isEnabled(tracerProvider) { + return handler + } + if propagator == nil { + propagator = propagation.TraceContext{} + } + if DebugMode() { + handler = &debugHTTPHandler{handler: handler} + } + return otelhttp.NewHandler( + handler, + operation, + otelhttp.WithTracerProvider(tracerProvider), + otelhttp.WithPropagators(propagator), + ) +} + +type debugHTTPClientRequestStateKey struct{} + +// debugHTTPClientSpanTransport stores response finalization here through the request context. +// debugHTTPClientTransport runs it before otelhttp ends the span when the body closes before EOF. +type debugHTTPClientRequestState struct { + finalizeResponse func() +} + +type debugHTTPClientTransport struct { + rt http.RoundTripper +} + +var _ http.RoundTripper = (*debugHTTPClientTransport)(nil) + +func (t *debugHTTPClientTransport) RoundTrip(req *http.Request) (*http.Response, error) { + state := &debugHTTPClientRequestState{} + req = req.WithContext(context.WithValue(req.Context(), debugHTTPClientRequestStateKey{}, state)) + resp, err := t.rt.RoundTrip(req) + if resp == nil || state.finalizeResponse == nil { + return resp, err + } + resp.Body = newCloseFinalizingReadCloser(resp.Body, state.finalizeResponse) + return resp, err +} + +type debugHTTPSpan struct { + trace.Span +} + +func (s debugHTTPSpan) annotateHeaders(prefix string, headers http.Header) { + // Debug mode is explicit opt-in for diagnostics, so all header values, including sensitive ones, + // are recorded verbatim. + for key, values := range headers { + s.SetAttributes(attribute.StringSlice(prefix+strings.ToLower(key), values)) + } +} + +func (s debugHTTPSpan) payloadAnnotator(key string) func(string) { + return func(payload string) { + s.SetAttributes(attribute.String(key, payload)) + } +} + +type debugHTTPClientSpanTransport struct { + rt http.RoundTripper +} + +var _ http.RoundTripper = (*debugHTTPClientSpanTransport)(nil) + +func (t *debugHTTPClientSpanTransport) RoundTrip(req *http.Request) (*http.Response, error) { + span := debugHTTPSpan{Span: trace.SpanFromContext(req.Context())} + // Non-recording spans discard attributes, so skip the debug-capture work. + if !span.IsRecording() { + return t.rt.RoundTrip(req) + } + span.annotateHeaders("http.request.headers.", req.Header) + req.Body, _ = newPayloadCapturingReadCloser( + req.Body, + req.ContentLength, + span.payloadAnnotator("http.request.payload"), + ) + + resp, err := t.rt.RoundTrip(req) + if resp == nil { + return resp, err + } + + span.annotateHeaders("http.response.headers.", resp.Header) + var responseCapture *payloadCapture + resp.Body, responseCapture = newPayloadCapturingReadCloser( + resp.Body, + resp.ContentLength, + span.payloadAnnotator("http.response.payload"), + ) + if state, ok := req.Context().Value(debugHTTPClientRequestStateKey{}).(*debugHTTPClientRequestState); ok && responseCapture != nil { + state.finalizeResponse = responseCapture.finalize + } + return resp, err +} + +type debugHTTPHandler struct { + handler http.Handler +} + +func (h *debugHTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + span := debugHTTPSpan{Span: trace.SpanFromContext(r.Context())} + // Non-recording spans discard attributes, so skip the debug-capture work. + if !span.IsRecording() { + h.handler.ServeHTTP(w, r) + return + } + span.annotateHeaders("http.request.headers.", r.Header) + var requestCapture *payloadCapture + r.Body, requestCapture = newPayloadCapturingReadCloser( + r.Body, + r.ContentLength, + span.payloadAnnotator("http.request.payload"), + ) + + responseBody := payloadCapture{onFinalize: span.payloadAnnotator("http.response.payload")} + w = newPayloadCapturingResponseWriter(w, &responseBody) + + h.handler.ServeHTTP(w, r) + // Finalize after the handler returns because an unknown-length request may be fully consumed + // without a read that returns EOF. + if requestCapture != nil { + requestCapture.finalize() + } + + span.annotateHeaders("http.response.headers.", w.Header()) + responseBody.finalize() +} + +// Use httpsnoop to intercept writes without dropping optional ResponseWriter interfaces. +func newPayloadCapturingResponseWriter(w http.ResponseWriter, capture *payloadCapture) http.ResponseWriter { + return httpsnoop.Wrap(w, httpsnoop.Hooks{ + Write: func(next httpsnoop.WriteFunc) httpsnoop.WriteFunc { + return func(p []byte) (int, error) { + n, err := next(p) + if n > 0 { + _, _ = capture.Write(p[:n]) + } + return n, err + } + }, + ReadFrom: func(httpsnoop.ReadFromFunc) httpsnoop.ReadFromFunc { + return func(src io.Reader) (int64, error) { + // Prevent io.Copy from using ReaderFrom so bytes pass through Write, where the debug + // response wrapper captures payloads and otelhttp counts response size. + return io.Copy(struct{ io.Writer }{w}, src) + } + }, + }) +} + +// Debug mode favors complete diagnostics, so it buffers all observed payload bytes in memory +// without a size limit. +type payloadCapture struct { + bytes.Buffer + finalized bool + onFinalize func(string) +} + +func (c *payloadCapture) finalize() { + if c.finalized { + return + } + c.finalized = true + if c.Len() > 0 { + c.onFinalize(c.String()) + } +} + +type payloadCapturingReadCloser struct { + io.ReadCloser + capture payloadCapture + contentLength int64 +} + +func newPayloadCapturingReadCloser( + body io.ReadCloser, + contentLength int64, + onFinalize func(string), +) (io.ReadCloser, *payloadCapture) { + if body == nil || body == http.NoBody { + return body, nil + } + capturingBody := &payloadCapturingReadCloser{ + ReadCloser: body, + capture: payloadCapture{onFinalize: onFinalize}, + contentLength: contentLength, + } + return preserveBodyWriter(body, capturingBody), &capturingBody.capture +} + +func (r *payloadCapturingReadCloser) Read(p []byte) (int, error) { + n, err := r.ReadCloser.Read(p) + if n > 0 { + _, _ = r.capture.Write(p[:n]) + } + // Finalize after ContentLength bytes because callers may stop without another read that returns EOF. + if err == io.EOF || r.contentLength > 0 && int64(r.capture.Len()) == r.contentLength { + r.capture.finalize() + } + return n, err +} + +func (r *payloadCapturingReadCloser) Close() error { + r.capture.finalize() + return r.ReadCloser.Close() +} + +type closeFinalizingReadCloser struct { + io.ReadCloser + finalize func() +} + +func newCloseFinalizingReadCloser(body io.ReadCloser, finalize func()) io.ReadCloser { + finalizingBody := &closeFinalizingReadCloser{ + ReadCloser: body, + finalize: finalize, + } + return preserveBodyWriter(body, finalizingBody) +} + +func (r *closeFinalizingReadCloser) Close() error { + r.finalize() + return r.ReadCloser.Close() +} + +func preserveBodyWriter(body io.ReadCloser, wrapped io.ReadCloser) io.ReadCloser { + if writer, ok := body.(io.Writer); ok { + return struct { + io.ReadCloser + io.Writer + }{wrapped, writer} + } + return wrapped +} diff --git a/common/telemetry/http_test.go b/common/telemetry/http_test.go new file mode 100644 index 0000000000..853334ad87 --- /dev/null +++ b/common/telemetry/http_test.go @@ -0,0 +1,633 @@ +package telemetry + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + oteltrace "go.opentelemetry.io/otel/trace" + "go.opentelemetry.io/otel/trace/noop" +) + +type roundTripperFunc func(*http.Request) (*http.Response, error) + +func (f roundTripperFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } + +type readerFromResponseRecorder struct { + *httptest.ResponseRecorder +} + +func (r *readerFromResponseRecorder) ReadFrom(src io.Reader) (int64, error) { + return io.Copy(r.Body, src) +} + +type fixedIDGenerator struct{} + +func (fixedIDGenerator) NewIDs(context.Context) (oteltrace.TraceID, oteltrace.SpanID) { + return oteltrace.TraceID{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16}, + oteltrace.SpanID{1, 2, 3, 4, 5, 6, 7, 8} +} + +func (fixedIDGenerator) NewSpanID(context.Context, oteltrace.TraceID) oteltrace.SpanID { + return oteltrace.SpanID{1, 2, 3, 4, 5, 6, 7, 8} +} + +type readTrackingCloser struct { + io.Reader + read bool +} + +func (r *readTrackingCloser) Read(p []byte) (int, error) { + r.read = true + return r.Reader.Read(p) +} + +func (r *readTrackingCloser) Close() error { return nil } + +type readWriteCloser struct { + io.Reader + written bytes.Buffer +} + +func (r *readWriteCloser) Write(p []byte) (int, error) { return r.written.Write(p) } + +func (r *readWriteCloser) Close() error { return nil } + +type failingReadCloser struct { + payload []byte + read bool +} + +func (r *failingReadCloser) Read(p []byte) (int, error) { + if r.read { + return 0, io.EOF + } + r.read = true + return copy(p, r.payload), errTestBodyRead +} + +func (r *failingReadCloser) Close() error { return nil } + +type httpTraceEnv struct { + t *testing.T + recorder *tracetest.SpanRecorder + tracerProvider *trace.TracerProvider +} + +func newHTTPTraceEnv(t *testing.T, options ...trace.TracerProviderOption) *httpTraceEnv { + t.Helper() + recorder := tracetest.NewSpanRecorder() + options = append(options, trace.WithSpanProcessor(recorder)) + return &httpTraceEnv{ + t: t, + recorder: recorder, + tracerProvider: trace.NewTracerProvider(options...), + } +} + +func (env *httpTraceEnv) newClientTransport(rt http.RoundTripper) http.RoundTripper { + return NewHTTPClientTransport(rt, env.tracerProvider, nil) +} + +func (env *httpTraceEnv) newHandler(handler http.Handler) http.Handler { + return NewHTTPHandler(handler, "test-handler", env.tracerProvider, nil) +} + +func (env *httpTraceEnv) requireSpans(count int) []trace.ReadOnlySpan { + env.t.Helper() + spans := env.recorder.Ended() + require.Len(env.t, spans, count) + return spans +} + +func (env *httpTraceEnv) spanAttrs() map[string]any { + env.t.Helper() + spans := env.requireSpans(1) + attrs := spans[0].Attributes() + attrsByKey := make(map[string]any, len(attrs)) + for _, attr := range attrs { + attrsByKey[string(attr.Key)] = attr.Value.AsInterface() + } + return attrsByKey +} + +var errTestBodyRead = errors.New("body read failed") + +// Verifies client transport construction, propagation, and debug body handling. +func TestNewHTTPClientTransport(t *testing.T) { + // A nil tracer provider disables instrumentation without changing the transport identity. + t.Run("Disabled", func(t *testing.T) { + t.Parallel() + + rt := http.DefaultTransport + require.Same(t, rt, NewHTTPClientTransport(rt, nil, nil)) + }) + + // Debug-only headers and payloads must remain absent unless debug mode is enabled. + t.Run("SkipsHeadersAndPayloadsByDefault", func(t *testing.T) { + t.Parallel() + + traceEnv := newHTTPTraceEnv(t) + + var traceparent string + rt := traceEnv.newClientTransport(roundTripperFunc(func(r *http.Request) (*http.Response, error) { + traceparent = r.Header.Get("traceparent") + payload, err := io.ReadAll(r.Body) + require.NoError(t, err) + require.Equal(t, "request body", string(payload)) + return &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(strings.NewReader("response body")), + Header: http.Header{ + "Response-Header": []string{"response-value"}, + }, + Request: r, + }, nil + })) + + req := httptest.NewRequest(http.MethodPost, "http://example.com", strings.NewReader("request body")) + req.Header.Set("Request-Header", "request-value") + resp, err := rt.RoundTrip(req) + require.NoError(t, err) + + body, err := io.ReadAll(resp.Body) + require.NoError(t, err) + require.NoError(t, resp.Body.Close()) + require.Equal(t, "response body", string(body)) + + require.NotEmpty(t, traceparent) + attrs := traceEnv.spanAttrs() + require.NotContains(t, attrs, "http.request.payload") + require.NotContains(t, attrs, "http.response.payload") + require.NotContains(t, attrs, "http.request.headers.request-header") + require.NotContains(t, attrs, "http.response.headers.response-header") + }) + + // Debug mode adds diagnostic HTTP headers and payloads to client spans. + t.Run("DebugMode", func(t *testing.T) { + t.Setenv("TEMPORAL_OTEL_DEBUG", "true") + + // Non-recording client spans should not pay debug payload-capture costs. + t.Run("SkipsCaptureForNonRecordingSpan", func(t *testing.T) { + requestBody := &readTrackingCloser{Reader: strings.NewReader("request body")} + responseBody := &readTrackingCloser{Reader: strings.NewReader("response body")} + var receivedBody io.ReadCloser + transport := &debugHTTPClientSpanTransport{rt: roundTripperFunc(func(r *http.Request) (*http.Response, error) { + receivedBody = r.Body + return &http.Response{ + StatusCode: http.StatusOK, + Body: responseBody, + Header: http.Header{}, + Request: r, + }, nil + })} + + req := httptest.NewRequest(http.MethodPost, "http://example.com", nil) + req.Body = requestBody + _, span := noop.NewTracerProvider().Tracer("test").Start(req.Context(), "test") + req = req.WithContext(oteltrace.ContextWithSpan(req.Context(), span)) + + resp, err := transport.RoundTrip(req) + require.NoError(t, err) + require.Same(t, requestBody, receivedBody) + require.Same(t, responseBody, resp.Body) + require.NoError(t, resp.Body.Close()) + }) + + // Debug spans must contain the headers and payloads needed to diagnose an exchange. + t.Run("AnnotatesHeadersAndPayloads", func(t *testing.T) { + const ( + requestPayload = "request body" + responsePayload = "response body" + ) + + traceEnv := newHTTPTraceEnv(t, trace.WithIDGenerator(fixedIDGenerator{})) + + rt := traceEnv.newClientTransport(roundTripperFunc(func(r *http.Request) (*http.Response, error) { + payload, err := io.ReadAll(r.Body) + require.NoError(t, err) + require.Equal(t, requestPayload, string(payload)) + return &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(bytes.NewBufferString(responsePayload)), + Header: http.Header{ + "Response-Header": []string{"response-value"}, + }, + Request: r, + }, nil + })) + + req := httptest.NewRequest(http.MethodPost, "http://example.com", bytes.NewBufferString(requestPayload)) + req.Header.Set("Request-Header", "request-value") + + resp, err := rt.RoundTrip(req) + require.NoError(t, err) + body, err := io.ReadAll(resp.Body) + require.NoError(t, err) + require.Equal(t, responsePayload, string(body)) + require.NoError(t, resp.Body.Close()) + + require.Equal(t, map[string]any{ + "http.request.headers.request-header": []string{"request-value"}, + "http.request.headers.traceparent": []string{"00-0102030405060708090a0b0c0d0e0f10-0102030405060708-01"}, + "http.request.method": "POST", + "http.request.payload": requestPayload, + "http.response.headers.response-header": []string{"response-value"}, + "http.response.payload": responsePayload, + "http.response.status_code": int64(http.StatusOK), + "network.protocol.version": "1.1", + "server.address": "example.com", + "url.full": "http://example.com", + }, traceEnv.spanAttrs()) + }) + + // After an HTTP 101 response, callers use Response.Body to read from and write to the upgraded connection. + t.Run("PreservesUpgradedConnection", func(t *testing.T) { + const ( + serverMessage = "server message" + clientMessage = "client message" + ) + + traceEnv := newHTTPTraceEnv(t) + upgradedConnection := &readWriteCloser{Reader: bytes.NewBufferString(serverMessage)} + rt := traceEnv.newClientTransport(roundTripperFunc(func(r *http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: http.StatusSwitchingProtocols, // HTTP 101 + Body: upgradedConnection, + Header: http.Header{}, + Request: r, + }, nil + })) + + resp, err := rt.RoundTrip(httptest.NewRequest(http.MethodGet, "http://example.com", nil)) + require.NoError(t, err) + + // The wrapped connection must still receive server messages. + connection, ok := resp.Body.(io.ReadWriteCloser) + require.True(t, ok) + responsePayload, err := io.ReadAll(connection) + require.NoError(t, err) + require.Equal(t, serverMessage, string(responsePayload)) + + // The wrapped connection must still send client messages. + written, err := connection.Write([]byte(clientMessage)) + require.NoError(t, err) + require.Equal(t, len(clientMessage), written) + require.Equal(t, clientMessage, upgradedConnection.written.String()) + require.NoError(t, connection.Close()) + }) + + // Decoders may stop after a complete value without reading a chunked body to EOF. + t.Run("AnnotatesChunkedResponsePayloadOnClose", func(t *testing.T) { + traceEnv := newHTTPTraceEnv(t) + rt := traceEnv.newClientTransport(roundTripperFunc(func(r *http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: http.StatusOK, + Body: &readTrackingCloser{Reader: bytes.NewBufferString(`{"ok":true}`)}, + ContentLength: -1, + Header: http.Header{}, + Request: r, + }, nil + })) + + resp, err := rt.RoundTrip(httptest.NewRequest(http.MethodGet, "http://example.com", nil)) + require.NoError(t, err) + var decoded map[string]bool + require.NoError(t, json.NewDecoder(resp.Body).Decode(&decoded)) + require.NoError(t, resp.Body.Close()) + + attrs := traceEnv.spanAttrs() + require.Equal(t, `{"ok":true}`, attrs["http.response.payload"]) + }) + + // Instrumentation must not consume a streaming response before application code reads it. + t.Run("DoesNotReadResponseBodyBeforeCaller", func(t *testing.T) { + traceEnv := newHTTPTraceEnv(t) + body := &readTrackingCloser{Reader: bytes.NewBufferString("response body")} + rt := traceEnv.newClientTransport(roundTripperFunc(func(r *http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: http.StatusOK, + Body: body, + Header: http.Header{}, + Request: r, + }, nil + })) + + resp, err := rt.RoundTrip(httptest.NewRequest(http.MethodGet, "http://example.com", nil)) + require.NoError(t, err) + require.False(t, body.read) + + _, err = io.ReadAll(resp.Body) + require.NoError(t, err) + require.NoError(t, resp.Body.Close()) + }) + + // io.ReadFull can consume the declared length without performing the read that returns EOF. + t.Run("AnnotatesFixedLengthPayloadsWithoutEOF", func(t *testing.T) { + const ( + requestPayload = "request body" + responsePayload = "response body" + ) + + traceEnv := newHTTPTraceEnv(t) + rt := traceEnv.newClientTransport(roundTripperFunc(func(r *http.Request) (*http.Response, error) { + requestBody := make([]byte, r.ContentLength) + requestReadBytes, err := io.ReadFull(r.Body, requestBody) + require.NoError(t, err) + require.Equal(t, len(requestPayload), requestReadBytes) + require.Equal(t, requestPayload, string(requestBody)) + return &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(bytes.NewBufferString(responsePayload)), + ContentLength: int64(len(responsePayload)), + Header: http.Header{}, + Request: r, + }, nil + })) + + req := httptest.NewRequest(http.MethodPost, "http://example.com", bytes.NewBufferString(requestPayload)) + resp, err := rt.RoundTrip(req) + require.NoError(t, err) + responseBody := make([]byte, resp.ContentLength) + responseReadBytes, err := io.ReadFull(resp.Body, responseBody) + require.NoError(t, err) + require.Equal(t, len(responsePayload), responseReadBytes) + require.Equal(t, responsePayload, string(responseBody)) + require.NoError(t, resp.Body.Close()) + + attrs := traceEnv.spanAttrs() + require.Equal(t, requestPayload, attrs["http.request.payload"]) + require.Equal(t, responsePayload, attrs["http.response.payload"]) + }) + + // Wrapping the request body must not mask errors returned by the original body. + t.Run("PreservesRequestBodyReadErrors", func(t *testing.T) { + traceEnv := newHTTPTraceEnv(t) + rt := traceEnv.newClientTransport(roundTripperFunc(func(r *http.Request) (*http.Response, error) { + _, err := io.ReadAll(r.Body) + return nil, err + })) + + req := httptest.NewRequest(http.MethodPost, "http://example.com", nil) + req.Body = &failingReadCloser{payload: []byte("partial request")} + _, err := rt.RoundTrip(req) + require.ErrorIs(t, err, errTestBodyRead) + }) + + // Wrapping the response body must not mask errors returned by the original body. + t.Run("PreservesResponseBodyReadErrors", func(t *testing.T) { + traceEnv := newHTTPTraceEnv(t) + rt := traceEnv.newClientTransport(roundTripperFunc(func(r *http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: http.StatusOK, + Body: &failingReadCloser{payload: []byte("partial response")}, + Header: http.Header{}, + Request: r, + }, nil + })) + + resp, err := rt.RoundTrip(httptest.NewRequest(http.MethodGet, "http://example.com", nil)) + require.NoError(t, err) + _, err = io.ReadAll(resp.Body) + require.ErrorIs(t, err, errTestBodyRead) + require.NoError(t, resp.Body.Close()) + }) + }) +} + +// Verifies server handler construction, standard tracing, and debug body handling. +func TestNewHTTPHandler(t *testing.T) { + // Debug-only headers and payloads must remain absent unless debug mode is enabled. + t.Run("SkipsHeadersAndPayloadsByDefault", func(t *testing.T) { + t.Parallel() + + traceEnv := newHTTPTraceEnv(t) + var handlerErr error + handler := traceEnv.newHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, handlerErr = io.ReadAll(r.Body) + if handlerErr != nil { + return + } + w.Header().Set("Response-Header", "response-value") + _, handlerErr = w.Write([]byte("response body")) + })) + + req := httptest.NewRequest(http.MethodPost, "http://example.com", bytes.NewBufferString("request body")) + req.Header.Set("Request-Header", "request-value") + rec := httptest.NewRecorder() + handler.ServeHTTP(rec, req) + require.NoError(t, handlerErr) + require.Equal(t, "response body", rec.Body.String()) + + attrs := traceEnv.spanAttrs() + require.NotContains(t, attrs, "http.request.payload") + require.NotContains(t, attrs, "http.response.payload") + require.NotContains(t, attrs, "http.request.headers.request-header") + require.NotContains(t, attrs, "http.response.headers.response-header") + }) + + // Debug mode adds diagnostic HTTP headers and payloads to server spans. + t.Run("DebugMode", func(t *testing.T) { + t.Setenv("TEMPORAL_OTEL_DEBUG", "true") + + // Non-recording server spans should not pay debug payload-capture costs. + t.Run("SkipsCaptureForNonRecordingSpan", func(t *testing.T) { + requestBody := &readTrackingCloser{Reader: strings.NewReader("request body")} + recorder := httptest.NewRecorder() + var receivedBody io.ReadCloser + var receivedWriter http.ResponseWriter + handler := &debugHTTPHandler{handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + receivedBody = r.Body + receivedWriter = w + })} + + req := httptest.NewRequest(http.MethodPost, "http://example.com", nil) + req.Body = requestBody + _, span := noop.NewTracerProvider().Tracer("test").Start(req.Context(), "test") + req = req.WithContext(oteltrace.ContextWithSpan(req.Context(), span)) + handler.ServeHTTP(recorder, req) + + require.Same(t, requestBody, receivedBody) + require.Same(t, recorder, receivedWriter) + }) + + // Debug spans must contain the headers and payloads needed to diagnose an exchange. + t.Run("AnnotatesHeadersAndPayloads", func(t *testing.T) { + const ( + requestPayload = "request body" + responsePayload = "response body" + ) + + traceEnv := newHTTPTraceEnv(t) + handler := traceEnv.newHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + payload, err := io.ReadAll(r.Body) + if err != nil { + t.Errorf("ReadAll() error = %v", err) + } + if string(payload) != requestPayload { + t.Errorf("payload = %q, want %q", string(payload), requestPayload) + } + w.Header().Set("Response-Header", "response-value") + _, err = w.Write([]byte(responsePayload)) + if err != nil { + t.Errorf("Write() error = %v", err) + } + })) + + req := httptest.NewRequest(http.MethodPost, "http://example.com", bytes.NewBufferString(requestPayload)) + req.Header.Set("Request-Header", "request-value") + rec := httptest.NewRecorder() + handler.ServeHTTP(rec, req) + require.Equal(t, responsePayload, rec.Body.String()) + + require.Equal(t, map[string]any{ + "client.address": "192.0.2.1", + "http.request.body.size": int64(len(requestPayload)), + "http.request.headers.request-header": []string{"request-value"}, + "http.request.method": "POST", + "http.request.payload": requestPayload, + "http.response.body.size": int64(len(responsePayload)), + "http.response.headers.response-header": []string{"response-value"}, + "http.response.payload": responsePayload, + "http.response.status_code": int64(http.StatusOK), + "network.peer.address": "192.0.2.1", + "network.peer.port": int64(1234), + "network.protocol.version": "1.1", + "server.address": "example.com", + "url.scheme": "http", + }, traceEnv.spanAttrs()) + }) + + // Instrumentation must leave request consumption under the application handler's control. + t.Run("DoesNotReadRequestBodyBeforeHandler", func(t *testing.T) { + traceEnv := newHTTPTraceEnv(t) + body := &readTrackingCloser{Reader: bytes.NewBufferString("request body")} + var readBeforeHandler bool + var handlerErr error + var handlerPayload []byte + handler := traceEnv.newHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + readBeforeHandler = body.read + handlerPayload, handlerErr = io.ReadAll(r.Body) + })) + + req := httptest.NewRequest(http.MethodPost, "http://example.com", nil) + req.Body = body + rec := httptest.NewRecorder() + handler.ServeHTTP(rec, req) + require.False(t, readBeforeHandler) + require.NoError(t, handlerErr) + require.Equal(t, "request body", string(handlerPayload)) + }) + + // Handlers may consume the expected bytes from an unknown-length body without reading EOF. + t.Run("AnnotatesChunkedRequestPayloadWithoutEOF", func(t *testing.T) { + const requestPayload = "request body" + + traceEnv := newHTTPTraceEnv(t) + handlerPayload := make([]byte, len(requestPayload)) + var handlerReadBytes int + var handlerErr error + handler := traceEnv.newHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + handlerReadBytes, handlerErr = io.ReadFull(r.Body, handlerPayload) + })) + + req := httptest.NewRequest(http.MethodPost, "http://example.com", bytes.NewBufferString(requestPayload)) + req.ContentLength = -1 + req.TransferEncoding = []string{"chunked"} + handler.ServeHTTP(httptest.NewRecorder(), req) + require.NoError(t, handlerErr) + require.Equal(t, len(requestPayload), handlerReadBytes) + require.Equal(t, requestPayload, string(handlerPayload)) + + attrs := traceEnv.spanAttrs() + require.Equal(t, requestPayload, attrs["http.request.payload"]) + }) + + // io.Copy can use ReaderFrom and bypass the Write hook that tracks standard response size. + t.Run("AnnotatesResponseSizeWhenUsingReadFrom", func(t *testing.T) { + const responsePayload = "response body" + + traceEnv := newHTTPTraceEnv(t) + var handlerErr error + handler := traceEnv.newHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, handlerErr = io.Copy(w, io.LimitReader(bytes.NewBufferString(responsePayload), int64(len(responsePayload)))) + })) + + rec := &readerFromResponseRecorder{ResponseRecorder: httptest.NewRecorder()} + handler.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "http://example.com", nil)) + require.NoError(t, handlerErr) + require.Equal(t, responsePayload, rec.Body.String()) + + attrs := traceEnv.spanAttrs() + require.Equal(t, int64(len(responsePayload)), attrs["http.response.body.size"]) + }) + }) +} + +// Verifies end-to-end span propagation and protocol attribution over TLS-negotiated HTTP/2. +func TestHTTP2Instrumentation(t *testing.T) { + type handlerResult struct { + payload string + err error + } + + traceEnv := newHTTPTraceEnv(t) + resultCh := make(chan handlerResult, 1) + handler := traceEnv.newHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + payload, err := io.ReadAll(r.Body) + if err == nil { + _, err = w.Write([]byte("response body")) + } + resultCh <- handlerResult{payload: string(payload), err: err} + })) + + server := httptest.NewUnstartedServer(handler) + server.EnableHTTP2 = true + server.StartTLS() + t.Cleanup(server.Close) + + client := server.Client() + client.Transport = traceEnv.newClientTransport(client.Transport) + req, err := http.NewRequest(http.MethodPost, server.URL, bytes.NewBufferString("request body")) + require.NoError(t, err) + resp, err := client.Do(req) + require.NoError(t, err) + responsePayload, err := io.ReadAll(resp.Body) + require.NoError(t, err) + require.NoError(t, resp.Body.Close()) + + result := <-resultCh + require.NoError(t, result.err) + require.Equal(t, "request body", result.payload) + require.Equal(t, "response body", string(responsePayload)) + require.NotNil(t, resp.TLS) + require.Equal(t, 2, resp.ProtoMajor) + + var clientSpan, serverSpan trace.ReadOnlySpan + for _, span := range traceEnv.requireSpans(2) { + switch span.SpanKind() { + case oteltrace.SpanKindClient: + clientSpan = span + case oteltrace.SpanKindServer: + serverSpan = span + default: + continue + } + } + require.NotNil(t, clientSpan) + require.NotNil(t, serverSpan) + require.Equal(t, clientSpan.SpanContext().TraceID(), serverSpan.SpanContext().TraceID()) + require.Equal(t, clientSpan.SpanContext().SpanID(), serverSpan.Parent().SpanID()) + + require.Contains(t, serverSpan.Attributes(), attribute.String("network.protocol.version", "2.0")) +} diff --git a/go.mod b/go.mod index cbaeaa3af9..7a9d0965bb 100644 --- a/go.mod +++ b/go.mod @@ -153,7 +153,7 @@ require ( github.com/envoyproxy/go-control-plane/envoy v1.37.0 // indirect github.com/envoyproxy/protoc-gen-validate v1.3.3 // indirect github.com/facebookgo/clock v0.0.0-20150410010913-600d898af40a // indirect - github.com/felixge/httpsnoop v1.0.4 // indirect + github.com/felixge/httpsnoop v1.0.4 github.com/fxamacker/cbor/v2 v2.9.1 // indirect github.com/go-logr/logr v1.4.3 // indirect github.com/go-logr/stdr v1.2.2 // indirect @@ -207,7 +207,7 @@ require ( github.com/xrash/smetrics v0.0.0-20250705151800-55b8f293f342 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/contrib/detectors/gcp v1.43.0 // indirect - go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.68.0 // indirect + go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.68.0 go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.43.0 go.opentelemetry.io/proto/otlp v1.10.0 go.uber.org/atomic v1.11.0 // indirect