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()); +}