feat(tracing): auto register tracing layer (#324)

* feat: auto register tracing layer

* feat: construct a new `LogItem` for tracing

This also takes care of the formatting in the `log()` function.

* feat(logger): change mpsc to crossbeam_channel

This also removes the `Logger.filter` field since filtering is already handled by the EnvFilter

* refactor(logger): filter fields before inserting them

* tests(logger): add test

* refactor(logger): quick cleanup

* refactor(logger): cargo fmt

* refactor: cargo sort

* test: update for tracing

* test: update

* test: work stealing

Co-authored-by: chesedo <pieter@chesedo.me>

.gitignore

@@ -1,4 +1,5 @@
 # Rust
+.cargo
 # Generated by Cargo
 # will have compiled files and executables
 **/target/

api/Cargo.toml

@@ -12,6 +12,7 @@ cargo = "0.64.0"
 cargo-util = "0.2.1"
 chrono = "0.4.22"
 clap = { version = "3.2.17", features = ["derive"] }
+crossbeam-channel = "0.5.6"
 env_logger = "0.9.0"
 fqdn = "0.2.2"
 futures = "0.3.23"

api/src/deployment.rs

@@ -21,7 +21,6 @@ use shuttle_proto::provisioner::provisioner_client::ProvisionerClient;
 use shuttle_service::loader::Loader;
 use shuttle_service::logger::Log;
 use shuttle_service::ServeHandle;
-use tokio::sync::mpsc::UnboundedSender;
 use tokio::sync::{mpsc, RwLock};
 use tonic::transport::{Channel, Endpoint};
 
@@ -108,7 +107,11 @@ impl Deployment {
 
     /// Tries to advance the deployment one stage. Does nothing if the deployment
     /// is in a terminal state.
-    pub(crate) async fn advance(&self, context: &Context, run_logs_tx: UnboundedSender<Log>) {
+    pub(crate) async fn advance(
+        &self,
+        context: &Context,
+        run_logs_tx: crossbeam_channel::Sender<Log>,
+    ) {
         {
             trace!("waiting to get write on the state");
             let meta = self.meta().await;
@@ -307,7 +310,7 @@ struct JobQueue {
 }
 
 impl JobQueue {
-    async fn new(context: Context, run_logs_tx: UnboundedSender<Log>) -> Self {
+    async fn new(context: Context, run_logs_tx: crossbeam_channel::Sender<Log>) -> Self {
         let (send, mut recv) = mpsc::channel::<Arc<Deployment>>(JOB_QUEUE_SIZE);
 
         log::debug!("starting job processor task");
@@ -360,7 +363,7 @@ impl DeploymentSystem {
         version_req: VersionReq,
     ) -> Self {
         let router: Arc<Router> = Default::default();
-        let (tx, mut rx) = mpsc::unbounded_channel::<Log>();
+        let (tx, rx) = crossbeam_channel::unbounded::<Log>();
 
         let deployments = Arc::new(RwLock::new(
             Self::initialise_from_fs(&build_system.fs_root(), &fqdn).await,
@@ -369,7 +372,7 @@ impl DeploymentSystem {
         let deployments_log = deployments.clone();
 
         tokio::spawn(async move {
-            while let Some(log) = rx.recv().await {
+            while let Ok(log) = rx.recv() {
                 let mut deployments_log = deployments_log.write().await;
 
                 if let Some(deployment) = deployments_log.get_mut(&log.deployment_id) {

cargo-shuttle/Cargo.toml

@@ -16,6 +16,7 @@ cargo_metadata = "0.15.0"
 chrono = "0.4.22"
 clap = { version = "3.2.17", features = ["derive", "env"] }
 colored = "2.0.0"
+crossbeam-channel = "0.5.6"
 crossterm = "0.25.0"
 dirs = "4.0.0"
 env_logger = "0.9.0"

cargo-shuttle/src/lib.rs

@@ -26,7 +26,6 @@ use semver::{Version, VersionReq};
 use shuttle_secrets::SecretStore;
 use shuttle_service::loader::{build_crate, Loader};
 use shuttle_service::Factory;
-use tokio::sync::mpsc;
 use toml_edit::Document;
 use uuid::Uuid;
 
@@ -193,7 +192,7 @@ impl Shuttle {
         let mut factory = LocalFactory::new(self.ctx.project_name().clone())?;
         let addr = SocketAddr::new(Ipv4Addr::LOCALHOST.into(), run_args.port);
         let deployment_id = Uuid::new_v4();
-        let (tx, mut rx) = mpsc::unbounded_channel();
+        let (tx, rx) = crossbeam_channel::unbounded();
 
         // Load secrets
         let secrets = self.ctx.secrets();
@@ -222,7 +221,7 @@ impl Shuttle {
         let (handle, so) = loader.load(&mut factory, addr, tx, deployment_id).await?;
 
         tokio::spawn(async move {
-            while let Some(log) = rx.recv().await {
+            while let Ok(log) = rx.recv() {
                 print::log(log.datetime, log.item);
             }
         });

cargo-shuttle/src/print.rs

@@ -1,27 +1,44 @@
 use chrono::{DateTime, Local, Utc};
 use colored::{ColoredString, Colorize};
-use log::Level;
+use serde_json::{Map, Value};
 use shuttle_common::LogItem;
 
 pub fn log(datetime: DateTime<Utc>, log_item: LogItem) {
     let datetime: DateTime<Local> = DateTime::from(datetime);
+
+    let mut fields: Map<String, Value> = serde_json::from_slice(&log_item.fields).unwrap();
+
+    let message = fields
+        .remove("message")
+        .map_or("".to_owned(), |msg| format!(" {}", msg.as_str().unwrap()));
+
     println!(
-        "{}{} {:<5} {}{} {}",
+        "{}{} {:<5} {}{}{} {}",
         "[".bright_black(),
-        datetime.format("%Y-%m-%dT%H:%M:%SZ"),
+        datetime.format("%Y-%m-%dT%H:%M:%SZ").to_string().dimmed(),
         get_colored_level(&log_item.level),
-        log_item.target,
+        log_item.target.dimmed(),
         "]".bright_black(),
-        log_item.body
+        message,
+        fmt_fields(&fields).dimmed()
     );
 }
 
-fn get_colored_level(level: &Level) -> ColoredString {
-    match level {
-        Level::Trace => level.to_string().bright_black(),
-        Level::Debug => level.to_string().blue(),
-        Level::Info => level.to_string().green(),
-        Level::Warn => level.to_string().yellow(),
-        Level::Error => level.to_string().red(),
+fn get_colored_level(level: &String) -> ColoredString {
+    match &level.to_uppercase()[..] {
+        "TRACE" => level.bright_black(),
+        "DEBUG" => level.blue(),
+        "INFO" => level.green(),
+        "WARN" => level.yellow(),
+        "ERROR" => level.red(),
+        _ => level.bright_black(), // TODO: should this panic?
     }
 }
+
+fn fmt_fields(fields: &serde_json::Map<String, serde_json::Value>) -> String {
+    fields
+        .iter()
+        .map(|(field, value)| format!("{}={}", field.italic(), value))
+        .collect::<Vec<_>>()
+        .join(" ")
+}

cargo-shuttle/tests/integration/run.rs

@@ -54,7 +54,7 @@ async fn cargo_shuttle_run(working_directory: &str) -> u16 {
     port
 }
 
-#[tokio::test]
+#[tokio::test(flavor = "multi_thread")]
 async fn rocket_hello_world() {
     let port = cargo_shuttle_run("../examples/rocket/hello-world").await;
 
@@ -71,7 +71,7 @@ async fn rocket_hello_world() {
 }
 
 // This example uses a shared Postgres. Thus local runs should create a docker container for it.
-#[tokio::test]
+#[tokio::test(flavor = "multi_thread")]
 async fn rocket_postgres() {
     let port = cargo_shuttle_run("../examples/rocket/postgres").await;
     let client = reqwest::Client::new();
@@ -111,7 +111,7 @@ async fn rocket_postgres() {
     assert_eq!(request_text, "the contents of my API key");
 }
 
-#[tokio::test]
+#[tokio::test(flavor = "multi_thread")]
 async fn rocket_authentication() {
     let port = cargo_shuttle_run("../examples/rocket/authentication").await;
     let client = reqwest::Client::new();
@@ -167,7 +167,7 @@ async fn rocket_authentication() {
     );
 }
 
-#[tokio::test]
+#[tokio::test(flavor = "multi_thread")]
 async fn axum_hello_world() {
     let port = cargo_shuttle_run("../examples/axum/hello-world").await;
 
@@ -183,7 +183,7 @@ async fn axum_hello_world() {
     assert_eq!(request_text, "Hello, world!");
 }
 
-#[tokio::test]
+#[tokio::test(flavor = "multi_thread")]
 async fn tide_hello_world() {
     let port = cargo_shuttle_run("../examples/tide/hello-world").await;
 
@@ -199,7 +199,7 @@ async fn tide_hello_world() {
     assert_eq!(request_text, "Hello, world!");
 }
 
-#[tokio::test]
+#[tokio::test(flavor = "multi_thread")]
 async fn tower_hello_world() {
     let port = cargo_shuttle_run("../examples/tower/hello-world").await;
 
@@ -215,7 +215,7 @@ async fn tower_hello_world() {
     assert_eq!(request_text, "Hello, world!");
 }
 
-#[tokio::test]
+#[tokio::test(flavor = "multi_thread")]
 async fn poem_hello_world() {
     let port = cargo_shuttle_run("../examples/poem/hello-world").await;
 
@@ -232,7 +232,7 @@ async fn poem_hello_world() {
 }
 
 // This example uses a shared Postgres. Thus local runs should create a docker container for it.
-#[tokio::test]
+#[tokio::test(flavor = "multi_thread")]
 async fn poem_postgres() {
     let port = cargo_shuttle_run("../examples/poem/postgres").await;
     let client = reqwest::Client::new();
@@ -263,7 +263,7 @@ async fn poem_postgres() {
 }
 
 // This example uses a shared MongoDb. Thus local runs should create a docker container for it.
-#[tokio::test]
+#[tokio::test(flavor = "multi_thread")]
 async fn poem_mongodb() {
     let port = cargo_shuttle_run("../examples/poem/mongodb").await;
     let client = reqwest::Client::new();
@@ -295,7 +295,7 @@ async fn poem_mongodb() {
     assert_eq!(request_text, "{\"note\":\"Deploy to shuttle\"}");
 }
 
-#[tokio::test]
+#[tokio::test(flavor = "multi_thread")]
 async fn salvo_hello_world() {
     let port = cargo_shuttle_run("../examples/salvo/hello-world").await;