telemetry_rust/instrumentations/http/hyper/
mod.rs

1//! Async hyper connection instrumentation helpers.
2//!
3//! This module instruments `hyper::client::conn::http1::SendRequest`,
4//! `hyper::client::conn::http2::SendRequest`, and, with the
5//! `hyper-client-legacy` feature, `hyper_util::client::legacy::Client`.
6//!
7//! # Example
8//!
9//! ```no_run
10//! use bytes::Bytes;
11//! use http_body_util::Empty;
12//! use hyper::{Request, header::HOST};
13//! use hyper_util::rt::TokioIo;
14//! use telemetry_rust::instrumentations::http::hyper::HyperSendRequestInstrument;
15//! use tokio::net::TcpStream;
16//!
17//! # async fn example() -> Result<(), Box<dyn std::error::Error>> {
18//! # #[cfg(feature = "hyper-http1")]
19//! # {
20//! let stream = TcpStream::connect("127.0.0.1:3000").await?;
21//! let io = TokioIo::new(stream);
22//! let (send_request, connection) = hyper::client::conn::http1::handshake(io).await?;
23//! tokio::spawn(async move {
24//!     let _ = connection.await;
25//! });
26//!
27//! // Create an instrumented sender once and reuse it for requests on this connection.
28//! let mut instrumented_sender = send_request.instrument();
29//! let request = Request::builder()
30//!     .uri("/health")
31//!     .header(HOST, "127.0.0.1:3000")
32//!     .body(Empty::<Bytes>::new())?;
33//! let response = instrumented_sender.send_request(request).await?;
34//! # let _ = response;
35//! # }
36//! # Ok(())
37//! # }
38//! ```
39
40use crate::{
41    Context, Value,
42    instrumentations::http::client::{HttpClientSpanBuilder, HttpError, UrlParts},
43};
44
45/// Async instrumentation helpers for `hyper_util::client::legacy::Client`.
46#[cfg(feature = "hyper-client-legacy")]
47pub mod legacy_client;
48
49#[cfg(feature = "hyper-client-legacy")]
50pub use legacy_client::HyperLegacyClientInstrument;
51
52impl UrlParts for hyper::Uri {
53    fn full_url(&self) -> Option<impl Into<Value>> {
54        self.host().map(|_| self.to_string())
55    }
56
57    fn path(&self) -> Option<impl Into<Value>> {
58        Some(self.path().to_owned())
59    }
60
61    fn host(&self) -> Option<impl Into<Value>> {
62        self.host().map(ToOwned::to_owned)
63    }
64
65    fn scheme(&self) -> Option<impl Into<Value>> {
66        self.scheme_str().map(ToOwned::to_owned)
67    }
68
69    fn port(&self) -> Option<impl Into<Value>> {
70        self.port_u16().map(i64::from)
71    }
72
73    fn query(&self) -> Option<impl Into<Value>> {
74        self.query().map(ToOwned::to_owned)
75    }
76}
77
78impl HttpError for hyper::Error {
79    fn error_type(&self) -> &'static str {
80        if self.is_timeout() {
81            "timeout"
82        } else if self.is_closed() {
83            "closed"
84        } else if self.is_canceled() {
85            "canceled"
86        } else if self.is_parse_status() {
87            "parse_status"
88        } else if self.is_parse() {
89            "parse"
90        } else if self.is_incomplete_message() {
91            "incomplete_message"
92        } else if self.is_body_write_aborted() {
93            "body_write_aborted"
94        } else if self.is_shutdown() {
95            "shutdown"
96        } else if self.is_user() {
97            "user"
98        } else {
99            "_OTHER"
100        }
101    }
102}
103
104impl<B> From<&hyper::Request<B>> for HttpClientSpanBuilder {
105    fn from(request: &hyper::Request<B>) -> Self {
106        Self::from_parts(request.method(), request.headers(), request.uri())
107    }
108}
109
110/// A trait for creating instrumented hyper connection senders with OpenTelemetry tracing.
111pub trait HyperSendRequestInstrument
112where
113    Self: Sized,
114{
115    /// Wraps this sender in an [`InstrumentedSendRequest`] that can be reused
116    /// to send traced requests on the same connection.
117    fn instrument(self) -> InstrumentedSendRequest<Self>;
118}
119
120#[cfg(feature = "hyper-http1")]
121impl<B> HyperSendRequestInstrument for hyper::client::conn::http1::SendRequest<B> {
122    fn instrument(self) -> InstrumentedSendRequest<Self> {
123        InstrumentedSendRequest::new(self)
124    }
125}
126
127#[cfg(feature = "hyper-http2")]
128impl<B> HyperSendRequestInstrument for hyper::client::conn::http2::SendRequest<B> {
129    fn instrument(self) -> InstrumentedSendRequest<Self> {
130        InstrumentedSendRequest::new(self)
131    }
132}
133
134/// A reusable wrapper around hyper `SendRequest` that records client spans for
135/// each request sent through the connection.
136#[must_use = "SendRequest does nothing until you call send_request()"]
137pub struct InstrumentedSendRequest<S> {
138    inner: S,
139    context: Option<Context>,
140}
141
142impl<S> InstrumentedSendRequest<S> {
143    /// Creates a new instrumented hyper sender.
144    pub fn new(inner: S) -> Self {
145        Self {
146            inner,
147            context: None,
148        }
149    }
150
151    /// Sets the OpenTelemetry context for requests sent by this wrapper.
152    pub fn context(mut self, context: &Context) -> Self {
153        self.context = Some(context.clone());
154        self
155    }
156
157    /// Sets the optional OpenTelemetry context for requests sent by this wrapper.
158    pub fn set_context(mut self, context: Option<&Context>) -> Self {
159        self.context = context.cloned();
160        self
161    }
162
163    /// Returns the wrapped hyper sender.
164    pub fn into_inner(self) -> S {
165        self.inner
166    }
167}
168
169impl<S> Clone for InstrumentedSendRequest<S>
170where
171    S: Clone,
172{
173    fn clone(&self) -> Self {
174        Self {
175            inner: self.inner.clone(),
176            context: self.context.clone(),
177        }
178    }
179}
180
181#[cfg(any(feature = "hyper-http1", feature = "hyper-http2"))]
182macro_rules! impl_instrumented_send_request {
183    ($http:ident) => {
184        mod $http {
185            use std::future::Future;
186            use std::task::{Context as TaskContext, Poll};
187
188            use hyper::{
189                Request, Response, Result,
190                body::{Body, Incoming},
191                client::conn::$http::SendRequest,
192            };
193
194            use super::InstrumentedSendRequest;
195            use crate::{
196                future::InstrumentedFuture, http,
197                instrumentations::http::client::HttpClientSpanBuilder,
198            };
199
200            impl<B: Body + 'static> InstrumentedSendRequest<SendRequest<B>> {
201                /// Polls until the underlying sender is ready to send a request.
202                pub fn poll_ready(
203                    &mut self,
204                    cx: &mut TaskContext<'_>,
205                ) -> Poll<Result<()>> {
206                    self.inner.poll_ready(cx)
207                }
208
209                /// Waits until the underlying sender is ready to send a request.
210                pub async fn ready(&mut self) -> Result<()> {
211                    self.inner.ready().await
212                }
213
214                /// Returns whether the underlying sender appears ready.
215                pub fn is_ready(&self) -> bool {
216                    self.inner.is_ready()
217                }
218
219                /// Returns whether the underlying connection has closed.
220                pub fn is_closed(&self) -> bool {
221                    self.inner.is_closed()
222                }
223
224                /// Sends a request and records an outbound HTTP client span around it.
225                pub fn send_request(
226                    &mut self,
227                    mut request: Request<B>,
228                ) -> impl Future<Output = Result<Response<Incoming>>> + '_ {
229                    let span = HttpClientSpanBuilder::from(&request).start(&self.context);
230
231                    http::inject_context_on_context(
232                        span.context(),
233                        request.headers_mut(),
234                    );
235
236                    let future = self.inner.send_request(request);
237                    InstrumentedFuture::new(future, span)
238                }
239            }
240        }
241    };
242}
243
244#[cfg(feature = "hyper-http1")]
245impl_instrumented_send_request!(http1);
246#[cfg(feature = "hyper-http2")]
247impl_instrumented_send_request!(http2);
248
249#[cfg(all(test, any(feature = "hyper-http1", feature = "hyper-http2")))]
250mod tests {
251    use super::HyperSendRequestInstrument;
252    use crate::{Context, instrumentations::http::test_utils::*, semconv};
253    use axum::http::StatusCode;
254    use bytes::Bytes;
255    use http_body_util::Empty;
256    use hyper::{
257        Request,
258        header::{HOST, USER_AGENT},
259    };
260    use hyper_util::rt::TokioIo;
261    use opentelemetry::trace::{Span, SpanKind, TraceContextExt, Tracer, TracerProvider};
262    use serial_test::serial;
263    use tokio::net::TcpStream;
264    use tracing_subscriber::{Registry, layer::SubscriberExt};
265
266    #[cfg(feature = "hyper-http1")]
267    mod http1 {
268        use super::*;
269        use assert2::assert;
270
271        #[tokio::test]
272        #[serial]
273        async fn instruments_successful_requests() {
274            let telemetry = configure_test_tracing();
275            let server = spawn_server().await;
276            let io = TokioIo::new(TcpStream::connect(server.addr).await.unwrap());
277            let (send_request, connection) =
278                hyper::client::conn::http1::handshake(io).await.unwrap();
279
280            tokio::spawn(async move {
281                connection.await.unwrap();
282            });
283
284            let mut send_request = send_request.instrument();
285            let request_url = format!("{}/ok?ready=true", server.base_url);
286            let response = send_request
287                .send_request(
288                    Request::builder()
289                        .uri(&request_url)
290                        .header(HOST, server.authority())
291                        .header(USER_AGENT, "telemetry-rust-tests")
292                        .body(Empty::<Bytes>::new())
293                        .unwrap(),
294                )
295                .await
296                .unwrap();
297
298            assert!(response.status() == StatusCode::OK);
299
300            let spans = force_flush_and_get_spans(&telemetry);
301            let span = find_span(&spans, "GET");
302            let traceparent = server.state.traceparent_for("/ok").unwrap();
303            let (trace_id, span_id) = traceparent_ids(&traceparent);
304
305            assert!(span.span_kind == SpanKind::Client);
306            assert!(span.span_context.trace_id().to_string() == trace_id);
307            assert!(span.span_context.span_id().to_string() == span_id);
308            assert!(matches!(span.status, opentelemetry::trace::Status::Unset));
309            assert!(string_attr(span, semconv::HTTP_REQUEST_METHOD) == Some("GET"));
310            assert!(string_attr(span, semconv::URL_SCHEME) == Some("http"));
311            assert!(string_attr(span, semconv::SERVER_ADDRESS) == Some("127.0.0.1"));
312            assert!(
313                i64_attr(span, semconv::SERVER_PORT)
314                    == Some(i64::from(server.addr.port()))
315            );
316            assert!(string_attr(span, semconv::URL_PATH) == Some("/ok"));
317            assert!(string_attr(span, semconv::URL_QUERY) == Some("ready=true"));
318            assert!(
319                string_attr(span, semconv::USER_AGENT_ORIGINAL)
320                    == Some("telemetry-rust-tests")
321            );
322            assert!(string_attr(span, semconv::URL_FULL) == Some(request_url.as_str()));
323            assert!(i64_attr(span, semconv::HTTP_RESPONSE_STATUS_CODE) == Some(200));
324            assert!(string_attr(span, semconv::NETWORK_PROTOCOL_VERSION).is_some());
325            assert!(string_attr(span, semconv::NETWORK_PEER_ADDRESS).is_none());
326            assert!(i64_attr(span, semconv::NETWORK_PEER_PORT).is_none());
327        }
328
329        #[tokio::test]
330        #[serial]
331        async fn falls_back_to_host_header_for_url_full_on_origin_form_uri() {
332            let telemetry = configure_test_tracing();
333            let server = spawn_server().await;
334            let io = TokioIo::new(TcpStream::connect(server.addr).await.unwrap());
335            let (send_request, connection) =
336                hyper::client::conn::http1::handshake(io).await.unwrap();
337
338            tokio::spawn(async move {
339                connection.await.unwrap();
340            });
341
342            let mut send_request = send_request.instrument();
343            let response = send_request
344                .send_request(
345                    Request::builder()
346                        .uri("/ok?ready=true")
347                        .header(HOST, server.authority())
348                        .header(USER_AGENT, "telemetry-rust-tests")
349                        .body(Empty::<Bytes>::new())
350                        .unwrap(),
351                )
352                .await
353                .unwrap();
354
355            assert!(response.status() == StatusCode::OK);
356
357            let spans = force_flush_and_get_spans(&telemetry);
358            let span = find_span(&spans, "GET");
359
360            let expected_url = format!("//{}/ok?ready=true", server.authority());
361            assert!(string_attr(span, semconv::URL_FULL) == Some(expected_url.as_str()));
362            assert!(string_attr(span, semconv::SERVER_ADDRESS) == Some("127.0.0.1"));
363            assert!(
364                i64_attr(span, semconv::SERVER_PORT)
365                    == Some(i64::from(server.addr.port()))
366            );
367        }
368
369        #[tokio::test]
370        #[serial]
371        async fn falls_back_to_host_header_without_port_for_origin_form_uri() {
372            let telemetry = configure_test_tracing();
373            let server = spawn_server().await;
374            let io = TokioIo::new(TcpStream::connect(server.addr).await.unwrap());
375            let (send_request, connection) =
376                hyper::client::conn::http1::handshake(io).await.unwrap();
377
378            tokio::spawn(async move {
379                connection.await.unwrap();
380            });
381
382            let mut send_request = send_request.instrument();
383            let response = send_request
384                .send_request(
385                    Request::builder()
386                        .uri("/ok?ready=true")
387                        .header(HOST, "localhost")
388                        .header(USER_AGENT, "telemetry-rust-tests")
389                        .body(Empty::<Bytes>::new())
390                        .unwrap(),
391                )
392                .await
393                .unwrap();
394
395            assert!(response.status() == StatusCode::OK);
396
397            let spans = force_flush_and_get_spans(&telemetry);
398            let span = find_span(&spans, "GET");
399
400            assert!(
401                string_attr(span, semconv::URL_FULL) == Some("//localhost/ok?ready=true")
402            );
403            assert!(string_attr(span, semconv::SERVER_ADDRESS) == Some("localhost"));
404            assert!(i64_attr(span, semconv::SERVER_PORT).is_none());
405        }
406
407        #[tokio::test]
408        #[serial]
409        async fn ignores_invalid_host_header_for_origin_form_uri() {
410            let telemetry = configure_test_tracing();
411            let server = spawn_server().await;
412            let io = TokioIo::new(TcpStream::connect(server.addr).await.unwrap());
413            let (send_request, connection) =
414                hyper::client::conn::http1::handshake(io).await.unwrap();
415
416            tokio::spawn(async move {
417                connection.await.unwrap();
418            });
419
420            let mut send_request = send_request.instrument();
421            let response = send_request
422                .send_request(
423                    Request::builder()
424                        .uri("/ok?ready=true")
425                        .header(HOST, "bad host value")
426                        .header(USER_AGENT, "telemetry-rust-tests")
427                        .body(Empty::<Bytes>::new())
428                        .unwrap(),
429                )
430                .await
431                .unwrap();
432
433            assert!(response.status() == StatusCode::OK);
434
435            let spans = force_flush_and_get_spans(&telemetry);
436            let span = find_span(&spans, "GET");
437
438            assert!(string_attr(span, semconv::URL_FULL).is_none());
439            assert!(string_attr(span, semconv::SERVER_ADDRESS).is_none());
440            assert!(i64_attr(span, semconv::SERVER_PORT).is_none());
441        }
442
443        #[tokio::test]
444        #[serial]
445        async fn uri_values_take_precedence_over_host_header_fallback() {
446            let telemetry = configure_test_tracing();
447            let server = spawn_server().await;
448            let io = TokioIo::new(TcpStream::connect(server.addr).await.unwrap());
449            let (send_request, connection) =
450                hyper::client::conn::http1::handshake(io).await.unwrap();
451
452            tokio::spawn(async move {
453                connection.await.unwrap();
454            });
455
456            let mut send_request = send_request.instrument();
457            let request_url = format!("{}/ok?ready=true", server.base_url);
458            let response = send_request
459                .send_request(
460                    Request::builder()
461                        .uri(&request_url)
462                        .header(HOST, "example.com:1234")
463                        .header(USER_AGENT, "telemetry-rust-tests")
464                        .body(Empty::<Bytes>::new())
465                        .unwrap(),
466                )
467                .await
468                .unwrap();
469
470            assert!(response.status() == StatusCode::OK);
471
472            let spans = force_flush_and_get_spans(&telemetry);
473            let span = find_span(&spans, "GET");
474
475            assert!(string_attr(span, semconv::URL_FULL) == Some(request_url.as_str()));
476            assert!(string_attr(span, semconv::SERVER_ADDRESS) == Some("127.0.0.1"));
477            assert!(
478                i64_attr(span, semconv::SERVER_PORT)
479                    == Some(i64::from(server.addr.port()))
480            );
481        }
482
483        #[tokio::test]
484        #[serial]
485        async fn uses_explicit_parent_context_when_provided() {
486            let telemetry = configure_test_tracing();
487            let server = spawn_server().await;
488            let io = TokioIo::new(TcpStream::connect(server.addr).await.unwrap());
489            let (send_request, connection) =
490                hyper::client::conn::http1::handshake(io).await.unwrap();
491            let tracer = telemetry.provider.tracer("hyper-tests");
492            let explicit_parent = tracer.start("explicit-parent");
493            let explicit_parent_span_id = explicit_parent.span_context().span_id();
494            let explicit_parent_cx = Context::current_with_span(explicit_parent);
495            let tracing_tracer = telemetry.provider.tracer("tracing-tests");
496            let subscriber = Registry::default()
497                .with(tracing_opentelemetry::layer().with_tracer(tracing_tracer));
498            let guard = tracing::subscriber::set_default(subscriber);
499            let current_parent = tracing::info_span!("current-parent");
500
501            tokio::spawn(async move {
502                connection.await.unwrap();
503            });
504
505            tracing::Instrument::instrument(
506                async {
507                    let mut send_request =
508                        send_request.instrument().context(&explicit_parent_cx);
509                    send_request
510                        .send_request(
511                            Request::builder()
512                                .uri(format!("{}/ok", server.base_url))
513                                .header(HOST, server.authority())
514                                .body(Empty::<Bytes>::new())
515                                .unwrap(),
516                        )
517                        .await
518                        .unwrap();
519                },
520                current_parent,
521            )
522            .await;
523
524            drop(guard);
525            explicit_parent_cx.span().end();
526
527            let spans = force_flush_and_get_spans(&telemetry);
528            let hyper_span = find_span(&spans, "GET");
529            let current_span = find_span(&spans, "current-parent");
530
531            assert!(hyper_span.parent_span_id == explicit_parent_span_id);
532            assert!(hyper_span.parent_span_id != current_span.span_context.span_id());
533        }
534
535        #[tokio::test]
536        #[serial]
537        async fn marks_error_responses_as_errors() {
538            let telemetry = configure_test_tracing();
539            let server = spawn_server().await;
540            let io = TokioIo::new(TcpStream::connect(server.addr).await.unwrap());
541            let (send_request, connection) =
542                hyper::client::conn::http1::handshake(io).await.unwrap();
543
544            tokio::spawn(async move {
545                connection.await.unwrap();
546            });
547
548            let mut send_request = send_request.instrument();
549            let response = send_request
550                .send_request(
551                    Request::builder()
552                        .uri(format!("{}/not-found", server.base_url))
553                        .header(HOST, server.authority())
554                        .body(Empty::<Bytes>::new())
555                        .unwrap(),
556                )
557                .await
558                .unwrap();
559
560            assert!(response.status() == StatusCode::NOT_FOUND);
561
562            let spans = force_flush_and_get_spans(&telemetry);
563            let span = find_span(&spans, "GET");
564
565            assert!(matches!(
566                span.status,
567                opentelemetry::trace::Status::Error { .. }
568            ));
569            assert!(i64_attr(span, semconv::HTTP_RESPONSE_STATUS_CODE) == Some(404));
570            assert!(string_attr(span, semconv::ERROR_TYPE) == Some("404"));
571        }
572    }
573
574    #[cfg(feature = "hyper-http2")]
575    mod http2 {
576        use super::*;
577        use assert2::assert;
578        use hyper_util::rt::TokioExecutor;
579
580        #[tokio::test]
581        #[serial]
582        async fn instruments_successful_requests() {
583            let telemetry = configure_test_tracing();
584            let server = spawn_server().await;
585            let io = TokioIo::new(TcpStream::connect(server.addr).await.unwrap());
586            let (send_request, connection) =
587                hyper::client::conn::http2::Builder::new(TokioExecutor::new())
588                    .handshake(io)
589                    .await
590                    .unwrap();
591
592            tokio::spawn(async move {
593                connection.await.unwrap();
594            });
595
596            let mut send_request = send_request.instrument();
597            let request_url = format!("{}/ok?ready=true", server.base_url);
598            let response = send_request
599                .send_request(
600                    Request::builder()
601                        .uri(&request_url)
602                        .header(HOST, server.authority())
603                        .header(USER_AGENT, "telemetry-rust-tests")
604                        .body(Empty::<Bytes>::new())
605                        .unwrap(),
606                )
607                .await
608                .unwrap();
609
610            assert!(response.status() == StatusCode::OK);
611
612            let spans = force_flush_and_get_spans(&telemetry);
613            let span = find_span(&spans, "GET");
614            let traceparent = server.state.traceparent_for("/ok").unwrap();
615            let (trace_id, span_id) = traceparent_ids(&traceparent);
616
617            assert!(span.span_kind == SpanKind::Client);
618            assert!(span.span_context.trace_id().to_string() == trace_id);
619            assert!(span.span_context.span_id().to_string() == span_id);
620            assert!(matches!(span.status, opentelemetry::trace::Status::Unset));
621            assert!(string_attr(span, semconv::HTTP_REQUEST_METHOD) == Some("GET"));
622            assert!(string_attr(span, semconv::URL_SCHEME) == Some("http"));
623            assert!(string_attr(span, semconv::SERVER_ADDRESS) == Some("127.0.0.1"));
624            assert!(
625                i64_attr(span, semconv::SERVER_PORT)
626                    == Some(i64::from(server.addr.port()))
627            );
628            assert!(string_attr(span, semconv::URL_PATH) == Some("/ok"));
629            assert!(string_attr(span, semconv::URL_QUERY) == Some("ready=true"));
630            assert!(string_attr(span, semconv::URL_FULL) == Some(request_url.as_str()));
631            assert!(i64_attr(span, semconv::HTTP_RESPONSE_STATUS_CODE) == Some(200));
632            assert!(string_attr(span, semconv::NETWORK_PROTOCOL_VERSION).is_some());
633        }
634    }
635}
telemetry_rust/instrumentations/http/hyper/mod.rs

telemetry_rust/instrumentations/http/hyper/
mod.rs
