Prepare for running the dist suite as an unprivileged user

src/bin/cachepot-dist/main.rs

@@ -16,11 +16,11 @@ use cachepot::dist::{
     SchedulerIncoming, SchedulerOutgoing, SchedulerStatusResult, SubmitToolchainResult, TcCache,
     Toolchain, ToolchainReader, UpdateJobStateResult, WorkerIncoming, WorkerNonce, WorkerOutgoing,
 };
+use cachepot::init_logging;
 use cachepot::util::daemonize;
 use jsonwebtoken as jwt;
 use rand::{rngs::OsRng, RngCore};
 use std::collections::{btree_map, BTreeMap, HashMap, HashSet};
-use std::env;
 use std::io;
 use std::path::Path;
 use std::path::PathBuf;
@@ -388,15 +388,6 @@ async fn run(command: Command) -> Result<i32> {
     }
 }
 
-fn init_logging() {
-    if env::var("RUST_LOG").is_ok() {
-        match env_logger::try_init() {
-            Ok(_) => (),
-            Err(e) => panic!("Failed to initalize logging: {:?}", e),
-        }
-    }
-}
-
 const MAX_PER_CORE_LOAD: f64 = 10f64;
 const WORKER_REMEMBER_ERROR_TIMEOUT: Duration = Duration::from_secs(300);
 const UNCLAIMED_PENDING_TIMEOUT: Duration = Duration::from_secs(300);

src/commands.rs

@@ -76,7 +76,9 @@ fn run_coordinator_process() -> Result<CoordinatorStartup> {
     trace!("run_coordinator_process");
     let tempdir = tempfile::Builder::new().prefix("cachepot").tempdir()?;
     let socket_path = tempdir.path().join("sock");
-    let runtime = Runtime::new()?;
+    let runtime = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()?;
     let exe_path = env::current_exe()?;
     let workdir = exe_path.parent().expect("executable path has no parent?!");
     let _child = process::Command::new(&exe_path)

src/lib.rs

@@ -62,8 +62,6 @@ pub mod util;
 
 use std::env;
 
-const LOGGING_ENV: &str = "CACHEPOT_LOG";
-
 pub fn main() {
     init_logging();
     std::process::exit(match cmdline::parse() {
@@ -88,9 +86,96 @@ pub fn main() {
     });
 }
 
-fn init_logging() {
+pub fn init_logging() {
+    const LOGGING_ENV: &str = "CACHEPOT_LOG";
+
+    use env_logger::fmt::{Color, Style};
+    use log::Level;
+    use std::io::Write;
+
+    /// The available service type that cachepot can run as.
+    #[derive(Copy, Clone)]
+    enum Kind {
+        /// A service that connects a coordinator and a remote worker to execute
+        /// a remote build.
+        DistScheduler,
+        /// A service that's used to directly perform remote sandbox compilation
+        DistWorker,
+        /// A background service used by the cachepot compilation wrapper (client)
+        /// to either re-use local compilation cache or schedule a remote
+        /// compilation via a remote scheduler
+        Coordinator,
+        /// A wrapper that masquerades as a compiler but spawns (or talks to) a
+        /// coordinator to perform the actual compilation locally or offload it
+        /// to a distributed cluster (in both cases we can re-use cached artifacts)
+        Client,
+    }
+
+    impl std::fmt::Display for Kind {
+        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+            write!(
+                f,
+                "{}",
+                match self {
+                    Kind::DistScheduler => "dist scheduler",
+                    Kind::DistWorker => "dist worker",
+                    Kind::Coordinator => "coordinator",
+                    Kind::Client => "client",
+                }
+            )
+        }
+    }
+
+    // TODO: That's a rough heuristic - share detection logic from cmdline.rs
+    let kind = if env::var("CACHEPOT_START_COORDINATOR").is_ok()
+        || env::args_os().any(|a| a == "--start-coordinator")
+    {
+        Kind::Coordinator
+    } else {
+        match std::env::args().nth(1).as_deref() {
+            Some("scheduler") => Kind::DistScheduler,
+            Some("worker") => Kind::DistWorker,
+            _ => Kind::Client,
+        }
+    };
+
+    let color_for_kind = |kind| match kind {
+        Kind::DistScheduler => Color::Yellow,
+        Kind::DistWorker => Color::Cyan,
+        Kind::Coordinator => Color::Blue,
+        Kind::Client => Color::Green,
+    };
+
+    let default_level_style = |mut level_style: Style, level: Level| {
+        match level {
+            Level::Trace => level_style.set_color(Color::Cyan),
+            Level::Debug => level_style.set_color(Color::Blue),
+            Level::Info => level_style.set_color(Color::Green),
+            Level::Warn => level_style.set_color(Color::Yellow),
+            Level::Error => level_style.set_color(Color::Red).set_bold(true),
+        };
+        level_style
+    };
+
     if env::var(LOGGING_ENV).is_ok() {
-        match env_logger::Builder::from_env(LOGGING_ENV).try_init() {
+        let mut builder = env_logger::Builder::from_env(LOGGING_ENV);
+        // That's mostly what env_logger does by default but we also attach the
+        // PID and kind of the cachepot executable due to its multi-process nature
+        builder.format(move |f, record| {
+            write!(
+                f,
+                "{}",
+                chrono::Local::now().format("%Y-%m-%dT%H:%M:%S%.3f"),
+            )?;
+            let style = default_level_style(f.style(), record.level());
+            write!(f, " {:<5}", style.value(record.level()))?;
+            write!(f, " [PID {}]", std::process::id())?;
+            let mut style = f.style();
+            style.set_color(color_for_kind(kind));
+            write!(f, " {:>14}", style.value(kind))?;
+            writeln!(f, " {}", record.args())
+        });
+        match builder.try_init() {
             Ok(_) => (),
             Err(e) => panic!("Failed to initalize logging: {:?}", e),
         }

systemd/config/scheduler.conf

@@ -1,6 +1,6 @@
 public_addr = "http://127.0.0.1:10600"
 
-[server_auth]
+[worker_auth]
 type = "token"
 token = "server_xxxxxxxxxxxxxx"