Init tracing

kanin/Cargo.toml

@@ -13,13 +13,17 @@ readme = "../README.md"
 
 [dependencies]
 # Derive macros for traits in kanin.
-kanin_derive = "0.5.0"
+# kanin_derive = "0.5.1"
+kanin_derive = { path = "../kanin_derive" }
 
 # Lower level AMQP framework.
 lapin = "2.1.0"
 
-# Generalized logging framework.
-log = "0.4"
+# Generalized tracing framework.
+tracing = "0.1.37"
+
+# Used to create unique request IDs.
+uuid = { version = "1.4.1", features = ["v4"] }
 
 # Asynchronous runtime.
 tokio = { version = "1.18.0", features = ["rt", "rt-multi-thread", "macros"] }

kanin/src/app.rs

@@ -7,8 +7,8 @@ use std::{any::Any, sync::Arc};
 use anymap::Map;
 use futures::future::{select_all, SelectAll};
 use lapin::{self, Connection, ConnectionProperties};
-use log::{debug, error, info, trace};
 use tokio::task::JoinHandle;
+use tracing::{debug, error, info, trace};
 
 use self::task::TaskFactory;
 use crate::{extract::State, Error, Handler, HandlerConfig, Respond, Result};
@@ -111,7 +111,9 @@ impl App {
  #[inline]
  pub async fn run(self, amqp_addr: &str) -> Result<()> {
  debug!("Connecting to AMQP on address: {amqp_addr:?} ...");
- let conn = Connection::connect(amqp_addr, ConnectionProperties::default()).await?;
+ let conn = Connection::connect(amqp_addr, ConnectionProperties::default())
+ .await
+ .map_err(Error::Lapin)?;
  trace!("Connected to AMQP on address: {amqp_addr:?}");
  self.run_with_connection(&conn).await
  }
@@ -177,7 +179,10 @@ impl App {
  );
 
  // Construct the task from the factory. This produces a pinned future which we can then spawn.
- let task = task_factory.build(conn, state.clone()).await?;
+ let task = task_factory
+ .build(conn, state.clone())
+ .await
+ .map_err(Error::Lapin)?;
 
  // Spawn the task and save the join handle.
  join_handles.push(tokio::spawn(task));

kanin/src/app/task.rs

@@ -9,7 +9,7 @@ use lapin::{
  types::FieldTable,
  BasicProperties, Channel, Connection, Consumer,
 };
-use log::{debug, error, trace, warn};
+use tracing::{debug, error, info, info_span, trace, warn, Instrument};
 
 use crate::{Handler, HandlerConfig, Request, Respond};
 
@@ -25,6 +25,7 @@ use super::StateMap;
 pub(super) type HandlerTask = Pin<Box<dyn Future<Output = String> + Send>>;
 
 /// Creates the handler task for the given handler and routing key. See [`HandlerTask`].
+#[tracing::instrument(skip_all, field(routing_key = routing_key))]
 fn handler_task<H, Args, Res>(
  routing_key: String,
  handler: H,
@@ -47,7 +48,7 @@ where
  let delivery = tokio::select! {
  // Listen on new deliveries.
  delivery = consumer.next() => match delivery {
- // Received a delivery succesfully, just unwrap it from the option.
+ // Received a delivery successfully, just unwrap it from the option.
  Some(delivery) => delivery,
  // We should only ever get to this point if the consumer is cancelled.
  // We'll just return the routing key - might be a help for the user to see which
@@ -79,7 +80,11 @@ where
  // Requests are handled and replied to concurrently.
  // This allows each handler task to process multiple requests at once.
  tasks.push(tokio::spawn(async move {
- handle_request(req, handler, channel, should_reply).await;
+ let span = info_span!("request", req_id = %req.req_id);
+
+ handle_request(req, handler, channel, should_reply)
+ .instrument(span)
+ .await;
  }));
  }
  })
@@ -100,12 +105,14 @@ async fn handle_request<H, Args, Res>(
  let properties = req.properties().cloned();
  let reply_to = properties.as_ref().and_then(|p| p.reply_to().clone());
  let correlation_id = properties.as_ref().and_then(|p| p.correlation_id().clone());
+ let app_id = req.app_id().unwrap_or("<unknown>");
+
+ let handler_name = std::any::type_name::<H>();
+ info!("Received request on handler {handler_name:?} from {app_id}");
+
  // Call the handler with the request.
  let response = handler.call(&mut req).await;
- debug!(
- "Handler {:?} produced response: {response:?}",
- std::any::type_name::<H>()
- );
+ info!("Handler produced response: {response:?}");
 
  let bytes_response = response.respond();
 

kanin/src/error.rs

@@ -2,9 +2,9 @@
 
 use std::convert::Infallible;
 
-use log::{error, warn};
 use prost::DecodeError;
 use thiserror::Error as ThisError;
+use tracing::{error, warn};
 
 /// Errors that may be returned by `kanin`, especially when the app runs.
 #[derive(Debug, ThisError)]
@@ -14,19 +14,19 @@ pub enum Error {
  NoHandlers,
  /// An error from an underlying lapin call.
  #[error("An underlying `lapin` call failed: {0}")]
- Lapin(#[from] lapin::Error),
+ Lapin(lapin::Error),
 }
 
 /// Errors that may be produced by handlers. Failing extractors provided by `kanin` return this error.
 #[derive(Debug, ThisError)]
 pub enum HandlerError {
  /// Errors due to invalid requests.
  #[error("Invalid Request: {0:#}")]
- InvalidRequest(#[from] RequestError),
+ InvalidRequest(RequestError),
 
  /// Internal server errors that are not the fault of clients.
  #[error("Internal Server Error: {0:#}")]
- Internal(#[from] ServerError),
+ Internal(ServerError),
 }
 
 impl HandlerError {
@@ -42,7 +42,7 @@ pub enum RequestError {
  ///
  /// This error is left as an opaque error as that is what is provided by [`prost`].
  #[error("Message could not be decoded into the required type: {0:#}")]
- DecodeError(#[from] DecodeError),
+ DecodeError(DecodeError),
 }
 
 /// Errors due to bad configuration or usage from the server-side.
@@ -91,7 +91,7 @@ where
 
 impl From<DecodeError> for HandlerError {
  fn from(e: DecodeError) -> Self {
- RequestError::from(e).into()
+ HandlerError::InvalidRequest(RequestError::DecodeError(e))
  }
 }
 

kanin/src/extract.rs

@@ -1,16 +1,18 @@
 //! Interface for types that can extract themselves from requests.
 
 mod message;
+mod req_id;
 mod state;
 
-use std::convert::Infallible;
+use std::{convert::Infallible, error::Error};
 
 use async_trait::async_trait;
 use lapin::{acker::Acker, message::Delivery, Channel};
 
 use crate::{error::HandlerError, Request};
 
 pub use message::Msg;
+pub use req_id::ReqId;
 pub use state::State;
 
 /// A trait for types that can be extracted from [requests](`Request`).
@@ -20,7 +22,7 @@ pub use state::State;
 #[async_trait]
 pub trait Extract: Sized {
  /// The error to return in case extraction fails.
- type Error;
+ type Error: Error;
 
  /// Extract the type from the request.
  async fn extract(req: &mut Request) -> Result<Self, Self::Error>;

kanin/src/extract/req_id.rs

@@ -0,0 +1,78 @@
+//! Request IDs.
+
+use core::fmt;
+use std::convert::Infallible;
+
+use async_trait::async_trait;
+use lapin::{
+ message::Delivery,
+ types::{AMQPValue, LongString},
+};
+use uuid::Uuid;
+
+use crate::{Extract, Request};
+
+/// Request IDs allow concurrent logs to be associated with a unique request. It can also enable requests
+/// to be traced between different services by propagating the request IDs when calling other services.
+/// This type implements [`Extract`], so it can be used in handlers.
+#[derive(Debug, Clone)]
+pub struct ReqId(AMQPValue);
+
+impl ReqId {
+ /// Create a new [`ReqId`] as a random UUID.
+ fn new() -> Self {
+ let uuid = Uuid::new_v4();
+ let amqp_value = AMQPValue::LongString(LongString::from(uuid.to_string()));
+ Self(amqp_value)
+ }
+}
+
+/// [`AMQPValue`] does not implement `Display` but we provide a `Display` implementation for
+/// `ReqId` to allow it to be used in tracing spans (see the `tracing` crate).
+impl fmt::Display for ReqId {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ match &self.0 {
+ AMQPValue::LongString(req_id) => req_id.fmt(f),
+ AMQPValue::Boolean(b) => b.fmt(f),
+ AMQPValue::ShortShortInt(v) => v.fmt(f),
+ AMQPValue::ShortShortUInt(v) => v.fmt(f),
+ AMQPValue::ShortInt(v) => v.fmt(f),
+ AMQPValue::ShortUInt(v) => v.fmt(f),
+ AMQPValue::LongInt(v) => v.fmt(f),
+ AMQPValue::LongUInt(v) => v.fmt(f),
+ AMQPValue::LongLongInt(v) => v.fmt(f),
+ AMQPValue::Float(v) => v.fmt(f),
+ AMQPValue::Double(v) => v.fmt(f),
+ AMQPValue::DecimalValue(v) => write!(f, "{v:?}"),
+ AMQPValue::ShortString(v) => write!(f, "{v:?}"),
+ AMQPValue::FieldArray(v) => write!(f, "{v:?}"),
+ AMQPValue::Timestamp(v) => write!(f, "{v:?}"),
+ AMQPValue::FieldTable(v) => write!(f, "{v:?}"),
+ AMQPValue::ByteArray(v) => write!(f, "{v:?}"),
+ AMQPValue::Void => write!(f, "Void"),
+ }
+ }
+}
+
+impl From<&Delivery> for ReqId {
+ fn from(delivery: &Delivery) -> Self {
+ let Some(headers) = delivery.properties.headers() else {
+ return Self::new();
+ };
+
+ let Some(req_id) = headers.inner().get("req_id") else {
+ return Self::new();
+ };
+
+ Self(req_id.clone())
+ }
+}
+
+#[async_trait]
+impl Extract for ReqId {
+ type Error = Infallible;
+
+ async fn extract(req: &mut Request) -> Result<Self, Self::Error> {
+ Ok(req.req_id.clone())
+ }
+}

kanin/src/extract/state.rs

@@ -4,7 +4,7 @@ use std::sync::Arc;
 
 use async_trait::async_trait;
 use derive_more::{Deref, DerefMut};
-use log::error;
+use tracing::error;
 
 use crate::{error::ServerError, Extract, HandlerError, Request};
 

kanin/src/handler.rs

@@ -46,7 +46,10 @@ macro_rules! impl_handler {
  $(
  let $ty = match $ty::extract(req).await {
  Ok(value) => value,
- Err(error) => return Res::from_error(error),
+ Err(error) => {
+ tracing::error!("{error}");
+ return Res::from_error(error);
+ }
  };
  )*
 

kanin/src/lib.rs

@@ -146,7 +146,7 @@ mod tests {
  use std::time::Duration;
 
  use lapin::{Connection, ConnectionProperties};
- use log::warn;
+ use tracing::warn;
 
  const TEST_AMQP_ADDR: &str = "amqp://localhost";
 

kanin/src/request.rs

@@ -7,6 +7,7 @@ use lapin::protocol::basic::AMQPProperties;
 use lapin::{message::Delivery, Channel};
 
 use crate::app::StateMap;
+use crate::extract::ReqId;
 
 /// An AMQP request.
 #[derive(Debug)]
@@ -15,6 +16,9 @@ pub struct Request {
  state: Arc<StateMap>,
  /// The channel the message was received on.
  channel: Channel,
+ /// Request ID. This is a unique ID for every request. Either a newly created uuid or whatever
+ /// is found in the header of the incoming amqp message.
+ pub(crate) req_id: ReqId,
  /// The message delivery.
  pub(crate) delivery: Option<Delivery>,
 }
@@ -25,6 +29,7 @@ impl Request {
  Self {
  state,
  channel,
+ req_id: ReqId::from(&delivery),
  delivery: Some(delivery),
  }
  }
@@ -44,4 +49,11 @@ impl Request {
  pub fn properties(&self) -> Option<&AMQPProperties> {
  self.delivery.as_ref().map(|d| &d.properties)
  }
+
+ /// Return `app_id` of the sender of the request.
+ pub fn app_id(&self) -> Option<&str> {
+ self.properties()
+ .and_then(|p| p.app_id().as_ref())
+ .map(|app_id| app_id.as_str())
+ }
 }