From 90035c881adc979a5dd11ecc185c7f4a21f018d0 Mon Sep 17 00:00:00 2001 From: Charles Sibbald Date: Tue, 27 Oct 2026 17:22:08 +0200 Subject: [PATCH] feat(client): optional tracing spans (feature `tracing`, off by default) With the feature on, each request through the pooled Client is a span tree, target `may_minihttp::client`: - http_client_request (INFO), the whole request until the body is read or the response dropped: http.request.method, server.address, server.port, url.scheme, http.response.status_code, error.type, http.request.resend_count, may_minihttp.connection_reused; - http_client_connect (INFO), DNS + connect + TLS of a new connection; - http_client_write, http_client_wait (time to first byte) and http_client_body (DEBUG); - DEBUG events pool_checkout (hit/miss/wait) and pool_evict (expired/closed). No paths, query strings or headers are recorded. The parent is the span given to RequestBuilder::parent_span, else Span::current() at send. RequestBuilder::request_span hands the client a caller's span to use as the request span instead: the client records the fields that span declares and opens its phase spans under it, so a caller with its own per-request client span gets one span, not two. When the request span is disabled, no phase spans or events are created (they would be orphan roots). The client never enters a span: a may coroutine can yield mid-request and resume on another thread, so spans get an explicit parent and close on drop. With the feature off there is no tracing dependency, the macros expand to nothing and the span fields are zero-sized. Also: ClientErrorKind::as_str(); HttpClient::send_request split into crate-private write_request/read_response so the write and the wait are timed separately; bench_client honours BENCH_TRACING=info|debug when built with the feature. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Y3TSRmEZAYU6RBUXhkn7WF --- Cargo.toml | 9 + README.md | 19 ++ examples/bench_client.rs | 18 ++ src/client/client_impl.rs | 13 +- src/client/mod.rs | 1 + src/client/rich.rs | 260 ++++++++++++++++++++++--- src/client/trace.rs | 145 ++++++++++++++ tests/client_tracing.rs | 391 ++++++++++++++++++++++++++++++++++++++ 8 files changed, 825 insertions(+), 31 deletions(-) create mode 100644 src/client/trace.rs create mode 100644 tests/client_tracing.rs diff --git a/Cargo.toml b/Cargo.toml index b500546..bd47dcf 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -29,6 +29,7 @@ rustls-platform-verifier = { version = "0.7", optional = true } url = { version = "2", optional = true } serde = { version = "1", optional = true } serde_json = { version = "1", optional = true } +tracing = { version = "0.1", default-features = false, features = ["std"], optional = true } [dev-dependencies] atoi = "2" @@ -51,6 +52,8 @@ testcontainers = "0.15" tokio = { version = "1", features = ["full"] } rcgen = "0.14" flate2 = "1.1" +tracing = "0.1" +tracing-subscriber = { version = "0.3", default-features = false, features = ["registry", "std"] } [target.'cfg(unix)'.dev-dependencies] may_postgres = { git = "https://github.com/Xudong-Huang/may_postgres.git", default-features = false } @@ -61,6 +64,8 @@ default = ["may/default"] client = ["http", "may/io_timeout", "dep:generator", "dep:rustls", "dep:rustls-platform-verifier", "dep:url"] # Typed JSON request/response helpers. Kept optional so the core HTTP transport stays lean. json = ["client", "dep:serde", "dep:serde_json"] +# `tracing` spans and events in the client (see `client::trace`). Off: nothing is compiled in. +tracing = ["client", "dep:tracing"] # Examples that use the client module — automatically enable the `client` feature. [[example]] @@ -153,6 +158,10 @@ required-features = ["client"] name = "tcp_nodelay_stall" required-features = ["client"] +[[test]] +name = "client_tracing" +required-features = ["tracing"] + [profile.release] opt-level = 3 codegen-units = 1 diff --git a/README.md b/README.md index 5a7ff50..5e27ec9 100644 --- a/README.md +++ b/README.md @@ -128,6 +128,25 @@ let client = Client::builder().build_with_interceptor(AttemptId)?; # Ok::<(), Box>(()) ``` +#### Tracing (feature `tracing`) + +With the `tracing` feature, each request through `Client` is a span tree under target +`may_minihttp::client`: + +- `http_client_request` (INFO) has method, server address and port, status, `error.type`, resend + count and connection reuse. +- `http_client_connect` (INFO) covers a new connection. +- `http_client_write`, `http_client_wait` (time to first byte) and `http_client_body` are at + DEBUG level. +- `pool_checkout` and `pool_evict` are DEBUG events. + +Paths, query strings and headers are not recorded. + +Spans are never entered, because a coroutine can resume on another thread. Give the parent +explicitly with `RequestBuilder::parent_span`, or hand the client your own per-request span with +`RequestBuilder::request_span`. Without the feature, nothing is compiled in, and `tracing` is not a +dependency. + Multipart filesystem helpers are deliberately named `blocking_*` and must be called outside latency-sensitive may coroutines. Automatic stale-socket retry is limited to one attempt for idempotent requests with replayable bodies. diff --git a/examples/bench_client.rs b/examples/bench_client.rs index 253e800..257d55f 100644 --- a/examples/bench_client.rs +++ b/examples/bench_client.rs @@ -18,6 +18,10 @@ //! `header-bytes` adds an `X-Pad` header of that size (a bearer token with claims is 1-6 KiB); //! `body-bytes > 0` makes the request a POST with a body of that size (`/echo` on bench_server). //! Workers are may coroutines. Pair with `examples/bench_server.rs`. +//! +//! Built with `--features tracing`, `BENCH_TRACING=info` (or `debug`) installs a +//! `tracing-subscriber` registry at that level, so the client's spans are created and closed; +//! unset, the feature is compiled in with no subscriber. use std::alloc::{GlobalAlloc, Layout, System}; use std::io::{self, Read}; @@ -143,6 +147,20 @@ fn main() -> io::Result<()> { let body_bytes: usize = arg(args.next(), 0); may::config().set_stack_size(0x8000); + #[cfg(feature = "tracing")] + { + use tracing_subscriber::filter::LevelFilter; + use tracing_subscriber::layer::SubscriberExt; + let level = match std::env::var("BENCH_TRACING").as_deref() { + Ok("info") => Some(LevelFilter::INFO), + Ok("debug") => Some(LevelFilter::DEBUG), + _ => None, + }; + if let Some(level) = level { + tracing::subscriber::set_global_default(tracing_subscriber::registry().with(level)) + .expect("install the tracing subscriber"); + } + } let parsed: Uri = url.parse().map_err(io::Error::other)?; let path: Uri = parsed .path_and_query() diff --git a/src/client/client_impl.rs b/src/client/client_impl.rs index 33649d3..31adcec 100644 --- a/src/client/client_impl.rs +++ b/src/client/client_impl.rs @@ -356,7 +356,13 @@ impl HttpClient { /// Send a request built from this client and read the response. #[inline] - pub fn send_request(&mut self, mut req: Request) -> io::Result { + pub fn send_request(&mut self, req: Request) -> io::Result { + self.write_request(req)?; + self.get_rsp() + } + + /// The first half of [`send_request`](Self::send_request): finish writing the request. + pub(crate) fn write_request(&mut self, mut req: Request) -> io::Result<()> { debug_assert!( self.conn.ptr_eq(req.conn()), "client and request must share the same connection" @@ -364,6 +370,11 @@ impl HttpClient { self.expect_body = req.expect_body_request(); req.finish()?; drop(req); + Ok(()) + } + + /// The second half of [`send_request`](Self::send_request): read the response head. + pub(crate) fn read_response(&mut self) -> io::Result { self.get_rsp() } diff --git a/src/client/mod.rs b/src/client/mod.rs index 2d92f5c..c28279f 100644 --- a/src/client/mod.rs +++ b/src/client/mod.rs @@ -17,6 +17,7 @@ mod response; mod rich; mod shared; mod tls; +mod trace; pub use cancellation::CancellationToken; pub use client_impl::HttpClient; diff --git a/src/client/rich.rs b/src/client/rich.rs index 1797116..4441eae 100644 --- a/src/client/rich.rs +++ b/src/client/rich.rs @@ -19,6 +19,7 @@ use url::Url; use super::cancellation::{cancelled_error, is_cancelled_error, run_cancellable}; use super::interceptor::{Interceptor, NoInterceptor, OutgoingRequest, ResponseHead, Timing}; +use super::trace::{self, client_event, client_span, phase_span, span_record}; use super::{ CancellationToken, ClientEvent, ClientObserver, ObservedOrigin, RequestMetadata, RequestMetadataContext, RequestMetadataProvider, Resolver, SystemResolver, @@ -391,6 +392,24 @@ pub enum ClientErrorKind { } impl ClientErrorKind { + /// A stable snake_case name, suitable as a metric label or an `error.type` attribute. + pub fn as_str(self) -> &'static str { + match self { + Self::Cancelled => "cancelled", + Self::Metadata => "metadata", + Self::InvalidRequest => "invalid_request", + Self::Dns => "dns", + Self::Connection => "connection", + Self::Tls => "tls", + Self::Timeout => "timeout", + Self::Protocol => "protocol", + Self::BodyTooLarge => "body_too_large", + Self::BodyNotReplayable => "body_not_replayable", + Self::Redirect => "redirect", + Self::Io => "io", + } + } + fn classify(source: &io::Error) -> Self { let message = source.to_string().to_ascii_lowercase(); match source.kind() { @@ -493,10 +512,12 @@ impl fmt::Debug for Client { } } -#[derive(Debug, Clone, Copy)] +#[derive(Debug, Clone)] struct RequestTrace { request_id: u64, started: Instant, + /// The request's span (feature `tracing`; zero-sized otherwise). + span: trace::Span, } struct StreamingExecution<'a> { @@ -508,6 +529,13 @@ struct StreamingExecution<'a> { tls_snapshot: &'a mut Option, } +/// What `RequestBuilder::parent_span` / `request_span` chose (feature `tracing`). +#[cfg_attr(not(feature = "tracing"), allow(dead_code))] +enum SpanChoice { + Parent(trace::Span), + Own(trace::Span), +} + struct AttemptInfo<'a> { method: &'a Method, url: &'a Url, @@ -556,6 +584,7 @@ impl Client { let url = parse_url(url)?; Ok(RequestBuilder { client: self.clone(), + span: None, method, url, headers: HeaderMap::new(), @@ -578,19 +607,26 @@ impl Client { self.inner.stats.snapshot() } - fn execute(&self, request: RequestBuilder) -> io::Result { - let trace = self.begin_request(&request.method, &request.url); + fn execute(&self, mut request: RequestBuilder) -> io::Result { + let trace = self.begin_request(&request.method, &request.url, request.span.take()); let cancellation = request.cancellation.clone(); let result = match cancellation { Some(token) => run_cancellable(&token, || self.execute_buffered(request, &trace)), None => self.execute_buffered(request, &trace), }; match &result { - Ok(response) => self.inner.observe(ClientEvent::RequestCompleted { - request_id: trace.request_id, - status: response.status, - total_duration: trace.started.elapsed(), - }), + Ok(response) => { + span_record!( + trace.span, + "http.response.status_code", + response.status.as_u16() + ); + self.inner.observe(ClientEvent::RequestCompleted { + request_id: trace.request_id, + status: response.status, + total_duration: trace.started.elapsed(), + }) + } Err(error) => self.observe_failure(&trace, error), } result @@ -726,12 +762,14 @@ impl Client { } } - fn execute_streaming(&self, request: RequestBuilder) -> io::Result { - let trace = self.begin_request(&request.method, &request.url); + fn execute_streaming(&self, mut request: RequestBuilder) -> io::Result { + let trace = self.begin_request(&request.method, &request.url, request.span.take()); let cancellation = request.cancellation.clone(); let result = match cancellation { - Some(token) => run_cancellable(&token, || self.execute_streaming_inner(request, trace)), - None => self.execute_streaming_inner(request, trace), + Some(token) => { + run_cancellable(&token, || self.execute_streaming_inner(request, &trace)) + } + None => self.execute_streaming_inner(request, &trace), }; if let Err(error) = &result { self.observe_failure(&trace, error); @@ -742,7 +780,7 @@ impl Client { fn execute_streaming_inner( &self, request: RequestBuilder, - trace: RequestTrace, + trace: &RequestTrace, ) -> io::Result { if self.inner.config.redirect_policy != RedirectPolicy::None { return Err(io::Error::new( @@ -769,7 +807,7 @@ impl Client { &mut body, StreamingExecution { deadline, - trace: &trace, + trace, cancellation, attempt: &mut attempt, sensitive_headers: &mut sensitive_headers, @@ -822,6 +860,14 @@ impl Client { )?; let mut lease = self.checkout(&key, Arc::clone(&tls_config), deadline, trace)?; let reused_idle_connection = lease.reused_idle_connection; + span_record!( + trace.span, + "may_minihttp.connection_reused", + reused_idle_connection + ); + if *attempt > 1 { + span_record!(trace.span, "http.request.resend_count", *attempt - 1); + } let timing = self.before_send( &mut attempt_headers, AttemptInfo { @@ -835,8 +881,15 @@ impl Client { }, ); let result = (|| { - let mut response = - self.send_on_lease(&mut lease, method, url, &attempt_headers, body, deadline)?; + let mut response = self.send_on_lease( + &mut lease, + method, + url, + &attempt_headers, + body, + deadline, + &trace.span, + )?; let status = response.status(); if I::ENABLED { @@ -879,6 +932,12 @@ impl Client { .and_then(|value| value.trim().parse::().ok()) .map_or(0, |length| length.min(limit)); let mut bytes = Vec::with_capacity(expected); + let body_span = phase_span!( + DEBUG, + parent: &trace.span, + "http_client_body", + "http.response.body.size" = tracing::field::Empty + ); loop { if response.body_complete() { break; @@ -904,6 +963,8 @@ impl Client { )); } } + span_record!(body_span, "http.response.body.size", bytes.len() as u64); + drop(body_span); drop(response); Ok(( BufferedResponse { @@ -1001,6 +1062,14 @@ impl Client { )?; let mut lease = self.checkout(&key, Arc::clone(&tls_config), deadline, trace)?; let reused_idle_connection = lease.reused_idle_connection; + span_record!( + trace.span, + "may_minihttp.connection_reused", + reused_idle_connection + ); + if *attempt > 1 { + span_record!(trace.span, "http.request.resend_count", *attempt - 1); + } let timing = self.before_send( &mut attempt_headers, AttemptInfo { @@ -1013,8 +1082,15 @@ impl Client { reused_connection: reused_idle_connection, }, ); - let sent = - self.send_on_lease(&mut lease, method, url, &attempt_headers, body, deadline); + let sent = self.send_on_lease( + &mut lease, + method, + url, + &attempt_headers, + body, + deadline, + &trace.span, + ); if I::ENABLED { if let Some(timing) = &timing { match &sent { @@ -1034,6 +1110,7 @@ impl Client { match sent { Ok(response) => { let status = response.status(); + span_record!(trace.span, "http.response.status_code", status.as_u16()); self.inner.observe(ClientEvent::ResponseHeaders { request_id: trace.request_id, origin: observed_url(url), @@ -1061,7 +1138,8 @@ impl Client { headers: response_headers, final_url: url.clone(), inner: Arc::clone(&self.inner), - trace: *trace, + trace: trace.clone(), + body_span: phase_span!(DEBUG, parent: &trace.span, "http_client_body"), cancellation, terminal_observed: false, reused_connection: reused_idle_connection, @@ -1310,6 +1388,7 @@ impl Client { Ok(Cow::Owned(headers)) } + #[allow(clippy::too_many_arguments)] fn send_on_lease( &self, lease: &mut PoolLease, @@ -1318,7 +1397,9 @@ impl Client { headers: &HeaderMap, body: &mut RequestBody, deadline: Instant, + span: &trace::Span, ) -> io::Result { + let _ = span; if headers.contains_key(HOST) { return Err(io::Error::new( io::ErrorKind::InvalidInput, @@ -1341,6 +1422,7 @@ impl Client { .client .set_max_response_header_bytes(self.inner.config.max_response_header_bytes)?; let target = origin_form(url)?; + let write_span = phase_span!(DEBUG, parent: span, "http_client_write"); let mut request = lease .connection_mut() .client @@ -1349,15 +1431,9 @@ impl Client { request.headers_mut().append(name, value.clone()); } match body { - RequestBody::Empty => lease.connection_mut().client.send_request(request), - RequestBody::Bytes(bytes) => { - request.send(bytes)?; - lease.connection_mut().client.send_request(request) - } - RequestBody::Multipart(form) => { - request.send_multipart(form)?; - lease.connection_mut().client.send_request(request) - } + RequestBody::Empty => {} + RequestBody::Bytes(bytes) => request.send(bytes)?, + RequestBody::Multipart(form) => request.send_multipart(form)?, RequestBody::Reader { reader, content_length, @@ -1375,15 +1451,59 @@ impl Client { ) })?; request.send_reader(&mut *reader, *content_length)?; - lease.connection_mut().client.send_request(request) } } + let client = &mut lease.connection_mut().client; + client.write_request(request)?; + drop(write_span); + let wait_span = phase_span!( + DEBUG, + parent: span, + "http_client_wait", + "http.response.status_code" = tracing::field::Empty + ); + let response = client.read_response()?; + span_record!( + wait_span, + "http.response.status_code", + response.status().as_u16() + ); + drop(wait_span); + Ok(response) } - fn begin_request(&self, method: &Method, url: &Url) -> RequestTrace { + fn begin_request( + &self, + method: &Method, + url: &Url, + choice: Option, + ) -> RequestTrace { + let span = match choice { + Some(SpanChoice::Own(span)) => span, + choice => { + let _parent = match choice { + Some(SpanChoice::Parent(parent)) => parent, + _ => trace::current_parent(), + }; + client_span!( + INFO, + parent: &_parent, + "http_client_request", + "http.request.method" = method.as_str(), + "server.address" = url.host_str().unwrap_or_default(), + "server.port" = url.port_or_known_default().unwrap_or_default(), + "url.scheme" = url.scheme(), + "http.response.status_code" = tracing::field::Empty, + "error.type" = tracing::field::Empty, + "http.request.resend_count" = tracing::field::Empty, + "may_minihttp.connection_reused" = tracing::field::Empty, + ) + } + }; let trace = RequestTrace { request_id: self.inner.next_request_id.fetch_add(1, Ordering::Relaxed), started: Instant::now(), + span, }; self.inner.observe(ClientEvent::RequestStarted { request_id: trace.request_id, @@ -1395,6 +1515,7 @@ impl Client { fn observe_failure(&self, trace: &RequestTrace, error: &io::Error) { let error = ClientErrorKind::classify(error); + span_record!(trace.span, "error.type", error.as_str()); if error == ClientErrorKind::Cancelled { self.inner.observe(ClientEvent::RequestCancelled { request_id: trace.request_id, @@ -1433,6 +1554,12 @@ impl Client { .stats .connections_expired .fetch_add(expired as u64, Ordering::Relaxed); + client_event!( + parent: &trace.span, + reason = "expired", + count = expired as u64, + "pool_evict" + ); } let idle = state.idle.get_mut(key).and_then(Vec::pop); if let Some(connection) = idle { @@ -1454,6 +1581,12 @@ impl Client { { // dropping the lease releases its slot and counts it discarded drop(lease); + client_event!( + parent: &trace.span, + reason = "closed", + count = 1_u64, + "pool_evict" + ); continue; } self.inner @@ -1464,6 +1597,7 @@ impl Client { request_id: trace.request_id, origin: observed_key(key), }); + client_event!(parent: &trace.span, result = "hit", "pool_checkout"); return Ok(lease); } let per_origin = state.per_origin.get(key).copied().unwrap_or(0); @@ -1476,6 +1610,15 @@ impl Client { let mut lease = PoolLease::reserved(Arc::clone(&self.inner), key.clone(), trace.request_id); + client_event!(parent: &trace.span, result = "miss", "pool_checkout"); + let connect_span = phase_span!( + INFO, + parent: &trace.span, + "http_client_connect", + "server.address" = &*key.host, + "server.port" = key.port, + "error.type" = tracing::field::Empty + ); let connect_budget = self.inner.config.connect_timeout.min(remaining(deadline)?); let connect_deadline = @@ -1496,6 +1639,13 @@ impl Client { source: resolution.as_ref().ok().map(|value| value.source), error: resolution.as_ref().err().map(ClientErrorKind::classify), }); + if let Err(_error) = &resolution { + span_record!( + connect_span, + "error.type", + ClientErrorKind::classify(_error).as_str() + ); + } let addresses = resolution?.addresses; let timeout = connect_deadline .checked_duration_since(Instant::now()) @@ -1517,6 +1667,14 @@ impl Client { tls: key.scheme == "https", error: client.as_ref().err().map(ClientErrorKind::classify), }); + if let Err(_error) = &client { + span_record!( + connect_span, + "error.type", + ClientErrorKind::classify(_error).as_str() + ); + } + drop(connect_span); let client = client?; lease.connection = Some(PooledConnection { client, @@ -1532,6 +1690,7 @@ impl Client { let wait = remaining(deadline)?; self.inner.stats.pool_waits.fetch_add(1, Ordering::Relaxed); + client_event!(parent: &trace.span, result = "wait", "pool_checkout"); let wait_started = Instant::now(); let (state_after_wait, timeout) = self .inner @@ -1558,6 +1717,8 @@ impl Client { /// Request builder whose body explicitly records whether it can be replayed. pub struct RequestBuilder { client: Client, + /// Parent of the request's span, or the request's span itself (feature `tracing`). + span: Option, method: Method, url: Url, headers: HeaderMap, @@ -1637,6 +1798,25 @@ impl RequestBuilder { self } + /// Make `span` the parent of this request's spans (feature `tracing`). Without it the parent + /// is [`tracing::Span::current()`] at the time of sending. Coroutine code should pass the + /// parent explicitly: a span entered on a thread is not a coroutine's context. + #[cfg(feature = "tracing")] + pub fn parent_span(mut self, span: &tracing::Span) -> Self { + self.span = Some(SpanChoice::Parent(span.clone())); + self + } + + /// Use `span` as this request's span instead of creating `http_client_request` (feature + /// `tracing`): the client records its request fields on it (only those the span declares, + /// e.g. as `tracing::field::Empty`) and opens its connect/write/wait/body spans under it. + /// For a caller that wants one span per request with its own name and attributes. + #[cfg(feature = "tracing")] + pub fn request_span(mut self, span: &tracing::Span) -> Self { + self.span = Some(SpanChoice::Own(span.clone())); + self + } + pub fn send(self) -> io::Result { let client = self.client.clone(); client.execute(self) @@ -1750,6 +1930,9 @@ pub struct StreamingResponse { cancellation: Option, terminal_observed: bool, reused_connection: bool, + /// Open while the body is read (feature `tracing`). + #[cfg_attr(not(feature = "tracing"), allow(dead_code))] + body_span: trace::Span, } impl fmt::Debug for StreamingResponse { @@ -1794,6 +1977,7 @@ impl StreamingResponse { } fn complete(&mut self) { + self.close_spans(None); drop(self.response.take()); if let Some(lease) = self.lease.take() { if self.reusable { @@ -1810,6 +1994,20 @@ impl StreamingResponse { } } + /// End the body and request spans at the end of the body rather than when the response is + /// dropped. No-op without the `tracing` feature. + fn close_spans(&mut self, error: Option) { + let _ = error; + #[cfg(feature = "tracing")] + { + if let Some(error) = error { + self.trace.span.record("error.type", error.as_str()); + } + self.body_span = tracing::Span::none(); + self.trace.span = tracing::Span::none(); + } + } + fn discard_connection(&mut self) { if let Some(response) = self.response.as_mut() { response.abandon_body(); @@ -1819,6 +2017,7 @@ impl StreamingResponse { } fn fail(&mut self, error: &io::Error) { + self.close_spans(Some(ClientErrorKind::classify(error))); self.discard_connection(); if !self.terminal_observed { self.inner.observe(ClientEvent::RequestFailed { @@ -1831,6 +2030,7 @@ impl StreamingResponse { } fn cancel(&mut self) -> io::Error { + self.close_spans(Some(ClientErrorKind::Cancelled)); self.discard_connection(); if !self.terminal_observed { self.inner.observe(ClientEvent::RequestCancelled { diff --git a/src/client/trace.rs b/src/client/trace.rs new file mode 100644 index 0000000..600dc39 --- /dev/null +++ b/src/client/trace.rs @@ -0,0 +1,145 @@ +//! Optional [`tracing`](https://docs.rs/tracing) instrumentation of the pooled client (feature +//! `tracing`, off by default). +//! +//! With the feature off, every item here is a zero-sized no-op and the macros expand to nothing: +//! no `tracing` dependency, no code, no fields evaluated. +//! +//! With it on, each request made through [`Client`](super::Client) gets one span, target +//! `may_minihttp::client`: +//! +//! | span (level) | when | fields | +//! |---|---|---| +//! | `http_client_request` (INFO) | the whole request, until the body is read or the response dropped | `http.request.method`, `server.address`, `server.port`, `url.scheme`, `http.response.status_code`, `error.type`, `http.request.resend_count`, `may_minihttp.connection_reused` | +//! | `http_client_connect` (INFO) | DNS and connect (TLS handshake included) of a new connection | `server.address`, `server.port`, `error.type` | +//! | `http_client_write` (DEBUG) | writing the request | | +//! | `http_client_wait` (DEBUG) | from the request written to the response head: time to first byte | `http.response.status_code` | +//! | `http_client_body` (DEBUG) | reading the response body | `http.response.body.size` | +//! +//! and DEBUG events on the request span: `pool_checkout` (`result` = `hit`, `miss` or `wait`) and +//! `pool_evict` (`reason` = `expired` or `closed`, `count`). +//! +//! Paths, query strings and headers are never recorded. +//! +//! The request span's parent is the span passed to `RequestBuilder::parent_span`, or else +//! [`tracing::Span::current()`] when the request is sent. `RequestBuilder::request_span` instead +//! hands the client a span to use *as* the request span (a caller's own client span, with its own +//! name and attributes): the client records the request fields that span declares and opens the +//! phase spans under it. When the request span is disabled, no phase spans or events are created. The client never *enters* its spans: a +//! may coroutine can yield inside the request and resume on another thread, and an entered span +//! would leak into whatever runs on the first thread meanwhile. Spans are created with an explicit +//! parent and closed when dropped, which is all a span exporter needs for the tree and the timing. + +#[cfg(feature = "tracing")] +pub(crate) type Span = tracing::Span; + +/// [`tracing::Span::current()`], skipped (a disabled span) when nothing records INFO spans: with +/// no subscriber, the request path does not touch the dispatcher's thread-local. +#[cfg(feature = "tracing")] +#[inline] +pub(crate) fn current_parent() -> Span { + if tracing::level_enabled!(tracing::Level::INFO) { + tracing::Span::current() + } else { + tracing::Span::none() + } +} + +/// Stand-in for `tracing::Span` with the feature off. +#[cfg(not(feature = "tracing"))] +#[derive(Clone, Debug, Default)] +pub(crate) struct Span; + +#[cfg(not(feature = "tracing"))] +#[inline(always)] +pub(crate) fn current_parent() -> Span { + Span +} + +/// `tracing::span!` with the client's target and an explicit parent; a zero-sized no-op without +/// the feature (nothing in the invocation is evaluated). +#[cfg(feature = "tracing")] +macro_rules! client_span { + ($level:ident, parent: $parent:expr, $name:literal $(, $($fields:tt)*)?) => { + tracing::span!( + target: "may_minihttp::client", + parent: $parent, + tracing::Level::$level, + $name + $(, $($fields)*)? + ) + }; +} + +#[cfg(not(feature = "tracing"))] +macro_rules! client_span { + ($($ignored:tt)*) => { + $crate::client::trace::Span + }; +} + +/// A span under the request's span, or none when the request's span is disabled (a child of a +/// disabled span would otherwise be exported as a root of its own). +#[cfg(feature = "tracing")] +macro_rules! phase_span { + ($level:ident, parent: $parent:expr, $name:literal $(, $($fields:tt)*)?) => {{ + let parent: &tracing::Span = $parent; + if parent.is_disabled() { + tracing::Span::none() + } else { + tracing::span!( + target: "may_minihttp::client", + parent: parent, + tracing::Level::$level, + $name + $(, $($fields)*)? + ) + } + }}; +} + +#[cfg(not(feature = "tracing"))] +macro_rules! phase_span { + ($($ignored:tt)*) => { + $crate::client::trace::Span + }; +} + +/// A DEBUG event on `parent` (skipped when `parent` is disabled); nothing without the feature. +#[cfg(feature = "tracing")] +macro_rules! client_event { + (parent: $parent:expr, $($rest:tt)*) => {{ + let parent: &tracing::Span = $parent; + if !parent.is_disabled() { + tracing::event!( + target: "may_minihttp::client", + parent: parent, + tracing::Level::DEBUG, + $($rest)* + ); + } + }}; +} + +#[cfg(not(feature = "tracing"))] +macro_rules! client_event { + ($($ignored:tt)*) => { + () + }; +} + +/// `span.record(field, value)`; nothing (the value is not evaluated) without the feature. +#[cfg(feature = "tracing")] +macro_rules! span_record { + ($span:expr, $field:literal, $value:expr) => {{ + $span.record($field, $value); + }}; +} + +#[cfg(not(feature = "tracing"))] +macro_rules! span_record { + ($($ignored:tt)*) => { + () + }; +} + +pub(crate) use {client_event, client_span, phase_span, span_record}; diff --git a/tests/client_tracing.rs b/tests/client_tracing.rs new file mode 100644 index 0000000..091969b --- /dev/null +++ b/tests/client_tracing.rs @@ -0,0 +1,391 @@ +//! The `tracing` feature: the client's span tree, its fields, and silence when disabled. + +use std::fmt::Write as _; +use std::io::{Read, Write}; +use std::net::TcpListener; +use std::sync::{Arc, Mutex}; +use std::thread; +use std::time::Duration; + +use may_minihttp::client::{Client, RedirectPolicy}; +use tracing::field::{Field, Visit}; +use tracing::span::{Attributes, Id, Record}; +use tracing::{Event, Subscriber}; +use tracing_subscriber::filter::{LevelFilter, Targets}; +use tracing_subscriber::layer::{Context, SubscriberExt}; +use tracing_subscriber::registry::LookupSpan; +use tracing_subscriber::Layer; + +/// Records `new parent= fields`, `record fields`, `event +/// parent= fields` and `close `, in order. +#[derive(Clone, Default)] +struct Recorder { + lines: Arc>>, +} + +impl Recorder { + fn lines(&self) -> Vec { + self.lines.lock().unwrap().clone() + } + + fn push(&self, line: String) { + self.lines.lock().unwrap().push(line); + } +} + +#[derive(Default)] +struct Fields { + message: String, + text: String, +} + +impl Visit for Fields { + fn record_str(&mut self, field: &Field, value: &str) { + if field.name() == "message" { + self.message = value.to_owned(); + } else { + let _ = write!(self.text, " {}={}", field.name(), value); + } + } + + fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { + if field.name() == "message" { + self.message = format!("{value:?}"); + } else { + let _ = write!(self.text, " {}={:?}", field.name(), value); + } + } +} + +impl Layer for Recorder +where + S: Subscriber + for<'a> LookupSpan<'a>, +{ + fn on_new_span(&self, attrs: &Attributes<'_>, id: &Id, ctx: Context<'_, S>) { + let span = ctx.span(id).expect("new span"); + let parent = span + .parent() + .map_or_else(|| "-".to_owned(), |parent| parent.name().to_owned()); + let mut fields = Fields::default(); + attrs.record(&mut fields); + self.push(format!( + "new {} parent={}{}", + span.name(), + parent, + fields.text + )); + } + + fn on_record(&self, id: &Id, values: &Record<'_>, ctx: Context<'_, S>) { + let span = ctx.span(id).expect("recorded span"); + let mut fields = Fields::default(); + values.record(&mut fields); + self.push(format!("record {}{}", span.name(), fields.text)); + } + + fn on_event(&self, event: &Event<'_>, ctx: Context<'_, S>) { + let parent = ctx + .event_span(event) + .map_or_else(|| "-".to_owned(), |span| span.name().to_owned()); + let mut fields = Fields::default(); + event.record(&mut fields); + self.push(format!( + "event {} parent={}{}", + fields.message, parent, fields.text + )); + } + + fn on_close(&self, id: Id, ctx: Context<'_, S>) { + let span = ctx.span(&id).expect("closed span"); + self.push(format!("close {}", span.name())); + } +} + +fn read_head(stream: &mut impl Read) -> String { + let mut request = Vec::new(); + let mut byte = [0_u8; 1]; + while !request.ends_with(b"\r\n\r\n") { + stream.read_exact(&mut byte).unwrap(); + request.push(byte[0]); + } + String::from_utf8(request).unwrap() +} + +/// Answers `count` requests on one keep-alive connection with `200 ok`. +fn server(count: usize) -> (u16, thread::JoinHandle<()>) { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let handle = thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + for _ in 0..count { + let _ = read_head(&mut stream); + stream + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok") + .unwrap(); + stream.flush().unwrap(); + } + }); + (port, handle) +} + +fn client() -> Client { + Client::builder() + .redirect_policy(RedirectPolicy::None) + .request_timeout(Duration::from_secs(2)) + .build() + .unwrap() +} + +fn client_lines(lines: &[String]) -> Vec { + lines + .iter() + .filter(|line| !line.contains(" caller")) + .cloned() + .collect() +} + +#[test] +fn a_request_is_a_span_tree_under_the_given_parent() { + let recorder = Recorder::default(); + let subscriber = tracing_subscriber::registry().with(recorder.clone()); + let (port, server) = server(2); + tracing::subscriber::with_default(subscriber, || { + let caller = tracing::info_span!("caller"); + let client = client(); + for path in ["/one?secret=1", "/two"] { + let response = client + .get(&format!("http://127.0.0.1:{port}{path}")) + .unwrap() + .parent_span(&caller) + .send() + .unwrap(); + assert_eq!(response.body(), b"ok"); + } + }); + server.join().unwrap(); + + let lines = client_lines(&recorder.lines()); + let text = lines.join("\n"); + let request = format!( + "new http_client_request parent=caller http.request.method=GET server.address=127.0.0.1 server.port={port} url.scheme=http" + ); + assert_eq!( + lines.iter().filter(|line| **line == request).count(), + 2, + "{text}" + ); + // The first request opens a connection, the second reuses it. + let first: Vec<&str> = lines[..lines + .iter() + .position(|line| line == "close http_client_request") + .unwrap() + + 1] + .iter() + .map(String::as_str) + .collect(); + assert_eq!( + first, + [ + request.as_str(), + "event pool_checkout parent=http_client_request result=miss", + &format!("new http_client_connect parent=http_client_request server.address=127.0.0.1 server.port={port}"), + "close http_client_connect", + "record http_client_request may_minihttp.connection_reused=false", + "new http_client_write parent=http_client_request", + "close http_client_write", + "new http_client_wait parent=http_client_request", + "record http_client_wait http.response.status_code=200", + "close http_client_wait", + "new http_client_body parent=http_client_request", + "record http_client_body http.response.body.size=2", + "close http_client_body", + "record http_client_request http.response.status_code=200", + "close http_client_request", + ], + "{text}" + ); + assert!( + text.contains("event pool_checkout parent=http_client_request result=hit"), + "{text}" + ); + assert!( + text.contains("record http_client_request may_minihttp.connection_reused=true"), + "{text}" + ); + // Paths and query strings are never recorded. + assert!(!text.contains("/one") && !text.contains("secret"), "{text}"); +} + +#[test] +fn without_an_explicit_parent_the_current_span_is_the_parent() { + let recorder = Recorder::default(); + let subscriber = tracing_subscriber::registry().with(recorder.clone()); + let (port, server) = server(1); + tracing::subscriber::with_default(subscriber, || { + let caller = tracing::info_span!("caller"); + let _entered = caller.enter(); + client() + .get(&format!("http://127.0.0.1:{port}/")) + .unwrap() + .send() + .unwrap(); + }); + server.join().unwrap(); + assert!(recorder + .lines() + .iter() + .any(|line| line.starts_with("new http_client_request parent=caller "))); +} + +#[test] +fn a_streaming_request_span_ends_with_the_body_not_the_drop() { + let recorder = Recorder::default(); + let subscriber = tracing_subscriber::registry().with(recorder.clone()); + let (port, server) = server(1); + tracing::subscriber::with_default(subscriber, || { + let mut response = client() + .get(&format!("http://127.0.0.1:{port}/")) + .unwrap() + .send_streaming() + .unwrap(); + let mut body = Vec::new(); + response.read_to_end(&mut body).unwrap(); + assert_eq!(body, b"ok"); + assert!(recorder + .lines() + .contains(&"close http_client_request".to_owned())); + drop(response); + }); + server.join().unwrap(); + let lines = recorder.lines(); + assert!( + lines.contains(&"close http_client_body".to_owned()), + "{lines:#?}" + ); +} + +#[test] +fn a_refused_connect_records_the_error_type() { + let port = { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + listener.local_addr().unwrap().port() + }; + let recorder = Recorder::default(); + let subscriber = tracing_subscriber::registry().with(recorder.clone()); + tracing::subscriber::with_default(subscriber, || { + assert!(client() + .get(&format!("http://127.0.0.1:{port}/")) + .unwrap() + .send() + .is_err()); + }); + let text = recorder.lines().join("\n"); + assert!( + text.contains("record http_client_connect error.type="), + "{text}" + ); + assert!( + text.contains("record http_client_request error.type="), + "{text}" + ); +} + +#[test] +fn nothing_is_emitted_when_the_target_is_disabled() { + let recorder = Recorder::default(); + let subscriber = tracing_subscriber::registry().with( + recorder + .clone() + .with_filter(Targets::new().with_target("may_minihttp", LevelFilter::OFF)), + ); + let (port, server) = server(2); + tracing::subscriber::with_default(subscriber, || { + let client = client(); + for _ in 0..2 { + client + .get(&format!("http://127.0.0.1:{port}/")) + .unwrap() + .send() + .unwrap(); + } + }); + server.join().unwrap(); + assert_eq!(recorder.lines(), Vec::::new()); +} + +#[test] +fn at_info_only_the_request_and_connect_spans_are_emitted() { + let recorder = Recorder::default(); + let subscriber = + tracing_subscriber::registry().with(recorder.clone().with_filter(LevelFilter::INFO)); + let (port, server) = server(1); + tracing::subscriber::with_default(subscriber, || { + client() + .get(&format!("http://127.0.0.1:{port}/")) + .unwrap() + .send() + .unwrap(); + }); + server.join().unwrap(); + let names: Vec = recorder + .lines() + .iter() + .filter(|line| line.starts_with("new ") || line.starts_with("event ")) + .map(|line| line.split(' ').nth(1).unwrap().to_owned()) + .collect(); + assert_eq!(names, ["http_client_request", "http_client_connect"]); +} + +#[test] +fn a_caller_span_can_be_the_request_span() { + let recorder = Recorder::default(); + let subscriber = tracing_subscriber::registry().with(recorder.clone()); + let (port, server) = server(1); + tracing::subscriber::with_default(subscriber, || { + let mine = tracing::info_span!( + "my_client_span", + "http.response.status_code" = tracing::field::Empty, + "may_minihttp.connection_reused" = tracing::field::Empty, + ); + client() + .get(&format!("http://127.0.0.1:{port}/")) + .unwrap() + .request_span(&mine) + .send() + .unwrap(); + }); + server.join().unwrap(); + let lines = recorder.lines(); + let text = lines.join("\n"); + assert!(!text.contains("http_client_request"), "{text}"); + for expected in [ + "new http_client_connect parent=my_client_span", + "new http_client_write parent=my_client_span", + "new http_client_wait parent=my_client_span", + "new http_client_body parent=my_client_span", + "record my_client_span may_minihttp.connection_reused=false", + "record my_client_span http.response.status_code=200", + ] { + assert!( + lines.iter().any(|line| line.starts_with(expected)), + "{expected}\n{text}" + ); + } +} + +#[test] +fn a_disabled_request_span_gets_no_orphan_children() { + let recorder = Recorder::default(); + let subscriber = tracing_subscriber::registry().with(recorder.clone()); + let (port, server) = server(1); + tracing::subscriber::with_default(subscriber, || { + client() + .get(&format!("http://127.0.0.1:{port}/")) + .unwrap() + .request_span(&tracing::Span::none()) + .send() + .unwrap(); + }); + server.join().unwrap(); + assert_eq!(recorder.lines(), Vec::::new()); +}