zebra_rpc/lightwalletd/
tracing.rs

1//! Request tracing for the lightwalletd gRPC server.
2//!
3//! The generated `CompactTxStreamer` service has no logging of its own, so a node
4//! serving light clients gives no sign of what it was asked for, even at `trace`. This
5//! layer logs every request and its outcome, in the same shape as the JSON-RPC
6//! server's [`RpcTracingMiddleware`](crate::server::rpc_tracing::RpcTracingMiddleware):
7//! a `SERVER` span per request carrying the method name, so traces from both servers
8//! group the same way in Jaeger.
9
10use std::{
11    task::{Context, Poll},
12    time::Instant,
13};
14
15use futures::future::BoxFuture;
16use tonic::codegen::http;
17use tower::{Layer, Service};
18use tracing::{debug, info_span, Instrument};
19
20/// Adds a [`GrpcTracing`] to a gRPC service.
21#[derive(Clone, Copy, Debug, Default)]
22pub struct GrpcTracingLayer;
23
24impl<S> Layer<S> for GrpcTracingLayer {
25    type Service = GrpcTracing<S>;
26
27    fn layer(&self, service: S) -> Self::Service {
28        GrpcTracing { service }
29    }
30}
31
32/// Logs each gRPC request and its outcome, inside a span naming the method.
33#[derive(Clone, Debug)]
34pub struct GrpcTracing<S> {
35    service: S,
36}
37
38impl<S, RequestBody, ResponseBody> Service<http::Request<RequestBody>> for GrpcTracing<S>
39where
40    S: Service<http::Request<RequestBody>, Response = http::Response<ResponseBody>>
41        + Clone
42        + Send
43        + 'static,
44    S::Future: Send + 'static,
45    RequestBody: Send + 'static,
46{
47    type Response = S::Response;
48    type Error = S::Error;
49    type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
50
51    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
52        self.service.poll_ready(cx)
53    }
54
55    fn call(&mut self, request: http::Request<RequestBody>) -> Self::Future {
56        let method = grpc_method(request.uri().path()).to_string();
57
58        // Build the span outside the async block, so it is registered with the
59        // subscriber before the future is polled.
60        let span = info_span!(
61            "lightwalletd_grpc_request",
62            otel.kind = "server",
63            rpc.method = %method,
64            rpc.system = "grpc",
65            rpc.grpc.status_code = tracing::field::Empty,
66        );
67        let span_for_record = span.clone();
68
69        // `poll_ready` was called on `self.service`, so the readiness belongs to it and
70        // not to a clone. Swap the ready service out and leave the clone behind.
71        let clone = self.service.clone();
72        let mut service = std::mem::replace(&mut self.service, clone);
73
74        Box::pin(
75            async move {
76                debug!("handling lightwalletd gRPC request");
77
78                let start = Instant::now();
79                let response = service.call(request).await;
80                let elapsed_ms = start.elapsed().as_millis();
81
82                match &response {
83                    Ok(response) => {
84                        // tonic reports the final status in the trailers, for streaming and
85                        // unary responses alike, and those are not visible here. The header
86                        // is only set when the request failed before the handler ran, so a
87                        // status is logged when there is one, and left out otherwise.
88                        let status = response
89                            .headers()
90                            .get("grpc-status")
91                            .and_then(|status| status.to_str().ok());
92
93                        match status {
94                            Some(status) => {
95                                span_for_record.record("rpc.grpc.status_code", status);
96                                debug!(
97                                    grpc_status = status,
98                                    elapsed_ms, "rejected lightwalletd gRPC request",
99                                );
100                            }
101                            // A streaming method responds as soon as its headers are sent,
102                            // so this is the time to the first response, not to the last
103                            // message.
104                            None => debug!(elapsed_ms, "responded to lightwalletd gRPC request"),
105                        }
106                    }
107                    Err(_) => debug!(
108                        elapsed_ms,
109                        "lightwalletd gRPC request failed before a response was produced",
110                    ),
111                }
112
113                response
114            }
115            .instrument(span),
116        )
117    }
118}
119
120/// Returns the method name from a gRPC request path.
121///
122/// tonic routes on `/<package>.<service>/<Method>`, so the method is the last segment.
123fn grpc_method(path: &str) -> &str {
124    path.rsplit('/').next().unwrap_or(path)
125}
126
127#[cfg(test)]
128mod tests {
129    use super::*;
130
131    #[test]
132    fn grpc_method_is_the_last_path_segment() {
133        assert_eq!(
134            grpc_method("/cash.z.wallet.sdk.rpc.CompactTxStreamer/GetBlock"),
135            "GetBlock",
136        );
137        // Reflection and any other service the server hosts are named the same way.
138        assert_eq!(
139            grpc_method("/grpc.reflection.v1.ServerReflection/ServerReflectionInfo"),
140            "ServerReflectionInfo",
141        );
142        // A path that isn't a gRPC route must not panic or return an empty name.
143        assert_eq!(grpc_method("nonsense"), "nonsense");
144    }
145
146    /// Logging a request must not change it, or consume the service's readiness.
147    ///
148    /// `call` hands the polled-ready service to the response future and leaves a clone
149    /// behind, so a second request has to be served as well as the first.
150    #[tokio::test]
151    async fn requests_pass_through_unchanged() {
152        use tower::ServiceExt;
153
154        let inner = tower::service_fn(|_request: http::Request<()>| async {
155            Ok::<_, std::convert::Infallible>(
156                http::Response::builder()
157                    .status(200)
158                    .body(())
159                    .expect("a 200 response is valid"),
160            )
161        });
162
163        let mut service = GrpcTracingLayer.layer(inner);
164
165        for _ in 0..2 {
166            let request = http::Request::builder()
167                .uri("/cash.z.wallet.sdk.rpc.CompactTxStreamer/GetBlock")
168                .body(())
169                .expect("the request is valid");
170
171            let response = service
172                .ready()
173                .await
174                .expect("the service is always ready")
175                .call(request)
176                .await
177                .expect("the inner service always succeeds");
178
179            assert_eq!(response.status(), 200);
180        }
181    }
182}
zebra_rpc/lightwalletd/tracing.rs

zebra_rpc/lightwalletd/
tracing.rs
