path: root/src/logging.rs

//! Logging: a compact fmt logger whose filter can be replaced while the node runs
//! (`/admin/log`), and, with the `tokio-console` feature, a tokio-console layer.
//!
//! The filter starts from `RUST_LOG`, or `info` without it. It applies to the fmt
//! logger only, so the console layer keeps getting the runtime events it needs.
//!
//! Each line is written with members named by their alias (`Renaming`): wherever a
//! member's node ID or one of its addresses appears, in a field, a message or a dump of
//! a value, it reads `<scheme>:<alias>` (`continuum:<alias>` for the node ID,
//! `libp2p:<alias>` for its libp2p peer ID, with its scheme or bare). The names and
//! aliases are as `set_aliases` last gave them (`swarm::topology`, whenever it looks at
//! the member list); at start, as the state file remembers them, so they show from the
//! first line (stale, perhaps: a member's alias that changed meanwhile is logged as such
//! once the member list says so). `--log-raw-ids` leaves IDs as they are.
//!
//! Under systemd's journal (`--log-format journal`, or `auto` when stdout goes to the
//! journal: `JOURNAL_STREAM`), lines go without a timestamp or colours, each with its
//! level as a syslog priority prefix (`<3>` for errors … `<7>` for debug), which the
//! journal takes for the entry's priority (`journalctl -p warning`).

use std::borrow::Cow;
use std::collections::{BTreeMap, HashMap};
use std::fmt;
use std::io::Write;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{OnceLock, RwLock};

use anyhow::{Result, anyhow};
use tracing::{Level, Metadata};
use tracing_subscriber::fmt::MakeWriter;
use tracing_subscriber::fmt::format::Writer;
use tracing_subscriber::fmt::time::{FormatTime, SystemTime};
use tracing_subscriber::{EnvFilter, Registry, prelude::*, reload};

use crate::locks::RwLockExt;

type FilterHandle = reload::Handle<EnvFilter, Registry>;

static FILTER: OnceLock<FilterHandle> = OnceLock::new();

/// Members' aliases, by the names that stand for them (`set_aliases`), and the
/// replacements made of them, longest first.
static ALIASES: RwLock<Aliases> = RwLock::new(Aliases {
    given: None,
    renames: Vec::new(),
});
/// `--log-raw-ids`.
static RAW_IDS: AtomicBool = AtomicBool::new(false);
/// Lines for the journal (`set_format`).
static JOURNAL: AtomicBool = AtomicBool::new(false);

/// `--log-format`.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum LogFormat {
    /// The journal's if stdout goes to it, else text.
    Auto,
    /// Timestamped lines, coloured on a terminal.
    Text,
    /// For systemd's journal (module docs).
    Journal,
}

impl LogFormat {
    pub(crate) const NAMES: [&str; 3] = ["auto", "text", "journal"];

    pub(crate) fn parse(name: &str) -> Option<Self> {
        match name {
            "auto" => Some(Self::Auto),
            "text" => Some(Self::Text),
            "journal" => Some(Self::Journal),
            _ => None,
        }
    }
}

/// Writes lines in `format` from now on.
pub(crate) fn set_format(format: LogFormat) {
    let journal = match format {
        LogFormat::Auto => to_journal(),
        LogFormat::Text => false,
        LogFormat::Journal => true,
    };
    JOURNAL.store(journal, Ordering::Relaxed);
}

/// Whether stdout is the journal's stream: systemd says which (`JOURNAL_STREAM`, its
/// device and inode) when it connects one.
fn to_journal() -> bool {
    let Some(stream) = std::env::var_os("JOURNAL_STREAM") else {
        return false;
    };
    let Some((device, inode)) = stream.to_str().and_then(|stream| stream.split_once(':')) else {
        return false;
    };
    let mut stat = std::mem::MaybeUninit::<libc::stat>::uninit();
    // SAFETY: fstat writes the struct for a valid descriptor, and says so.
    if unsafe { libc::fstat(libc::STDOUT_FILENO, stat.as_mut_ptr()) } != 0 {
        return false;
    }
    // SAFETY: fstat succeeded.
    let stat = unsafe { stat.assume_init() };
    device.parse::<u64>().ok() == Some(stat.st_dev as u64)
        && inode.parse::<u64>().ok() == Some(stat.st_ino as u64)
}

/// The fmt logger's timestamps, but none for the journal (it has its own).
struct Timestamps;

impl FormatTime for Timestamps {
    fn format_time(&self, writer: &mut Writer<'_>) -> fmt::Result {
        if JOURNAL.load(Ordering::Relaxed) {
            return Ok(());
        }
        SystemTime.format_time(writer)
    }
}

