diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/config/mod.rs | 19 | ||||
| -rw-r--r-- | src/logging.rs | 163 | ||||
| -rw-r--r-- | src/main.rs | 3 | ||||
| -rw-r--r-- | src/transport/libp2p.rs | 109 |
4 files changed, 284 insertions, 10 deletions
diff --git a/src/config/mod.rs b/src/config/mod.rs index 9210922..5837d1e 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -110,6 +110,13 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[ "HTTP API users (from htpasswd) who may write protocol data under /store/_continuum (remake records, members, forget records), e.g. to repair a stuck remake. Such writes act on the whole swarm and are logged. Epoch metadata and the swarm ID are never writable.", ), option( + "log-format", + Kind::Text, + "auto|text|journal", + Some("auto"), + "How log lines look: `text` (timestamped), `journal` (for systemd's journal: no timestamp or colours, the level as each line's priority, for journalctl -p), or `auto`: journal when stdout goes to the journal, else text.", + ), + option( "log-raw-ids", Kind::Flag, "BOOL", @@ -390,6 +397,8 @@ pub(crate) struct Settings { pub(crate) protocol_writers: Vec<String>, #[serde(default)] pub(crate) debug_api: bool, + #[serde(default = "default_log_format")] + pub(crate) log_format: String, #[serde(default)] pub(crate) log_raw_ids: bool, #[serde(default)] @@ -499,6 +508,9 @@ pub(crate) const TRANSPORTS: &[(&str, bool)] = &[ ("libp2p", cfg!(feature = "libp2p")), ("tor", cfg!(feature = "tor")), ]; +fn default_log_format() -> String { + "auto".to_owned() +} fn default_tor_port() -> u16 { 9662 } @@ -535,6 +547,13 @@ impl Settings { } fn check(&self) -> Result<()> { + if crate::logging::LogFormat::parse(&self.log_format).is_none() { + bail!( + "log-format {:?}: one of {}", + self.log_format, + crate::logging::LogFormat::NAMES.join(", ") + ); + } if let Some(url) = &self.mycelium_rest_url { url::Url::parse(url).with_context(|| format!("mycelium-rest-url {url:?}"))?; } diff --git a/src/logging.rs b/src/logging.rs index 093d49e..216c7f9 100644 --- a/src/logging.rs +++ b/src/logging.rs @@ -12,15 +12,24 @@ //! 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; @@ -37,6 +46,109 @@ static ALIASES: RwLock<Aliases> = RwLock::new(Aliases { }); /// `--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>>, @@ -133,12 +245,26 @@ impl<'a> MakeWriter<'a> for Renaming { type Writer = RenamingLine; fn make_writer(&'a self) -> Self::Writer { - RenamingLine(Vec::new()) + RenamingLine(Vec::new(), &Level::INFO) + } + + fn make_writer_for(&'a self, meta: &Metadata<'_>) -> Self::Writer { + RenamingLine(Vec::new(), priority_level(meta.level())) } } -/// One event's text, written renamed when the logger is done with it. -struct RenamingLine(Vec<u8>); +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> { @@ -154,7 +280,13 @@ impl Write for RenamingLine { impl Drop for RenamingLine { fn drop(&mut self) { let text = String::from_utf8_lossy(&self.0); - let _ = std::io::stdout().write_all(rename(&text).as_bytes()); + 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()); } } @@ -163,9 +295,14 @@ 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), ); @@ -208,6 +345,24 @@ 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([ diff --git a/src/main.rs b/src/main.rs index f16138e..8af1c29 100644 --- a/src/main.rs +++ b/src/main.rs @@ -52,6 +52,9 @@ async fn main() -> Result<()> { return Ok(()); } let args = &config.settings; + if let Some(format) = logging::LogFormat::parse(&args.log_format) { + logging::set_format(format); + } if args.relay_only { return relay_only(args).await; } diff --git a/src/transport/libp2p.rs b/src/transport/libp2p.rs index 7351b16..d621c8e 100644 --- a/src/transport/libp2p.rs +++ b/src/transport/libp2p.rs @@ -252,6 +252,8 @@ struct Book { failed_relays: HashMap<PeerId, tokio::time::Instant>, /// Relays that granted us a slot we hold. slots: std::collections::HashSet<PeerId>, + /// Peers holding a slot on this node, as a relay. + slots_here: std::collections::HashSet<PeerId>, /// Open connections: to whom, whether through a relay, and the remote address. connections: HashMap<ConnectionId, (PeerId, bool, Multiaddr)>, /// What AutoNAT says, and the address it confirmed while public. @@ -305,6 +307,46 @@ fn describe_dial_error(error: &::libp2p::swarm::DialError) -> String { described.join("; ") } +/// Why a slot on a relay ended, or was never granted, shortly: libp2p nests the +/// relay's answer several types deep (`Left(Left(Left(Reservation(...))))`), and the +/// common cases deserve a sentence. +fn relay_failure(reason: &std::io::Result<()>) -> String { + let Err(error) = reason else { + return "closed".to_owned(); + }; + relay_failure_text(&format!("{error:?}"), &innermost_causes(error)) +} + +/// `relay_failure` on the error's debug form, and its innermost cause as text. +fn relay_failure_text(debug: &str, innermost: &str) -> String { + let known = [ + ( + "NoAddressesInReservation", + "the relay gave no address to be reached through it", + ), + ("ResourceLimitExceeded", "the relay has no slot left"), + ("PermissionDenied", "the relay doesn't let us reserve"), + ("NoReservation", "the relay holds no slot for us"), + ( + "connection gated", + "the relay refused the connection (connection gated)", + ), + ( + "UnsupportedProtocol", + "the relay doesn't relay (no circuit relay protocol)", + ), + ("Timeout", "the relay didn't answer in time"), + ]; + if let Some((_, said)) = known.iter().find(|(pattern, _)| debug.contains(pattern)) { + return (*said).to_owned(); + } + // The protocol refused, but only after its first data (`swarm_config`). + if innermost == "Failed" { + return "the relay refused the circuit relay protocol (is it a relay?)".to_owned(); + } + innermost.to_owned() +} + /// The innermost of `error`'s causes, each distinct one once: those libp2p lists in a /// "Multiple dial errors occurred:" text (one per line, ` - <cause>: <its cause>...`), /// or the last of the source chain. @@ -365,7 +407,15 @@ impl Book { /// The connections to `peer` to close, of several: all but one, the oldest direct /// one (the oldest relayed one if none is direct). - fn duplicates(&self, peer: &PeerId) -> Vec<ConnectionId> { + /// + /// None between a relay and a node with a slot on it, or reserving one (`reserving`): + /// a slot lives on one connection, which libp2p doesn't tell, and closing it would + /// end the slot, and with it every circuit to the node (the relay then answers + /// `NoReservation`). + fn duplicates(&self, peer: &PeerId, reserving: bool) -> Vec<ConnectionId> { + if reserving || self.slots.contains(peer) || self.slots_here.contains(peer) { + return Vec::new(); + } let mut connections: Vec<(bool, ConnectionId)> = self .connections .iter() @@ -1127,7 +1177,8 @@ async fn run( .collect(); for peer in due { dedupe_due.remove(&peer); - let duplicates = lock(&book).duplicates(&peer); + let reserving = picked.values().any(|relay| *relay == peer); + let duplicates = lock(&book).duplicates(&peer, reserving); for connection in duplicates { if swarm.close_connection(connection) { closing_duplicates.insert(connection); @@ -1327,13 +1378,15 @@ async fn run( } lost }; + let reason = relay_failure(&reason); if lost { - debug!(relay = %Addressed(SCHEME, &relay), reason = ?reason, "Lost the slot on a relay"); + debug!(relay = %Addressed(SCHEME, &relay), reason = %reason, "Lost the slot on a relay"); } else { info!( relay = %Addressed(SCHEME, &relay), pause = ?RELAY_PAUSE, - "A relay didn't grant a slot (unreachable, or another peer at its address); picking others for a while" + reason = %reason, + "A relay didn't grant a slot; picking others for a while" ); } } @@ -1374,6 +1427,20 @@ async fn run( nat = new; } SwarmEvent::Behaviour(BehaviourEvent::Relay(event)) => { + { + use ::libp2p::relay::Event; + let mut book = lock(&book); + match &event { + Event::ReservationReqAccepted { src_peer_id, .. } => { + book.slots_here.insert(*src_peer_id); + } + Event::ReservationClosed { src_peer_id } + | Event::ReservationTimedOut { src_peer_id } => { + book.slots_here.remove(src_peer_id); + } + _ => {} + } + } debug!(event = ?event, "Relay event"); } SwarmEvent::Behaviour(BehaviourEvent::Dcutr(event)) => match event.result { @@ -2218,6 +2285,29 @@ mod tests { assert_eq!(book.prune(later, |_| false), 1); // `connected`, once it isn't. } + #[test] + fn relay_failures_are_short() { + assert_eq!( + relay_failure_text( + "Custom { kind: Other, error: Other(Left(Left(Left(Reservation(Protocol(NoAddressesInReservation)))))) }", + "x" + ), + "the relay gave no address to be reached through it" + ); + assert_eq!( + relay_failure_text( + "Codec(Error(Custom { kind: Other, error: Failed }))", + "Failed" + ), + "the relay refused the circuit relay protocol (is it a relay?)" + ); + assert_eq!( + relay_failure_text("Io(…)", "connection reset"), + "connection reset" + ); + assert_eq!(relay_failure(&Ok(())), "closed"); + } + /// libp2p's nested dial errors come out as each address's cause, once. #[test] fn dial_errors_are_short() { @@ -2300,12 +2390,19 @@ mod tests { // Of several, the oldest direct one stays; a relayed one only without a direct. connect(&mut book, both, false); assert_eq!( - book.duplicates(&both), + book.duplicates(&both, false), [ ConnectionId::new_unchecked(4), ConnectionId::new_unchecked(2) ] ); - assert!(book.duplicates(&relayed_only).is_empty()); + assert!(book.duplicates(&relayed_only, false).is_empty()); + // Not while one of them carries a slot, ours or the peer's, or we reserve one. + assert!(book.duplicates(&both, true).is_empty()); + book.slots.insert(both); + assert!(book.duplicates(&both, false).is_empty()); + book.slots.clear(); + book.slots_here.insert(both); + assert!(book.duplicates(&both, false).is_empty()); } } |