cloudpub-common 3.6.1173

use anyhow::{Context as _, Result};
use rolling_file::{BasicRollingFileAppender, RollingConditionBasic};
use std::path::Path;
use tracing::debug;
pub use tracing_appender::non_blocking::WorkerGuard;
use tracing_log::LogTracer;
use tracing_subscriber::fmt;
use tracing_subscriber::fmt::time::FormatTime;
use tracing_subscriber::prelude::*;

/// The default trace-filter directives, RUST_LOG aside.
///
/// The public listener faces the whole internet, and pingora reports every
/// failed TLS handshake and every garbage HTTP request a scanner sends at
/// ERROR - hundreds of thousands of lines a day on prod. Those modules are
/// silenced outright; the rest of pingora_core is capped at ERROR to drop its
/// routine idle-connection and response-header warnings - CLOUDPUB-1281.
pub fn trace_filter(level: &str) -> String {
    format!(
        "{},hyper=error,tokio_postgres=info,pingora_core=error,pingora_core::services::listening=off,pingora_core::apps=off,pingora_proxy=off,pingora_pool=error,tungstenite=info",
        level
    )
}

/// The ceiling for the `log` crate, derived from the configured level.
///
/// A `log` macro evaluates its arguments whenever `log::max_level()` admits
/// the record, and pingora logs raw upstream response headers with
/// `trace!("Raw Response header: {:?}", str::from_utf8(..).unwrap())`. Left
/// at Trace - which is what a bare `LogTracer::init()` does, deferring all
/// filtering to the tracing subscriber - that unwrap runs on every upstream
/// response and panics the proxy worker on the first header carrying
/// non-UTF-8 bytes (CLOUDPUB-1507). Capping the `log` crate at the level the
/// operator actually configured keeps those arguments unevaluated, and saves
/// the formatting work besides.
///
/// Parsed the way the tracing `EnvFilter` built from the same string parses
/// it - case-insensitively, whitespace trimmed - so the ceiling never lands
/// below the configured level and drops records the operator asked for.
/// Anything unparseable falls back to Info: the fallback must not be the one
/// value that reinstates the panic.
pub fn log_max_level(level: &str) -> log::LevelFilter {
    match level.trim().to_ascii_lowercase().as_str() {
        "off" => log::LevelFilter::Off,
        "error" => log::LevelFilter::Error,
        "warn" => log::LevelFilter::Warn,
        "debug" => log::LevelFilter::Debug,
        "trace" => log::LevelFilter::Trace,
        _ => log::LevelFilter::Info,
    }
}

/// The zone that renders the timestamp at the head of a log line.
///
/// The two variants serve two different audiences, and that is deliberate -
/// not a divergence waiting to be unified:
///
/// - The server logs in [`LogZone::Utc`]: server.log is read next to
///   backend.log and security.log, which Django writes in UTC (USE_TZ), and
///   a local-zone stamp skews the incident timeline across the files by the
///   zone offset - three hours on the prod MSK hosts (CLOUDPUB-1474).
/// - The client logs in [`LogZone::Local`]: its log is read by one person on
///   their own machine, correlating entries with what they just did by hand.
///   The stamp format carries no offset of its own, so a UTC stamp would
///   show that person a shifted time with no sign it is UTC - it would read
///   as broken clocks, not as UTC. Showing the zone in the format is the
///   prerequisite for ever moving the client to UTC, and the format is
///   deliberately outside this ticket's scope.
pub enum LogZone {
    Utc,
    Local,
}

/// The timestamp put at the head of every log line: `YYYY-MM-DD HH:MM:SS`,
/// in the zone `LogZone` asks for. The format string carries no offset of
/// its own, so the zone is invisible in the line and only observable in the
/// rendered digits - see the test below, which pins the server's choice.
///
/// Extracted from `init_log` so that zone is pinnable by a test.
fn log_timer(zone: LogZone) -> Result<impl FormatTime + Clone> {
    let format = time::format_description::parse_borrowed::<2>(
        "[year]-[month padding:zero]-[day padding:zero] [hour]:[minute]:[second]",
    )
    .context("Failed to parse time format")?;

    let offset = match zone {
        LogZone::Utc => time::UtcOffset::UTC,
        LogZone::Local => time::UtcOffset::current_local_offset().unwrap_or(time::UtcOffset::UTC),
    };
    Ok(fmt::time::OffsetTime::new(offset, format))
}

