From 6576442e7e6ed532cf4fc2b0d149108bb3784fd4 Mon Sep 17 00:00:00 2001 From: Alessandro Pagnin Date: Fri, 24 Jul 2026 18:27:44 +0200 Subject: [PATCH 1/3] feat: add last request byte timing --- docs-website/router/configuration.mdx | 8 +- .../configuration/template-expressions.mdx | 2 + .../router/metrics-and-monitoring.mdx | 6 + .../prometheus-metric-reference.mdx | 6 + router-tests/observability/prometheus_test.go | 8 + .../observability/structured_logging_test.go | 64 ++ .../telemetry/connection_metrics_test.go | 73 +++ router/internal/expr/expr.go | 2 + router/internal/traceclient/traceclient.go | 300 ++++++++- .../internal/traceclient/traceclient_test.go | 599 +++++++++++++++++- router/pkg/config/config.schema.json | 4 +- router/pkg/metric/connection_measurements.go | 20 + router/pkg/metric/connection_metric_store.go | 36 ++ router/pkg/metric/noop_connection_metrics.go | 10 + .../metric/oltp_connection_metric_store.go | 12 + .../metric/prom_connection_metric_store.go | 12 + 16 files changed, 1133 insertions(+), 29 deletions(-) diff --git a/docs-website/router/configuration.mdx b/docs-website/router/configuration.mdx index ebfc748316..959f2fe72d 100644 --- a/docs-website/router/configuration.mdx +++ b/docs-website/router/configuration.mdx @@ -711,7 +711,7 @@ This option may change or be removed in future versions as the OpenTelemetry SDK | METRICS_OTLP_EXCLUDE_METRICS | exclude_metrics | | The metrics to exclude from the OTEL metrics. Accepts a list of Go regular expressions. Use https://regex101.com/ to test your regular expressions. | [] | | METRICS_OTLP_EXCLUDE_METRIC_LABELS | exclude_metric_labels | | The metric labels to exclude from the OTEL metrics. Accepts a list of Go regular expressions. Use https://regex101.com/ to test your regular expressions. | [] | | METRICS_OTLP_CONNECTION_STATS | connection_stats | | Enable connection metrics. | false | -| METRICS_OTLP_NETWORK_ENABLED | network.enabled | | Enable per-request subgraph HTTP phase metrics: DNS lookup, TCP connect, TLS handshake, time-to-first-request-byte, and time-to-first-byte histograms. | false | +| METRICS_OTLP_NETWORK_ENABLED | network.enabled | | Enable per-request subgraph HTTP phase metrics: DNS lookup, TCP connect, TLS handshake, time-to-first-request-byte, time-to-last-request-byte (request transfer duration), time-to-first-byte, and time-to-last-byte (response transfer duration) histograms. | false | | METRICS_OTLP_RESOLVER_ENABLED | resolver.enabled | | Enable resolver metrics: resolver concurrency gauges and the resolver acquire duration histogram. | false | | METRICS_OTLP_CIRCUIT_BREAKER | circuit_breaker | | Ensure that circuit breaker metrics are enabled for OTEL. | false | | METRICS_OTLP_STREAM | streams | | Enable Cosmo Streams metrics. | false | @@ -750,7 +750,7 @@ telemetry: router_runtime: true graphql_cache: true connection_stats: false - # Per-request subgraph HTTP phase metrics (DNS, TCP connect, TLS handshake, time to first request byte, time to first byte) + # Per-request subgraph HTTP phase metrics (DNS, TCP connect, TLS handshake, time to first request byte, request transfer duration, time to first response byte, response transfer duration) network: enabled: false # Resolver concurrency gauges and resolver acquire duration histogram @@ -787,7 +787,7 @@ telemetry: | PROMETHEUS_LISTEN_ADDR | listen_addr | | The address to listen on for the prometheus metrics endpoint. | "127.0.0.1:8088" | | PROMETHEUS_GRAPHQL_CACHE | graphql_cache | | Enable the collection of metrics for the GraphQL operation router caches. | false | | PROMETHEUS_CONNECTION_STATS | connection_stats | | Enable connection metrics. | false | -| PROMETHEUS_NETWORK_ENABLED | network.enabled | | Enable per-request subgraph HTTP phase metrics: DNS lookup, TCP connect, TLS handshake, time-to-first-request-byte, and time-to-first-byte histograms. | false | +| PROMETHEUS_NETWORK_ENABLED | network.enabled | | Enable per-request subgraph HTTP phase metrics: DNS lookup, TCP connect, TLS handshake, time-to-first-request-byte, time-to-last-request-byte (request transfer duration), time-to-first-byte, and time-to-last-byte (response transfer duration) histograms. | false | | PROMETHEUS_RESOLVER_ENABLED | resolver.enabled | | Enable resolver metrics: resolver concurrency gauges and the resolver acquire duration histogram. | false | | PROMETHEUS_EXCLUDE_METRICS | exclude_metrics | | | | | PROMETHEUS_EXCLUDE_METRIC_LABELS | exclude_metric_labels | | | | @@ -811,7 +811,7 @@ telemetry: listen_addr: "127.0.0.1:8088" graphql_cache: true connection_stats: false - # Per-request subgraph HTTP phase metrics (DNS, TCP connect, TLS handshake, time to first request byte, time to first byte) + # Per-request subgraph HTTP phase metrics (DNS, TCP connect, TLS handshake, time to first request byte, request transfer duration, time to first response byte, response transfer duration) network: enabled: false # Resolver concurrency gauges and resolver acquire duration histogram diff --git a/docs-website/router/configuration/template-expressions.mdx b/docs-website/router/configuration/template-expressions.mdx index 596503f25c..8436105d29 100644 --- a/docs-website/router/configuration/template-expressions.mdx +++ b/docs-website/router/configuration/template-expressions.mdx @@ -217,7 +217,9 @@ Client trace values describe the individual subgraph fetch that is being logged - `subgraph.request.clientTrace.tcpConnectDuration` (time.Duration): The duration of the TCP connect (dial) for the subgraph request. It is 0 when an existing connection is reused. In case of retries, the value of the last attempt that dialed is recorded. - `subgraph.request.clientTrace.tlsHandshakeDuration` (time.Duration): The duration of the TLS handshake for the subgraph request. It is 0 when an existing connection is reused or the subgraph is reached over plaintext. In case of retries, the value of the last attempt that performed a handshake is recorded. - `subgraph.request.clientTrace.timeToFirstRequestByte` (time.Duration): The duration from the start of the HTTP attempt to writing the first request byte to the subgraph. It includes acquiring the connection. In case of retries, the value of the last attempt that wrote a request byte is recorded. +- `subgraph.request.clientTrace.timeToLastRequestByte` (time.Duration): The duration between the first request-header write callback and a successful request-complete callback from Go's HTTP transport. This approximates the time from the first to the last request byte. The value is 0 unless both callbacks are observed and the request-complete callback reports no error. If request completion is reported only after response processing has finished, the histogram still receives a sample but the expression value remains 0. In case of retries, the value of the last attempt with both callbacks is recorded. The value is also 0 for single-flight followers because they do not write a separate subgraph request. - `subgraph.request.clientTrace.timeToFirstByte` (time.Duration): The duration from completing the request write to receiving the first response byte from the subgraph. In case of retries, only the last attempt is recorded. +- `subgraph.request.clientTrace.timeToLastByte` (time.Duration): The duration from receiving the first response byte to consuming the last response byte. The value is 0 unless the first response byte is observed and the full response body is consumed, either to EOF or through the declared `Content-Length`. A response that cannot carry a body completes when its headers have been read. The value remains 0 for bodies that are closed early or end in a read error, upgraded or streaming subscription responses, and single-flight followers that do not independently consume the origin response body. In case of retries, the value of the last attempt whose response is fully consumed is recorded. This object is available in template expressions and can be accessed using the `subgraph` identifier. For example, you can access the subgraph name using `subgraph.name` or check for errors using `subgraph.request.error`. diff --git a/docs-website/router/metrics-and-monitoring.mdx b/docs-website/router/metrics-and-monitoring.mdx index 43f05d58f2..90de84a429 100644 --- a/docs-website/router/metrics-and-monitoring.mdx +++ b/docs-website/router/metrics-and-monitoring.mdx @@ -348,6 +348,8 @@ We use the following standard dimensions These metrics break down each outgoing subgraph request into its individual HTTP phases. They are collected per request using Go's `httptrace`, so unlike [Connection Metrics](#connection-metrics) they do not depend on the connection pool dialer. Use them to attribute subgraph latency to DNS resolution, connection setup, the TLS handshake, or the subgraph itself. This is useful for diagnosing issues such as slow DNS lookups. +The histograms record actual outgoing subgraph HTTP requests. When single-flight request deduplication coalesces fetches, only the fetch that sends the subgraph request produces samples. Followers do not independently write the request or consume the origin response body. + ```bash config.yaml telemetry: @@ -371,8 +373,12 @@ telemetry: * `router.http.client.time_to_first_request_byte`: Histogram (ms) of the time from the start of the HTTP attempt to writing the first request byte to the subgraph. It includes acquiring the connection: pool wait, or DNS, TCP connect, and TLS handshake for new connections. +* `router.http.client.time_to_last_request_byte`: Histogram (ms) of the duration between the first request-header write callback and a successful request-complete callback from Go's HTTP transport. This approximates the time from the first to the last request byte. Recorded only when both callbacks are observed and the request-complete callback reports no error. + * `router.http.client.time_to_first_byte`: Histogram (ms) of the time from completing the request write to receiving the first response byte from the subgraph. +* `router.http.client.time_to_last_byte`: Histogram (ms) of the duration from receiving the first response byte to consuming the last response byte. Recorded only when a first response byte is observed and the full body is consumed, either to EOF or through the declared `Content-Length`. A response that cannot carry a body completes when its headers have been read. No sample is emitted when the body is closed early, when reading the body ends with an error, or for upgraded or streaming subscription responses. + ## Custom Attributes You can also add custom attributes to OTEL and Prometheus. Please refer to the [Custom Attributes](/router/open-telemetry/custom-attributes) section. diff --git a/docs-website/router/metrics-and-monitoring/prometheus-metric-reference.mdx b/docs-website/router/metrics-and-monitoring/prometheus-metric-reference.mdx index ab2ac8a46d..81b4169c2d 100644 --- a/docs-website/router/metrics-and-monitoring/prometheus-metric-reference.mdx +++ b/docs-website/router/metrics-and-monitoring/prometheus-metric-reference.mdx @@ -219,6 +219,8 @@ You can find more examples here under [`Connection`](/router/metrics-and-monitor These metrics break down each outgoing subgraph request into its individual HTTP phases, collected per request via Go's `httptrace`. Use them to attribute subgraph latency to DNS resolution, connection setup, the TLS handshake, or the subgraph itself. +The histograms record actual outgoing subgraph HTTP requests. When single-flight request deduplication coalesces fetches, only the fetch that sends the subgraph request produces samples. Followers do not independently write the request or consume the origin response body. + #### Enable Network Metrics ```yaml config.yaml @@ -239,8 +241,12 @@ telemetry: * `router_http_client_time_to_first_request_byte`: The time in ms from the start of the HTTP attempt to writing the first request byte to the subgraph. It includes acquiring the connection: pool wait, or DNS, TCP connect, and TLS handshake for new connections. +* `router_http_client_time_to_last_request_byte`: The duration in ms between the first request-header write callback and a successful request-complete callback from Go's HTTP transport. This approximates the time from the first to the last request byte. Recorded only when both callbacks are observed and the request-complete callback reports no error. + * `router_http_client_time_to_first_byte`: The time in ms from completing the request write to receiving the first response byte from the subgraph. +* `router_http_client_time_to_last_byte`: The duration in ms from receiving the first response byte to consuming the last response byte. Recorded only when a first response byte is observed and the full body is consumed, either to EOF or through the declared `Content-Length`. A response that cannot carry a body completes when its headers have been read. No sample is emitted when the body is closed early, when reading the body ends with an error, or for upgraded or streaming subscription responses. + ### Go Runtime Metrics These metrics help monitor application memory usage, concurrency, and garbage collection efficiency: diff --git a/router-tests/observability/prometheus_test.go b/router-tests/observability/prometheus_test.go index 43534c7ae5..9bb32ebcc0 100644 --- a/router-tests/observability/prometheus_test.go +++ b/router-tests/observability/prometheus_test.go @@ -5341,10 +5341,18 @@ func TestFlakyPrometheusRouterConnectionMetrics(t *testing.T) { require.NotNil(t, ttfb) require.NotEmpty(t, ttfb.GetMetric()) + ttlb := findMetricFamilyByName(mf, "router_http_client_time_to_last_byte") + require.NotNil(t, ttlb) + require.NotEmpty(t, ttlb.GetMetric()) + firstRequestByte := findMetricFamilyByName(mf, "router_http_client_time_to_first_request_byte") require.NotNil(t, firstRequestByte) require.NotEmpty(t, firstRequestByte.GetMetric()) + lastRequestByte := findMetricFamilyByName(mf, "router_http_client_time_to_last_request_byte") + require.NotNil(t, lastRequestByte) + require.NotEmpty(t, lastRequestByte.GetMetric()) + tcpConnect := findMetricFamilyByName(mf, "router_http_client_tcp_connect_duration") require.NotNil(t, tcpConnect) require.NotEmpty(t, tcpConnect.GetMetric()) diff --git a/router-tests/observability/structured_logging_test.go b/router-tests/observability/structured_logging_test.go index 929e523dbe..e7164bbb6a 100644 --- a/router-tests/observability/structured_logging_test.go +++ b/router-tests/observability/structured_logging_test.go @@ -3653,6 +3653,70 @@ func TestFlakyAccessLogs(t *testing.T) { }) }) + t.Run("verify timeToLastRequestByte value is attached", func(t *testing.T) { + t.Parallel() + + testenv.Run(t, &testenv.Config{ + SubgraphAccessLogsEnabled: true, + SubgraphAccessLogFields: []config.CustomAttribute{ + { + Key: "time_to_last_request_byte", + ValueFrom: &config.CustomDynamicAttribute{ + Expression: "subgraph.request.clientTrace.timeToLastRequestByte", + }, + }, + }, + LogObservation: testenv.LogObservationConfig{ + Enabled: true, + LogLevel: zapcore.InfoLevel, + }, + }, func(t *testing.T, xEnv *testenv.Environment) { + xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{ + Query: `query myQuery { employees { id } }`, + }) + requestLog := xEnv.Observer().FilterMessage("/graphql") + requestLogAll := requestLog.All() + requestContextMap := requestLogAll[0].ContextMap() + + timeToLastRequestByte, ok := requestContextMap["time_to_last_request_byte"].(time.Duration) + require.True(t, ok) + + require.Greater(t, int(timeToLastRequestByte), 0) + }) + }) + + t.Run("verify timeToLastByte value is attached", func(t *testing.T) { + t.Parallel() + + testenv.Run(t, &testenv.Config{ + SubgraphAccessLogsEnabled: true, + SubgraphAccessLogFields: []config.CustomAttribute{ + { + Key: "time_to_last_byte", + ValueFrom: &config.CustomDynamicAttribute{ + Expression: "subgraph.request.clientTrace.timeToLastByte", + }, + }, + }, + LogObservation: testenv.LogObservationConfig{ + Enabled: true, + LogLevel: zapcore.InfoLevel, + }, + }, func(t *testing.T, xEnv *testenv.Environment) { + xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{ + Query: `query myQuery { employees { id } }`, + }) + requestLog := xEnv.Observer().FilterMessage("/graphql") + requestLogAll := requestLog.All() + requestContextMap := requestLogAll[0].ContextMap() + + timeToLastByte, ok := requestContextMap["time_to_last_byte"].(time.Duration) + require.True(t, ok) + + require.Greater(t, int(timeToLastByte), 0) + }) + }) + t.Run("verify connAcquireDuration value is attached for multiple subgraph calls", func(t *testing.T) { t.Parallel() diff --git a/router-tests/telemetry/connection_metrics_test.go b/router-tests/telemetry/connection_metrics_test.go index 421c9a8664..da7a61631f 100644 --- a/router-tests/telemetry/connection_metrics_test.go +++ b/router-tests/telemetry/connection_metrics_test.go @@ -154,6 +154,79 @@ func TestConnectionMetrics(t *testing.T) { }) }) + t.Run("validate transfer duration metrics are present when network metrics are enabled", func(t *testing.T) { + t.Parallel() + + metricReader := metric.NewManualReader() + testenv.Run(t, &testenv.Config{ + MetricReader: metricReader, + MetricOptions: testenv.MetricOptions{ + EnableOTLPNetworkMetrics: true, + }, + }, func(t *testing.T, xEnv *testenv.Environment) { + xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{ + Query: `query { employees { id } }`, + }) + + rm := metricdata.ResourceMetrics{} + err := metricReader.Collect(context.Background(), &rm) + require.NoError(t, err) + + scopeMetric := testutils.GetMetricScopeByName(rm.ScopeMetrics, "cosmo.router.connections") + require.NotNil(t, scopeMetric) + excludePortFromMetrics(t, rm.ScopeMetrics) + + expectedAttributes := attribute.NewSet( + otel.ServerAddress.String("127.0.0.1"), + otel.WgClientReusedConnection.Bool(false), + otel.WgSubgraphName.String("employees"), + ) + + tests := []struct { + name string + description string + }{ + { + name: "router.http.client.time_to_last_request_byte", + description: "Time from the first request byte write event to the successful request write completion event for outgoing subgraph requests", + }, + { + name: "router.http.client.time_to_last_byte", + description: "Time from the first response byte to the last response byte from subgraph", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + actual := testutils.GetMetricByName(scopeMetric, tc.name) + require.NotNil(t, actual) + + actualHistogram, ok := actual.Data.(metricdata.Histogram[float64]) + require.True(t, ok) + require.Len(t, actualHistogram.DataPoints, 1) + require.Greater(t, actualHistogram.DataPoints[0].Count, uint64(0)) + require.Greater(t, actualHistogram.DataPoints[0].Sum, 0.0) + + expected := metricdata.Metrics{ + Name: tc.name, + Description: tc.description, + Unit: "ms", + Data: metricdata.Histogram[float64]{ + Temporality: metricdata.CumulativeTemporality, + DataPoints: []metricdata.HistogramDataPoint[float64]{ + { + Attributes: expectedAttributes, + }, + }, + }, + } + + metricdatatest.AssertEqual(t, expected, *actual, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreValue()) + }) + } + }) + }) + t.Run("verify custom subgraph transport configs", func(t *testing.T) { t.Parallel() diff --git a/router/internal/expr/expr.go b/router/internal/expr/expr.go index 1a540df165..1b7a40e3f9 100644 --- a/router/internal/expr/expr.go +++ b/router/internal/expr/expr.go @@ -166,7 +166,9 @@ type ClientTrace struct { TCPConnectDuration time.Duration `expr:"tcpConnectDuration"` TLSHandshakeDuration time.Duration `expr:"tlsHandshakeDuration"` TimeToFirstRequestByte time.Duration `expr:"timeToFirstRequestByte"` + TimeToLastRequestByte time.Duration `expr:"timeToLastRequestByte"` TimeToFirstByte time.Duration `expr:"timeToFirstByte"` + TimeToLastByte time.Duration `expr:"timeToLastByte"` } // Subgraph Related diff --git a/router/internal/traceclient/traceclient.go b/router/internal/traceclient/traceclient.go index 07e27c7edd..a515740a99 100644 --- a/router/internal/traceclient/traceclient.go +++ b/router/internal/traceclient/traceclient.go @@ -3,8 +3,11 @@ package traceclient import ( "context" "crypto/tls" + "io" + "mime" "net/http" "net/http/httptrace" + "strings" "sync" "time" @@ -34,6 +37,7 @@ type phaseDurations struct { TCPConnect time.Duration TLSHandshake time.Duration TimeToFirstRequestByte time.Duration + TimeToLastRequestByte time.Duration TimeToFirstByte time.Duration } @@ -45,7 +49,11 @@ type ClientTrace struct { connectStart map[string]time.Time tlsStart time.Time wroteFirstByte time.Time + attemptFirstByte time.Time wroteRequest time.Time + gotFirstRespByte time.Time + + timeToLastRequestByteObserver func(time.Duration) durations phaseDurations } @@ -59,6 +67,13 @@ func (c *ClientTrace) HttpClientTrace() *httptrace.ClientTrace { Time: time.Now(), HostPort: hostPort, } + // GetConn starts a new transport attempt. Keep wroteFirstByte for the + // existing time-to-first-request-byte metric, but reset the + // first-to-last request span so retries never pair timestamps from + // different attempts. + c.attemptFirstByte = time.Time{} + c.wroteRequest = time.Time{} + c.durations.TimeToLastRequestByte = 0 }, GotConn: func(info httptrace.GotConnInfo) { c.mu.Lock() @@ -120,24 +135,45 @@ func (c *ClientTrace) HttpClientTrace() *httptrace.ClientTrace { WroteHeaderField: func(_ string, _ []string) { c.mu.Lock() defer c.mu.Unlock() + now := time.Now() + if c.attemptFirstByte.IsZero() { + c.attemptFirstByte = now + } // Only the first header field marks the first request byte written if !c.wroteFirstByte.IsZero() { return } - c.wroteFirstByte = time.Now() + c.wroteFirstByte = now if c.ConnectionGet != nil && c.wroteFirstByte.After(c.ConnectionGet.Time) { c.durations.TimeToFirstRequestByte = c.wroteFirstByte.Sub(c.ConnectionGet.Time) } }, - WroteRequest: func(_ httptrace.WroteRequestInfo) { + WroteRequest: func(info httptrace.WroteRequestInfo) { c.mu.Lock() - defer c.mu.Unlock() + if info.Err != nil { + c.mu.Unlock() + return + } + c.wroteRequest = time.Now() + var duration time.Duration + if !c.attemptFirstByte.IsZero() && c.wroteRequest.After(c.attemptFirstByte) { + duration = c.wroteRequest.Sub(c.attemptFirstByte) + c.durations.TimeToLastRequestByte = duration + } + observer := c.timeToLastRequestByteObserver + c.timeToLastRequestByteObserver = nil + c.mu.Unlock() + + if observer != nil && duration > 0 { + observer(duration) + } }, GotFirstResponseByte: func() { c.mu.Lock() defer c.mu.Unlock() now := time.Now() + c.gotFirstRespByte = now if !c.wroteRequest.IsZero() && now.After(c.wroteRequest) { c.durations.TimeToFirstByte = now.Sub(c.wroteRequest) } @@ -147,12 +183,26 @@ func (c *ClientTrace) HttpClientTrace() *httptrace.ClientTrace { // snapshot returns a consistent view of the observed state. The transport's // write loop can still fire callbacks concurrently with (and after) RoundTrip -// returning, so readers must not access the fields directly. Phases that -// complete after the snapshot are not recorded. -func (c *ClientTrace) snapshot() (*GetConnection, *AcquiredConnection, phaseDurations) { +// returning, so readers must not access the fields directly. gotFirstRespByte +// is the baseline for the first-to-last response-byte measurement, which can +// only be completed once the response body is fully consumed. +func (c *ClientTrace) snapshot() (*GetConnection, *AcquiredConnection, time.Time, phaseDurations) { c.mu.Lock() defer c.mu.Unlock() - return c.ConnectionGet, c.ConnectionAcquired, c.durations + return c.ConnectionGet, c.ConnectionAcquired, c.gotFirstRespByte, c.durations +} + +// observeTimeToLastRequestByte either returns an already-completed request +// transfer span or installs an observer for a successful WroteRequest callback +// that arrives after RoundTrip returns. +func (c *ClientTrace) observeTimeToLastRequestByte(observer func(time.Duration)) time.Duration { + c.mu.Lock() + defer c.mu.Unlock() + if c.durations.TimeToLastRequestByte > 0 { + return c.durations.TimeToLastRequestByte + } + c.timeToLastRequestByteObserver = observer + return 0 } func NewClientTrace() *ClientTrace { @@ -206,12 +256,201 @@ func (t *TraceInjectingRoundTripper) RoundTrip(req *http.Request) (*http.Respons req = req.WithContext(httptrace.WithClientTrace(ctx, ec.HttpClientTrace())) trip, err := t.base.RoundTrip(req) - t.processConnectionMetrics(req.Context(), req, ec) + recorder := t.processConnectionMetrics(req.Context(), req, ec) + + // httptrace has no "last response byte" callback: the last byte is only + // observable once the caller has fully consumed the response body. Wrap + // finite response bodies so the first-to-last-byte span is recorded on a + // clean EOF (or after the declared Content-Length has been read). + // This is always before the subgraph access log is written, because the + // engine fully reads and closes the body during the fetch load phase, which + // completes before the log is emitted. + if recorder != nil { + switch { + case err != nil || !shouldMeasureResponseTransfer(trip): + // Upgraded and streaming responses are intentionally excluded. In + // particular, leaving HTTP 101 bodies untouched preserves their + // io.ReadWriteCloser contract for WebSocket clients. + recorder.cancel() + case responseHasNoBody(req, trip): + // RoundTrip returns after the response headers are read. For HEAD + // and other responses that cannot carry a body, that is also the + // point at which the last response byte has been consumed. + recorder.fire() + default: + trip.Body = &timedResponseBody{ + ReadCloser: trip.Body, + recorder: recorder, + contentLength: trip.ContentLength, + } + } + } return trip, err } -func (t *TraceInjectingRoundTripper) processConnectionMetrics(ctx context.Context, req *http.Request, trace *ClientTrace) { +func shouldMeasureResponseTransfer(resp *http.Response) bool { + if resp == nil || resp.Body == nil || resp.StatusCode == http.StatusSwitchingProtocols { + return false + } + mediaType, _, err := mime.ParseMediaType(resp.Header.Get("Content-Type")) + if err == nil && strings.EqualFold(mediaType, "text/event-stream") { + return false + } + return true +} + +func responseHasNoBody(req *http.Request, resp *http.Response) bool { + if req.Method == http.MethodHead || resp.Body == http.NoBody || resp.ContentLength == 0 { + return true + } + return resp.StatusCode >= 100 && resp.StatusCode <= 199 || + resp.StatusCode == http.StatusNoContent || + resp.StatusCode == http.StatusNotModified +} + +// requestByteRecorder keeps request-write completion independent from response +// completion. A successful WroteRequest callback can arrive after RoundTrip +// returns, including after an early response. Metrics still record that valid +// request span. Per-fetch expression results stop accepting updates when the +// response finishes so OnFinished can read them without racing a late callback. +type requestByteRecorder struct { + mu sync.Mutex + recorded bool + resultsOpen bool + results *expr.ClientTrace + recordMetric func(time.Duration) +} + +func (r *requestByteRecorder) record(duration time.Duration) { + if r == nil || duration <= 0 { + return + } + + r.mu.Lock() + if r.recorded { + r.mu.Unlock() + return + } + r.recorded = true + if r.resultsOpen { + r.results.TimeToLastRequestByte = duration + } + r.mu.Unlock() + + r.recordMetric(duration) +} + +// closeResults records a request completion already visible in the trace, then +// prevents later callbacks from mutating the expression result. The observer +// remains installed so a later successful WroteRequest can still emit a metric. +func (r *requestByteRecorder) closeResults(duration time.Duration) { + if r == nil { + return + } + + var recordMetric bool + r.mu.Lock() + if !r.recorded && duration > 0 { + r.recorded = true + if r.resultsOpen { + r.results.TimeToLastRequestByte = duration + } + recordMetric = true + } + r.resultsOpen = false + r.mu.Unlock() + + if recordMetric { + r.recordMetric(duration) + } +} + +// lastByteRecorder coordinates a request-write callback that may arrive after +// RoundTrip returns with response-body completion. +type lastByteRecorder struct { + mu sync.Mutex + done bool + trace *ClientTrace + request *requestByteRecorder + recordResponse func(time.Duration) +} + +func (r *lastByteRecorder) fire() { + if r == nil { + return + } + _, _, firstResponseByte, durations := r.trace.snapshot() + r.request.closeResults(durations.TimeToLastRequestByte) + + r.mu.Lock() + if r.done { + r.mu.Unlock() + return + } + r.done = true + if !firstResponseByte.IsZero() { + if duration := time.Since(firstResponseByte); duration > 0 { + r.recordResponse(duration) + } + } + r.mu.Unlock() +} + +func (r *lastByteRecorder) cancel() { + if r == nil { + return + } + _, _, _, durations := r.trace.snapshot() + r.request.closeResults(durations.TimeToLastRequestByte) + + r.mu.Lock() + r.done = true + r.mu.Unlock() +} + +// timedResponseBody records completion only after the full response body is +// consumed. Close by itself is cancellation, not evidence that the last byte +// was received. +type timedResponseBody struct { + io.ReadCloser + recorder *lastByteRecorder + contentLength int64 + bytesRead int64 +} + +func (b *timedResponseBody) Read(p []byte) (int, error) { + n, err := b.ReadCloser.Read(p) + b.bytesRead += int64(n) + + complete := false + switch { + case err == io.EOF: + // A short EOF on a response with an explicit Content-Length is a + // truncated response, not a successfully observed last byte. + complete = b.contentLength < 0 || b.bytesRead >= b.contentLength + case err == nil && b.contentLength > 0 && b.bytesRead >= b.contentLength: + // Some readers return the final bytes with a nil error and expect no + // subsequent read. Content-Length lets us recognize completion there. + complete = true + } + if complete { + b.recorder.fire() + } + return n, err +} + +func (b *timedResponseBody) Close() error { + err := b.ReadCloser.Close() + b.recorder.cancel() + return err +} + +// processConnectionMetrics records the connection and per-attempt request-phase +// metrics that are already observable when RoundTrip returns. It returns a +// recorder for the time-to-last-byte metric, which can only be measured once +// the response body has been consumed, or nil when there is nothing to record. +func (t *TraceInjectingRoundTripper) processConnectionMetrics(ctx context.Context, req *http.Request, trace *ClientTrace) *lastByteRecorder { var subgraph string subgraphCtxVal := ctx.Value(rcontext.CurrentSubgraphContextKey{}) if subgraphCtxVal != nil { @@ -226,21 +465,21 @@ func (t *TraceInjectingRoundTripper) processConnectionMetrics(ctx context.Contex } if trace == nil { - return + return nil } results := ClientTraceResultsFromContext(ctx) if results == nil { - return + return nil } - connectionGet, connectionAcquired, durations := trace.snapshot() + connectionGet, connectionAcquired, _, durations := trace.snapshot() // The transport can fail before it ever asks the pool for a connection, // in which case no phase was observed and there is nothing to record. if connectionGet == nil { - return + return nil } serverAttributes := rotel.GetServerAttributes(connectionGet.HostPort) @@ -304,6 +543,41 @@ func (t *TraceInjectingRoundTripper) processConnectionMetrics(ctx context.Contex serverAttributes..., ) } + + lastByteMetricStore, _ := t.connectionMetricStore.(metric.LastByteMetricStore) + requestRecorder := &requestByteRecorder{ + resultsOpen: true, + results: results, + recordMetric: func(duration time.Duration) { + if lastByteMetricStore != nil { + lastByteMetricStore.MeasureTimeToLastRequestByte( + ctx, + msFromDuration(duration), + serverAttributes..., + ) + } + }, + } + recorder := &lastByteRecorder{ + trace: trace, + request: requestRecorder, + recordResponse: func(duration time.Duration) { + results.TimeToLastByte = duration + if lastByteMetricStore != nil { + lastByteMetricStore.MeasureTimeToLastByte( + ctx, + msFromDuration(duration), + serverAttributes..., + ) + } + }, + } + + if duration := trace.observeTimeToLastRequestByte(requestRecorder.record); duration > 0 { + requestRecorder.record(duration) + } + + return recorder } func msFromDuration(d time.Duration) float64 { diff --git a/router/internal/traceclient/traceclient_test.go b/router/internal/traceclient/traceclient_test.go index 75545bd944..17e4d9e538 100644 --- a/router/internal/traceclient/traceclient_test.go +++ b/router/internal/traceclient/traceclient_test.go @@ -65,12 +65,98 @@ func (rt *hookFiringRoundTripper) RoundTrip(req *http.Request) (*http.Response, }, nil } +// bodyReturningRoundTripper fires a hook sequence and then returns a response +// whose body is the given reader, so the time-to-last-byte measurement (which +// is only observable once the body is read) can be exercised. +type bodyReturningRoundTripper struct { + fire func(ct *httptrace.ClientTrace) + body io.ReadCloser + statusCode int + header http.Header + contentLength int64 +} + +func (rt *bodyReturningRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) { + rt.fire(httptrace.ContextClientTrace(req.Context())) + statusCode := rt.statusCode + if statusCode == 0 { + statusCode = http.StatusOK + } + header := rt.header + if header == nil { + header = make(http.Header) + } + return &http.Response{ + StatusCode: statusCode, + Body: rt.body, + Header: header, + ContentLength: rt.contentLength, + Request: req, + }, nil +} + +// pacedBody yields its content one chunk per read, sleeping before each, so the +// time-to-last-byte window grows measurably as the body is consumed. +type pacedBody struct { + chunks [][]byte + delay time.Duration + i int + closed bool +} + +func (b *pacedBody) Read(p []byte) (int, error) { + if b.i >= len(b.chunks) { + return 0, io.EOF + } + time.Sleep(b.delay) + n := copy(p, b.chunks[b.i]) + b.i++ + return n, nil +} + +func (b *pacedBody) Close() error { + b.closed = true + return nil +} + +type failingBody struct { + read bool + closed bool +} + +func (b *failingBody) Read(p []byte) (int, error) { + if !b.read { + b.read = true + return copy(p, "partial"), nil + } + return 0, errors.New("response body read failed") +} + +func (b *failingBody) Close() error { + b.closed = true + return nil +} + +type readWriteBody struct { + *pacedBody +} + +func (b *readWriteBody) Write(p []byte) (int, error) { + return len(p), nil +} + +type roundTripperFunc func(*http.Request) (*http.Response, error) + +func (f roundTripperFunc) RoundTrip(req *http.Request) (*http.Response, error) { + return f(req) +} + // recordingConnectionMetricStore counts how many times each measurement is // recorded and keeps the last recorded value in milliseconds. type recordingConnectionMetricStore struct { - acquire, dns, tcp, tls, reqFirstByte, ttfb int + acquire, dns, tcp, tls, reqFirstByte, reqLastByte, ttfb, ttlb int - dnsMs, tcpMs float64 + dnsMs, tcpMs, reqLastByteMs float64 } func (s *recordingConnectionMetricStore) MeasureConnectionAcquireDuration(_ context.Context, _ float64, _ ...attribute.KeyValue) { @@ -90,9 +176,16 @@ func (s *recordingConnectionMetricStore) MeasureTLSHandshakeDuration(_ context.C func (s *recordingConnectionMetricStore) MeasureTimeToFirstRequestByte(_ context.Context, _ float64, _ ...attribute.KeyValue) { s.reqFirstByte++ } +func (s *recordingConnectionMetricStore) MeasureTimeToLastRequestByte(_ context.Context, value float64, _ ...attribute.KeyValue) { + s.reqLastByte++ + s.reqLastByteMs = value +} func (s *recordingConnectionMetricStore) MeasureTimeToFirstByte(_ context.Context, _ float64, _ ...attribute.KeyValue) { s.ttfb++ } +func (s *recordingConnectionMetricStore) MeasureTimeToLastByte(_ context.Context, _ float64, _ ...attribute.KeyValue) { + s.ttlb++ +} func (s *recordingConnectionMetricStore) Shutdown(_ context.Context) error { return nil } // roundTripThroughHooks runs a request whose context optionally carries a fresh @@ -122,6 +215,8 @@ func roundTripThroughHooks(t *testing.T, withContainer bool, fire func(ct *httpt resp, err := rt.RoundTrip(req) require.NoError(t, err) + _, err = io.Copy(io.Discard, resp.Body) + require.NoError(t, err) _ = resp.Body.Close() var results expr.ClientTrace @@ -169,7 +264,9 @@ func TestTraceInjectingRoundTripper(t *testing.T) { require.Equal(t, 1, store.tcp, "TCP connect duration should be recorded once") require.Equal(t, 1, store.tls, "TLS handshake duration should be recorded once") require.Equal(t, 1, store.reqFirstByte, "time to first request byte should be recorded once") + require.Equal(t, 1, store.reqLastByte, "time to last request byte should be recorded once") require.Equal(t, 1, store.ttfb, "time to first byte should be recorded once") + require.Equal(t, 1, store.ttlb, "time to last byte should be recorded once the body is read") }) t.Run("records connection phase timings without racing concurrent httptrace callbacks", func(t *testing.T) { @@ -222,7 +319,9 @@ func TestTraceInjectingRoundTripper(t *testing.T) { require.Greater(t, results.TCPConnectDuration, time.Duration(0)) require.Greater(t, results.TLSHandshakeDuration, time.Duration(0)) require.Greater(t, results.TimeToFirstRequestByte, time.Duration(0)) + require.Greater(t, results.TimeToLastRequestByte, time.Duration(0)) require.Greater(t, results.TimeToFirstByte, time.Duration(0)) + require.Greater(t, results.TimeToLastByte, time.Duration(0)) require.Zero(t, exprCtx.Subgraph.Request.ClientTrace, "the request-scoped expression context must stay untouched when a per-fetch container is present") @@ -231,7 +330,9 @@ func TestTraceInjectingRoundTripper(t *testing.T) { require.Equal(t, 1, store.tcp) require.Equal(t, 1, store.tls) require.Equal(t, 1, store.reqFirstByte) + require.Equal(t, 1, store.reqLastByte) require.Equal(t, 1, store.ttfb) + require.Equal(t, 1, store.ttlb) }) t.Run("keeps the last observation of each phase across retry attempts of one fetch", func(t *testing.T) { @@ -349,21 +450,28 @@ func TestTraceInjectingRoundTripper(t *testing.T) { // WroteHeaderField fires once per header field; only the first call // marks the first request byte. The duration spans from the connection // request (attempt start, not connection acquired) to that first field - // and must not grow with later fields. + // and must not grow with later fields. The first-to-last request span, + // by contrast, excludes connection acquisition. results, _, store := roundTripThroughHooks(t, true, func(ct *httptrace.ClientTrace) { ct.GetConn("subgraph.local:443") - time.Sleep(10 * time.Millisecond) + time.Sleep(30 * time.Millisecond) ct.GotConn(httptrace.GotConnInfo{}) - time.Sleep(10 * time.Millisecond) - ct.WroteHeaderField("Host", []string{"subgraph.local"}) time.Sleep(20 * time.Millisecond) + ct.WroteHeaderField("Host", []string{"subgraph.local"}) + time.Sleep(10 * time.Millisecond) ct.WroteHeaderField("Content-Type", []string{"application/json"}) ct.WroteRequest(httptrace.WroteRequestInfo{}) }) require.Equal(t, 1, store.reqFirstByte) - require.GreaterOrEqual(t, results.TimeToFirstRequestByte, 19*time.Millisecond, "must span from the connection request, including the acquisition, to the first header field") - require.Less(t, results.TimeToFirstRequestByte, 39*time.Millisecond, "must not span later header fields") + require.GreaterOrEqual(t, results.TimeToFirstRequestByte, 49*time.Millisecond, "must span from the connection request, including acquisition, to the first header field") + + // The last request byte is marked by a successful WroteRequest, and its + // duration starts at the first request byte rather than GetConn. + require.Equal(t, 1, store.reqLastByte) + require.GreaterOrEqual(t, results.TimeToLastRequestByte, 9*time.Millisecond) + require.Less(t, results.TimeToLastRequestByte, 30*time.Millisecond, "must exclude connection acquisition") + require.Less(t, results.TimeToLastRequestByte, results.TimeToFirstRequestByte) }) t.Run("ignores header bytes written without a connection request", func(t *testing.T) { @@ -376,14 +484,18 @@ func TestTraceInjectingRoundTripper(t *testing.T) { require.Equal(t, 0, store.reqFirstByte) require.Zero(t, results.TimeToFirstRequestByte) + require.Equal(t, 0, store.reqLastByte) + require.Zero(t, results.TimeToLastRequestByte) + require.Equal(t, 0, store.ttlb, "without a connection request there is nothing to anchor the last byte to") + require.Zero(t, results.TimeToLastByte) }) t.Run("keeps the first attempt's measurement when the transport retries inside one RoundTrip", func(t *testing.T) { // A reused keep-alive connection that turns out dead makes net/http // retry on a new connection within the same RoundTrip, reusing the // same trace. The first request byte was written by the first attempt: - // its measurement is kept, and the duration must never be recomputed - // against the redial's connection request. + // its time-to-first measurement is kept. The first-to-last request span + // must use only the successful retry's timestamps. results, _, store := roundTripThroughHooks(t, true, func(ct *httptrace.ClientTrace) { ct.GetConn("subgraph.local:443") ct.GotConn(httptrace.GotConnInfo{Reused: true}) @@ -394,12 +506,29 @@ func TestTraceInjectingRoundTripper(t *testing.T) { ct.GetConn("subgraph.local:443") ct.GotConn(httptrace.GotConnInfo{}) ct.WroteHeaderField("Host", []string{"subgraph.local"}) + time.Sleep(5 * time.Millisecond) ct.WroteRequest(httptrace.WroteRequestInfo{}) }) require.Equal(t, 1, store.reqFirstByte) require.GreaterOrEqual(t, results.TimeToFirstRequestByte, 4*time.Millisecond, "the first attempt's measurement is kept") require.Less(t, results.TimeToFirstRequestByte, 30*time.Millisecond, "must never span the dead attempt and the redial") + require.Equal(t, 1, store.reqLastByte) + require.GreaterOrEqual(t, results.TimeToLastRequestByte, 4*time.Millisecond) + require.Less(t, results.TimeToLastRequestByte, 20*time.Millisecond, "must not pair the first attempt's first byte with the retry's last byte") + }) + + t.Run("does not record a last request byte when writing the request fails", func(t *testing.T) { + results, _, store := roundTripThroughHooks(t, true, func(ct *httptrace.ClientTrace) { + ct.GetConn("subgraph.local:443") + ct.GotConn(httptrace.GotConnInfo{}) + ct.WroteHeaderField("Host", []string{"subgraph.local"}) + time.Sleep(time.Millisecond) + ct.WroteRequest(httptrace.WroteRequestInfo{Err: errors.New("request body write failed")}) + }) + + require.Equal(t, 0, store.reqLastByte) + require.Zero(t, results.TimeToLastRequestByte) }) t.Run("does record failed TLS handshakes and failed connects", func(t *testing.T) { @@ -419,4 +548,454 @@ func TestTraceInjectingRoundTripper(t *testing.T) { require.NotZero(t, results.TCPConnectDuration) require.NotZero(t, results.TLSHandshakeDuration) }) + + t.Run("measures the time to the last response byte once the body is fully read", func(t *testing.T) { + store := &recordingConnectionMetricStore{} + body := &pacedBody{ + chunks: [][]byte{[]byte("hello "), []byte("world")}, + delay: 10 * time.Millisecond, + } + rt := NewTraceInjectingRoundTripper( + &bodyReturningRoundTripper{ + fire: func(ct *httptrace.ClientTrace) { + ct.GetConn("subgraph.local:443") + ct.GotConn(httptrace.GotConnInfo{}) + ct.WroteHeaderField("Host", []string{"subgraph.local"}) + ct.WroteRequest(httptrace.WroteRequestInfo{}) + // Server wait belongs to TimeToFirstByte, not the + // first-to-last response transfer duration. + time.Sleep(40 * time.Millisecond) + ct.GotFirstResponseByte() + }, + body: body, + contentLength: -1, + }, + store, + func(ctx context.Context, req *http.Request) (*expr.Context, string) { + return &expr.Context{}, "employees" + }, + ) + + ctx := WithClientTraceResults(context.Background()) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, "https://subgraph.local/graphql", http.NoBody) + require.NoError(t, err) + + resp, err := rt.RoundTrip(req) + require.NoError(t, err) + + // Nothing is recorded until the caller consumes the response body. + require.Equal(t, 0, store.ttlb, "time to last byte must not be recorded before the body is read") + require.Zero(t, ClientTraceResultsFromContext(ctx).TimeToLastByte) + + read, err := io.ReadAll(resp.Body) + require.NoError(t, err) + require.NoError(t, resp.Body.Close()) + require.Equal(t, "hello world", string(read)) + require.True(t, body.closed, "the underlying body must still be closed through the wrapper") + + results := *ClientTraceResultsFromContext(ctx) + require.Equal(t, 1, store.ttlb, "time to last byte is recorded once, when the body is fully read") + require.GreaterOrEqual(t, results.TimeToLastByte, 19*time.Millisecond, "must span reading the whole body") + require.Less(t, results.TimeToLastByte, 40*time.Millisecond, "must exclude server wait before the first response byte") + require.Less(t, results.TimeToLastByte, results.TimeToFirstByte) + }) + + t.Run("records the time to the last response byte only once even if read and closed", func(t *testing.T) { + store := &recordingConnectionMetricStore{} + rt := NewTraceInjectingRoundTripper( + &bodyReturningRoundTripper{ + fire: func(ct *httptrace.ClientTrace) { + ct.GetConn("subgraph.local:443") + ct.GotConn(httptrace.GotConnInfo{}) + ct.WroteHeaderField("Host", []string{"subgraph.local"}) + ct.WroteRequest(httptrace.WroteRequestInfo{}) + ct.GotFirstResponseByte() + }, + body: &pacedBody{chunks: [][]byte{[]byte("data")}}, + contentLength: -1, + }, + store, + func(ctx context.Context, req *http.Request) (*expr.Context, string) { + return &expr.Context{}, "employees" + }, + ) + + ctx := WithClientTraceResults(context.Background()) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, "https://subgraph.local/graphql", http.NoBody) + require.NoError(t, err) + + resp, err := rt.RoundTrip(req) + require.NoError(t, err) + + // Read to EOF and then close: the EOF and the Close must not both record. + _, err = io.ReadAll(resp.Body) + require.NoError(t, err) + require.NoError(t, resp.Body.Close()) + + require.Equal(t, 1, store.ttlb, "reaching EOF and then closing must record the last byte only once") + }) + + t.Run("does not record the last response byte when the body is closed early", func(t *testing.T) { + store := &recordingConnectionMetricStore{} + body := &pacedBody{chunks: [][]byte{[]byte("first"), []byte("second")}} + rt := NewTraceInjectingRoundTripper( + &bodyReturningRoundTripper{ + fire: func(ct *httptrace.ClientTrace) { + ct.GetConn("subgraph.local:443") + ct.GotConn(httptrace.GotConnInfo{}) + ct.WroteHeaderField("Host", []string{"subgraph.local"}) + ct.WroteRequest(httptrace.WroteRequestInfo{}) + ct.GotFirstResponseByte() + }, + body: body, + contentLength: -1, + }, + store, + func(context.Context, *http.Request) (*expr.Context, string) { + return &expr.Context{}, "employees" + }, + ) + + ctx := WithClientTraceResults(context.Background()) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, "https://subgraph.local/graphql", http.NoBody) + require.NoError(t, err) + resp, err := rt.RoundTrip(req) + require.NoError(t, err) + + _, err = resp.Body.Read(make([]byte, len("first"))) + require.NoError(t, err) + require.NoError(t, resp.Body.Close()) + + require.True(t, body.closed) + require.Equal(t, 0, store.ttlb) + require.Zero(t, ClientTraceResultsFromContext(ctx).TimeToLastByte) + }) + + t.Run("does not record the last response byte after a body read error", func(t *testing.T) { + store := &recordingConnectionMetricStore{} + body := &failingBody{} + rt := NewTraceInjectingRoundTripper( + &bodyReturningRoundTripper{ + fire: func(ct *httptrace.ClientTrace) { + ct.GetConn("subgraph.local:443") + ct.GotConn(httptrace.GotConnInfo{}) + ct.WroteHeaderField("Host", []string{"subgraph.local"}) + ct.WroteRequest(httptrace.WroteRequestInfo{}) + ct.GotFirstResponseByte() + }, + body: body, + contentLength: -1, + }, + store, + func(context.Context, *http.Request) (*expr.Context, string) { + return &expr.Context{}, "employees" + }, + ) + + ctx := WithClientTraceResults(context.Background()) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, "https://subgraph.local/graphql", http.NoBody) + require.NoError(t, err) + resp, err := rt.RoundTrip(req) + require.NoError(t, err) + + _, err = io.ReadAll(resp.Body) + require.EqualError(t, err, "response body read failed") + require.NoError(t, resp.Body.Close()) + + require.True(t, body.closed) + require.Equal(t, 0, store.ttlb) + require.Zero(t, ClientTraceResultsFromContext(ctx).TimeToLastByte) + }) + + t.Run("records completion at Content-Length without requiring another EOF read", func(t *testing.T) { + store := &recordingConnectionMetricStore{} + body := &pacedBody{chunks: [][]byte{[]byte("data")}} + rt := NewTraceInjectingRoundTripper( + &bodyReturningRoundTripper{ + fire: func(ct *httptrace.ClientTrace) { + ct.GetConn("subgraph.local:443") + ct.GotConn(httptrace.GotConnInfo{}) + ct.WroteHeaderField("Host", []string{"subgraph.local"}) + ct.WroteRequest(httptrace.WroteRequestInfo{}) + ct.GotFirstResponseByte() + }, + body: body, + contentLength: int64(len("data")), + }, + store, + func(context.Context, *http.Request) (*expr.Context, string) { + return &expr.Context{}, "employees" + }, + ) + + ctx := WithClientTraceResults(context.Background()) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, "https://subgraph.local/graphql", http.NoBody) + require.NoError(t, err) + resp, err := rt.RoundTrip(req) + require.NoError(t, err) + + n, err := resp.Body.Read(make([]byte, len("data"))) + require.NoError(t, err) + require.Equal(t, len("data"), n) + require.Equal(t, 1, store.ttlb) + require.NoError(t, resp.Body.Close()) + require.Equal(t, 1, store.ttlb) + }) + + t.Run("records no-body responses when their headers have been read", func(t *testing.T) { + tests := []struct { + name string + method string + statusCode int + body io.ReadCloser + contentLength int64 + }{ + { + name: "HEAD with representation length", + method: http.MethodHead, + statusCode: http.StatusOK, + body: &pacedBody{}, + contentLength: 128, + }, + { + name: "no content", + method: http.MethodPost, + statusCode: http.StatusNoContent, + body: &pacedBody{}, + contentLength: -1, + }, + { + name: "not modified", + method: http.MethodGet, + statusCode: http.StatusNotModified, + body: &pacedBody{}, + contentLength: -1, + }, + { + name: "explicit zero length", + method: http.MethodPost, + statusCode: http.StatusOK, + body: http.NoBody, + contentLength: 0, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + store := &recordingConnectionMetricStore{} + rt := NewTraceInjectingRoundTripper( + &bodyReturningRoundTripper{ + fire: func(ct *httptrace.ClientTrace) { + ct.GetConn("subgraph.local:443") + ct.GotConn(httptrace.GotConnInfo{}) + ct.WroteHeaderField("Host", []string{"subgraph.local"}) + ct.WroteRequest(httptrace.WroteRequestInfo{}) + ct.GotFirstResponseByte() + }, + body: tt.body, + statusCode: tt.statusCode, + contentLength: tt.contentLength, + }, + store, + func(context.Context, *http.Request) (*expr.Context, string) { + return &expr.Context{}, "employees" + }, + ) + + ctx := WithClientTraceResults(context.Background()) + req, err := http.NewRequestWithContext(ctx, tt.method, "https://subgraph.local/graphql", http.NoBody) + require.NoError(t, err) + resp, err := rt.RoundTrip(req) + require.NoError(t, err) + + require.Equal(t, tt.body, resp.Body) + require.Equal(t, 1, store.ttlb) + require.Greater(t, ClientTraceResultsFromContext(ctx).TimeToLastByte, time.Duration(0)) + require.NoError(t, resp.Body.Close()) + require.Equal(t, 1, store.ttlb) + }) + } + }) + + t.Run("does not record a short EOF against Content-Length", func(t *testing.T) { + store := &recordingConnectionMetricStore{} + rt := NewTraceInjectingRoundTripper( + &bodyReturningRoundTripper{ + fire: func(ct *httptrace.ClientTrace) { + ct.GetConn("subgraph.local:443") + ct.GotConn(httptrace.GotConnInfo{}) + ct.WroteHeaderField("Host", []string{"subgraph.local"}) + ct.WroteRequest(httptrace.WroteRequestInfo{}) + ct.GotFirstResponseByte() + }, + body: &pacedBody{chunks: [][]byte{[]byte("short")}}, + contentLength: int64(len("longer-body")), + }, + store, + func(context.Context, *http.Request) (*expr.Context, string) { + return &expr.Context{}, "employees" + }, + ) + + ctx := WithClientTraceResults(context.Background()) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, "https://subgraph.local/graphql", http.NoBody) + require.NoError(t, err) + resp, err := rt.RoundTrip(req) + require.NoError(t, err) + + _, err = io.ReadAll(resp.Body) + require.NoError(t, err) + require.NoError(t, resp.Body.Close()) + + require.Equal(t, 0, store.ttlb) + require.Zero(t, ClientTraceResultsFromContext(ctx).TimeToLastByte) + }) + + t.Run("preserves upgraded response bodies and excludes subscriptions", func(t *testing.T) { + tests := []struct { + name string + statusCode int + header http.Header + }{ + { + name: "websocket upgrade", + statusCode: http.StatusSwitchingProtocols, + }, + { + name: "event stream", + statusCode: http.StatusOK, + header: http.Header{"Content-Type": []string{"text/event-stream; charset=utf-8"}}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + store := &recordingConnectionMetricStore{} + body := &readWriteBody{pacedBody: &pacedBody{}} + rt := NewTraceInjectingRoundTripper( + &bodyReturningRoundTripper{ + fire: func(ct *httptrace.ClientTrace) { + ct.GetConn("subgraph.local:443") + ct.GotConn(httptrace.GotConnInfo{}) + ct.WroteHeaderField("Host", []string{"subgraph.local"}) + ct.WroteRequest(httptrace.WroteRequestInfo{}) + ct.GotFirstResponseByte() + }, + body: body, + statusCode: tt.statusCode, + header: tt.header, + contentLength: -1, + }, + store, + func(context.Context, *http.Request) (*expr.Context, string) { + return &expr.Context{}, "employees" + }, + ) + + req, err := http.NewRequestWithContext(WithClientTraceResults(context.Background()), http.MethodPost, "https://subgraph.local/graphql", http.NoBody) + require.NoError(t, err) + resp, err := rt.RoundTrip(req) + require.NoError(t, err) + + require.Same(t, body, resp.Body) + _, ok := resp.Body.(io.ReadWriteCloser) + require.True(t, ok, "upgraded response capabilities must remain intact") + require.NoError(t, resp.Body.Close()) + require.Equal(t, 0, store.ttlb) + }) + } + }) + + t.Run("records a successful request write that completes after RoundTrip returns", func(t *testing.T) { + store := &recordingConnectionMetricStore{} + writeDone := make(chan struct{}) + base := roundTripperFunc(func(req *http.Request) (*http.Response, error) { + ct := httptrace.ContextClientTrace(req.Context()) + ct.GetConn("subgraph.local:443") + ct.GotConn(httptrace.GotConnInfo{}) + ct.GotFirstResponseByte() + go func() { + defer close(writeDone) + ct.WroteHeaderField("Host", []string{"subgraph.local"}) + time.Sleep(5 * time.Millisecond) + ct.WroteRequest(httptrace.WroteRequestInfo{}) + }() + return &http.Response{ + StatusCode: http.StatusOK, + Body: &pacedBody{chunks: [][]byte{[]byte("response")}, delay: 10 * time.Millisecond}, + Header: make(http.Header), + ContentLength: -1, + Request: req, + }, nil + }) + rt := NewTraceInjectingRoundTripper( + base, + store, + func(context.Context, *http.Request) (*expr.Context, string) { + return &expr.Context{}, "employees" + }, + ) + + ctx := WithClientTraceResults(context.Background()) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, "https://subgraph.local/graphql", http.NoBody) + require.NoError(t, err) + resp, err := rt.RoundTrip(req) + require.NoError(t, err) + _, err = io.ReadAll(resp.Body) + require.NoError(t, err) + require.NoError(t, resp.Body.Close()) + <-writeDone + + require.Equal(t, 1, store.reqLastByte) + require.GreaterOrEqual(t, ClientTraceResultsFromContext(ctx).TimeToLastRequestByte, 4*time.Millisecond) + }) + + t.Run("records a request metric when the successful write completes after the response", func(t *testing.T) { + store := &recordingConnectionMetricStore{} + startWrite := make(chan struct{}) + writeDone := make(chan struct{}) + base := roundTripperFunc(func(req *http.Request) (*http.Response, error) { + ct := httptrace.ContextClientTrace(req.Context()) + ct.GetConn("subgraph.local:443") + ct.GotConn(httptrace.GotConnInfo{}) + ct.GotFirstResponseByte() + go func() { + defer close(writeDone) + <-startWrite + ct.WroteHeaderField("Host", []string{"subgraph.local"}) + time.Sleep(time.Millisecond) + ct.WroteRequest(httptrace.WroteRequestInfo{}) + }() + return &http.Response{ + StatusCode: http.StatusOK, + Body: &pacedBody{chunks: [][]byte{[]byte("response")}}, + Header: make(http.Header), + ContentLength: -1, + Request: req, + }, nil + }) + rt := NewTraceInjectingRoundTripper( + base, + store, + func(context.Context, *http.Request) (*expr.Context, string) { + return &expr.Context{}, "employees" + }, + ) + + ctx := WithClientTraceResults(context.Background()) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, "https://subgraph.local/graphql", http.NoBody) + require.NoError(t, err) + resp, err := rt.RoundTrip(req) + require.NoError(t, err) + _, err = io.ReadAll(resp.Body) + require.NoError(t, err) + require.NoError(t, resp.Body.Close()) + + close(startWrite) + <-writeDone + + require.Equal(t, 1, store.reqLastByte) + require.Greater(t, store.reqLastByteMs, 0.0) + require.Zero(t, ClientTraceResultsFromContext(ctx).TimeToLastRequestByte, "late callbacks must not race the completed fetch's expression results") + }) } diff --git a/router/pkg/config/config.schema.json b/router/pkg/config/config.schema.json index 7a9ca9ab96..902d5e9401 100644 --- a/router/pkg/config/config.schema.json +++ b/router/pkg/config/config.schema.json @@ -1496,7 +1496,7 @@ "enabled": { "type": "boolean", "default": false, - "description": "Enable per-request subgraph HTTP phase metrics, including DNS lookup, TCP connect, TLS handshake, time-to-first-request-byte, and time-to-first-byte histograms. The default value is false." + "description": "Enable per-request subgraph HTTP phase metrics, including DNS lookup, TCP connect, TLS handshake, time-to-first-request-byte, time-to-last-request-byte, time-to-first-byte, and time-to-last-byte histograms. The default value is false." } } }, @@ -1682,7 +1682,7 @@ "enabled": { "type": "boolean", "default": false, - "description": "Enable per-request subgraph HTTP phase metrics, including DNS lookup, TCP connect, TLS handshake, time-to-first-request-byte, and time-to-first-byte histograms. The default value is false." + "description": "Enable per-request subgraph HTTP phase metrics, including DNS lookup, TCP connect, TLS handshake, time-to-first-request-byte, time-to-last-request-byte, time-to-first-byte, and time-to-last-byte histograms. The default value is false." } } }, diff --git a/router/pkg/metric/connection_measurements.go b/router/pkg/metric/connection_measurements.go index cd40b81004..3b03288449 100644 --- a/router/pkg/metric/connection_measurements.go +++ b/router/pkg/metric/connection_measurements.go @@ -16,7 +16,9 @@ const ( tcpConnectDuration = "router.http.client.tcp_connect_duration" tlsHandshakeDuration = "router.http.client.tls_handshake_duration" timeToFirstRequestByte = "router.http.client.time_to_first_request_byte" + timeToLastRequestByte = "router.http.client.time_to_last_request_byte" timeToFirstByte = "router.http.client.time_to_first_byte" + timeToLastByte = "router.http.client.time_to_last_byte" ) var ( @@ -53,10 +55,20 @@ var ( otelmetric.WithDescription("Time from the start of the HTTP attempt to the first request byte written for outgoing subgraph requests"), } + timeToLastRequestByteOptions = []otelmetric.Float64HistogramOption{ + otelmetric.WithUnit("ms"), + otelmetric.WithDescription("Time from the first request byte write event to the successful request write completion event for outgoing subgraph requests"), + } + timeToFirstByteOptions = []otelmetric.Float64HistogramOption{ otelmetric.WithUnit("ms"), otelmetric.WithDescription("Time from request write completion to first response byte from subgraph"), } + + timeToLastByteOptions = []otelmetric.Float64HistogramOption{ + otelmetric.WithUnit("ms"), + otelmetric.WithDescription("Time from the first response byte to the last response byte from subgraph"), + } ) type connectionInstruments struct { @@ -69,7 +81,9 @@ type connectionInstruments struct { tcpConnectDuration otelmetric.Float64Histogram tlsHandshakeDuration otelmetric.Float64Histogram timeToFirstRequestByte otelmetric.Float64Histogram + timeToLastRequestByte otelmetric.Float64Histogram timeToFirstByte otelmetric.Float64Histogram + timeToLastByte otelmetric.Float64Histogram } func newConnectionInstruments(meter otelmetric.Meter, enhancedConnectionStats bool) (*connectionInstruments, error) { @@ -119,9 +133,15 @@ func newConnectionInstruments(meter otelmetric.Meter, enhancedConnectionStats bo if ci.timeToFirstRequestByte, err = meter.Float64Histogram(timeToFirstRequestByte, timeToFirstRequestByteOptions...); err != nil { return nil, fmt.Errorf("failed to create time to first request byte histogram: %w", err) } + if ci.timeToLastRequestByte, err = meter.Float64Histogram(timeToLastRequestByte, timeToLastRequestByteOptions...); err != nil { + return nil, fmt.Errorf("failed to create time to last request byte histogram: %w", err) + } if ci.timeToFirstByte, err = meter.Float64Histogram(timeToFirstByte, timeToFirstByteOptions...); err != nil { return nil, fmt.Errorf("failed to create time to first byte histogram: %w", err) } + if ci.timeToLastByte, err = meter.Float64Histogram(timeToLastByte, timeToLastByteOptions...); err != nil { + return nil, fmt.Errorf("failed to create time to last byte histogram: %w", err) + } return ci, nil } diff --git a/router/pkg/metric/connection_metric_store.go b/router/pkg/metric/connection_metric_store.go index 68af12e92f..9b971aceca 100644 --- a/router/pkg/metric/connection_metric_store.go +++ b/router/pkg/metric/connection_metric_store.go @@ -26,6 +26,14 @@ type ConnectionMetricProvider interface { Shutdown() error } +// lastByteMetricProvider is an optional extension implemented by the built-in +// OTLP and Prometheus providers. Keeping it separate avoids adding required +// methods to the exported ConnectionMetricProvider interface. +type lastByteMetricProvider interface { + MeasureTimeToLastRequestByte(ctx context.Context, duration float64, opts ...otelmetric.RecordOption) + MeasureTimeToLastByte(ctx context.Context, duration float64, opts ...otelmetric.RecordOption) +} + // ConnectionMetricStore is the interface for connection and pool metrics only. type ConnectionMetricStore interface { MeasureConnectionAcquireDuration(ctx context.Context, duration float64, attrs ...attribute.KeyValue) @@ -37,6 +45,14 @@ type ConnectionMetricStore interface { Shutdown(ctx context.Context) error } +// LastByteMetricStore is an optional extension for first-to-last-byte transfer +// metrics. Implementations of ConnectionMetricStore that predate these metrics +// remain source-compatible. +type LastByteMetricStore interface { + MeasureTimeToLastRequestByte(ctx context.Context, duration float64, attrs ...attribute.KeyValue) + MeasureTimeToLastByte(ctx context.Context, duration float64, attrs ...attribute.KeyValue) +} + type ConnectionMetrics struct { baseAttributes []attribute.KeyValue logger *zap.Logger @@ -121,12 +137,32 @@ func (c *ConnectionMetrics) MeasureTimeToFirstRequestByte(ctx context.Context, d c.promConnectionMetrics.MeasureTimeToFirstRequestByte(ctx, duration, opts) } +func (c *ConnectionMetrics) MeasureTimeToLastRequestByte(ctx context.Context, duration float64, attrs ...attribute.KeyValue) { + opts := c.recordOpts(attrs) + if provider, ok := c.otlpConnectionMetrics.(lastByteMetricProvider); ok { + provider.MeasureTimeToLastRequestByte(ctx, duration, opts) + } + if provider, ok := c.promConnectionMetrics.(lastByteMetricProvider); ok { + provider.MeasureTimeToLastRequestByte(ctx, duration, opts) + } +} + func (c *ConnectionMetrics) MeasureTimeToFirstByte(ctx context.Context, duration float64, attrs ...attribute.KeyValue) { opts := c.recordOpts(attrs) c.otlpConnectionMetrics.MeasureTimeToFirstByte(ctx, duration, opts) c.promConnectionMetrics.MeasureTimeToFirstByte(ctx, duration, opts) } +func (c *ConnectionMetrics) MeasureTimeToLastByte(ctx context.Context, duration float64, attrs ...attribute.KeyValue) { + opts := c.recordOpts(attrs) + if provider, ok := c.otlpConnectionMetrics.(lastByteMetricProvider); ok { + provider.MeasureTimeToLastByte(ctx, duration, opts) + } + if provider, ok := c.promConnectionMetrics.(lastByteMetricProvider); ok { + provider.MeasureTimeToLastByte(ctx, duration, opts) + } +} + func (c *ConnectionMetrics) recordOpts(attrs []attribute.KeyValue) otelmetric.RecordOption { copied := append([]attribute.KeyValue{}, c.baseAttributes...) return otelmetric.WithAttributes(append(copied, attrs...)...) diff --git a/router/pkg/metric/noop_connection_metrics.go b/router/pkg/metric/noop_connection_metrics.go index 61f8d6cd3e..15fd4b4ef8 100644 --- a/router/pkg/metric/noop_connection_metrics.go +++ b/router/pkg/metric/noop_connection_metrics.go @@ -28,9 +28,15 @@ func (h *noopConnectionMetricProvider) MeasureTLSHandshakeDuration(ctx context.C func (h *noopConnectionMetricProvider) MeasureTimeToFirstRequestByte(ctx context.Context, duration float64, opts ...otelmetric.RecordOption) { } +func (h *noopConnectionMetricProvider) MeasureTimeToLastRequestByte(ctx context.Context, duration float64, opts ...otelmetric.RecordOption) { +} + func (h *noopConnectionMetricProvider) MeasureTimeToFirstByte(ctx context.Context, duration float64, opts ...otelmetric.RecordOption) { } +func (h *noopConnectionMetricProvider) MeasureTimeToLastByte(ctx context.Context, duration float64, opts ...otelmetric.RecordOption) { +} + func (h *noopConnectionMetricProvider) Flush(ctx context.Context) error { return nil } @@ -51,7 +57,11 @@ func (h *NoopConnectionMetricStore) MeasureTLSHandshakeDuration(ctx context.Cont } func (h *NoopConnectionMetricStore) MeasureTimeToFirstRequestByte(ctx context.Context, duration float64, attrs ...attribute.KeyValue) { } +func (h *NoopConnectionMetricStore) MeasureTimeToLastRequestByte(ctx context.Context, duration float64, attrs ...attribute.KeyValue) { +} func (h *NoopConnectionMetricStore) MeasureTimeToFirstByte(ctx context.Context, duration float64, attrs ...attribute.KeyValue) { } +func (h *NoopConnectionMetricStore) MeasureTimeToLastByte(ctx context.Context, duration float64, attrs ...attribute.KeyValue) { +} func (h *NoopConnectionMetricStore) Flush(ctx context.Context) error { return nil } func (h *NoopConnectionMetricStore) Shutdown(ctx context.Context) error { return nil } diff --git a/router/pkg/metric/oltp_connection_metric_store.go b/router/pkg/metric/oltp_connection_metric_store.go index dde622f76d..0a1c90303e 100644 --- a/router/pkg/metric/oltp_connection_metric_store.go +++ b/router/pkg/metric/oltp_connection_metric_store.go @@ -117,12 +117,24 @@ func (h *otlpConnectionMetrics) MeasureTimeToFirstRequestByte(ctx context.Contex } } +func (h *otlpConnectionMetrics) MeasureTimeToLastRequestByte(ctx context.Context, duration float64, opts ...otelmetric.RecordOption) { + if h.instruments.timeToLastRequestByte != nil { + h.instruments.timeToLastRequestByte.Record(ctx, duration, opts...) + } +} + func (h *otlpConnectionMetrics) MeasureTimeToFirstByte(ctx context.Context, duration float64, opts ...otelmetric.RecordOption) { if h.instruments.timeToFirstByte != nil { h.instruments.timeToFirstByte.Record(ctx, duration, opts...) } } +func (h *otlpConnectionMetrics) MeasureTimeToLastByte(ctx context.Context, duration float64, opts ...otelmetric.RecordOption) { + if h.instruments.timeToLastByte != nil { + h.instruments.timeToLastByte.Record(ctx, duration, opts...) + } +} + func (h *otlpConnectionMetrics) Flush(ctx context.Context) error { return h.meterProvider.ForceFlush(ctx) } diff --git a/router/pkg/metric/prom_connection_metric_store.go b/router/pkg/metric/prom_connection_metric_store.go index 8ce1dfcc8b..4aa489117a 100644 --- a/router/pkg/metric/prom_connection_metric_store.go +++ b/router/pkg/metric/prom_connection_metric_store.go @@ -116,12 +116,24 @@ func (m *promConnectionMetrics) MeasureTimeToFirstRequestByte(ctx context.Contex } } +func (m *promConnectionMetrics) MeasureTimeToLastRequestByte(ctx context.Context, duration float64, opts ...otelmetric.RecordOption) { + if m.instruments.timeToLastRequestByte != nil { + m.instruments.timeToLastRequestByte.Record(ctx, duration, opts...) + } +} + func (m *promConnectionMetrics) MeasureTimeToFirstByte(ctx context.Context, duration float64, opts ...otelmetric.RecordOption) { if m.instruments.timeToFirstByte != nil { m.instruments.timeToFirstByte.Record(ctx, duration, opts...) } } +func (m *promConnectionMetrics) MeasureTimeToLastByte(ctx context.Context, duration float64, opts ...otelmetric.RecordOption) { + if m.instruments.timeToLastByte != nil { + m.instruments.timeToLastByte.Record(ctx, duration, opts...) + } +} + func (m *promConnectionMetrics) Flush(ctx context.Context) error { return m.meterProvider.ForceFlush(ctx) } From 07a33ad646757e71806a776b9e9b40ee72dec864 Mon Sep 17 00:00:00 2001 From: Alessandro Pagnin Date: Thu, 6 Aug 2026 12:17:00 +0200 Subject: [PATCH 2/3] fix: better doc --- docs-website/router/configuration/template-expressions.mdx | 2 +- docs-website/router/metrics-and-monitoring.mdx | 2 +- .../metrics-and-monitoring/prometheus-metric-reference.mdx | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/docs-website/router/configuration/template-expressions.mdx b/docs-website/router/configuration/template-expressions.mdx index 8436105d29..f32f1ef386 100644 --- a/docs-website/router/configuration/template-expressions.mdx +++ b/docs-website/router/configuration/template-expressions.mdx @@ -217,7 +217,7 @@ Client trace values describe the individual subgraph fetch that is being logged - `subgraph.request.clientTrace.tcpConnectDuration` (time.Duration): The duration of the TCP connect (dial) for the subgraph request. It is 0 when an existing connection is reused. In case of retries, the value of the last attempt that dialed is recorded. - `subgraph.request.clientTrace.tlsHandshakeDuration` (time.Duration): The duration of the TLS handshake for the subgraph request. It is 0 when an existing connection is reused or the subgraph is reached over plaintext. In case of retries, the value of the last attempt that performed a handshake is recorded. - `subgraph.request.clientTrace.timeToFirstRequestByte` (time.Duration): The duration from the start of the HTTP attempt to writing the first request byte to the subgraph. It includes acquiring the connection. In case of retries, the value of the last attempt that wrote a request byte is recorded. -- `subgraph.request.clientTrace.timeToLastRequestByte` (time.Duration): The duration between the first request-header write callback and a successful request-complete callback from Go's HTTP transport. This approximates the time from the first to the last request byte. The value is 0 unless both callbacks are observed and the request-complete callback reports no error. If request completion is reported only after response processing has finished, the histogram still receives a sample but the expression value remains 0. In case of retries, the value of the last attempt with both callbacks is recorded. The value is also 0 for single-flight followers because they do not write a separate subgraph request. +- `subgraph.request.clientTrace.timeToLastRequestByte` (time.Duration): The duration between the first request-header write callback and a successful request-complete callback from Go's HTTP transport. This approximates the time from the first to the last request byte. The value is 0 unless both callbacks are observed and the request-complete callback reports no error. In case of retries, the value of the last attempt with both callbacks is recorded. The value is also 0 for single-flight followers because they do not write a separate subgraph request. - `subgraph.request.clientTrace.timeToFirstByte` (time.Duration): The duration from completing the request write to receiving the first response byte from the subgraph. In case of retries, only the last attempt is recorded. - `subgraph.request.clientTrace.timeToLastByte` (time.Duration): The duration from receiving the first response byte to consuming the last response byte. The value is 0 unless the first response byte is observed and the full response body is consumed, either to EOF or through the declared `Content-Length`. A response that cannot carry a body completes when its headers have been read. The value remains 0 for bodies that are closed early or end in a read error, upgraded or streaming subscription responses, and single-flight followers that do not independently consume the origin response body. In case of retries, the value of the last attempt whose response is fully consumed is recorded. diff --git a/docs-website/router/metrics-and-monitoring.mdx b/docs-website/router/metrics-and-monitoring.mdx index 90de84a429..5a3ebe4fad 100644 --- a/docs-website/router/metrics-and-monitoring.mdx +++ b/docs-website/router/metrics-and-monitoring.mdx @@ -377,7 +377,7 @@ telemetry: * `router.http.client.time_to_first_byte`: Histogram (ms) of the time from completing the request write to receiving the first response byte from the subgraph. -* `router.http.client.time_to_last_byte`: Histogram (ms) of the duration from receiving the first response byte to consuming the last response byte. Recorded only when a first response byte is observed and the full body is consumed, either to EOF or through the declared `Content-Length`. A response that cannot carry a body completes when its headers have been read. No sample is emitted when the body is closed early, when reading the body ends with an error, or for upgraded or streaming subscription responses. +* `router.http.client.time_to_last_byte`: Histogram (ms) of the duration from receiving the first response byte to consuming the last response byte. Recorded only when a first response byte is observed and the full body is consumed, either to EOF or through the declared `Content-Length`. A response that cannot carry a body completes when its headers have been read. ## Custom Attributes diff --git a/docs-website/router/metrics-and-monitoring/prometheus-metric-reference.mdx b/docs-website/router/metrics-and-monitoring/prometheus-metric-reference.mdx index 81b4169c2d..9f219c6f17 100644 --- a/docs-website/router/metrics-and-monitoring/prometheus-metric-reference.mdx +++ b/docs-website/router/metrics-and-monitoring/prometheus-metric-reference.mdx @@ -245,7 +245,7 @@ telemetry: * `router_http_client_time_to_first_byte`: The time in ms from completing the request write to receiving the first response byte from the subgraph. -* `router_http_client_time_to_last_byte`: The duration in ms from receiving the first response byte to consuming the last response byte. Recorded only when a first response byte is observed and the full body is consumed, either to EOF or through the declared `Content-Length`. A response that cannot carry a body completes when its headers have been read. No sample is emitted when the body is closed early, when reading the body ends with an error, or for upgraded or streaming subscription responses. +* `router_http_client_time_to_last_byte`: The duration in ms from receiving the first response byte to consuming the last response byte. Recorded only when a first response byte is observed and the full body is consumed, either to EOF or through the declared `Content-Length`. A response that cannot carry a body completes when its headers have been read. ### Go Runtime Metrics From 2b9ebc176b90e1c8b98374218628bdddc129b923 Mon Sep 17 00:00:00 2001 From: Alessandro Pagnin Date: Fri, 7 Aug 2026 17:19:23 +0200 Subject: [PATCH 3/3] fix: remove useless structures --- router/internal/traceclient/traceclient.go | 66 +++++++++---------- .../internal/traceclient/traceclient_test.go | 2 +- router/pkg/metric/connection_metric_store.go | 36 +++------- 3 files changed, 40 insertions(+), 64 deletions(-) diff --git a/router/internal/traceclient/traceclient.go b/router/internal/traceclient/traceclient.go index a515740a99..12fecc2a24 100644 --- a/router/internal/traceclient/traceclient.go +++ b/router/internal/traceclient/traceclient.go @@ -257,6 +257,9 @@ func (t *TraceInjectingRoundTripper) RoundTrip(req *http.Request) (*http.Respons trip, err := t.base.RoundTrip(req) recorder := t.processConnectionMetrics(req.Context(), req, ec) + if recorder == nil { + return trip, err + } // httptrace has no "last response byte" callback: the last byte is only // observable once the caller has fully consumed the response body. Wrap @@ -265,24 +268,22 @@ func (t *TraceInjectingRoundTripper) RoundTrip(req *http.Request) (*http.Respons // This is always before the subgraph access log is written, because the // engine fully reads and closes the body during the fetch load phase, which // completes before the log is emitted. - if recorder != nil { - switch { - case err != nil || !shouldMeasureResponseTransfer(trip): - // Upgraded and streaming responses are intentionally excluded. In - // particular, leaving HTTP 101 bodies untouched preserves their - // io.ReadWriteCloser contract for WebSocket clients. - recorder.cancel() - case responseHasNoBody(req, trip): - // RoundTrip returns after the response headers are read. For HEAD - // and other responses that cannot carry a body, that is also the - // point at which the last response byte has been consumed. - recorder.fire() - default: - trip.Body = &timedResponseBody{ - ReadCloser: trip.Body, - recorder: recorder, - contentLength: trip.ContentLength, - } + switch { + case err != nil || !shouldMeasureResponseTransfer(trip): + // Upgraded and streaming responses are intentionally excluded. In + // particular, leaving HTTP 101 bodies untouched preserves their + // io.ReadWriteCloser contract for WebSocket clients. + recorder.cancel() + case responseHasNoBody(trip): + // RoundTrip returns after the response headers are read. For responses + // that cannot carry a body, that is also the point at which the last + // response byte has been consumed. + recorder.fire() + default: + trip.Body = &timedResponseBody{ + ReadCloser: trip.Body, + recorder: recorder, + contentLength: trip.ContentLength, } } @@ -300,8 +301,8 @@ func shouldMeasureResponseTransfer(resp *http.Response) bool { return true } -func responseHasNoBody(req *http.Request, resp *http.Response) bool { - if req.Method == http.MethodHead || resp.Body == http.NoBody || resp.ContentLength == 0 { +func responseHasNoBody(resp *http.Response) bool { + if resp.Body == http.NoBody || resp.ContentLength == 0 { return true } return resp.StatusCode >= 100 && resp.StatusCode <= 199 || @@ -544,18 +545,15 @@ func (t *TraceInjectingRoundTripper) processConnectionMetrics(ctx context.Contex ) } - lastByteMetricStore, _ := t.connectionMetricStore.(metric.LastByteMetricStore) requestRecorder := &requestByteRecorder{ resultsOpen: true, results: results, recordMetric: func(duration time.Duration) { - if lastByteMetricStore != nil { - lastByteMetricStore.MeasureTimeToLastRequestByte( - ctx, - msFromDuration(duration), - serverAttributes..., - ) - } + t.connectionMetricStore.MeasureTimeToLastRequestByte( + ctx, + msFromDuration(duration), + serverAttributes..., + ) }, } recorder := &lastByteRecorder{ @@ -563,13 +561,11 @@ func (t *TraceInjectingRoundTripper) processConnectionMetrics(ctx context.Contex request: requestRecorder, recordResponse: func(duration time.Duration) { results.TimeToLastByte = duration - if lastByteMetricStore != nil { - lastByteMetricStore.MeasureTimeToLastByte( - ctx, - msFromDuration(duration), - serverAttributes..., - ) - } + t.connectionMetricStore.MeasureTimeToLastByte( + ctx, + msFromDuration(duration), + serverAttributes..., + ) }, } diff --git a/router/internal/traceclient/traceclient_test.go b/router/internal/traceclient/traceclient_test.go index 17e4d9e538..1f71ae4f8d 100644 --- a/router/internal/traceclient/traceclient_test.go +++ b/router/internal/traceclient/traceclient_test.go @@ -754,7 +754,7 @@ func TestTraceInjectingRoundTripper(t *testing.T) { name: "HEAD with representation length", method: http.MethodHead, statusCode: http.StatusOK, - body: &pacedBody{}, + body: http.NoBody, contentLength: 128, }, { diff --git a/router/pkg/metric/connection_metric_store.go b/router/pkg/metric/connection_metric_store.go index f2effc3a91..646e4fb3f3 100644 --- a/router/pkg/metric/connection_metric_store.go +++ b/router/pkg/metric/connection_metric_store.go @@ -23,16 +23,10 @@ type ConnectionMetricProvider interface { MeasureTCPConnectDuration(ctx context.Context, duration float64, opts ...otelmetric.RecordOption) MeasureTLSHandshakeDuration(ctx context.Context, duration float64, opts ...otelmetric.RecordOption) MeasureTimeToFirstRequestByte(ctx context.Context, duration float64, opts ...otelmetric.RecordOption) - MeasureTimeToFirstByte(ctx context.Context, duration float64, opts ...otelmetric.RecordOption) - Shutdown() error -} - -// lastByteMetricProvider is an optional extension implemented by the built-in -// OTLP and Prometheus providers. Keeping it separate avoids adding required -// methods to the exported ConnectionMetricProvider interface. -type lastByteMetricProvider interface { MeasureTimeToLastRequestByte(ctx context.Context, duration float64, opts ...otelmetric.RecordOption) + MeasureTimeToFirstByte(ctx context.Context, duration float64, opts ...otelmetric.RecordOption) MeasureTimeToLastByte(ctx context.Context, duration float64, opts ...otelmetric.RecordOption) + Shutdown() error } // ConnectionMetricStore is the interface for connection and pool metrics only. @@ -42,16 +36,10 @@ type ConnectionMetricStore interface { MeasureTCPConnectDuration(ctx context.Context, duration float64, attrs ...attribute.KeyValue) MeasureTLSHandshakeDuration(ctx context.Context, duration float64, attrs ...attribute.KeyValue) MeasureTimeToFirstRequestByte(ctx context.Context, duration float64, attrs ...attribute.KeyValue) - MeasureTimeToFirstByte(ctx context.Context, duration float64, attrs ...attribute.KeyValue) - Shutdown(ctx context.Context) error -} - -// LastByteMetricStore is an optional extension for first-to-last-byte transfer -// metrics. Implementations of ConnectionMetricStore that predate these metrics -// remain source-compatible. -type LastByteMetricStore interface { MeasureTimeToLastRequestByte(ctx context.Context, duration float64, attrs ...attribute.KeyValue) + MeasureTimeToFirstByte(ctx context.Context, duration float64, attrs ...attribute.KeyValue) MeasureTimeToLastByte(ctx context.Context, duration float64, attrs ...attribute.KeyValue) + Shutdown(ctx context.Context) error } type ConnectionMetrics struct { @@ -141,12 +129,8 @@ func (c *ConnectionMetrics) MeasureTimeToFirstRequestByte(ctx context.Context, d func (c *ConnectionMetrics) MeasureTimeToLastRequestByte(ctx context.Context, duration float64, attrs ...attribute.KeyValue) { opts := c.recordOpts(attrs) - if provider, ok := c.otlpConnectionMetrics.(lastByteMetricProvider); ok { - provider.MeasureTimeToLastRequestByte(ctx, duration, opts) - } - if provider, ok := c.promConnectionMetrics.(lastByteMetricProvider); ok { - provider.MeasureTimeToLastRequestByte(ctx, duration, opts) - } + c.otlpConnectionMetrics.MeasureTimeToLastRequestByte(ctx, duration, opts) + c.promConnectionMetrics.MeasureTimeToLastRequestByte(ctx, duration, opts) } func (c *ConnectionMetrics) MeasureTimeToFirstByte(ctx context.Context, duration float64, attrs ...attribute.KeyValue) { @@ -157,12 +141,8 @@ func (c *ConnectionMetrics) MeasureTimeToFirstByte(ctx context.Context, duration func (c *ConnectionMetrics) MeasureTimeToLastByte(ctx context.Context, duration float64, attrs ...attribute.KeyValue) { opts := c.recordOpts(attrs) - if provider, ok := c.otlpConnectionMetrics.(lastByteMetricProvider); ok { - provider.MeasureTimeToLastByte(ctx, duration, opts) - } - if provider, ok := c.promConnectionMetrics.(lastByteMetricProvider); ok { - provider.MeasureTimeToLastByte(ctx, duration, opts) - } + c.otlpConnectionMetrics.MeasureTimeToLastByte(ctx, duration, opts) + c.promConnectionMetrics.MeasureTimeToLastByte(ctx, duration, opts) } func (c *ConnectionMetrics) recordOpts(attrs []attribute.KeyValue) otelmetric.RecordOption {