From dabe904ccabed0f50a501a20cfe5eb4d1a1fc7ef Mon Sep 17 00:00:00 2001 From: Nick Furfaro Date: Tue, 22 Jul 2025 12:08:08 -0600 Subject: [PATCH] chore: add tracing to api --- Cargo.lock | 1 + crates/sigma-api/Cargo.toml | 1 + crates/sigma-api/src/lib.rs | 2 +- crates/sigma-api/src/server.rs | 60 ++++++++++++++++++++++++++++------ 4 files changed, 53 insertions(+), 11 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 23ff91f..2e9a32e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3655,6 +3655,7 @@ dependencies = [ "tokio", "tower", "tower-http", + "tracing", ] [[package]] diff --git a/crates/sigma-api/Cargo.toml b/crates/sigma-api/Cargo.toml index 1bca6c2..8bf1f8e 100644 --- a/crates/sigma-api/Cargo.toml +++ b/crates/sigma-api/Cargo.toml @@ -19,3 +19,4 @@ thiserror.workspace = true tokio.workspace = true tower.workspace = true tower-http.workspace = true +tracing.workspace = true diff --git a/crates/sigma-api/src/lib.rs b/crates/sigma-api/src/lib.rs index 7c57964..1972a5f 100644 --- a/crates/sigma-api/src/lib.rs +++ b/crates/sigma-api/src/lib.rs @@ -65,7 +65,7 @@ impl CurrentHeight { } /// Converts the current height and another height into a [`Height`] [`Range`]. - /// Not the range is exclusive but the other height is included. + /// Note: the range is exclusive but the other height is included. pub fn to_range(&self, other: &Self) -> Range { match (self, other) { (CurrentHeight::Empty, CurrentHeight::Empty) => 0..0, diff --git a/crates/sigma-api/src/server.rs b/crates/sigma-api/src/server.rs index f445a97..9eb5109 100644 --- a/crates/sigma-api/src/server.rs +++ b/crates/sigma-api/src/server.rs @@ -24,9 +24,15 @@ pub type ServeConnError = Box; /// This constructs a new `JoinSet` to use for limiting connections and then /// calls [`serve_next_conn`] in a loop. Any outstanding connections will not be /// counted toward the connection limit. +#[tracing::instrument( + name = "sigma-api::serve", + level = "info", + skip(router, listener, conn_limit) +)] pub async fn serve(router: &Router, listener: &TcpListener, conn_limit: usize) { let mut conn_set = JoinSet::new(); loop { + tracing::info!("Waiting for next connection"); serve_next_conn(router, listener, conn_limit, &mut conn_set).await; } } @@ -38,23 +44,44 @@ pub async fn serve(router: &Router, listener: &TcpListener, conn_limit: usize) { /// /// If we're at the connection limit, this first awaits for a connection task to /// become available. + +#[tracing::instrument( + name = "sigma_api::serve_next_conn", + level = "info", + skip(router, listener, conn_set), + fields(conn_limit) +)] pub async fn serve_next_conn( router: &Router, listener: &TcpListener, conn_limit: usize, conn_set: &mut JoinSet<()>, ) { + tracing::debug!("Waiting for next connection"); + // Await the next connection. let stream = match next_conn(listener, conn_limit, conn_set).await { - Ok((stream, _remote_addr)) => stream, - Err(_err) => { + Ok((stream, remote_addr)) => { + tracing::info!(%remote_addr, "Connection accepted"); + stream + } + Err(err) => { + tracing::error!(%err, "Failed to accept connection"); return; } }; // Serve the acquired connection. let router = router.clone(); - conn_set.spawn(async move { if let Err(_err) = serve_conn(&router, stream).await {} }); + conn_set.spawn(async move { + if let Err(err) = serve_conn(&router, stream).await { + tracing::error!(%err, "Connection handler exited with error"); + } else { + tracing::debug!("Connection handler completed successfully"); + } + }); + + tracing::debug!("Connection task spawned"); } /// Accept and return the next TCP stream connection. @@ -68,6 +95,7 @@ pub async fn next_conn( ) -> io::Result<(TcpStream, SocketAddr)> { // If the `conn_set` size currently exceeds the limit, wait for the next to join. if conn_set.len() >= conn_limit { + tracing::debug!("Connection limit reached, waiting for a connection to finish"); conn_set.join_next().await.expect("set cannot be empty")?; } // Await another connection. @@ -75,24 +103,36 @@ pub async fn next_conn( } /// Serve a newly accepted TCP stream. +#[tracing::instrument( + name = "sigma_api::serve_conn", + level = "info", + skip(router, stream), + fields(peer_addr = ?stream.peer_addr().ok()) +)] pub async fn serve_conn(router: &Router, stream: TcpStream) -> Result<(), ServeConnError> { - // Hyper has its own `AsyncRead` and `AsyncWrite` traits and doesn't use - // tokio. `TokioIo` converts between them. + tracing::info!("Starting to serve connection"); + let stream = hyper_util::rt::TokioIo::new(stream); - // Hyper also has its own `Service` trait and doesn't use tower. We can use - // `hyper::service::service_fn` to create a hyper `Service` that calls our - // app through `tower::Service::call`. let hyper_service = hyper::service::service_fn( move |request: axum::extract::Request| { tower::Service::call(&mut router.clone(), request) }, ); - // `TokioExecutor` tells hyper to use `tokio::spawn` to spawn tasks. let executor = hyper_util::rt::TokioExecutor::new(); let conn = hyper_util::server::conn::auto::Builder::new(executor).http2_only(); - conn.serve_connection(stream, hyper_service).await + + match conn.serve_connection(stream, hyper_service).await { + Ok(()) => { + tracing::info!("Connection served successfully"); + Ok(()) + } + Err(e) => { + tracing::error!(error = %e, "Failed to serve connection"); + Err(e) + } + } } /// Construct the endpoint router with the node [`endpoint`]s, CORS layer and DB