runner logging using simple logger (#47755)

* runner using runner logger

Author: sima-fastly

src/bin/pushpin.rs

@@ -15,18 +15,32 @@
  */
 
 use clap::Parser;
-use log::info;
-use pushpin::runner::{ArgsData, CliArgs, Settings};
+use log::{info, LevelFilter};
+use pushpin::runner::{get_runner_logger, ArgsData, CliArgs, Settings};
 use pushpin::service::start_services;
 use std::error::Error;
 use std::process;
 
 fn process_args_and_run(args: CliArgs) -> Result<(), Box<dyn Error>> {
     let args_data = ArgsData::new(args)?;
     let settings = Settings::new(args_data)?;
+
+    log::set_logger(get_runner_logger()).unwrap();
+    let ll = settings
+        .log_levels
+        .get("")
+        .unwrap_or(settings.log_levels.get("default").unwrap());
+    let level = match ll {
+        0 => LevelFilter::Error,
+        1 => LevelFilter::Warn,
+        2 => LevelFilter::Info,
+        3 => LevelFilter::Debug,
+        4..=u8::MAX => LevelFilter::Trace,
+    };
+    log::set_max_level(level);
+
     info!("using config: {:?}", settings.config_file.display());
     start_services(settings);
-    //To be implemented in the next PR
 
     Ok(())
 }

src/runner.rs

@@ -16,14 +16,21 @@
 
 use clap::{ArgAction, Parser};
 use log::{error, warn};
+use log::{Level, Log, Metadata, Record};
 use serde::Deserialize;
 use std::collections::HashMap;
 use std::env;
 use std::error::Error;
 use std::fs;
+use std::io;
+use std::mem;
 use std::net::{IpAddr, Ipv4Addr, SocketAddr};
 use std::path::{Path, PathBuf};
+use std::str;
 use std::string::String;
+use std::sync::Once;
+use time::macros::format_description;
+use time::{OffsetDateTime, UtcOffset};
 use url::Url;
 
 use crate::config::{get_config_file, CustomConfig};
@@ -887,3 +894,81 @@ mod tests {
         }
     }
 }
+
+struct RunnerLogger {
+    local_offset: UtcOffset,
+}
+
+impl Log for RunnerLogger {
+    fn enabled(&self, metadata: &Metadata) -> bool {
+        metadata.level() <= Level::Trace
+    }
+
+    fn log(&self, record: &Record) {
+        if !self.enabled(record.metadata()) {
+            return;
+        }
+
+        let binding = record.args().to_string();
+        let msg_parts: Vec<&str> = binding.split_whitespace().collect();
+
+        if msg_parts.len() >= 4 {
+            println!(
+                "{} {} {} [{}] {}",
+                msg_parts[0],
+                msg_parts[1],
+                msg_parts[2],
+                record.target(),
+                msg_parts[3..].join(" ")
+            );
+        } else {
+            let now = OffsetDateTime::now_utc().to_offset(self.local_offset);
+
+            let format = format_description!(
+                "[year]-[month]-[day] [hour]:[minute]:[second].[subsecond digits:3]"
+            );
+
+            let mut ts = [0u8; 64];
+
+            let size = {
+                let mut ts = io::Cursor::new(&mut ts[..]);
+
+                now.format_into(&mut ts, &format)
+                    .expect("failed to write timestamp");
+
+                ts.position() as usize
+            };
+
+            let ts = str::from_utf8(&ts[..size]).expect("timestamp is not utf-8");
+
+            let lname = match record.level() {
+                log::Level::Error => "ERR",
+                log::Level::Warn => "WARN",
+                log::Level::Info => "INFO",
+                log::Level::Debug => "DEBUG",
+                log::Level::Trace => "TRACE",
+            };
+
+            println!("[{}] {} [{}] {}", lname, ts, record.target(), record.args());
+        }
+    }
+
+    fn flush(&self) {}
+}
+
+static mut LOGGER: mem::MaybeUninit<RunnerLogger> = mem::MaybeUninit::uninit();
+
+pub fn get_runner_logger() -> &'static impl Log {
+    static INIT: Once = Once::new();
+
+    unsafe {
+        INIT.call_once(|| {
+            let local_offset =
+                UtcOffset::current_local_offset().expect("failed to get local time offset");
+
+            LOGGER.write(RunnerLogger { local_offset });
+        });
+
+        LOGGER.as_ptr().as_ref().unwrap()
+    }
+}