pub fn init_log(
    level: &str,
    log_file: &Path,
    stderr: bool,
    zone: LogZone,
    max_size: usize,
    max_files: usize,
) -> Result<WorkerGuard> {
    let file_appender = BasicRollingFileAppender::new(
        log_file,
        RollingConditionBasic::new().max_size(max_size as u64), // 10 MB
        max_files,
    )
    .context("Failed to create rolling file appender")?;

    let (file_writer, guard) = tracing_appender::non_blocking(file_appender);

    let trace_cfg = format!(
        "{},{}",
        trace_filter(level),
        std::env::var("RUST_LOG").unwrap_or_default()
    );

    let timer = log_timer(zone).context("Failed to build log timer")?;

    // Create the file layer with common settings
    let file_layer = fmt::Layer::default()
        .with_timer(timer.clone())
        .with_ansi(false)
        .with_writer(file_writer);

    // Build a registry with the file layer and conditional stderr layer
    let registry = tracing_subscriber::registry()
        .with(tracing_subscriber::EnvFilter::new(trace_cfg))
        .with(file_layer);

    // Add stderr layer only if requested
    if stderr {
        let stderr_layer = fmt::Layer::default()
            .with_timer(timer)
            .with_ansi(cfg!(unix))
            .with_writer(std::io::stderr);

        tracing::subscriber::set_global_default(registry.with(stderr_layer))
            .context("Failed to set global default subscriber")?;
    } else {
        tracing::subscriber::set_global_default(registry)
            .context("Failed to set global default subscriber")?;
    }

    LogTracer::builder()
        .with_max_level(log_max_level(level))
        .init()
        .context("Failed to initialize log tracer")?;
    log_panics::init();

    debug!("Tracing initialized ({})", level);
    Ok(guard)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io;
    use std::sync::{Arc, Mutex};
    use tracing::subscriber::with_default;
    use tracing_subscriber::fmt::MakeWriter;
    use tracing_subscriber::layer::{Context, Layer};
    use tracing_subscriber::registry::LookupSpan;

    #[derive(Clone, Default)]
    struct Capture(Arc<Mutex<Vec<String>>>);

    impl<S: tracing::Subscriber + for<'a> LookupSpan<'a>> Layer<S> for Capture {
        fn on_event(&self, event: &tracing::Event<'_>, _ctx: Context<'_, S>) {
            self.0
                .lock()
                .unwrap()
                .push(event.metadata().target().to_string());
        }
    }

    /// Internet scanners hammer the public TLS port and pingora logs every
    /// failed handshake and every garbage HTTP request at ERROR - hundreds of
    /// thousands of lines a day on prod, drowning out real errors. The trace
    /// filter has to drop that noise while keeping the server's own errors.
    /// STATUS: FIXED - see CLOUDPUB-1281
    #[test]
    fn trace_filter_suppresses_scanner_noise() {
        let capture = Capture::default();
        let filter = tracing_subscriber::EnvFilter::new(trace_filter("error"));
        let subscriber = tracing_subscriber::registry()
            .with(filter)
            .with(capture.clone());

        with_default(subscriber, || {
            // Scanner noise pingora reports at ERROR/WARN level
            tracing::error!(
                target: "pingora_core::services::listening",
                "Downstream handshake error: TLS accept() failed"
            );
            tracing::error!(
                target: "pingora_core::apps::http_app",
                "HTTP server fails to read from downstream: InvalidHTTPHeader"
            );
            tracing::warn!(
                target: "pingora_core::connectors",
                "Unexpected data read in idle connection"
            );
            tracing::warn!(
                target: "pingora_core::protocols::http::v1::server",
                "Respond header is already sent, cannot send again"
            );
            // The server's own errors must keep flowing
            tracing::error!(target: "server::control", "real error");
        });

        let targets = capture.0.lock().unwrap();
        assert!(
            !targets.iter().any(|t| t.starts_with("pingora_core")),
            "pingora scanner noise must be filtered out, got: {:?}",
            targets
        );
        assert!(
            targets.iter().any(|t| t == "server::control"),
            "the server's own errors must not be filtered out, got: {:?}",
            targets
        );
    }

    /// In-memory sink for the formatted log line: the fmt layer writes the
    /// whole `timestamp LEVEL target: message` record here, and the test reads
    /// the stamp the way a log consumer grepping the file would.
    #[derive(Clone, Default)]
    struct LineSink(Arc<Mutex<Vec<u8>>>);

    impl<'a> MakeWriter<'a> for LineSink {
        type Writer = LineWriter;

        fn make_writer(&'a self) -> LineWriter {
            LineWriter(self.0.clone())
        }
    }

    struct LineWriter(Arc<Mutex<Vec<u8>>>);

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

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

    /// Decodes a `YYYY-MM-DD HH:MM:SS` stamp taken off a log line into an
    /// instant on the UTC timeline. The digits are split and rebuilt as a
    /// calendar date by hand, independently of the format machinery under
    /// test: whatever the timer wrote, it is read back as UTC - which is the
    /// contract, since the format string carries no offset of its own.
    fn stamp_as_utc(line: &str) -> time::OffsetDateTime {
        let stamp = &line[..19]; // "YYYY-MM-DD HH:MM:SS" is exactly 19 chars
        let fields: Vec<&str> = stamp.split(['-', ' ', ':']).collect();
        assert_eq!(fields.len(), 6, "malformed stamp {stamp:?} in {line:?}");
        let number = |i: usize| {
            fields[i]
                .parse::<u32>()
                .unwrap_or_else(|e| panic!("field {i} of {stamp:?}: {e}"))
        };
        let month = u8::try_from(number(1)).unwrap();
        time::Date::from_calendar_date(
            number(0) as i32,
            time::Month::try_from(month).unwrap(),
            u8::try_from(number(2)).unwrap(),
        )
        .unwrap()
        .with_hms(
            u8::try_from(number(3)).unwrap(),
            u8::try_from(number(4)).unwrap(),
            u8::try_from(number(5)).unwrap(),
        )
        .unwrap()
        .assume_utc()
    }

    /// The stamp on a rust log line must render its instant in UTC, not in
    /// the host's local zone. The backend logs (backend.log, security.log)
    /// come from Django with USE_TZ and are therefore UTC, and a local-zone
    /// server.log skews against them by the zone offset - three hours on the
    /// prod MSK hosts - so an incident timeline cannot be spliced across the
    /// files without remembering the shift.
    /// HYPOTHESIS: `log_timer()` builds its timer with
    /// `UtcOffset::current_local_offset()`, so on this +07 dev host a line
    /// stamped now renders now+7h and lands far off the UTC clock.
    /// SOLUTION: render the stamp with `fmt::time::UtcTime` so every host,
    /// whatever its local zone, logs UTC.
    /// RESULT: the rendered stamp lands within two minutes of `now_utc()`; a
    /// local-zone timer is a whole zone offset away (25200s here), which no
    /// scheduling jitter can bridge.
    /// STATUS: FIXED - see CLOUDPUB-1474
    #[test]
    fn log_timestamp_is_rendered_in_utc() {
        let sink = LineSink::default();
        let layer = fmt::Layer::default()
            .with_timer(log_timer(LogZone::Utc).expect("failed to build log timer"))
            .with_ansi(false)
            .with_writer(sink.clone());
        let subscriber = tracing_subscriber::registry().with(layer);

        with_default(subscriber, || {
            tracing::info!("timestamp zone probe");
        });

        let line = String::from_utf8(std::mem::take(&mut *sink.0.lock().unwrap()))
            .expect("log line must be valid UTF-8");
        let logged = stamp_as_utc(&line);
        let skew =
            (logged.unix_timestamp() - time::OffsetDateTime::now_utc().unix_timestamp()).abs();
        assert!(
            skew <= 120,
            "log stamp {logged} is {skew}s away from UTC now - \
             stamps must render UTC, got line: {line:?}"
        );
    }

    /// The `log` crate's max level must follow the configured level, so that
    /// `log::trace!`/`debug!` arguments are never evaluated above it.
    ///
    /// HYPOTHESIS: prod (server 3.5.1056, pingora-core 0.8.1) panicked three
    /// times on 07.09 in the HTTP proxy worker - `Utf8Error { valid_up_to:
    /// 153 }` out of `pingora-core-0.8.1/src/protocols/http/v1/client.rs:760`,
    /// which is `trace!("Raw Response header: {:?}",
    /// str::from_utf8(self.get_headers_raw()).unwrap())`. A `log` macro
    /// evaluates its arguments whenever `log::max_level()` admits the record,
    /// and `LogTracer::init()` sets that ceiling to Trace, leaving all
    /// filtering to the tracing subscriber - so the unwrap ran on every
    /// upstream response no matter how the server was configured, and any
    /// upstream header carrying non-UTF-8 bytes (a cp1251 filename in
    /// Content-Disposition) tore the request down.
    /// SOLUTION: derive the ceiling from the same level string init_log
    /// receives and hand it to `LogTracer::builder().with_max_level(..)`, so
    /// the argument is not evaluated at all below it.
    /// RESULT: red before the fix - `log_max_level` did not exist and the
    /// module failed to compile; green once it does.
    ///
    /// The assertion stops at the pure function on purpose: `log::max_level()`
    /// is process-global state set by `LogTracer::init`, and the suite shares
    /// one process with one global subscriber, so an assertion on it after
    /// `init_log` would depend on which test installed the logger first.
    /// STATUS: FIXED - see CLOUDPUB-1507
    #[test]
    fn log_max_level_follows_configured_level() {
        use log::LevelFilter;

        assert_eq!(log_max_level("error"), LevelFilter::Error);
        assert_eq!(log_max_level("warn"), LevelFilter::Warn);
        assert_eq!(log_max_level("info"), LevelFilter::Info);
        assert_eq!(log_max_level("debug"), LevelFilter::Debug);
        assert_eq!(log_max_level("trace"), LevelFilter::Trace);
        // "off" is a level the EnvFilter next door accepts too.
        assert_eq!(log_max_level("off"), LevelFilter::Off);

        // The level reaches init_log from a config file (server.toml
        // log_level) or the --log-level flag, so it is free-form user input:
        // parse it the way the tracing EnvFilter next to it does, which is
        // case-insensitively. Getting this wrong would cap the `log` crate
        // BELOW the configured level and silently drop records the operator
        // asked for.
        assert_eq!(log_max_level("INFO"), LevelFilter::Info);
        assert_eq!(log_max_level("Trace"), LevelFilter::Trace);
        assert_eq!(log_max_level(" debug "), LevelFilter::Debug);

        // Anything unparseable falls back to Info rather than to Trace: the
        // fallback must not be the one value that reinstates the panic.
        assert_eq!(log_max_level("bogus"), LevelFilter::Info);
        assert_eq!(log_max_level(""), LevelFilter::Info);
    }
}