telemetry_rust/instrumentations/http/hyper/
legacy_client.rs

1//! Async hyper-util legacy client instrumentation helpers.
2//!
3//! This module instruments `hyper_util::client::legacy::Client` requests
4//! with OpenTelemetry client spans and trace-context propagation.
5//!
6//! # Example
7//!
8//! ```no_run
9//! use bytes::Bytes;
10//! use http_body_util::Empty;
11//! use hyper::{Request, header::USER_AGENT};
12//! use hyper_util::rt::TokioExecutor;
13//! use telemetry_rust::instrumentations::http::hyper::HyperLegacyClientInstrument;
14//!
15//! # async fn example() -> Result<(), Box<dyn std::error::Error>> {
16//! let client = hyper_util::client::legacy::Client::builder(TokioExecutor::new())
17//!     .build_http::<Empty<Bytes>>()
18//!     .instrument();
19//!
20//! let response = client
21//!     .request(
22//!         Request::builder()
23//!             .uri("http://127.0.0.1:3000/health")
24//!             .header(USER_AGENT, "telemetry-rust")
25//!             .body(Empty::<Bytes>::new())?,
26//!     )
27//!     .await?;
28//! # let _ = response;
29//! # Ok(())
30//! # }
31//! ```
32
33use std::error::Error as StdError;
34use std::future::Future;
35
36use hyper::{
37    Request, Response,
38    body::{Body, Incoming},
39};
40use hyper_util::client::legacy;
41
42use crate::{
43    Context,
44    future::InstrumentedFuture,
45    http,
46    instrumentations::http::client::{HttpClientSpanBuilder, HttpError},
47};
48
49impl HttpError for legacy::Error {
50    fn error_type(&self) -> &'static str {
51        if self.is_connect() {
52            "connect"
53        } else if let Some(hyper_error) = self
54            .source()
55            .and_then(|source| source.downcast_ref::<hyper::Error>())
56        {
57            hyper_error.error_type()
58        } else {
59            "_OTHER"
60        }
61    }
62}
63
64/// A trait for creating instrumented hyper-util legacy clients with
65/// OpenTelemetry tracing.
66pub trait HyperLegacyClientInstrument
67where
68    Self: Sized,
69{
70    /// The legacy client's connector type.
71    type Connector;
72
73    /// The legacy client's request body type.
74    type Body;
75
76    /// Wraps this client in an [`InstrumentedLegacyClient`] that can be reused
77    /// to send traced requests.
78    fn instrument(self) -> InstrumentedLegacyClient<Self::Connector, Self::Body>;
79}
80
81impl<C, B> HyperLegacyClientInstrument for legacy::Client<C, B> {
82    type Connector = C;
83    type Body = B;
84
85    fn instrument(self) -> InstrumentedLegacyClient<C, B> {
86        InstrumentedLegacyClient::new(self)
87    }
88}
89
90/// A reusable wrapper around `hyper_util::client::legacy::Client` that records
91/// client spans for each request sent through the client.
92#[must_use = "Client does nothing until you call request()"]
93pub struct InstrumentedLegacyClient<C, B> {
94    inner: legacy::Client<C, B>,
95    context: Option<Context>,
96}
97
98impl<C, B> InstrumentedLegacyClient<C, B> {
99    /// Creates a new instrumented legacy hyper client.
100    pub fn new(inner: legacy::Client<C, B>) -> Self {
101        Self {
102            inner,
103            context: None,
104        }
105    }
106
107    /// Sets the OpenTelemetry context for requests sent by this wrapper.
108    pub fn context(mut self, context: &Context) -> Self {
109        self.context = Some(context.clone());
110        self
111    }
112
113    /// Sets the optional OpenTelemetry context for requests sent by this wrapper.
114    pub fn set_context(mut self, context: Option<&Context>) -> Self {
115        self.context = context.cloned();
116        self
117    }
118
119    /// Returns the wrapped legacy hyper client.
120    pub fn into_inner(self) -> legacy::Client<C, B> {
121        self.inner
122    }
123}
124
125impl<C, B> Clone for InstrumentedLegacyClient<C, B>
126where
127    legacy::Client<C, B>: Clone,
128{
129    fn clone(&self) -> Self {
130        Self {
131            inner: self.inner.clone(),
132            context: self.context.clone(),
133        }
134    }
135}
136
137impl<C, B> InstrumentedLegacyClient<C, B>
138where
139    C: legacy::connect::Connect + Clone + Send + Sync + 'static,
140    B: Body + Send + Unpin + 'static,
141    B::Data: Send,
142    B::Error: Into<Box<dyn StdError + Send + Sync>>,
143{
144    /// Sends a constructed request and records an outbound HTTP client span
145    /// around it.
146    pub fn request(
147        &self,
148        mut request: Request<B>,
149    ) -> impl Future<Output = Result<Response<Incoming>, legacy::Error>> + '_ {
150        let span = HttpClientSpanBuilder::from(&request).start(&self.context);
151
152        http::inject_context_on_context(span.context(), request.headers_mut());
153
154        let future = self.inner.request(request);
155        InstrumentedFuture::new(future, span)
156    }
157
158    /// Sends a GET request to the supplied URI and records an outbound HTTP
159    /// client span around it.
160    pub fn get(
161        &self,
162        uri: hyper::Uri,
163    ) -> impl Future<Output = Result<Response<Incoming>, legacy::Error>> + '_
164    where
165        B: Default,
166    {
167        let mut request = Request::new(B::default());
168        *request.uri_mut() = uri;
169        self.request(request)
170    }
171}
172
173#[cfg(test)]
174mod tests {
175    use super::HyperLegacyClientInstrument;
176    use crate::{Context, instrumentations::http::test_utils::*, semconv};
177    use assert2::assert;
178    use axum::http::StatusCode;
179    use bytes::Bytes;
180    use http_body_util::Empty;
181    use hyper::{Request, header::USER_AGENT};
182    use hyper_util::rt::TokioExecutor;
183    use opentelemetry::trace::{Span, SpanKind, TraceContextExt, Tracer, TracerProvider};
184    use serial_test::serial;
185    use tokio::net::TcpListener;
186    use tracing_subscriber::{Registry, layer::SubscriberExt};
187
188    #[tokio::test]
189    #[serial]
190    async fn instruments_successful_legacy_client_requests() {
191        let telemetry = configure_test_tracing();
192        let server = spawn_server().await;
193        let client = hyper_util::client::legacy::Client::builder(TokioExecutor::new())
194            .build_http::<Empty<Bytes>>()
195            .instrument();
196        let request_url = format!("{}/ok?ready=true", server.base_url);
197        let response = client
198            .request(
199                Request::builder()
200                    .uri(&request_url)
201                    .header(USER_AGENT, "telemetry-rust-tests")
202                    .body(Empty::<Bytes>::new())
203                    .unwrap(),
204            )
205            .await
206            .unwrap();
207
208        assert!(response.status() == StatusCode::OK);
209
210        let spans = force_flush_and_get_spans(&telemetry);
211        let span = find_span(&spans, "GET");
212        let traceparent = server.state.traceparent_for("/ok").unwrap();
213        let (trace_id, span_id) = traceparent_ids(&traceparent);
214
215        assert!(span.span_kind == SpanKind::Client);
216        assert!(span.span_context.trace_id().to_string() == trace_id);
217        assert!(span.span_context.span_id().to_string() == span_id);
218        assert!(matches!(span.status, opentelemetry::trace::Status::Unset));
219        assert!(string_attr(span, semconv::HTTP_REQUEST_METHOD) == Some("GET"));
220        assert!(string_attr(span, semconv::URL_SCHEME) == Some("http"));
221        assert!(string_attr(span, semconv::SERVER_ADDRESS) == Some("127.0.0.1"));
222        assert!(
223            i64_attr(span, semconv::SERVER_PORT) == Some(i64::from(server.addr.port()))
224        );
225        assert!(string_attr(span, semconv::URL_PATH) == Some("/ok"));
226        assert!(string_attr(span, semconv::URL_QUERY) == Some("ready=true"));
227        assert!(
228            string_attr(span, semconv::USER_AGENT_ORIGINAL)
229                == Some("telemetry-rust-tests")
230        );
231        assert!(string_attr(span, semconv::URL_FULL) == Some(request_url.as_str()));
232        assert!(i64_attr(span, semconv::HTTP_RESPONSE_STATUS_CODE) == Some(200));
233        assert!(string_attr(span, semconv::NETWORK_PROTOCOL_VERSION).is_some());
234    }
235
236    #[tokio::test]
237    #[serial]
238    async fn legacy_client_uses_explicit_parent_context_when_provided() {
239        let telemetry = configure_test_tracing();
240        let server = spawn_server().await;
241        let client = hyper_util::client::legacy::Client::builder(TokioExecutor::new())
242            .build_http::<Empty<Bytes>>();
243        let tracer = telemetry.provider.tracer("hyper-legacy-tests");
244        let explicit_parent = tracer.start("explicit-parent");
245        let explicit_parent_span_id = explicit_parent.span_context().span_id();
246        let explicit_parent_cx = Context::current_with_span(explicit_parent);
247        let tracing_tracer = telemetry.provider.tracer("tracing-tests");
248        let subscriber = Registry::default()
249            .with(tracing_opentelemetry::layer().with_tracer(tracing_tracer));
250        let guard = tracing::subscriber::set_default(subscriber);
251        let current_parent = tracing::info_span!("current-parent");
252
253        tracing::Instrument::instrument(
254            async {
255                let client = client.instrument().context(&explicit_parent_cx);
256                client
257                    .request(
258                        Request::builder()
259                            .uri(format!("{}/ok", server.base_url))
260                            .body(Empty::<Bytes>::new())
261                            .unwrap(),
262                    )
263                    .await
264                    .unwrap();
265            },
266            current_parent,
267        )
268        .await;
269
270        drop(guard);
271        explicit_parent_cx.span().end();
272
273        let spans = force_flush_and_get_spans(&telemetry);
274        let hyper_span = find_span(&spans, "GET");
275        let current_span = find_span(&spans, "current-parent");
276
277        assert!(hyper_span.parent_span_id == explicit_parent_span_id);
278        assert!(hyper_span.parent_span_id != current_span.span_context.span_id());
279    }
280
281    #[tokio::test]
282    #[serial]
283    async fn legacy_client_marks_error_responses_as_errors() {
284        let telemetry = configure_test_tracing();
285        let server = spawn_server().await;
286        let client = hyper_util::client::legacy::Client::builder(TokioExecutor::new())
287            .build_http::<Empty<Bytes>>()
288            .instrument();
289        let response = client
290            .request(
291                Request::builder()
292                    .uri(format!("{}/not-found", server.base_url))
293                    .body(Empty::<Bytes>::new())
294                    .unwrap(),
295            )
296            .await
297            .unwrap();
298
299        assert!(response.status() == StatusCode::NOT_FOUND);
300
301        let spans = force_flush_and_get_spans(&telemetry);
302        let span = find_span(&spans, "GET");
303
304        assert!(matches!(
305            span.status,
306            opentelemetry::trace::Status::Error { .. }
307        ));
308        assert!(i64_attr(span, semconv::HTTP_RESPONSE_STATUS_CODE) == Some(404));
309        assert!(string_attr(span, semconv::ERROR_TYPE) == Some("404"));
310    }
311
312    #[tokio::test]
313    #[serial]
314    async fn legacy_client_marks_transport_errors_as_errors() {
315        let telemetry = configure_test_tracing();
316        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
317        let addr = listener.local_addr().unwrap();
318        drop(listener);
319        let client = hyper_util::client::legacy::Client::builder(TokioExecutor::new())
320            .build_http::<Empty<Bytes>>()
321            .instrument();
322        let result = client
323            .request(
324                Request::builder()
325                    .uri(format!("http://{addr}/unreachable"))
326                    .body(Empty::<Bytes>::new())
327                    .unwrap(),
328            )
329            .await;
330
331        assert!(result.is_err());
332
333        let spans = force_flush_and_get_spans(&telemetry);
334        let span = find_span(&spans, "GET");
335
336        assert!(matches!(
337            span.status,
338            opentelemetry::trace::Status::Error { .. }
339        ));
340        assert!(string_attr(span, semconv::ERROR_TYPE) == Some("connect"));
341    }
342}
telemetry_rust/instrumentations/http/hyper/legacy_client.rs

telemetry_rust/instrumentations/http/hyper/
legacy_client.rs
