Json logging capability (#2624)

* Support configuring the log format as json or plain.

Separately test json and plain logger. They would be competing on the
same global subscriber otherwise.

* Implement log_format for pageserver config

* Implement configurable log format for safekeeper.

Cargo.lock

@@ -3932,6 +3932,16 @@ dependencies = [
  "tracing-core",
 ]
 
+[[package]]
+name = "tracing-serde"
+version = "0.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bc6b213177105856957181934e4920de57730fc69bf42c37ee5bb664d406d9e1"
+dependencies = [
+ "serde",
+ "tracing-core",
+]
+
 [[package]]
 name = "tracing-subscriber"
 version = "0.3.16"
@@ -3942,12 +3952,15 @@ dependencies = [
  "nu-ansi-term",
  "once_cell",
  "regex",
+ "serde",
+ "serde_json",
  "sharded-slab",
  "smallvec",
  "thread_local",
  "tracing",
  "tracing-core",
  "tracing-log",
+ "tracing-serde",
 ]
 
 [[package]]
@@ -4042,6 +4055,8 @@ dependencies = [
  "serde_json",
  "serde_with",
  "signal-hook",
+ "strum",
+ "strum_macros",
  "tempfile",
  "thiserror",
  "tokio",

libs/utils/Cargo.toml

@@ -19,7 +19,7 @@ thiserror = "1.0"
 tokio = { version = "1.17", features = ["macros"]}
 tokio-rustls = "0.23"
 tracing = "0.1"
-tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }
 nix = "0.25"
 signal-hook = "0.3.10"
 rand = "0.8.3"
@@ -30,6 +30,8 @@ rustls-split = "0.3.0"
 git-version = "0.3.5"
 serde_with = "2.0"
 once_cell = "1.13.0"
+strum = "0.24"
+strum_macros = "0.24"
 
 
 metrics = { path = "../metrics" }

libs/utils/src/logging.rs

@@ -1,11 +1,35 @@
 use std::{
     fs::{File, OpenOptions},
     path::Path,
+    str::FromStr,
 };
 
 use anyhow::{Context, Result};
+use strum_macros::{EnumString, EnumVariantNames};
 
-pub fn init(log_filename: impl AsRef<Path>, daemonize: bool) -> Result<File> {
+#[derive(EnumString, EnumVariantNames, Eq, PartialEq, Debug, Clone, Copy)]
+#[strum(serialize_all = "snake_case")]
+pub enum LogFormat {
+    Plain,
+    Json,
+}
+
+impl LogFormat {
+    pub fn from_config(s: &str) -> anyhow::Result<LogFormat> {
+        use strum::VariantNames;
+        LogFormat::from_str(s).with_context(|| {
+            format!(
+                "Unrecognized log format. Please specify one of: {:?}",
+                LogFormat::VARIANTS
+            )
+        })
+    }
+}
+pub fn init(
+    log_filename: impl AsRef<Path>,
+    daemonize: bool,
+    log_format: LogFormat,
+) -> Result<File> {
     // Don't open the same file for output multiple times;
     // the different fds could overwrite each other's output.
     let log_file = OpenOptions::new()
@@ -21,22 +45,50 @@ pub fn init(log_filename: impl AsRef<Path>, daemonize: bool) -> Result<File> {
     let env_filter = tracing_subscriber::EnvFilter::try_from_default_env()
         .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new(default_filter_str));
 
+    let x: File = log_file.try_clone().unwrap();
     let base_logger = tracing_subscriber::fmt()
         .with_env_filter(env_filter)
-        .with_target(false) // don't include event targets
-        .with_ansi(false); // don't use colors in log file;
+        .with_target(false)
+        .with_ansi(false)
+        .with_writer(move || -> Box<dyn std::io::Write> {
+            // we are cloning and returning log file in order to allow redirecting daemonized stdout and stderr to it
+            // if we do not use daemonization (e.g. in docker) it is better to log to stdout directly
+            // for example to be in line with docker log command which expects logs comimg from stdout
+            if daemonize {
+                Box::new(x.try_clone().unwrap())
+            } else {
+                Box::new(std::io::stdout())
+            }
+        });
 
-    // we are cloning and returning log file in order to allow redirecting daemonized stdout and stderr to it
-    // if we do not use daemonization (e.g. in docker) it is better to log to stdout directly
-    // for example to be in line with docker log command which expects logs comimg from stdout
-    if daemonize {
-        let x = log_file.try_clone().unwrap();
-        base_logger
-            .with_writer(move || x.try_clone().unwrap())
-            .init();
-    } else {
-        base_logger.init();
+    match log_format {
+        LogFormat::Json => base_logger.json().init(),
+        LogFormat::Plain => base_logger.init(),
     }
 
     Ok(log_file)
 }
+
+// #[cfg(test)]
+// Due to global logger, can't run tests in same process.
+// So until there's a non-global one, the tests are in ../tests/ as separate files.
+#[macro_export(local_inner_macros)]
+macro_rules! test_init_file_logger {
+    ($log_level:expr, $log_format:expr) => {{
+        use std::str::FromStr;
+        std::env::set_var("RUST_LOG", $log_level);
+
+        let tmp_dir = tempfile::TempDir::new().unwrap();
+        let log_file_path = tmp_dir.path().join("logfile");
+
+        let log_format = $crate::logging::LogFormat::from_str($log_format).unwrap();
+        let _log_file = $crate::logging::init(&log_file_path, true, log_format).unwrap();
+
+        let log_file = std::fs::OpenOptions::new()
+            .read(true)
+            .open(&log_file_path)
+            .unwrap();
+
+        log_file
+    }};
+}

libs/utils/tests/logger_json_test.rs

@@ -0,0 +1,36 @@
+// This could be in ../src/logging.rs but since the logger is global, these
+// can't be run in threads of the same process
+use std::fs::File;
+use std::io::{BufRead, BufReader, Lines};
+use tracing::*;
+use utils::test_init_file_logger;
+
+fn read_lines(file: File) -> Lines<BufReader<File>> {
+    BufReader::new(file).lines()
+}
+
+#[test]
+fn test_json_format_has_message_and_custom_field() {
+    std::env::set_var("RUST_LOG", "info");
+
+    let log_file = test_init_file_logger!("info", "json");
+
+    let custom_field: &str = "hi";
+    trace!(custom = %custom_field, "test log message");
+    debug!(custom = %custom_field, "test log message");
+    info!(custom = %custom_field, "test log message");
+    warn!(custom = %custom_field, "test log message");
+    error!(custom = %custom_field, "test log message");
+
+    let lines = read_lines(log_file);
+    for line in lines {
+        let content = line.unwrap();
+        let json_object = serde_json::from_str::<serde_json::Value>(&content).unwrap();
+
+        assert_eq!(json_object["fields"]["custom"], "hi");
+        assert_eq!(json_object["fields"]["message"], "test log message");
+
+        assert_ne!(json_object["level"], "TRACE");
+        assert_ne!(json_object["level"], "DEBUG");
+    }
+}

libs/utils/tests/logger_plain_test.rs

@@ -0,0 +1,36 @@
+// This could be in ../src/logging.rs but since the logger is global, these
+// can't be run in threads of the same process
+use std::fs::File;
+use std::io::{BufRead, BufReader, Lines};
+use tracing::*;
+use utils::test_init_file_logger;
+
+fn read_lines(file: File) -> Lines<BufReader<File>> {
+    BufReader::new(file).lines()
+}
+
+#[test]
+fn test_plain_format_has_message_and_custom_field() {
+    std::env::set_var("RUST_LOG", "warn");
+
+    let log_file = test_init_file_logger!("warn", "plain");
+
+    let custom_field: &str = "hi";
+    trace!(custom = %custom_field, "test log message");
+    debug!(custom = %custom_field, "test log message");
+    info!(custom = %custom_field, "test log message");
+    warn!(custom = %custom_field, "test log message");
+    error!(custom = %custom_field, "test log message");
+
+    let lines = read_lines(log_file);
+    for line in lines {
+        let content = line.unwrap();
+        serde_json::from_str::<serde_json::Value>(&content).unwrap_err();
+        assert!(content.contains("custom=hi"));
+        assert!(content.contains("test log message"));
+
+        assert!(!content.contains("TRACE"));
+        assert!(!content.contains("DEBUG"));
+        assert!(!content.contains("INFO"));
+    }
+}

pageserver/src/bin/pageserver.rs

@@ -199,7 +199,7 @@ fn initialize_config(
 
 fn start_pageserver(conf: &'static PageServerConf, daemonize: bool) -> Result<()> {
     // Initialize logger
-    let log_file = logging::init(LOG_FILE_NAME, daemonize)?;
+    let log_file = logging::init(LOG_FILE_NAME, daemonize, conf.log_format)?;
 
     info!("version: {}", version());
 

pageserver/src/config.rs

@@ -17,6 +17,7 @@ use toml_edit::{Document, Item};
 use url::Url;
 use utils::{
     id::{NodeId, TenantId, TimelineId},
+    logging::LogFormat,
     postgres_backend::AuthType,
 };
 
@@ -45,6 +46,8 @@ pub mod defaults {
     pub const DEFAULT_PAGE_CACHE_SIZE: usize = 8192;
     pub const DEFAULT_MAX_FILE_DESCRIPTORS: usize = 100;
 
+    pub const DEFAULT_LOG_FORMAT: &str = "plain";
+
     ///
     /// Default built-in configuration file.
     ///
@@ -63,6 +66,7 @@ pub mod defaults {
 # initial superuser role name to use when creating a new tenant
 #initial_superuser_name = '{DEFAULT_SUPERUSER}'
 
+#log_format = '{DEFAULT_LOG_FORMAT}'
 # [tenant_config]
 #checkpoint_distance = {DEFAULT_CHECKPOINT_DISTANCE} # in bytes
 #checkpoint_timeout = {DEFAULT_CHECKPOINT_TIMEOUT}
@@ -126,6 +130,8 @@ pub struct PageServerConf {
 
     /// Etcd broker endpoints to connect to.
     pub broker_endpoints: Vec<Url>,
+
+    pub log_format: LogFormat,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -192,6 +198,8 @@ struct PageServerConfigBuilder {
     profiling: BuilderValue<ProfilingConfig>,
     broker_etcd_prefix: BuilderValue<String>,
     broker_endpoints: BuilderValue<Vec<Url>>,
+
+    log_format: BuilderValue<LogFormat>,
 }
 
 impl Default for PageServerConfigBuilder {
@@ -219,6 +227,7 @@ impl Default for PageServerConfigBuilder {
             profiling: Set(ProfilingConfig::Disabled),
             broker_etcd_prefix: Set(etcd_broker::DEFAULT_NEON_BROKER_ETCD_PREFIX.to_string()),
             broker_endpoints: Set(Vec::new()),
+            log_format: Set(LogFormat::from_str(DEFAULT_LOG_FORMAT).unwrap()),
         }
     }
 }
@@ -291,6 +300,10 @@ impl PageServerConfigBuilder {
         self.profiling = BuilderValue::Set(profiling)
     }
 
+    pub fn log_format(&mut self, log_format: LogFormat) {
+        self.log_format = BuilderValue::Set(log_format)
+    }
+
     pub fn build(self) -> anyhow::Result<PageServerConf> {
         let broker_endpoints = self
             .broker_endpoints
@@ -335,6 +348,7 @@ impl PageServerConfigBuilder {
             broker_etcd_prefix: self
                 .broker_etcd_prefix
                 .ok_or(anyhow!("missing broker_etcd_prefix"))?,
+            log_format: self.log_format.ok_or(anyhow!("missing log_format"))?,
         })
     }
 }
@@ -459,6 +473,9 @@ impl PageServerConf {
                         })
                         .collect::<anyhow::Result<_>>()?,
                 ),
+                "log_format" => builder.log_format(
+                    LogFormat::from_config(&parse_toml_string(key, item)?)?
+                ),
                 _ => bail!("unrecognized pageserver option '{key}'"),
             }
         }
@@ -571,6 +588,7 @@ impl PageServerConf {
             default_tenant_conf: TenantConf::dummy_conf(),
             broker_endpoints: Vec::new(),
             broker_etcd_prefix: etcd_broker::DEFAULT_NEON_BROKER_ETCD_PREFIX.to_string(),
+            log_format: LogFormat::from_str(defaults::DEFAULT_LOG_FORMAT).unwrap(),
         }
     }
 }
