Merge pull request #1069 from sozu-proxy/binary-access-logs

Protobuf access logs

bin/config.toml

@@ -33,7 +33,12 @@ log_target = "stdout"
 
 # optional different target for access logs (IP addresses, domains, URI, HTTP status, etc)
 # It supports the same options as log_target
-# log_access_target = "file:///var/logs/sozu-access.log"
+# access_logs_target = "file:///var/logs/sozu-access.log"
+
+# format of the access logs. Defaults to ascii.
+# - ascii
+# - protobuf (defined in [sozu_command_lib::proto::command::ProtobufAccessLog])
+# access_logs_format = "ascii"
 
 # path to the unix socket file used to send commands to sozu
 # default value points to "sozu.sock" file in the current directory

bin/src/cli.rs

@@ -130,8 +130,8 @@ pub enum SubCmd {
     },
     #[clap(name = "logging", about = "change logging level")]
     Logging {
-        #[clap(subcommand)]
-        level: LoggingLevel,
+        #[clap(name = "filter")]
+        filter: String,
     },
     #[clap(name = "state", about = "state management")]
     State {
@@ -231,28 +231,6 @@ pub enum MetricsCmd {
     },
 }
 
-#[derive(Subcommand, PartialEq, Eq, Clone, Debug)]
-pub enum LoggingLevel {
-    #[clap(name = "trace", about = "Displays a LOT of logs")]
-    Trace,
-    #[clap(
-        name = "debug",
-        about = "Displays more logs about the inner workings of Sōzu"
-    )]
-    Debug,
-    #[clap(name = "error", about = "Displays occurring errors")]
-    Error,
-    #[clap(name = "warn", about = "Displays warnings about non-critical errors")]
-    Warn,
-    #[clap(name = "info", about = "Displays logs about normal behaviour of Sōzu")]
-    Info,
-}
-impl std::fmt::Display for LoggingLevel {
-    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
-        write!(f, "{self:?}")
-    }
-}
-
 #[derive(Subcommand, PartialEq, Eq, Clone, Debug)]
 pub enum StateCmd {
     #[clap(name = "save", about = "Save state to that file")]

bin/src/command/requests.rs

@@ -1,5 +1,6 @@
 use std::{
     collections::{BTreeMap, HashMap},
+    env,
     fs::File,
     io::{ErrorKind, Read},
 };
@@ -216,17 +217,28 @@ fn save_state(server: &mut Server, client: &mut ClientSession, path: &str) {
 /// change logging level on the main process, and on all workers
 fn set_logging_level(server: &mut Server, client: &mut ClientSession, logging_filter: String) {
     debug!("Changing main process log level to {}", logging_filter);
-    logging::LOGGER.with(|l| {
-        let directives = logging::parse_logging_spec(&logging_filter);
-        l.borrow_mut().set_directives(directives);
+    let (directives, errors) = logging::parse_logging_spec(&logging_filter);
+    if !errors.is_empty() {
+        client.finish_failure(format!(
+            "Error parsing logging filter:\n- {}",
+            errors
+                .iter()
+                .map(logging::LogSpecParseError::to_string)
+                .collect::<Vec<String>>()
+                .join("\n- ")
+        ));
+        return;
+    }
+    logging::LOGGER.with(|logger| {
+        logger.borrow_mut().set_directives(directives);
     });
 
     // also change / set the content of RUST_LOG so future workers / main thread
     // will have the new logging filter value
-    ::std::env::set_var("RUST_LOG", &logging_filter);
+    env::set_var("RUST_LOG", &logging_filter);
     debug!(
         "Logging level now: {}",
-        ::std::env::var("RUST_LOG").unwrap_or("could get RUST_LOG from env".to_string())
+        env::var("RUST_LOG").unwrap_or("could get RUST_LOG from env".to_string())
     );
 
     worker_request(server, client, RequestType::Logging(logging_filter));

bin/src/command/server.rs

@@ -1,6 +1,6 @@
 use std::{
     collections::{HashMap, HashSet},
-    fmt::Debug,
+    fmt::{self, Debug},
     io::Error as IoError,
     ops::{Deref, DerefMut},
     os::fd::{AsRawFd, FromRawFd},
@@ -223,8 +223,8 @@ impl CommandHub {
         let channel = Channel::new(stream, 4096, u64::MAX);
         let id = self.next_client_id();
         let session = ClientSession::new(channel, id, token);
-        info!("register new client: {}", id);
-        debug!("new client: {:?}", session);
+        info!("Register new client: {}", id);
+        debug!("{:#?}", session);
         self.clients.insert(token, session);
     }
 
@@ -295,7 +295,7 @@ impl CommandHub {
     /// - manage timeouts of tasks
     pub fn run(&mut self) {
         let mut events = Events::with_capacity(100);
-        debug!("running the command hub: {:?}", self);
+        debug!("running the command hub: {:#?}", self);
 
         loop {
             let run_state = self.run_state;
@@ -400,7 +400,8 @@ impl CommandHub {
                                     server.handle_client_request(client, request);
                                 }
                                 ClientResult::CloseSession => {
-                                    info!("Closing client {:#?}", client);
+                                    info!("Closing client {}", client.id);
+                                    debug!("{:#?}", client);
                                     self.event_subscribers.remove(&token);
                                     self.clients.remove(&token);
                                 }
@@ -511,7 +512,6 @@ pub enum ServerState {
 /// - scatter to workers
 /// - gather worker responses
 /// - trigger a finishing function when all responses are gathered
-#[derive(Debug)]
 pub struct Server {
     pub config: Config,
     /// Sōzu clients that subscribed to events
@@ -857,3 +857,23 @@ impl Server {
         }
     }
 }
+
+impl Debug for Server {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Server")
+            .field("config", &self.config)
+            .field("event_subscribers", &self.event_subscribers)
+            .field("executable_path", &self.executable_path)
+            .field("in_flight", &self.in_flight)
+            .field("next_client_id", &self.next_client_id)
+            .field("next_session_id", &self.next_session_id)
+            .field("next_task_id", &self.next_task_id)
+            .field("next_worker_id", &self.next_worker_id)
+            .field("poll", &self.poll)
+            .field("queued_tasks", &self.queued_tasks)
+            .field("run_state", &self.run_state)
+            .field("unix_listener", &self.unix_listener)
+            .field("workers", &self.workers)
+            .finish()
+    }
+}

bin/src/ctl/mod.rs

@@ -126,7 +126,7 @@ impl CommandManager {
                 } => self.get_metrics(list, refresh, names, clusters, backends, no_clusters),
                 _ => self.configure_metrics(cmd),
             },
-            SubCmd::Logging { level } => self.logging_filter(&level),
+            SubCmd::Logging { filter } => self.logging_filter(filter),
             SubCmd::State { cmd } => match cmd {
                 StateCmd::Save { file } => self.save_state(file),
                 StateCmd::Load { file } => self.load_state(file),

bin/src/ctl/request_builder.rs

@@ -18,8 +18,8 @@ use sozu_command_lib::{
 
 use crate::{
     cli::{
-        BackendCmd, ClusterCmd, HttpFrontendCmd, HttpListenerCmd, HttpsListenerCmd, LoggingLevel,
-        MetricsCmd, TcpFrontendCmd, TcpListenerCmd,
+        BackendCmd, ClusterCmd, HttpFrontendCmd, HttpListenerCmd, HttpsListenerCmd, MetricsCmd,
+        TcpFrontendCmd, TcpListenerCmd,
     },
     ctl::CommandManager,
 };
@@ -192,7 +192,7 @@ impl CommandManager {
                     }
 
                     let query_domain = QueryClusterByDomain {
-                        hostname: splitted.get(0).ok_or(CtlError::NeedClusterDomain)?.clone(),
+                        hostname: splitted.first().ok_or(CtlError::NeedClusterDomain)?.clone(),
                         path: splitted.get(1).cloned().map(|path| format!("/{path}")), // We add the / again because of the splitn removing it
                     };
 
@@ -484,8 +484,8 @@ impl CommandManager {
         )
     }
 
-    pub fn logging_filter(&mut self, filter: &LoggingLevel) -> Result<(), CtlError> {
-        self.send_request(RequestType::Logging(filter.to_string().to_lowercase()).into())
+    pub fn logging_filter(&mut self, filter: String) -> Result<(), CtlError> {
+        self.send_request(RequestType::Logging(filter).into())
     }
 
     pub fn add_certificate(

bin/src/upgrade.rs

@@ -7,7 +7,7 @@ use std::{
     process::Command,
 };
 
-use libc::{self, pid_t};
+use libc::pid_t;
 use mio::net::UnixStream;
 use nix::{
     errno::Errno,

bin/src/util.rs

@@ -10,7 +10,6 @@ use nix::{
     errno::Errno,
     fcntl::{fcntl, FcntlArg, FdFlag},
 };
-use thiserror;
 
 use sozu_command_lib::config::Config;
 use sozu_lib::metrics::{self, MetricError};
@@ -163,9 +162,10 @@ pub unsafe fn get_executable_path() -> Result<String, UtilError> {
     Ok(path_str)
 }
 
+
 #[cfg(target_os = "macos")]
 extern "C" {
-    pub fn _NSGetExecutablePath(buf: *mut c_char, size: *mut u32) -> i32;
+    pub fn _NSGetExecutablePath(buf: *mut libc::c_char, size: *mut u32) -> i32;
 }
 
 #[cfg(target_os = "macos")]

bin/src/worker.rs

@@ -10,14 +10,10 @@ use std::{
     process::Command,
 };
 
-#[cfg(target_os = "macos")]
-use libc::c_char;
-use libc::{self, pid_t};
-#[cfg(target_os = "freebsd")]
-use libc::{sysctl, CTL_KERN, KERN_PROC, KERN_PROC_PATHNAME, PATH_MAX};
+use libc::pid_t;
+
 use mio::net::UnixStream;
 use nix::{
-    self,
     errno::Errno,
     unistd::{fork, ForkResult},
 };
@@ -26,7 +22,7 @@ use tempfile::tempfile;
 use sozu_command_lib::{
     channel::{Channel, ChannelError},
     config::Config,
-    logging::setup_logging,
+    logging::{setup_logging, AccessLogFormat},
     proto::command::{ServerConfig, WorkerRequest, WorkerResponse},
     ready::Ready,
     request::{read_initial_state_from_file, RequestError},
@@ -110,10 +106,15 @@ pub fn begin_worker_process(
 
     let worker_id = format!("{}-{:02}", "WRK", id);
 
+    let access_log_format = AccessLogFormat::from(&worker_config.access_log_format());
+
     // do not try to log anything before this, or the logger will panic
     setup_logging(
         &worker_config.log_target,
-        worker_config.log_access_target.as_deref(),
+        worker_config.log_colored,
+        worker_config.access_logs_target.as_deref(),
+        Some(access_log_format),
+        Some(worker_config.log_colored),
         &worker_config.log_level,
         &worker_id,
     );

command/Cargo.toml

@@ -38,6 +38,7 @@ nix = { version = "^0.28.0", features = ["socket", "uio"] }
 nom = "^7.1.3"
 prost = "^0.12.3"
 rand = "^0.8.5"
+rusty_ulid = "^2.0.0"
 serde = { version = "^1.0.195", features = ["derive"] }
 serde_json = "^1.0.111"
 sha2 = "^0.10.8"
@@ -52,6 +53,7 @@ x509-parser = "^0.16.0"
 unstable = []
 logs-debug = []
 logs-trace = []
+logs-cache = []
 
 [badges]
 travis-ci = { repository = "sozu-proxy/sozu" }