src/service.rs

@@ -15,10 +15,12 @@
  */
 
 use crate::runner::Settings;
-use log::error;
+use log::{error, LevelFilter};
 use mpsc::{channel, Sender};
 use signal_hook::consts::{SIGINT, SIGTERM, TERM_SIGNALS};
 use signal_hook::iterator::Signals;
+use std::io::{BufRead, BufReader};
+use std::process::{ChildStderr, ChildStdout, Stdio};
 use std::sync::mpsc;
 use std::thread::JoinHandle;
 use std::{process::Command, thread};
@@ -31,6 +33,7 @@ pub enum ServiceError {
 
 pub struct Service {
     pub name: String,
+    pub log_level: u8,
 }
 
 pub fn start_services(settings: Settings) {
@@ -54,7 +57,7 @@ pub fn start_services(settings: Settings) {
     let (sender, receiver) = channel();
     let mut threads: Vec<Option<JoinHandle<()>>> = vec![];
     for mut service in services {
-        threads.push(service.start(sender.clone()));
+        threads.extend(service.start(sender.clone()));
     }
 
     // Spawn a signal handling thread
@@ -101,22 +104,49 @@ pub fn start_services(settings: Settings) {
 }
 
 impl Service {
-    fn new(name: String) -> Self {
-        Self { name }
+    fn new(name: String, log_level: u8) -> Self {
+        Self { name, log_level }
     }
-
     pub fn start(
         &mut self,
         args: Vec<String>,
         sender: Sender<Result<(), ServiceError>>,
-    ) -> Option<thread::JoinHandle<()>> {
+    ) -> Vec<Option<JoinHandle<()>>> {
         let name = self.name.clone();
+        let name_str = self.name.clone();
+
+        let level = match self.log_level {
+            0 => LevelFilter::Error,
+            1 => LevelFilter::Warn,
+            2 => LevelFilter::Info,
+            3 => LevelFilter::Debug,
+            4..=u8::MAX => LevelFilter::Trace,
+        };
+        log::set_max_level(level);
+
+        // Create a channel for sending thread handles back to main thread
+        let (handle_sender, handle_receiver) = channel();
+
+        let mut result: Vec<Option<JoinHandle<()>>> = Vec::new();
 
-        Some(thread::spawn(move || {
-            let mut command = Command::new(args[0].clone());
+        result.push(Some(thread::spawn(move || {
+            let mut command = Command::new(&args[0]);
             command.args(&args[1..]);
 
-            let status = command.status().expect("Failed to execute command");
+            // Capture stdout and stderr
+            command.stdout(Stdio::piped());
+            command.stderr(Stdio::piped());
+
+            let mut child = command.spawn().expect("Failed to execute command");
+
+            let stdout = child.stdout.take().unwrap();
+            let stderr = child.stderr.take().unwrap();
+            let handles = start_log_handler(stdout, stderr, name_str);
+
+            // Send the handles back to main thread
+            handle_sender.send(handles).unwrap();
+
+            let status = child.wait().expect("Failed to wait for command");
 
             if status.success() {
                 sender.send(Ok(())).unwrap();
@@ -126,12 +156,17 @@ impl Service {
                     .send(Err(ServiceError::ThreadError(error_message)))
                     .unwrap();
             }
-        }))
+        })));
+        // Receive the handles from the channel and add them to the result vector
+        let received_handles: Vec<Option<JoinHandle<()>>> = handle_receiver.recv().unwrap();
+        result.extend(received_handles);
+
+        result
     }
 }
 
 pub trait RunnerService {
-    fn start(&mut self, sender: Sender<Result<(), ServiceError>>) -> Option<JoinHandle<()>>;
+    fn start(&mut self, sender: Sender<Result<(), ServiceError>>) -> Vec<Option<JoinHandle<()>>>;
 }
 
 pub struct CondureService {
@@ -147,12 +182,10 @@ impl CondureService {
         args.push(settings.condure_bin.display().to_string());
 
         let log_level = match settings.log_levels.get(service_name) {
-            Some(&x) => x as i8,
-            None => settings.log_levels.get("default").unwrap().to_owned() as i8,
+            Some(&x) => x,
+            None => settings.log_levels.get("default").unwrap().to_owned(),
         };
-        if log_level >= 0 {
-            args.push(format!("--log-level={}", log_level));
-        }
+        args.push(format!("--log-level={}", log_level));
 
         args.push(format!("--buffer-size={}", settings.client_buffer_size));
         args.push(format!(
@@ -215,14 +248,14 @@ impl CondureService {
         }
 
         Self {
-            service: Service::new(String::from(service_name)),
+            service: Service::new(String::from(service_name), log_level),
             args,
         }
     }
 }
 
 impl RunnerService for CondureService {
-    fn start(&mut self, sender: Sender<Result<(), ServiceError>>) -> Option<JoinHandle<()>> {
+    fn start(&mut self, sender: Sender<Result<(), ServiceError>>) -> Vec<Option<JoinHandle<()>>> {
         self.service.start(self.args.clone(), sender)
     }
 }
@@ -244,26 +277,24 @@ impl PushpinProxyService {
             args.push(format!("--ipc-prefix={}", settings.ipc_prefix));
         }
         let log_level = match settings.log_levels.get("pushpin-proxy") {
-            Some(&x) => x as i8,
-            None => settings.log_levels.get("default").unwrap().to_owned() as i8,
+            Some(&x) => x,
+            None => settings.log_levels.get("default").unwrap().to_owned(),
         };
-        if log_level >= 0 {
-            args.push(format!("--loglevel={}", log_level));
-        }
+        args.push(format!("--loglevel={}", log_level));
 
         for route in settings.route_lines.clone() {
             args.push(format!("--route={}", route));
         }
 
         Self {
-            service: Service::new(String::from(service_name)),
+            service: Service::new(String::from(service_name), log_level),
             args,
         }
     }
 }
 
 impl RunnerService for PushpinProxyService {
-    fn start(&mut self, sender: Sender<Result<(), ServiceError>>) -> Option<JoinHandle<()>> {
+    fn start(&mut self, sender: Sender<Result<(), ServiceError>>) -> Vec<Option<JoinHandle<()>>> {
         self.service.start(self.args.clone(), sender)
     }
 }
@@ -288,22 +319,75 @@ impl PushpinHandlerService {
             args.push(format!("--ipc-prefix={}", settings.ipc_prefix));
         }
         let log_level = match settings.log_levels.get("pushpin-handler") {
-            Some(&x) => x as i8,
-            None => settings.log_levels.get("default").unwrap().to_owned() as i8,
+            Some(&x) => x,
+            None => settings.log_levels.get("default").unwrap().to_owned(),
         };
-        if log_level >= 0 {
-            args.push(format!("--loglevel={}", log_level));
-        }
+        args.push(format!("--loglevel={}", log_level));
 
         Self {
-            service: Service::new(String::from(service_name)),
+            service: Service::new(String::from(service_name), log_level),
             args,
         }
     }
 }
 
 impl RunnerService for PushpinHandlerService {
-    fn start(&mut self, sender: Sender<Result<(), ServiceError>>) -> Option<JoinHandle<()>> {
+    fn start(&mut self, sender: Sender<Result<(), ServiceError>>) -> Vec<Option<JoinHandle<()>>> {
         self.service.start(self.args.clone(), sender)
     }
 }
+
+fn start_log_handler(
+    stdout: ChildStdout,
+    stderr: ChildStderr,
+    name: String,
+) -> Vec<Option<JoinHandle<()>>> {
+    let mut result: Vec<Option<JoinHandle<()>>> = Vec::new();
+
+    let name_str = name.clone();
+    result.push(Some(thread::spawn(move || {
+        let reader = BufReader::new(stdout);
+        for line in reader.lines() {
+            match line {
+                Ok(msg) => log_message(&name_str, log::Level::Info, &msg),
+                Err(_) => {
+                    log_message(
+                        &name_str,
+                        log::Level::Error,
+                        "failed to read from standard out.",
+                    );
+                    break;
+                }
+            }
+        }
+    })));
+
+    result.push(Some(thread::spawn(move || {
+        let reader_err = BufReader::new(stderr);
+        for line in reader_err.lines() {
+            match line {
+                Ok(msg) => log_message(&name, log::Level::Error, &msg),
+                Err(_) => {
+                    log_message(
+                        &name,
+                        log::Level::Error,
+                        "failed to read from standard error.",
+                    );
+                    break;
+                }
+            }
+        }
+    })));
+
+    result
+}
+
+fn log_message(name: &str, level: log::Level, msg: &str) {
+    log::logger().log(
+        &log::Record::builder()
+            .level(level)
+            .target(name)
+            .args(format_args!("{}", msg))
+            .build(),
+    );
+}