@@ -665,6 +683,8 @@ max_file_descriptors = 333
 initial_superuser_name = 'zzzz'
 id = 10
 
+log_format = 'json'
+
 "#;
 
     #[test]
@@ -704,6 +724,7 @@ id = 10
                     .parse()
                     .expect("Failed to parse a valid broker endpoint URL")],
                 broker_etcd_prefix: etcd_broker::DEFAULT_NEON_BROKER_ETCD_PREFIX.to_string(),
+                log_format: LogFormat::from_str(defaults::DEFAULT_LOG_FORMAT).unwrap(),
             },
             "Correct defaults should be used when no config values are provided"
         );
@@ -748,6 +769,7 @@ id = 10
                     .parse()
                     .expect("Failed to parse a valid broker endpoint URL")],
                 broker_etcd_prefix: etcd_broker::DEFAULT_NEON_BROKER_ETCD_PREFIX.to_string(),
+                log_format: LogFormat::Json,
             },
             "Should be able to parse all basic config values correctly"
         );

safekeeper/src/bin/safekeeper.rs

@@ -32,8 +32,12 @@ use safekeeper::GlobalTimelines;
 use safekeeper::SafeKeeperConf;
 use utils::auth::JwtAuth;
 use utils::{
-    http::endpoint, id::NodeId, logging, project_git_version, shutdown::exit_now, signals,
-    tcp_listener,
+    http::endpoint,
+    id::NodeId,
+    logging::{self, LogFormat},
+    project_git_version,
+    shutdown::exit_now,
+    signals, tcp_listener,
 };
 
 const LOCK_FILE_NAME: &str = "safekeeper.lock";
