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 a unique req_id
+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,9 +9,9 @@ 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};
+use crate::{extract::ReqId, Extract, Handler, HandlerConfig, Request, Respond};
 
 use super::StateMap;
 
@@ -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
@@ -64,7 +65,7 @@ where
  },
  };
 
- let req = match delivery {
+ let mut req = match delivery {
  Err(e) => {
  error!("Error when receiving delivery on routing key \"{routing_key}\": {e:#}");
  continue;
@@ -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 req_id = ReqId::extract(&mut req).await.expect("infallible");
+
+ handle_request(req, handler, channel, should_reply)
+ .instrument(info_span!("request", %req_id))
+ .await;
  }));
  }
  })
@@ -100,12 +105,18 @@ 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 = properties
+ .as_ref()
+ .and_then(|p| p.app_id().as_ref())
+ .map(|app_id| app_id.as_str())
+ .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 {handler_name:?} 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,68 @@
+//! Req id ...
+
+use core::fmt;
+use std::convert::Infallible;
+
+use async_trait::async_trait;
+use lapin::types::{AMQPValue, LongString};
+use uuid::Uuid;
+
+use crate::{Extract, Request};
+
+/// Request id.
+pub struct ReqId(AMQPValue);
+
+impl ReqId {
+ /// Create a new ReqId with a random uuid.
+ fn new() -> Self {
+ let uuid = Uuid::new_v4();
+ let amqp_value = AMQPValue::LongString(LongString::from(uuid.to_string()));
+ Self(amqp_value)
+ }
+}
+
+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"),
+ }
+ }
+}
+
+/// Extract implementation for protobuf messages.
+#[async_trait]
+impl Extract for ReqId {
+ type Error = Infallible;
+
+ async fn extract(req: &mut Request) -> Result<Self, Self::Error> {
+ let Some(delivery) = &req.delivery else {
+ return Ok(Self::new());
+ };
+
+ let Some(headers) = delivery.properties.headers() else {
+ return Ok(Self::new());
+ };
+
+ match headers.inner().get("req_id") {
+ None => Ok(Self::new()),
+ Some(req_id) => Ok(Self(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/tests/send_recv.rs

@@ -2,13 +2,20 @@ use std::sync::{Arc, Mutex};
 
 use async_trait::async_trait;
 use futures::future::join_all;
-use lapin::{message::Delivery, options::BasicPublishOptions, BasicProperties, Channel};
-use log::info;
+use lapin::{
+ message::Delivery,
+ options::BasicPublishOptions,
+ types::{AMQPValue, FieldTable},
+ BasicProperties, Channel,
+};
 use tokio::sync::{mpsc::Sender, OnceCell};
+use tracing::info;
 
 use crate::{
- error::FromError, extract::State, tests::init_logging, App, Extract, HandlerError, Request,
- Respond,
+ error::FromError,
+ extract::{ReqId, State},
+ tests::init_logging,
+ App, Extract, HandlerError, Request, Respond,
 };
 
 use super::amqp_connect;
@@ -67,6 +74,12 @@ async fn handler_delivery(_delivery: Delivery) -> MyResponse {
  MyResponse("handler_delivery".into())
 }
 
+async fn handler_req_id(req_id: ReqId) -> MyResponse {
+ assert_eq!(format!("{req_id}"), "abc");
+ SYNC.get().unwrap().send(()).await.unwrap();
+ MyResponse("handler_req_id".into())
+}
+
 async fn handler_two_extractors(_channel: Channel, _delivery: Delivery) -> MyResponse {
  SYNC.get().unwrap().send(()).await.unwrap();
  MyResponse("handler_two_extractors".into())
@@ -107,6 +120,7 @@ async fn it_receives_various_messages_and_works_as_expected() {
  .handler("handler", handler)
  .handler("handler_channel", handler_channel)
  .handler("handler_delivery", handler_delivery)
+ .handler("handler_req_id", handler_req_id)
  .handler("handler_two_extractors", handler_two_extractors)
  .handler("handler_state_extractor", handler_state_extractor)
  .handler("listener", listener)
@@ -131,13 +145,18 @@ async fn it_receives_various_messages_and_works_as_expected() {
  .expect("failed to create channel");
 
  let send_msg = |routing_key: &'static str, reply_to: &'static str| async {
+ let mut headers = FieldTable::default();
+ headers.insert("req_id".into(), AMQPValue::LongString("abc".into()));
+
  channel
  .basic_publish(
  "",
  routing_key,
  BasicPublishOptions::default(),
  &[],
- BasicProperties::default().with_reply_to(reply_to.into()),
+ BasicProperties::default()
+ .with_reply_to(reply_to.into())
+ .with_headers(headers),
  )
  .await
  .expect("failed to publish");
@@ -161,6 +180,9 @@ async fn it_receives_various_messages_and_works_as_expected() {
  send_msg("handler_delivery", "handler_message_reply_to").await;
  recv.recv().await.unwrap();
  recv.recv().await.unwrap();
+ send_msg("handler_req_id", "handler_message_reply_to").await;
+ recv.recv().await.unwrap();
+ recv.recv().await.unwrap();
  send_msg("handler_two_extractors", "handler_message_reply_to").await;
  recv.recv().await.unwrap();
  recv.recv().await.unwrap();
@@ -213,6 +235,7 @@ async fn it_receives_various_messages_and_works_as_expected() {
  // "handler",
  // "handler_channel",
  // "handler_delivery",
+ // "handler_req_id",
  // "handler_two_extractors",
  "handler_state_extractor",
  "listener",
@@ -229,6 +252,7 @@ async fn it_receives_various_messages_and_works_as_expected() {
  "handler_message",
  "handler_message",
  "handler_message",
+ "handler_message",
  "shutdown",
  ],
  recv_calls.as_ref()