/// The syslog priority of `level`, as the journal reads it at the start of a line.
fn priority(level: &Level) -> &'static str {
    match *level {
        Level::ERROR => "<3>",
        Level::WARN => "<4>",
        Level::INFO => "<6>",
        _ => "<7>",
    }
}

/// `line` for the journal: without colours or leading blanks (where the timestamp went),
/// each of its lines with `priority` in front.
fn for_journal(line: &str, priority: &str) -> String {
    let mut plain = String::with_capacity(line.len());
    let mut chars = line.chars();
    while let Some(c) = chars.next() {
        if c == '\x1b' {
            // A colour: ESC [ ... m.
            for c in chars.by_ref() {
                if c == 'm' {
                    break;
                }
            }
        } else {
            plain.push(c);
        }
    }
    let mut out = String::with_capacity(plain.len() + 4);
    for part in plain.trim_start().split_inclusive('\n') {
        out.push_str(priority);
        out.push_str(part);
    }
    out
}

struct Aliases {
    given: Option<HashMap<String, String>>,
    renames: Vec<(String, String)>,
}

/// Leave IDs in the log as they are (`--log-raw-ids`).
pub(crate) fn set_raw_ids(raw: bool) {
    RAW_IDS.store(raw, Ordering::Relaxed);
}

/// The aliases to name members by from now on, by the names that stand for them: node
/// IDs and addresses (`<scheme>:<transport identity>`).
pub(crate) fn set_aliases(aliases: HashMap<String, String>) {
    let previous = {
        let mut current = ALIASES.write_or_recover();
        if current.given.as_ref() == Some(&aliases) {
            return;
        }
        current.renames = renames(&aliases);
        current.given.replace(aliases.clone())
    };
    // A member's alias that changed, after the lock: the line itself is renamed.
    let Some(previous) = previous else {
        return;
    };
    for (name, before) in &previous {
        if !crate::identity::is_node_id(name) {
            continue;
        }
        match aliases.get(name) {
            Some(after) if after != before => {
                tracing::info!(peer = %name, previous = %before, "A member's alias changed");
            }
            None => tracing::info!(peer = %name, previous = %before, "A member's alias is gone"),
            _ => {}
        }
    }
}

/// What each name becomes: `<scheme>:<alias>`, for the name with its scheme and bare.
/// Longest first, so that an address is renamed whole before its bare identity could be.
fn renames(aliases: &HashMap<String, String>) -> Vec<(String, String)> {
    let mut renames: BTreeMap<String, String> = BTreeMap::new();
    for (name, alias) in aliases {
        let Some((scheme, identity)) = name.split_once(':') else {
            continue;
        };
        let renamed = format!("{scheme}:{alias}");
        renames.insert(name.clone(), renamed.clone());
        // A node ID's hex alone is no name of it.
        if !crate::identity::is_node_id(name) && !identity.is_empty() {
            renames.entry(identity.to_owned()).or_insert(renamed);
        }
    }
    let mut renames: Vec<(String, String)> = renames.into_iter().collect();
    renames.sort_by(|a, b| b.0.len().cmp(&a.0.len()).then_with(|| a.0.cmp(&b.0)));
    renames
}

/// The aliases `set_aliases` was last given, sorted (for the state file).
pub(crate) fn aliases() -> BTreeMap<String, String> {
    ALIASES
        .read_or_recover()
        .given
        .iter()
        .flatten()
        .map(|(name, alias)| (name.clone(), alias.clone()))
        .collect()
}

/// `line` with members named by their alias (module docs).
fn rename(line: &str) -> Cow<'_, str> {
    if RAW_IDS.load(Ordering::Relaxed) {
        return Cow::Borrowed(line);
    }
    rename_with(line, &ALIASES.read_or_recover().renames)
}

fn rename_with<'a>(line: &'a str, renames: &[(String, String)]) -> Cow<'a, str> {
    let mut renamed = Cow::Borrowed(line);
    for (name, alias) in renames {
        if renamed.contains(name.as_str()) {
            renamed = Cow::Owned(renamed.replace(name.as_str(), alias));
        }
    }
    renamed
}

/// Writes the log to stdout, a line (one event) at a time, renamed (`rename`).
struct Renaming;

impl<'a> MakeWriter<'a> for Renaming {
    type Writer = RenamingLine;

    fn make_writer(&'a self) -> Self::Writer {
        RenamingLine(Vec::new(), &Level::INFO)
    }

