mysten_network/request_log/
service.rs

1// Copyright (c) Mysten Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4use std::task::{Context, Poll};
5
6use http::Request;
7use prost_reflect::DescriptorPool;
8use tower::Service;
9use tracing::trace_span;
10
11use super::TARGET;
12use super::body::{CaptureState, RequestLogBody};
13
14/// Middleware that captures decoded gRPC request messages on the [`TARGET`] tracing target.
15///
16/// See the [module docs](super) for how captures are enabled and collected.
17#[derive(Clone)]
18pub struct GrpcRequestLog<S> {
19    inner: S,
20    pool: DescriptorPool,
21    max_captured_message_size: usize,
22}
23
24impl<S> GrpcRequestLog<S> {
25    pub(crate) fn new(inner: S, pool: DescriptorPool, max_captured_message_size: usize) -> Self {
26        Self {
27            inner,
28            pool,
29            max_captured_message_size,
30        }
31    }
32
33    /// Decide whether to capture `request`, and if so set up the state the body wrapper needs:
34    /// the span carrying the `service`/`method` fields that `EnvFilter` span-field directives
35    /// filter on. Returns `None` (pass the body through untouched) when the target is disabled,
36    /// the request is not gRPC, or its path does not resolve to a method in the descriptor pool
37    /// (e.g. a service whose `FileDescriptorSet` was not registered).
38    fn capture_state<B>(&self, request: &Request<B>) -> Option<CaptureState> {
39        // Two probes, because they fail in complementary ways:
40        //
41        // - The event probe covers plain `grpc_request=trace` directives, and keeps working under
42        //   `telemetry-subscribers`, whose global level filter rejects every *span* callsite above
43        //   `TOKIO_SPAN_LEVEL` (default `info`) — trace events still flow, trace spans do not.
44        // - The span probe covers field-scoped directives like
45        //   `grpc_request[{service=...}]=trace`, for which `EnvFilter` disables events outside a
46        //   matching span, so the event probe — before any span exists — fails. It declares the
47        //   same field names as the real span below because `EnvFilter` only considers a
48        //   field-scoped directive for callsites that have those fields. Under
49        //   `telemetry-subscribers` this additionally requires `TOKIO_SPAN_LEVEL=trace` so the
50        //   span survives its global level filter.
51        //
52        // When no `grpc_request` directive is set at all, both probes short-circuit on cached
53        // callsite interest.
54        //
55        // Both probes are deliberately evaluated (no `||` short-circuit): under the full
56        // `telemetry-subscribers` stack, skipping the span probe when the event probe already
57        // passed was empirically observed to break field-scoped capture — the value-matched
58        // events stopped reaching any layer. Evaluating the span probe registers its
59        // fields-bearing callsite with every filter before the first `capture` span is created.
60        let event_enabled = tracing::event_enabled!(target: TARGET, tracing::Level::TRACE);
61        let span_enabled =
62            tracing::span_enabled!(target: TARGET, tracing::Level::TRACE, service, method);
63        if !(event_enabled || span_enabled) {
64            return None;
65        }
66
67        // Only gRPC requests carry the length-prefixed message framing the body wrapper parses.
68        // This also skips non-gRPC routes merged into the same router.
69        if !request
70            .headers()
71            .get(http::header::CONTENT_TYPE)?
72            .to_str()
73            .ok()?
74            .starts_with("application/grpc")
75        {
76            return None;
77        }
78
79        let path = request.uri().path();
80        let (service, method) = path.strip_prefix('/')?.split_once('/')?;
81        let service = self.pool.get_service_by_name(service)?;
82        let method = service.methods().find(|m| m.name() == method)?;
83
84        let span = trace_span!(
85            target: TARGET,
86            "capture",
87            service = %service.full_name(),
88            method = %method.name(),
89        );
90
91        // The probes above are callsite-level, so under a field-scoped directive they pass for
92        // *every* request regardless of its service/method values. Re-check with this request's
93        // span entered, so a global `EnvFilter` — which consults the span's recorded field values
94        // — prunes requests whose events would all be dropped anyway before any buffering or
95        // decoding happens. Per-layer filtered stacks (e.g. `telemetry-subscribers`) answer this
96        // aggregate check permissively and filter at dispatch instead; there the per-request
97        // message-count and message-size caps bound the wasted work.
98        if !span.in_scope(|| tracing::event_enabled!(target: TARGET, tracing::Level::TRACE)) {
99            return None;
100        }
101
102        Some(CaptureState::new(
103            span,
104            path.to_owned(),
105            self.max_captured_message_size,
106        ))
107    }
108}
109
110impl<S, ReqBody> Service<Request<ReqBody>> for GrpcRequestLog<S>
111where
112    S: Service<Request<RequestLogBody<ReqBody>>>,
113{
114    type Response = S::Response;
115    type Error = S::Error;
116    type Future = S::Future;
117
118    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
119        self.inner.poll_ready(cx)
120    }
121
122    fn call(&mut self, request: Request<ReqBody>) -> Self::Future {
123        let capture = self.capture_state(&request);
124        self.inner
125            .call(request.map(|body| RequestLogBody::new(body, capture)))
126    }
127}
mysten_network/request_log/service.rs

mysten_network/request_log/
service.rs