@@ -131,11 +135,15 @@ fn main() -> anyhow::Result<()> {
         .get_one::<String>("auth-validation-public-key-path")
         .map(PathBuf::from);
 
+    if let Some(log_format) = arg_matches.get_one::<String>("log-format") {
+        conf.log_format = LogFormat::from_config(log_format)?;
+    }
+
     start_safekeeper(conf, given_id, arg_matches.get_flag("init"))
 }
 
 fn start_safekeeper(mut conf: SafeKeeperConf, given_id: Option<NodeId>, init: bool) -> Result<()> {
-    let log_file = logging::init("safekeeper.log", conf.daemonize)?;
+    let log_file = logging::init("safekeeper.log", conf.daemonize, conf.log_format)?;
 
     info!("version: {GIT_VERSION}");
 
@@ -436,6 +444,11 @@ fn cli() -> Command {
                 .long("auth-validation-public-key-path")
                 .help("Path to an RSA .pem public key which is used to check JWT tokens")
         )
+        .arg(
+            Arg::new("log-format")
+                .long("log-format")
+                .help("Format for logging, either 'plain' or 'json'")
+        )
 }
 
 #[test]

safekeeper/src/lib.rs

@@ -7,7 +7,10 @@ use std::path::PathBuf;
 use std::time::Duration;
 use url::Url;
 
-use utils::id::{NodeId, TenantId, TenantTimelineId};
+use utils::{
+    id::{NodeId, TenantId, TenantTimelineId},
+    logging::LogFormat,
+};
 
 pub mod broker;
 pub mod control_file;
@@ -64,6 +67,7 @@ pub struct SafeKeeperConf {
     pub auth_validation_public_key_path: Option<PathBuf>,
     pub heartbeat_timeout: Duration,
     pub max_offloader_lag_bytes: u64,
+    pub log_format: LogFormat,
 }
 
 impl SafeKeeperConf {
@@ -97,6 +101,7 @@ impl Default for SafeKeeperConf {
             auth_validation_public_key_path: None,
             heartbeat_timeout: DEFAULT_HEARTBEAT_TIMEOUT,
             max_offloader_lag_bytes: DEFAULT_MAX_OFFLOADER_LAG_BYTES,
+            log_format: LogFormat::Plain,
         }
     }
 }