    fn make_writer_for(&'a self, meta: &Metadata<'_>) -> Self::Writer {
        RenamingLine(Vec::new(), priority_level(meta.level()))
    }
}

fn priority_level(level: &Level) -> &'static Level {
    match *level {
        Level::ERROR => &Level::ERROR,
        Level::WARN => &Level::WARN,
        Level::INFO => &Level::INFO,
        Level::DEBUG => &Level::DEBUG,
        _ => &Level::TRACE,
    }
}

/// One event's text, and its level, written renamed when the logger is done with it.
struct RenamingLine(Vec<u8>, &'static Level);

impl Write for RenamingLine {
    fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
        self.0.extend_from_slice(bytes);
        Ok(bytes.len())
    }

    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}

impl Drop for RenamingLine {
    fn drop(&mut self) {
        let text = String::from_utf8_lossy(&self.0);
        let renamed = rename(&text);
        let line = if JOURNAL.load(Ordering::Relaxed) {
            Cow::Owned(for_journal(&renamed, priority(self.1)))
        } else {
            renamed
        };
        let _ = std::io::stdout().write_all(line.as_bytes());
    }
}

/// Installs the global subscriber. Call once, first thing in main.
pub(crate) fn init() {
    let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
    let (filter, handle) = reload::Layer::new(filter);

    // Until the configuration says otherwise (`set_format`).
    set_format(LogFormat::Auto);
    let registry = tracing_subscriber::registry().with(
        tracing_subscriber::fmt::layer()
            .compact()
            .with_timer(Timestamps)
            // The journal gets none either way (`for_journal`).
            .with_ansi(!to_journal())
            .with_writer(Renaming)
            .with_filter(filter),
    );

    #[cfg(feature = "tokio-console")]
    let registry = registry
        .with(console_subscriber::spawn().with_filter(EnvFilter::new("tokio=trace,runtime=trace")));

    registry.init();
    let _ = FILTER.set(handle);
}

/// The filter in effect, in `RUST_LOG` syntax.
pub(crate) fn filter() -> Result<String> {
    handle()?
        .with_current(|filter| filter.to_string())
        .map_err(|error| anyhow!("Reading the log filter: {error}"))
}

/// Replaces the filter. `directives` uses `RUST_LOG` syntax, e.g.
/// `info,continuum::swarm::sync=trace`. An invalid one leaves the old filter in place.
pub(crate) fn set_filter(directives: &str) -> Result<String> {
    let filter = EnvFilter::try_new(directives)
        .map_err(|error| anyhow!("Invalid log filter {directives:?}: {error}"))?;
    let applied = filter.to_string();
    handle()?
        .reload(filter)
        .map_err(|error| anyhow!("Replacing the log filter: {error}"))?;
    Ok(applied)
}

fn handle() -> Result<&'static FilterHandle> {
    FILTER
        .get()
        .ok_or_else(|| anyhow!("Logging is not initialised"))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn journal_lines() {
        assert_eq!(
            for_journal(
                " \x1b[32m INFO\x1b[0m continuum: Ready peer=a\n",
                priority(&Level::INFO)
            ),
            "<6>INFO continuum: Ready peer=a\n"
        );
        assert_eq!(
            for_journal("ERROR x: one\ntwo\n", priority(&Level::ERROR)),
            "<3>ERROR x: one\n<3>two\n"
        );
        assert_eq!(priority(&Level::TRACE), "<7>");
        assert_eq!(LogFormat::parse("journal"), Some(LogFormat::Journal));
        assert_eq!(LogFormat::parse("syslog"), None);
    }

    #[test]
    fn members_are_named_by_their_alias() {
        let node = crate::identity::node_id_of(&[1; 32]);
        let renames = renames(&HashMap::from([
            (node.clone(), "kitchen".to_owned()),
            ("libp2p:12D3KooWexample".to_owned(), "kitchen".to_owned()),
            ("tor:abc.onion:9662".to_owned(), "kitchen".to_owned()),
        ]));
        let line = format!(
            "peer={node} to=libp2p:12D3KooWexample event=PeerId(\"12D3KooWexample\") at=abc.onion:9662 other=12D3KooWother"
        );
        assert_eq!(
            rename_with(&line, &renames),
            "peer=continuum:kitchen to=libp2p:kitchen event=PeerId(\"libp2p:kitchen\") at=tor:kitchen other=12D3KooWother"
        );
        // A node ID's hex alone stays.
        let hex = node.split_once(':').unwrap().1;
        assert_eq!(rename_with(hex, &renames), hex);
    }
}