diff --git a/crates/cli/README.md b/crates/cli/README.md index 0e32001e..72526f22 100644 --- a/crates/cli/README.md +++ b/crates/cli/README.md @@ -6,7 +6,7 @@ Pluto enables the operation of Ethereum validators in a fault tolerant manner by ## Commands (current) -Most flags below also read a `CHARON_*` environment variable (for example `--beacon-node-endpoints` reads `CHARON_BEACON_NODE_ENDPOINTS`), mirroring charon's environment surface. Exceptions: none of the `alpha test` flags have env bindings, nor do `enr --verbose`, `version --verbose` and `relay --log-color`. A variable that is set but empty is treated as unset. Run `pluto --help` for the authoritative list — flags with a binding show an `[env: ...]` line. +Most flags below also read a `CHARON_*` environment variable (for example `--beacon-node-endpoints` reads `CHARON_BEACON_NODE_ENDPOINTS`), mirroring charon's environment surface. A variable that is set but empty is treated as unset. Run `pluto --help` for the authoritative list — flags with a binding show an `[env: ...]` line. Some flags are accepted for charon compatibility but are not yet wired up. They are marked below as **[IGNORED]** (parsed, then dropped — `pluto run` logs a warning for most of them) or **[UNSUPPORTED]** (setting the flag makes the command exit at startup with an error). @@ -59,7 +59,7 @@ Starts the long-running Pluto middleware process to perform distributed validato - `--testnet-genesis-timestamp `: Genesis timestamp of the custom test network. - `--testnet-capella-hard-fork `: Capella hard fork version of the custom test network. - The custom network is only registered when the testnet flags are fully specified; a partial set is silently ignored and pluto falls back to the built-in network registry. -- Plus the [common P2P flags](#common-p2p-flags) and [common logging flags](#common-logging-flags) (including the Loki flags). +- Plus the [common P2P flags](#common-p2p-flags) and [common logging flags](#common-logging-flags). ### `pluto relay` @@ -69,13 +69,12 @@ Starts a libp2p circuit relay that charon clients can use to discover and connec - `--data-dir `: The directory where pluto will store all its internal data. (default: `.charon`) - `--http-address `: Listening address (ip and port) for the relay http server serving runtime ENR. (default: `127.0.0.1:3640`) - `--auto-p2pkey`: Automatically generate and persist a p2p key if one does not exist. Always on: it defaults to true and cannot be switched off on the command line (`--auto-p2pkey=false` is rejected); set `CHARON_AUTO_P2PKEY=false` to require an existing key. - - `--p2p-relay-loglevel `: Log level for the upstream `libp2p_relay` crate, letting its logs be quieted (`--p2p-relay-loglevel=error`) without lowering pluto's own verbosity. Takes the same values as `--log-level`; when unset the relay crate follows `--log-level`. - `--p2p-max-reservations `: Updates max circuit reservations per peer (each valid for 1 hour). (default: `512`) - `--p2p-max-connections `: Currently applied as the relay's total reservation limit; it does not cap inbound connections. (default: `16384`) - `--p2p-advertise-private-addresses`: Enable advertising of libp2p auto-detected private addresses. - `--monitoring-address `: Listening address (ip and port) for the monitoring API (prometheus). - `--debug-address `: **[IGNORED]** Parsed but no debug listener is started (no warning is emitted). -- Plus the [common P2P flags](#common-p2p-flags) and [common logging flags](#common-logging-flags) (including the Loki flags). Note that `--p2p-relays` is accepted but unused by the relay itself. +- Plus the [common P2P flags](#common-p2p-flags) and [common logging flags](#common-logging-flags). Note that `--p2p-relays` is accepted but unused by the relay itself. ### `pluto dkg` @@ -94,7 +93,7 @@ Participate in a distributed key generation ceremony for a specific cluster defi - `--publish-address `: The URL to publish the cluster to. (default: `https://api.obol.tech/v1`) - `--publish-timeout `: Timeout for publishing a cluster; increase for clusters with more than 200 validators. (default: `30s`) - `--zipped`: Create a tar archive compressed with gzip of the target directory after creation. -- Plus the [common P2P flags](#common-p2p-flags) and the [common logging flags](#common-logging-flags) (Loki flags are **not** available on `dkg`). +- Plus the [common P2P flags](#common-p2p-flags) and the [common logging flags](#common-logging-flags). ### `pluto enr` @@ -266,17 +265,19 @@ Shared by `run`, `relay`, `dkg` and `alpha test peers`. ### Common logging flags -Shared by `run`, `relay` and `dkg`. +Global: accepted by every command, and parsed identically before or after the subcommand (`pluto --log-level=debug run` and `pluto run --log-level=debug` are equivalent). -- `--log-format `: **[IGNORED]** Accepted but not yet applied — output is always console-formatted. (default: `console`) -- `--log-level `: Log level; `off`, `trace`, `debug`, `info`, `warn` or `error`. (default: `info`) -- `--log-color `: Log color; `auto`, `force` or `disable`. (default: `auto`) -- `--log-output-path `: **[IGNORED]** Accepted but not yet applied — no log file is written. +All log output goes to stderr, leaving each command's stdout free for its own data. -`run` and `relay` additionally support Loki output: +`RUST_LOG` is not consulted; `--log-level` (or its default) always decides the filter. +- `--log-format `: **[IGNORED]** Accepted but not yet applied — output is always console-formatted. (default: `console`) +- `--log-level `: Log level; `off`, `trace`, `debug`, `info`, `warn` or `error`. Charon accepts only the last four; the two extra levels are what `tracing`'s `EnvFilter` understands. (default: `info`) +- `--log-color `: Log color; `auto`, `force` or `disable`. `auto` means "unless `NO_COLOR` is set", not TTY detection. (default: `auto`) +- `--log-output-path `: **[IGNORED]** Accepted but not yet applied — no log file is written. - `--loki-addresses `: Enables sending of logfmt structured logs to a Loki log aggregation server, in addition to normal stderr logs. Only the first address is used; extra entries are ignored with a warning (charon fans out to every address). - `--loki-service `: Service label sent with logs to Loki. (default: `pluto`) +- `--p2p-relay-loglevel `: Log level for the upstream `libp2p_relay` crate, letting its logs be quieted (`--p2p-relay-loglevel=error`) without lowering pluto's own verbosity. Takes the same values as `--log-level`; when unset the relay crate follows `--log-level`. Charon scopes this to `relay`. ## Example diff --git a/crates/cli/src/cli.rs b/crates/cli/src/cli.rs index 121a201b..2e72ba7a 100644 --- a/crates/cli/src/cli.rs +++ b/crates/cli/src/cli.rs @@ -3,6 +3,7 @@ use clap::{Parser, Subcommand}; use crate::commands::{ + common::TracingArgs, create_cluster::CreateClusterArgs, create_dkg::CreateDkgArgs, create_enr::CreateEnrArgs, @@ -29,6 +30,10 @@ pub struct Cli { /// The subcommand to execute. #[command(subcommand)] pub command: Commands, + + /// Logging flags, shared by every subcommand. + #[command(flatten)] + pub tracing: TracingArgs, } /// Available commands. diff --git a/crates/cli/src/commands/common.rs b/crates/cli/src/commands/common.rs index fd2b7cb2..c4e07bd1 100644 --- a/crates/cli/src/commands/common.rs +++ b/crates/cli/src/commands/common.rs @@ -1,6 +1,6 @@ //! Shared helpers for CLI commands. -use std::fmt; +use std::{collections::HashMap, fmt, path::PathBuf}; use pluto_p2p::config::RelayAddr; use tracing::warn; @@ -53,32 +53,156 @@ impl fmt::Display for LogLevel { } } -/// Builds a tracing configuration for CLI commands, optionally enabling Loki. +/// Adds a `libp2p_relay` directive to the `base` env filter, which `EnvFilter` +/// prefix-matches against every `libp2p_relay::*` target. +fn relay_filter(base: LogLevel, relay_level: Option) -> String { + match relay_level { + Some(level) => format!("{base},libp2p_relay={level}"), + None => base.to_string(), + } +} + +/// Log output encoding +#[derive(clap::ValueEnum, Clone, Copy, Debug, Default, PartialEq, Eq)] +pub enum LogFormat { + /// Human-readable, optionally colored. + #[default] + Console, + /// Flat `key=value` pairs. + Logfmt, + /// One JSON object per event. + Json, +} + +/// Logging and Loki flags, accepted by every subcommand. /// -/// `loki` is `Some` when the caller wants events forwarded to a Loki endpoint -/// (e.g. via `--loki-addresses`), and `None` for commands that only need -/// console output. +/// These are `global`, so they parse identically before or after the +/// subcommand and are readable from the root [`crate::cli::Cli`] before any +/// command-specific config conversion runs. That ordering is what lets +/// `main` install the subscriber before validation starts. // TODO: wire `log-output-path` (file output) and `log-format` (logfmt/json) // into the tracing layers. `pluto_tracing` supports console + Loki only, so -// `run`/`dkg`/`relay` accept these flags but do not yet apply them. -pub fn build_console_tracing_config( - level: impl Into, - color: &ConsoleColor, - loki: Option, -) -> pluto_tracing::TracingConfig { - let mut builder = pluto_tracing::TracingConfig::builder().with_default_console(); - - builder = match color { - ConsoleColor::Auto => builder.console_with_ansi(std::env::var("NO_COLOR").is_err()), - ConsoleColor::Force => builder.console_with_ansi(true), - ConsoleColor::Disable => builder.console_with_ansi(false), - }; - - if let Some(loki) = loki { - builder = builder.loki(loki); +// these flags are accepted but not yet applied. +#[derive(clap::Args, Clone, Debug)] +#[command(next_help_heading = "Logging")] +pub struct TracingArgs { + #[arg( + long = "log-format", + env = "CHARON_LOG_FORMAT", + default_value = "console", + global = true, + ignore_case = true, + display_order = 1000, + help = "Log format; console, logfmt or json" + )] + pub log_format: LogFormat, + + #[arg( + long = "log-level", + env = "CHARON_LOG_LEVEL", + default_value = "info", + global = true, + ignore_case = true, + display_order = 1001, + help = "Log level" + )] + pub log_level: LogLevel, + + #[arg( + long = "log-color", + env = "CHARON_LOG_COLOR", + default_value = "auto", + global = true, + ignore_case = true, + display_order = 1002, + help = "Log color; auto, force, disable." + )] + pub log_color: ConsoleColor, + + #[arg( + long = "log-output-path", + env = "CHARON_LOG_OUTPUT_PATH", + global = true, + display_order = 1003, + help = "Path in which to write on-disk logs." + )] + pub log_output_path: Option, + + #[arg( + long = "loki-addresses", + env = "CHARON_LOKI_ADDRESSES", + value_delimiter = ',', + global = true, + display_order = 1004, + help = "Enables sending of logfmt structured logs to these Loki log aggregation server addresses. This is in addition to normal stderr logs." + )] + pub loki_addresses: Vec, + + #[arg( + long = "loki-service", + env = "CHARON_LOKI_SERVICE", + default_value = "pluto", + global = true, + display_order = 1005, + help = "Service label sent with logs to Loki." + )] + pub loki_service: String, + + #[arg( + long = "p2p-relay-loglevel", + env = "CHARON_P2P_RELAY_LOGLEVEL", + global = true, + ignore_case = true, + display_order = 1006, + help = "Libp2p circuit relay log level. Defaults to --log-level." + )] + pub p2p_relay_log_level: Option, +} + +impl TracingArgs { + /// Builds the subscriber configuration. + /// + /// Emits nothing: this runs before the subscriber exists, so any diagnostic + /// it produced would be dropped. Deferred warnings live in + /// [`TracingArgs::warn_unused`]. + pub fn tracing_config(&self) -> pluto_tracing::TracingConfig { + let ansi = match self.log_color { + ConsoleColor::Auto => std::env::var_os("NO_COLOR").is_none(), + ConsoleColor::Force => true, + ConsoleColor::Disable => false, + }; + + let mut builder = pluto_tracing::TracingConfig::builder() + .with_default_console() + .console_with_ansi(ansi) + .override_env_filter(relay_filter(self.log_level, self.p2p_relay_log_level)); + + // Only the first address is used; see `warn_unused`. + if let Some(loki_url) = self.loki_addresses.first() { + builder = builder.loki(pluto_tracing::LokiConfig { + loki_url: loki_url.clone(), + labels: HashMap::from([("service".to_string(), self.loki_service.clone())]), + extra_fields: HashMap::new(), + }); + } + + builder.build() } - builder.override_env_filter(level.into()).build() + /// Reports flag values that were accepted but not applied. + /// + /// Call once the subscriber is installed. + pub fn warn_unused(&self) { + // Charon fans logs out to every entry in `loki-addresses`, but + // `pluto_tracing::TracingConfig` supports a single Loki layer today. + let ignored = self.loki_addresses.len().saturating_sub(1); + if ignored > 0 { + warn!( + ignored, + "Additional --loki-addresses ignored; only the first is used" + ); + } + } } /// Parses the configured relay addresses, warning about insecure ones. @@ -120,6 +244,122 @@ pub fn parse_relay_addrs(relays: &[String]) -> std::result::Result::try_parse_from([ + "pluto", + "enr", + &format!("--log-level={level}"), + ]) + .unwrap_or_else(|err| panic!("--log-level={level} should parse: {err}")); + + assert_eq!( + cli.tracing.tracing_config().override_env_filter.as_deref(), + Some("debug") + ); + } + + for color in ["disable", "DISABLE", "Disable"] { + let cli = ::try_parse_from([ + "pluto", + "enr", + &format!("--log-color={color}"), + ]) + .unwrap_or_else(|err| panic!("--log-color={color} should parse: {err}")); + + assert!( + !cli.tracing + .tracing_config() + .console + .expect("console") + .with_ansi + ); + } + + for format in ["logfmt", "LOGFMT", "Logfmt"] { + let cli = ::try_parse_from([ + "pluto", + "enr", + &format!("--log-format={format}"), + ]) + .unwrap_or_else(|err| panic!("--log-format={format} should parse: {err}")); + + assert_eq!(cli.tracing.log_format, LogFormat::Logfmt); + } + } + + #[test] + fn log_flags_reject_unknown_values() { + for flag in [ + "--log-level=nonsense", + "--log-format=nonsense", + "--p2p-relay-loglevel=fatal", + ] { + let err = match ::try_parse_from(["pluto", "enr", flag]) { + Ok(_) => panic!("{flag} should be rejected"), + Err(err) => err, + }; + + assert_eq!(err.kind(), clap::error::ErrorKind::InvalidValue); + } + } + + /// Runs `f` with a subscriber that only lets `filter` through. + fn with_filter(filter: &str, f: impl FnOnce()) { + let filter = EnvFilter::from_str(filter).expect("relay filter should be a valid EnvFilter"); + tracing::subscriber::with_default(tracing_subscriber::registry().with(filter), f); + } + + #[test] + fn relay_filter_scopes_upstream_relay_logs() { + // An unset relay level leaves the base filter alone. + with_filter(&relay_filter(LogLevel::Info, None), || { + assert!(enabled!(target: "libp2p_relay::behaviour::handler", Level::WARN)); + }); + + // A relay level silences the upstream relay crate but not our own logs. + with_filter(&relay_filter(LogLevel::Info, Some(LogLevel::Error)), || { + assert!(!enabled!(target: "libp2p_relay::behaviour::handler", Level::WARN)); + assert!(enabled!(target: "pluto_relay_server::p2p", Level::INFO)); + }); + } + + #[test] + fn every_log_level_composes_into_a_valid_filter() { + for base in LogLevel::value_variants() { + for relay in LogLevel::value_variants() { + let filter = relay_filter(*base, Some(*relay)); + EnvFilter::from_str(&filter).unwrap_or_else(|e| panic!("{filter:?}: {e}")); + } + } + } + + #[test] + fn p2p_relay_loglevel_reaches_the_env_filter() { + // The flag is global, so it composes with `--log-level` from the root + // rather than from the `relay` subcommand that used to own it. + let cli = ::try_parse_from([ + "pluto", + "relay", + "--log-level=info", + "--p2p-relay-loglevel=error", + ]) + .expect("relay args should parse"); + + assert_eq!( + cli.tracing.tracing_config().override_env_filter.as_deref(), + Some("info,libp2p_relay=error") + ); + } // Per-address parsing is covered by `RelayAddr`'s own tests; what is left // to check here is the empty-value contract and the error wrapping. diff --git a/crates/cli/src/commands/dkg.rs b/crates/cli/src/commands/dkg.rs index cc806ca1..463e1344 100644 --- a/crates/cli/src/commands/dkg.rs +++ b/crates/cli/src/commands/dkg.rs @@ -3,7 +3,7 @@ use std::{future::Future, path::PathBuf}; use crate::{ - commands::common::{ConsoleColor, LICENSE, build_console_tracing_config, parse_relay_addrs}, + commands::common::{LICENSE, parse_relay_addrs}, duration::Duration, error::{CliError, Result}, }; @@ -56,9 +56,6 @@ pub struct DkgArgs { #[command(flatten)] pub p2p: DkgP2PArgs, - #[command(flatten)] - pub log: DkgLogArgs, - #[arg( long = "publish-address", env = "CHARON_PUBLISH_ADDRESS", @@ -123,8 +120,6 @@ impl TryFrom for pluto_dkg::dkg::Config { fn try_from(args: DkgArgs) -> Result { validate_p2p_args(&args.p2p)?; - let tracing_config = - build_console_tracing_config(args.log.level.clone(), &args.log.color, None); let p2p_config = { let relays = parse_relay_addrs(&args.p2p.relays)?; @@ -143,7 +138,6 @@ impl TryFrom for pluto_dkg::dkg::Config { .no_verify(args.no_verify) .data_dir(args.data_dir) .p2p(p2p_config) - .log(tracing_config) .keymanager( pluto_dkg::dkg::KeymanagerConfig::builder() .address(args.keymanager_address) @@ -217,41 +211,6 @@ pub struct DkgP2PArgs { pub disable_reuseport: bool, } -/// Logging arguments for the `dkg` command. -#[derive(clap::Args, Clone, Debug)] -pub struct DkgLogArgs { - #[arg( - long = "log-format", - env = "CHARON_LOG_FORMAT", - default_value = "console", - help = "Log format; console, logfmt or json" - )] - pub format: String, - - #[arg( - long = "log-level", - env = "CHARON_LOG_LEVEL", - default_value = "info", - help = "Log level; debug, info, warn or error" - )] - pub level: String, - - #[arg( - long = "log-color", - env = "CHARON_LOG_COLOR", - default_value = "auto", - help = "Log color; auto, force, disable." - )] - pub color: ConsoleColor, - - #[arg( - long = "log-output-path", - env = "CHARON_LOG_OUTPUT_PATH", - help = "Path in which to write on-disk logs." - )] - pub log_output_path: Option, -} - /// Runs the `dkg` command from an already-built configuration. pub async fn run(config: pluto_dkg::dkg::Config, ct: CancellationToken) -> Result<()> { run_with_runner(config, ct, pluto_dkg::dkg::run).await @@ -283,10 +242,13 @@ fn validate_p2p_args(args: &DkgP2PArgs) -> Result<()> { #[cfg(test)] mod tests { use super::*; - use crate::cli::{Cli, Commands}; + use crate::{ + cli::{Cli, Commands}, + commands::common::{LogFormat, LogLevel}, + }; use clap::Parser; use pluto_p2p::config::RelayAddr; - use std::{sync::Arc, time::Duration as StdDuration}; + use std::{path::Path, sync::Arc, time::Duration as StdDuration}; #[test] fn dkg_is_registered_as_top_level_subcommand() { @@ -302,6 +264,9 @@ mod tests { fn dkg_defaults_match_go() { let cli = Cli::try_parse_from(["pluto", "dkg"]).expect("dkg command should parse"); + assert_eq!(cli.tracing.log_level, LogLevel::Info); + assert_eq!(cli.tracing.log_format, LogFormat::Console); + let Commands::Dkg(args) = cli.command else { panic!("expected dkg command"); }; @@ -326,8 +291,6 @@ mod tests { args.p2p.relays, pluto_p2p::config::DEFAULT_RELAYS.map(String::from).to_vec(), ); - assert_eq!(args.log.level, "info"); - assert_eq!(args.log.format, "console"); } #[test] @@ -357,7 +320,6 @@ mod tests { ("keymanager-address", "CHARON_KEYMANAGER_ADDRESS"), ("keymanager-auth-token", "CHARON_KEYMANAGER_AUTH_TOKEN"), ("p2p-relays", "CHARON_P2P_RELAYS"), - ("log-level", "CHARON_LOG_LEVEL"), ("publish", "CHARON_PUBLISH"), ("publish-timeout", "CHARON_PUBLISH_TIMEOUT"), ("timeout", "CHARON_TIMEOUT"), @@ -406,6 +368,23 @@ mod tests { ]) .expect("dkg command should parse"); + // Log flags are global, so they land on the root rather than on `dkg`. + assert_eq!(cli.tracing.log_level, LogLevel::Debug); + assert_eq!(cli.tracing.log_format, LogFormat::Json); + assert_eq!( + cli.tracing.log_output_path.as_deref(), + Some(Path::new("/tmp/pluto.log")) + ); + let tracing_config = cli.tracing.tracing_config(); + assert_eq!(tracing_config.override_env_filter.as_deref(), Some("debug")); + assert!( + tracing_config + .console + .as_ref() + .expect("console config") + .with_ansi + ); + let Commands::Dkg(args) = cli.command else { panic!("expected dkg command"); }; @@ -430,9 +409,6 @@ mod tests { assert_eq!(config.p2p.tcp_addrs, vec!["0.0.0.0:9000".to_string()]); assert_eq!(config.p2p.udp_addrs, vec!["0.0.0.0:9000".to_string()]); assert!(config.p2p.disable_reuse_port); - assert_eq!(config.log.override_env_filter.as_deref(), Some("debug")); - let console = config.log.console.as_ref().expect("console config"); - assert!(console.with_ansi); assert!(config.publish.enabled); assert_eq!(config.publish.address, "https://api.example/v1"); assert_eq!(config.publish.timeout, StdDuration::from_secs(40)); @@ -453,6 +429,17 @@ mod tests { "--log-output-path=/tmp/pluto.log", ]) .expect("dkg command should parse"); + + let tracing_config = cli.tracing.tracing_config(); + assert_eq!(tracing_config.override_env_filter.as_deref(), Some("debug")); + assert!( + !tracing_config + .console + .as_ref() + .expect("console config") + .with_ansi + ); + let Commands::Dkg(args) = cli.command else { panic!("expected dkg command"); }; @@ -466,9 +453,6 @@ mod tests { move |config, token| async move { assert!(!token.is_cancelled()); assert_eq!(config.def_file, ".charon/cluster-definition.json"); - assert_eq!(config.log.override_env_filter.as_deref(), Some("debug")); - let console = config.log.console.as_ref().expect("console config"); - assert!(!console.with_ansi); events.lock().expect("lock").push("runner"); Ok(()) } diff --git a/crates/cli/src/commands/relay.rs b/crates/cli/src/commands/relay.rs index 08692ce5..f6e371a1 100644 --- a/crates/cli/src/commands/relay.rs +++ b/crates/cli/src/commands/relay.rs @@ -1,27 +1,12 @@ use crate::{ - commands::common::{ - ConsoleColor, LICENSE, LogLevel, build_console_tracing_config, parse_relay_addrs, - }, + commands::common::{LICENSE, parse_relay_addrs}, error::CliError, }; use pluto_p2p::k1; -use std::{collections::HashMap, path::PathBuf, time::Duration}; +use std::path::PathBuf; use tokio_util::sync::CancellationToken; use tracing::{error, info}; -/// Grace period given to the Loki background task to flush buffered logs -/// once `BackgroundTaskController::shutdown` has been signalled. -const LOKI_FLUSH_TIMEOUT: Duration = Duration::from_secs(3); - -/// Adds a `libp2p_relay` directive to the `base` env filter, which `EnvFilter` -/// prefix-matches against every `libp2p_relay::*` target. -fn relay_filter(base: LogLevel, relay_level: Option) -> String { - match relay_level { - Some(level) => format!("{base},libp2p_relay={level}"), - None => base.to_string(), - } -} - /// Arguments for the relay command. #[derive(clap::Args, Clone)] pub struct RelayArgs { @@ -36,12 +21,6 @@ pub struct RelayArgs { #[clap(flatten)] pub p2p: RelayP2PArgs, - - #[clap(flatten)] - pub log: RelayLogFlags, - - #[clap(flatten)] - pub loki: RelayLokiArgs, } impl TryInto for RelayArgs { @@ -76,40 +55,6 @@ impl TryInto for RelayArgs { } }; - let loki_config = match self.loki.loki_addresses.as_slice() { - [] => None, - [loki_url, rest @ ..] => { - if !rest.is_empty() { - // Charon fans logs out to every entry in `loki-addresses`, - // but `pluto_tracing::TracingConfig` - // only supports a single Loki - // layer today. `tracing::warn!` would be a no-op here - // because no subscriber is installed - // yet (init happens later inside - // `commands::relay::run`), so write directly to stderr. - eprintln!( - "warning: {extra} additional --loki-addresses ignored; only the first is used", - extra = rest.len(), - ); - } - - let labels = - HashMap::from([("service".to_string(), self.loki.loki_service.clone())]); - - Some(pluto_tracing::LokiConfig { - loki_url: loki_url.clone(), - labels, - extra_fields: HashMap::new(), - }) - } - }; - - let log_config = build_console_tracing_config( - relay_filter(self.log.level, self.relay.p2p_relay_log_level), - &self.log.color, - loki_config, - ); - let builder = pluto_relay_server::config::Config::builder() .data_dir(self.data_dir.data_dir) .http_addr(self.relay.http_address) @@ -122,8 +67,7 @@ impl TryInto for RelayArgs { .filter_private_addrs(!self.relay.advertise_priv) .maybe_monitoring_addr(self.debug_monitoring.monitor_addr) .maybe_debug_addr(self.debug_monitoring.debug_addr) - .p2p_config(p2p_config) - .log_config(log_config); + .p2p_config(p2p_config); Ok(builder.build()) } @@ -158,14 +102,6 @@ pub struct RelayRelayArgs { )] pub auto_p2p_key: bool, - #[arg( - long = "p2p-relay-loglevel", - env = "CHARON_P2P_RELAY_LOGLEVEL", - ignore_case = true, - help = "Libp2p circuit relay log level. Defaults to --log-level." - )] - pub p2p_relay_log_level: Option, - // TODO: Check if https://github.com/libp2p/go-libp2p/issues/1713 is relevant for the Rust libp2p implementation // If so, decrease defaults after this has been addressed #[arg( @@ -260,112 +196,16 @@ pub struct RelayP2PArgs { pub disable_reuseport: bool, } -#[derive(clap::Args, Clone)] -pub struct RelayLogFlags { - #[arg( - long = "log-format", - env = "CHARON_LOG_FORMAT", - default_value = "console", - help = "Log format; console, logfmt or json" - )] - pub format: String, - - #[arg( - long = "log-level", - env = "CHARON_LOG_LEVEL", - default_value = "info", - ignore_case = true, - help = "Log level" - )] - pub level: LogLevel, - - #[arg(long = "log-color", default_value = "auto", help = "Log color")] - pub color: ConsoleColor, - - #[arg( - long = "log-output-path", - env = "CHARON_LOG_OUTPUT_PATH", - help = "Path in which to write on-disk logs." - )] - pub log_output_path: Option, -} - -#[derive(clap::Args, Clone)] -pub struct RelayLokiArgs { - #[arg( - long = "loki-addresses", - env = "CHARON_LOKI_ADDRESSES", - value_delimiter = ',', - help = "Enables sending of logfmt structured logs to these Loki log aggregation server addresses. This is in addition to normal stderr logs." - )] - pub loki_addresses: Vec, - - #[arg( - long = "loki-service", - env = "CHARON_LOKI_SERVICE", - default_value = "pluto", - help = "Service label sent with logs to Loki." - )] - pub loki_service: String, -} - pub async fn run( config: pluto_relay_server::config::Config, ct: CancellationToken, -) -> Result<(), CliError> { - let loki_shutdown = match pluto_tracing::init(&config.log_config) { - Ok(Some(loki)) => { - let controller = loki.controller; - let handle = tokio::spawn(loki.task); - Some((controller, handle)) - } - Ok(None) => None, - // In tests, the global tracing subscriber is shared across runs in the - // same process, so reinitializing fails. In production this would mean - // the relay silently uses an unrelated subscriber and Loki forwarding - // is dropped — fail loudly instead. - #[cfg(test)] - Err(pluto_tracing::init::Error::Init(_)) => None, - Err(err) => return Err(err.into()), - }; - - // Run the relay in an inner scope so every early `?` / `return Err(..)` is - // captured into `result` and the Loki cleanup below always runs. - let result = serve_relay(&config, ct).await; - - if let Err(err) = &result { - // Surface the shutdown reason through the subscriber so it reaches - // Loki before we close the worker; `main` only `eprintln!`s the - // returned error and that path bypasses the tracing subscriber. - error!(error = %err, "relay exited with error"); - } - - // Drain the Loki worker under a single budget so a hung Loki endpoint - // (e.g. `controller.shutdown` blocked on a full mpsc) cannot wedge - // process exit. After the budget elapses we hard-abort the worker. - if let Some((controller, handle)) = loki_shutdown { - let abort_handle = handle.abort_handle(); - let _ = tokio::time::timeout(LOKI_FLUSH_TIMEOUT, async { - controller.shutdown().await; - let _ = handle.await; - }) - .await; - abort_handle.abort(); - } - - result -} - -async fn serve_relay( - config: &pluto_relay_server::config::Config, - ct: CancellationToken, ) -> Result<(), CliError> { info!("{LICENSE}"); info!(config = ?config); - let key = load_or_create_key(config)?; + let key = load_or_create_key(&config)?; - pluto_relay_server::p2p::run_relay_p2p_node(config, key, ct) + pluto_relay_server::p2p::run_relay_p2p_node(&config, key, ct) .await .map_err(Into::into) } @@ -403,7 +243,6 @@ fn load_or_create_key( #[cfg(test)] mod tests { - use clap::{Parser as _, ValueEnum as _}; use std::{ net::{Ipv4Addr, SocketAddr}, path::Path, @@ -413,10 +252,6 @@ mod tests { }; use tokio::{net, task::JoinHandle}; use tokio_util::sync::CancellationToken; - use tracing::{Level, enabled}; - use tracing_subscriber::{EnvFilter, layer::SubscriberExt as _}; - - use crate::cli::Cli; /// Args mirroring the clap defaults (notably `debug_addr: Some("")`), /// plus a TCP address so the baseline conversion succeeds. @@ -428,7 +263,6 @@ mod tests { relay: super::RelayRelayArgs { http_address: "127.0.0.1:3640".into(), auto_p2p_key: true, - p2p_relay_log_level: None, max_res_per_peer: 512, max_conns: 16384, advertise_priv: false, @@ -445,16 +279,6 @@ mod tests { udp_addrs: vec![], disable_reuseport: false, }, - log: super::RelayLogFlags { - format: "console".into(), - level: super::LogLevel::Error, - color: super::ConsoleColor::Disable, - log_output_path: None, - }, - loki: super::RelayLokiArgs { - loki_addresses: vec![], - loki_service: "pluto".into(), - }, } } @@ -524,10 +348,8 @@ mod tests { let dir = tempfile::tempdir().unwrap(); let args = relay_args(dir.path()); - // Covers the CLI entry point that the fixture bypasses: tracing init - // and the Loki drain. A pre-cancelled token is deterministic - // because the shutdown arm of the serve loop is the `biased` - // first branch. + // A pre-cancelled token is deterministic because the shutdown arm of + // the serve loop is the `biased` first branch. let ct = CancellationToken::new(); ct.cancel(); @@ -771,7 +593,7 @@ mod tests { static CLIENT: LazyLock = LazyLock::new(reqwest::Client::new); /// Relay arguments every test starts from: all listeners on [`ANY_ADDR`], - /// quiet logs, no relays to dial. + /// no relays to dial. /// /// `advertise_priv` is load-bearing: without it, `filter_private_addrs` /// drops the loopback listen addresses, and `/enr` answers 500 forever. @@ -783,7 +605,6 @@ mod tests { relay: super::RelayRelayArgs { http_address: ANY_ADDR.into(), auto_p2p_key: true, - p2p_relay_log_level: None, max_res_per_peer: 0, max_conns: 0, advertise_priv: true, @@ -800,16 +621,6 @@ mod tests { udp_addrs: vec![ANY_ADDR.into()], disable_reuseport: false, }, - log: super::RelayLogFlags { - format: "console".into(), - level: super::LogLevel::Error, - color: super::ConsoleColor::Disable, - log_output_path: None, - }, - loki: super::RelayLokiArgs { - loki_addresses: vec![], - loki_service: "".into(), - }, } } @@ -947,46 +758,4 @@ mod tests { let addr = listener.local_addr().unwrap().to_string(); (listener, addr) } - - /// Runs `f` with a subscriber that only lets `filter` through. - fn with_filter(filter: &str, f: impl FnOnce()) { - let filter = EnvFilter::from_str(filter).expect("relay filter should be a valid EnvFilter"); - tracing::subscriber::with_default(tracing_subscriber::registry().with(filter), f); - } - - #[test] - fn relay_filter_scopes_upstream_relay_logs() { - // An unset relay level leaves the base filter alone. - with_filter(&super::relay_filter(super::LogLevel::Info, None), || { - assert!(enabled!(target: "libp2p_relay::behaviour::handler", Level::WARN)); - }); - - // A relay level silences the upstream relay crate but not our own logs. - with_filter( - &super::relay_filter(super::LogLevel::Info, Some(super::LogLevel::Error)), - || { - assert!(!enabled!(target: "libp2p_relay::behaviour::handler", Level::WARN)); - assert!(enabled!(target: "pluto_relay_server::p2p", Level::INFO)); - }, - ); - } - - #[test] - fn every_log_level_composes_into_a_valid_filter() { - for base in super::LogLevel::value_variants() { - for relay in super::LogLevel::value_variants() { - let filter = super::relay_filter(*base, Some(*relay)); - EnvFilter::from_str(&filter).unwrap_or_else(|e| panic!("{filter:?}: {e}")); - } - } - } - #[test] - fn unknown_log_level_is_rejected() { - let err = match Cli::try_parse_from(["pluto", "relay", "--p2p-relay-loglevel=fatal"]) { - Ok(_) => panic!("`fatal` is not an EnvFilter level"), - Err(err) => err, - }; - - assert_eq!(err.kind(), clap::error::ErrorKind::InvalidValue); - } } diff --git a/crates/cli/src/commands/run.rs b/crates/cli/src/commands/run.rs index de40cb63..f9fa8f37 100644 --- a/crates/cli/src/commands/run.rs +++ b/crates/cli/src/commands/run.rs @@ -26,7 +26,6 @@ //! ignored with a warning. use std::{ - collections::HashMap, net::{SocketAddr, ToSocketAddrs}, path::{Path, PathBuf}, time::Duration as StdDuration, @@ -35,10 +34,10 @@ use std::{ use pluto_eth2util::helpers::validate_http_headers; use pluto_featureset::{Feature, FeaturesetError, Status}; use tokio_util::sync::CancellationToken; -use tracing::{error, info, warn}; +use tracing::{info, warn}; use crate::{ - commands::common::{ConsoleColor, LICENSE, build_console_tracing_config, parse_relay_addrs}, + commands::common::{LICENSE, parse_relay_addrs}, duration::Duration, error::{CliError, Result}, }; @@ -50,8 +49,6 @@ const MAX_GRAFFITI_BYTES: usize = 28; const MAX_GRAFFITI_BYTES_NO_APPEND: usize = 32; /// Maximum peer nickname length in bytes. const MAX_NICKNAME_BYTES: usize = 32; -/// Grace period for the Loki background task to flush buffered logs on exit. -const LOKI_FLUSH_TIMEOUT: StdDuration = StdDuration::from_secs(3); /// Default `--monitoring-address`. const DEFAULT_MONITORING_ADDR: &str = "127.0.0.1:3620"; /// Default `--simnet-validator-keys-dir` (matches Charon). @@ -83,12 +80,6 @@ pub struct RunArgs { #[command(flatten)] pub p2p: RunP2PArgs, - #[command(flatten)] - pub log: RunLogArgs, - - #[command(flatten)] - pub loki: RunLokiArgs, - #[command(flatten)] pub feature: RunFeatureArgs, } @@ -478,61 +469,6 @@ pub struct RunP2PArgs { pub disable_reuseport: bool, } -/// Logging flags. -#[derive(clap::Args, Clone, Debug)] -pub struct RunLogArgs { - #[arg( - long = "log-format", - env = "CHARON_LOG_FORMAT", - default_value = "console", - help = "Log format; console, logfmt or json" - )] - pub format: String, - - #[arg( - long = "log-level", - env = "CHARON_LOG_LEVEL", - default_value = "info", - help = "Log level; debug, info, warn or error" - )] - pub level: String, - - #[arg( - long = "log-color", - env = "CHARON_LOG_COLOR", - default_value = "auto", - help = "Log color; auto, force, disable." - )] - pub color: ConsoleColor, - - #[arg( - long = "log-output-path", - env = "CHARON_LOG_OUTPUT_PATH", - help = "Path in which to write on-disk logs." - )] - pub log_output_path: Option, -} - -/// Loki flags. -#[derive(clap::Args, Clone, Debug)] -pub struct RunLokiArgs { - #[arg( - long = "loki-addresses", - env = "CHARON_LOKI_ADDRESSES", - value_delimiter = ',', - help = "Enables sending of logfmt structured logs to these Loki log aggregation server addresses. This is in addition to normal stderr logs." - )] - pub loki_addresses: Vec, - - #[arg( - long = "loki-service", - env = "CHARON_LOKI_SERVICE", - default_value = "pluto", - help = "Service label sent with logs to Loki." - )] - pub loki_service: String, -} - /// Feature set flags. #[derive(clap::Args, Clone, Debug)] pub struct RunFeatureArgs { @@ -614,15 +550,13 @@ pub struct FeatureSetConfig { } /// Configuration for the `run` command — the settings produced from the parsed -/// flags, consumed by [`run`] (tracing/Loki) and bridged into +/// flags, consumed by [`run`] and bridged into /// [`pluto_app::node::AppConfig`] by [`build_app_config`]; `p2p_fuzz` is the /// single test-only field, set only via the hidden `unsafe run` command. #[derive(Debug)] pub struct RunConfig { /// P2P configuration built from [`RunP2PArgs`]. pub p2p: pluto_p2p::config::P2PConfig, - /// Tracing configuration built from [`RunLogArgs`]/[`RunLokiArgs`]. - pub log: pluto_tracing::TracingConfig, /// Feature set configuration. pub feature_set: FeatureSetConfig, /// Path to the cluster lock file. @@ -719,8 +653,6 @@ impl TryFrom for RunConfig { debug_monitoring, no_verify, p2p, - log, - loki, feature, } = args; @@ -793,12 +725,8 @@ impl TryFrom for RunConfig { disable_reuse_port: p2p.disable_reuseport, }; - let log_config = - build_console_tracing_config(log.level, &log.color, build_loki_config(&loki)); - Ok(Self { p2p: p2p_config, - log: log_config, feature_set: FeatureSetConfig { min_status: feature.feature_set, enabled: feature.feature_set_enable, @@ -897,78 +825,10 @@ fn validate_vc_tls(cert: &str, key: &str) -> Result<()> { Ok(()) } -/// Builds the optional Loki tracing configuration from the loki flags. -/// -/// Only a single Loki endpoint is supported today, so any extra -/// `--loki-addresses` entries are ignored with a warning. The warning goes to -/// stderr because no tracing subscriber is installed yet. -fn build_loki_config(loki: &RunLokiArgs) -> Option { - match loki.loki_addresses.as_slice() { - [] => None, - [loki_url, rest @ ..] => { - if !rest.is_empty() { - eprintln!( - "warning: {extra} additional --loki-addresses ignored; only the first is used", - extra = rest.len(), - ); - } - - Some(pluto_tracing::LokiConfig { - loki_url: loki_url.clone(), - labels: HashMap::from([("service".to_string(), loki.loki_service.clone())]), - extra_fields: HashMap::new(), - }) - } - } -} - /// Runs the `run` command from an already-built configuration. -/// -/// Initializes tracing and owns the Loki lifecycle for the command's lifetime: -/// when `--loki-addresses` is set, the background task is spawned here and -/// drained on exit so buffered logs are delivered. pub async fn run(config: RunConfig, ct: CancellationToken) -> Result<()> { - let loki_shutdown = match pluto_tracing::init(&config.log) { - Ok(Some(loki)) => Some((loki.controller, tokio::spawn(loki.task))), - Ok(None) => None, - // In tests the global subscriber is shared across runs in the same - // process, so reinitializing fails; treat that as "no Loki worker" - // rather than failing the command. - #[cfg(test)] - Err(pluto_tracing::init::Error::Init(_)) => None, - Err(err) => return Err(err.into()), - }; - info!("{LICENSE}"); - let result = run_workflow(config, ct).await; - - if let Err(err) = &result { - // Surface the failure through the subscriber so it reaches Loki before - // the worker is drained; `main` only `eprintln!`s the returned error - // and that path bypasses the tracing subscriber. - error!(error = %err, "run exited with error"); - } - - // Drain the Loki worker under a single budget so a hung endpoint cannot - // wedge process exit; hard-abort after the budget elapses. - if let Some((controller, handle)) = loki_shutdown { - let abort_handle = handle.abort_handle(); - let _ = tokio::time::timeout(LOKI_FLUSH_TIMEOUT, async { - controller.shutdown().await; - let _ = handle.await; - }) - .await; - abort_handle.abort(); - } - - result -} - -/// The long-running validator workflow: bridges the parsed [`RunConfig`] into -/// [`pluto_app::node::AppConfig`] and drives the node until `ct` fires (signal -/// handling lives in `main`). -async fn run_workflow(config: RunConfig, ct: CancellationToken) -> Result<()> { let app_config = build_app_config(config)?; pluto_app::node::App::new(app_config).run(ct).await?; Ok(()) @@ -1000,7 +860,6 @@ fn build_app_config(config: RunConfig) -> Result { // bridge behavior fails to compile instead of being silently dropped. let RunConfig { p2p, - log: _, feature_set: _, lock_file, manifest_file: _, @@ -1098,7 +957,6 @@ fn check_unsupported_flags(config: &RunConfig) -> Result<()> { } /// Warns about observability/availability-only flags the run workflow ignores. -/// Runs after tracing init (see [`run`]) so the warnings reach the subscriber. fn warn_ignored_flags(config: &RunConfig) { if !config.debug_addr.is_empty() { warn!( @@ -1176,12 +1034,16 @@ fn parse_socket_addr(flag: &str, addr: &str) -> Result { #[cfg(test)] mod tests { use super::*; - use crate::cli::{Cli, Commands, UnsafeCommands}; + use crate::{ + cli::{Cli, Commands, UnsafeCommands}, + commands::common::{LogFormat, LogLevel}, + }; use clap::{CommandFactory, Parser}; use std::{collections::BTreeSet, time::Duration as StdDuration}; - /// Every flag the safe `run` command must expose. - const EXPECTED_RUN_FLAGS: [&str; 54] = [ + /// Every flag the safe `run` command must expose itself. The log and Loki + /// flags are global and live on the root command. + const EXPECTED_RUN_FLAGS: [&str; 48] = [ // priv key "private-key-file", "private-key-file-lock", @@ -1232,14 +1094,6 @@ mod tests { "p2p-tcp-address", "p2p-udp-address", "p2p-disable-reuseport", - // log - "log-format", - "log-level", - "log-color", - "log-output-path", - // loki - "loki-addresses", - "loki-service", // feature "feature-set-enable", "feature-set-disable", @@ -1282,6 +1136,23 @@ mod tests { (*args).try_into() } + /// Parses safe `run` args and returns the tracing config the global log + /// flags produce. + fn parse_tracing(extra: &[&str]) -> pluto_tracing::TracingConfig { + let mut argv = vec![ + "pluto", + "run", + "--beacon-node-endpoints", + "http://beacon.node", + ]; + argv.extend_from_slice(extra); + + Cli::try_parse_from(argv) + .expect("run args should parse") + .tracing + .tracing_config() + } + /// Returns the `Display` string of the error from a failing `parse_run`. fn run_err(extra: &[&str]) -> String { parse_run(extra) @@ -1383,6 +1254,13 @@ mod tests { "http://beacon.node", ]) .expect("run command should parse"); + + // Log and Loki flags are global, so they land on the root. + assert_eq!(cli.tracing.log_level, LogLevel::Info); + assert_eq!(cli.tracing.log_format, LogFormat::Console); + assert_eq!(cli.tracing.loki_service, "pluto"); + assert!(cli.tracing.loki_addresses.is_empty()); + let Commands::Run(args) = cli.command else { panic!("expected run command"); }; @@ -1442,13 +1320,6 @@ mod tests { assert!(args.p2p.tcp_addrs.is_empty()); assert!(args.p2p.external_ip.is_none()); - // Log. - assert_eq!(args.log.level, "info"); - assert_eq!(args.log.format, "console"); - - // Loki. - assert_eq!(args.loki.loki_service, "pluto"); - // Feature. assert_eq!(args.feature.feature_set, "stable"); assert!(args.feature.feature_set_enable.is_empty()); @@ -1584,19 +1455,16 @@ mod tests { #[test] fn run_loki_config_built_from_addresses() { - // `--loki-addresses` must produce a Loki layer in the tracing config so - // `run` spawns/drains the Loki worker (regression for the lifecycle bug - // where the worker was initialized in `main` and then dropped). - let config = parse_run(&[ + // The Loki flags are global, but they must still reach the subscriber + // config when supplied on `run`. + let tracing_config = parse_tracing(&[ "--loki-addresses", "http://loki.test/push", "--loki-service", "svc", - ]) - .expect("config should build"); + ]); - let loki = config - .log + let loki = tracing_config .loki .as_ref() .expect("loki layer should be configured"); @@ -1604,13 +1472,7 @@ mod tests { assert_eq!(loki.labels.get("service").map(String::as_str), Some("svc")); // No `--loki-addresses` → no Loki layer (nothing to spawn). - assert!( - parse_run(&[]) - .expect("config should build") - .log - .loki - .is_none() - ); + assert!(parse_tracing(&[]).loki.is_none()); } #[test] @@ -1751,10 +1613,6 @@ mod tests { ); // p2p_fuzz is never set on the safe `run` path. assert!(!config.p2p_fuzz); - // `--log-color=force` forces ANSI on the console layer. - let console = config.log.console.as_ref().expect("console config"); - assert!(console.with_ansi); - assert_eq!(config.log.override_env_filter.as_deref(), Some("debug")); } /// Builds the app config from safe `run` flags. diff --git a/crates/cli/src/error.rs b/crates/cli/src/error.rs index c1ce261a..4bf2787b 100644 --- a/crates/cli/src/error.rs +++ b/crates/cli/src/error.rs @@ -116,14 +116,6 @@ pub enum CliError { #[error("Eth2util deposit error: {0}")] Eth2utilDepositError(#[from] pluto_eth2util::deposit::DepositError), - /// Tracing initialization error. - #[error("Tracing initialization error: {0}")] - TracingInitError(#[from] pluto_tracing::init::Error), - - /// Command parsing error. - #[error("Command parsing error: {0}")] - CommandParsingError(#[from] clap::Error), - /// Create DKG error. #[error("Create DKG error: {0}")] CreateDKGError(#[from] crate::commands::create_dkg::CreateDkgError), diff --git a/crates/cli/src/main.rs b/crates/cli/src/main.rs index 79bb40a7..a7bb6326 100644 --- a/crates/cli/src/main.rs +++ b/crates/cli/src/main.rs @@ -9,6 +9,7 @@ use clap::FromArgMatches; use cli::{AlphaCommands, Cli, Commands, CreateCommands, TestCommands, UnsafeCommands}; use std::process::ExitCode; use tokio_util::sync::CancellationToken; +use tracing::error; mod ascii; mod cli; @@ -18,19 +19,43 @@ mod error; #[tokio::main] async fn main() -> ExitCode { - match run().await { + let matches = cli::build_command().get_matches(); + + let cli = match Cli::from_arg_matches(&matches) { + Ok(cli) => cli, + Err(err) => { + eprintln!("{err}"); + return ExitCode::FAILURE; + } + }; + + let loki = match pluto_tracing::init(&cli.tracing.tracing_config()) { + Ok(loki) => loki, + Err(err) => { + eprintln!("{err}"); + return ExitCode::FAILURE; + } + }; + cli.tracing.warn_unused(); + + let result = run(cli.command).await; + + let exit = match &result { Ok(()) => ExitCode::SUCCESS, Err(err) => { - eprintln!("Error: {}", err); + error!(error = %err, "command exited with error"); ExitCode::FAILURE } + }; + + if let Some(loki) = loki { + loki.shutdown().await; } -} -async fn run() -> std::result::Result<(), CliError> { - let matches = cli::build_command().get_matches(); - let cli = Cli::from_arg_matches(&matches)?; + exit +} +async fn run(command: Commands) -> std::result::Result<(), CliError> { // Top level cancellation token for graceful shutdown on Ctrl+C / SIGTERM. let ct = CancellationToken::new(); tokio::spawn({ @@ -61,23 +86,18 @@ async fn run() -> std::result::Result<(), CliError> { }); let mut stdout = std::io::stdout(); - match cli.command { - Commands::Create(args) => { - pluto_tracing::init(&pluto_tracing::TracingConfig::default()) - .expect("Failed to initialize tracing"); - match args.command { - CreateCommands::Dkg(args) => commands::create_dkg::run(*args).await, - CreateCommands::Enr(args) => commands::create_enr::run(args), - CreateCommands::Cluster(args) => { - commands::create_cluster::run(&mut stdout, *args).await - } + match command { + Commands::Create(args) => match args.command { + CreateCommands::Dkg(args) => commands::create_dkg::run(*args).await, + CreateCommands::Enr(args) => commands::create_enr::run(args), + CreateCommands::Cluster(args) => { + commands::create_cluster::run(&mut stdout, *args).await } - } + }, Commands::Enr(args) => commands::enr::run(args), Commands::Version(args) => commands::version::run(args), Commands::Dkg(args) => { let config: pluto_dkg::dkg::Config = (*args).try_into()?; - pluto_tracing::init(&config.log).expect("Failed to initialize tracing"); commands::dkg::run(config, ct).await } Commands::Relay(args) => { @@ -86,7 +106,6 @@ async fn run() -> std::result::Result<(), CliError> { } Commands::Run(args) => { let config: commands::run::RunConfig = (*args).try_into()?; - // Tracing/Loki init is owned by `commands::run::run`. commands::run::run(config, ct).await } Commands::Unsafe(args) => match args.command { @@ -96,32 +115,26 @@ async fn run() -> std::result::Result<(), CliError> { } }, Commands::Alpha(args) => match args.command { - AlphaCommands::Test(args) => { - pluto_tracing::init(&pluto_tracing::TracingConfig::default()) - .expect("Failed to initialize tracing"); - match args.command { - TestCommands::Peers(args) => commands::test::peers::run(args, &mut stdout, ct) + AlphaCommands::Test(args) => match args.command { + TestCommands::Peers(args) => commands::test::peers::run(args, &mut stdout, ct) + .await + .map(|_| ()), + TestCommands::Beacon(args) => commands::test::beacon::run(args, &mut stdout, ct) + .await + .map(|_| ()), + TestCommands::Validator(args) => { + commands::test::validator::run(args, &mut stdout, ct) .await - .map(|_| ()), - TestCommands::Beacon(args) => { - commands::test::beacon::run(args, &mut stdout, ct) - .await - .map(|_| ()) - } - TestCommands::Validator(args) => { - commands::test::validator::run(args, &mut stdout, ct) - .await - .map(|_| ()) - } - TestCommands::Mev(args) => commands::test::mev::run(args, &mut stdout, ct) - .await - .map(|_| ()), - TestCommands::Infra(args) => commands::test::infra::run(args, &mut stdout, ct) - .await - .map(|_| ()), - TestCommands::All(args) => commands::test::all::run(*args, &mut stdout).await, + .map(|_| ()) } - } + TestCommands::Mev(args) => commands::test::mev::run(args, &mut stdout, ct) + .await + .map(|_| ()), + TestCommands::Infra(args) => commands::test::infra::run(args, &mut stdout, ct) + .await + .map(|_| ()), + TestCommands::All(args) => commands::test::all::run(*args, &mut stdout).await, + }, }, } } diff --git a/crates/dkg/Cargo.toml b/crates/dkg/Cargo.toml index eb1983bf..2dd3c6fc 100644 --- a/crates/dkg/Cargo.toml +++ b/crates/dkg/Cargo.toml @@ -34,7 +34,6 @@ pluto-parsigex.workspace = true pluto-peerinfo.workspace = true pluto-frost.workspace = true async-trait.workspace = true -pluto-tracing.workspace = true hex.workspace = true rand.workspace = true serde.workspace = true diff --git a/crates/dkg/src/dkg.rs b/crates/dkg/src/dkg.rs index 21f36195..0bca0615 100644 --- a/crates/dkg/src/dkg.rs +++ b/crates/dkg/src/dkg.rs @@ -36,7 +36,6 @@ use pluto_eth2util::keymanager::{self, KeymanagerError}; use pluto_p2p::{ bootnode::BootnodeError, config::P2PConfig, k1::key_path, p2p::P2PError, peer::Peer, }; -use pluto_tracing::TracingConfig; use url::Url; const DEFAULT_DATA_DIR: &str = ".charon"; @@ -275,10 +274,6 @@ pub struct Config { #[builder(default = default_p2p_config())] pub p2p: P2PConfig, - /// Shared tracing configuration for the DKG entrypoint. - #[builder(default = default_tracing_config())] - pub log: pluto_tracing::TracingConfig, - /// Keymanager configuration. #[builder(default)] pub keymanager: KeymanagerConfig, @@ -369,13 +364,6 @@ fn default_p2p_config() -> P2PConfig { } } -fn default_tracing_config() -> TracingConfig { - TracingConfig::builder() - .with_default_console() - .override_env_filter("info") - .build() -} - /// Runs the DKG entrypoint. pub async fn run(conf: Config, ct: CancellationToken) -> Result<(), DkgError> { if ct.is_cancelled() { @@ -1091,8 +1079,6 @@ mod tests { assert!(!config.no_verify); assert_eq!(config.data_dir, path::PathBuf::from(DEFAULT_DATA_DIR)); assert_eq!(config.p2p.relays, pluto_p2p::config::default_relays()); - assert_eq!(config.log.override_env_filter.as_deref(), Some("info")); - assert!(config.log.console.is_some()); assert_eq!(config.publish.address, DEFAULT_PUBLISH_ADDRESS); assert_eq!(config.publish.timeout, DEFAULT_PUBLISH_TIMEOUT); assert!(!config.publish.enabled); diff --git a/crates/peerinfo/examples/peerinfo.rs b/crates/peerinfo/examples/peerinfo.rs index c374ce73..35bdc395 100644 --- a/crates/peerinfo/examples/peerinfo.rs +++ b/crates/peerinfo/examples/peerinfo.rs @@ -237,8 +237,8 @@ async fn main() -> anyhow::Result<()> { // Initialize tracing with optional Loki support let tracing_config = build_tracing_config(&args); - if let Some(loki) = pluto_tracing::init(&tracing_config)? { - tokio::spawn(loki.task); + let loki = pluto_tracing::init(&tracing_config)?; + if loki.is_some() { tracing::info!("Loki logging enabled"); } @@ -404,5 +404,9 @@ async fn main() -> anyhow::Result<()> { } } + if let Some(loki) = loki { + loki.shutdown().await; + } + Ok(()) } diff --git a/crates/relay-server/Cargo.toml b/crates/relay-server/Cargo.toml index 8e04354e..1818ecb3 100644 --- a/crates/relay-server/Cargo.toml +++ b/crates/relay-server/Cargo.toml @@ -19,7 +19,6 @@ vise.workspace = true tokio.workspace = true tokio-util.workspace = true rand.workspace = true -pluto-tracing.workspace = true tracing.workspace = true pluto-p2p.workspace = true pluto-core.workspace = true @@ -27,6 +26,7 @@ pluto-core.workspace = true [dev-dependencies] reqwest = { workspace = true } serde_json = { workspace = true } +pluto-tracing.workspace = true [lints] workspace = true diff --git a/crates/relay-server/src/config.rs b/crates/relay-server/src/config.rs index 35b692e6..499b382c 100644 --- a/crates/relay-server/src/config.rs +++ b/crates/relay-server/src/config.rs @@ -3,7 +3,6 @@ use std::{num::NonZeroU32, path::PathBuf, time::Duration}; use bon::Builder; use libp2p::relay; use pluto_p2p::config::P2PConfig; -use pluto_tracing::TracingConfig; /// One hour in seconds. pub const ONE_HOUR_SECONDS: u64 = 60 * 60; @@ -36,9 +35,6 @@ pub struct Config { pub debug_addr: Option, /// The P2P configuration. pub p2p_config: P2PConfig, - /// The logging configuration. - #[builder(default)] - pub log_config: TracingConfig, /// Whether to automatically generate a P2P key. #[builder(default = false)] pub auto_p2p_key: bool, diff --git a/crates/tracing/examples/basic.rs b/crates/tracing/examples/basic.rs index 29cb35b7..70955775 100644 --- a/crates/tracing/examples/basic.rs +++ b/crates/tracing/examples/basic.rs @@ -22,7 +22,6 @@ async fn main() { // Initialize tracing with default console config let config = TracingConfig::builder() .with_default_console() - .with_metrics(true) .loki(LokiConfig { loki_url: "http://localhost:3100".to_string(), labels: HashMap::new(), @@ -35,8 +34,6 @@ async fn main() { .expect("Failed to initialize tracing") .expect("Loki background task should be Some"); - tokio::spawn(loki.task); - let bind_address = SocketAddr::from(([0, 0, 0, 0], 9464)); let exporter = MetricsExporter::default() @@ -71,7 +68,9 @@ async fn main() { info!("Processing completed"); // Wait for 10 seconds to see the logs in Loki - std::thread::sleep(std::time::Duration::from_secs(10)); + tokio::time::sleep(std::time::Duration::from_secs(10)).await; + + loki.shutdown().await; } #[instrument] diff --git a/crates/tracing/src/config.rs b/crates/tracing/src/config.rs index 3fde25b2..2223a77e 100644 --- a/crates/tracing/src/config.rs +++ b/crates/tracing/src/config.rs @@ -7,13 +7,10 @@ pub struct TracingConfig { /// logging is enabled. pub loki: Option, - /// Console configuration. Enables console logging if provided. If not - no - /// console logging is enabled. + /// Console layer options. Defaults are used when absent; console logging is + /// always enabled. pub console: Option, - /// Enables metrics logging. If not - no metrics logging is enabled. - pub metrics: bool, - /// Overrides the environment filter. If not - the environment filter is /// used. pub override_env_filter: Option, @@ -178,18 +175,6 @@ impl TracingConfigBuilder { self } - /// Enables metrics logging. - pub fn with_metrics(mut self, enabled: bool) -> Self { - self.tracing_config.metrics = enabled; - self - } - - /// Sets whether metrics logging is enabled. - pub fn metrics(mut self, enabled: bool) -> Self { - self.tracing_config.metrics = enabled; - self - } - /// Sets the environment filter override. pub fn override_env_filter(mut self, filter: impl Into) -> Self { self.tracing_config.override_env_filter = Some(filter.into()); diff --git a/crates/tracing/src/init.rs b/crates/tracing/src/init.rs index 8b116969..0b6469b2 100644 --- a/crates/tracing/src/init.rs +++ b/crates/tracing/src/init.rs @@ -1,8 +1,8 @@ -use std::str::FromStr; +use std::{str::FromStr, time::Duration}; use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64}; use percent_encoding::percent_decode_str; -use tracing_loki::{BackgroundTask, BackgroundTaskController, url::Url}; +use tracing_loki::{BackgroundTaskController, url::Url}; use tracing_subscriber::{ EnvFilter, Registry, layer::SubscriberExt as _, util::SubscriberInitExt as _, }; @@ -27,23 +27,48 @@ pub enum Error { type Result = std::result::Result; -/// Loki background task plus the controller used to signal graceful shutdown. +/// Grace period given to the Loki background task to flush buffered events once +/// shutdown has been signalled. +const FLUSH_TIMEOUT: Duration = Duration::from_secs(3); + +/// The running Loki background task, returned by [`init`] when Loki is +/// configured. /// -/// For long-lived services, hold onto `controller` and call -/// `controller.shutdown().await` followed by awaiting the spawned `task` -/// before exit so buffered events are drained. Short-lived programs (e.g. -/// examples, one-shot CLI subcommands) may drop the controller; any logs -/// not yet posted to Loki at process exit will be lost. -#[must_use = "the background `task` must be spawned for events to reach Loki"] -pub struct LokiInit { - /// Handle used to tell the background task to drain its queue and exit. - pub controller: BackgroundTaskController, - /// Future that ships buffered events to Loki; must be spawned to run. - pub task: BackgroundTask, +/// Dropping this rather than calling [`LokiWorker::shutdown`] loses every event +/// not yet posted to Loki. +pub struct LokiWorker { + controller: BackgroundTaskController, + handle: tokio::task::JoinHandle<()>, +} + +impl LokiWorker { + /// Drains buffered events and stops the worker. + /// + /// Signalling and draining share a single grace period so an unreachable + /// Loki endpoint cannot wedge process exit; the worker is hard-aborted once + /// the budget elapses. + pub async fn shutdown(self) { + let abort_handle = self.handle.abort_handle(); + + let _ = tokio::time::timeout(FLUSH_TIMEOUT, async { + self.controller.shutdown().await; + let _ = self.handle.await; + }) + .await; + + abort_handle.abort(); + } } /// Initializes the tracing subscriber. -pub fn init(config: &TracingConfig) -> Result> { +/// +/// When `config` enables Loki, the background task that ships events is spawned +/// here and returned so the caller can drain it before exit. +/// +/// # Panics +/// +/// Panics when Loki is configured and this is called outside a Tokio runtime. +pub fn init(config: &TracingConfig) -> Result> { let env_filter = if let Some(override_env_filter) = config.override_env_filter.as_ref() { EnvFilter::from_str(override_env_filter).unwrap_or_else(|_| default_env_filter()) } else { @@ -53,6 +78,8 @@ pub fn init(config: &TracingConfig) -> Result> { let console_config = config.console.clone().unwrap_or_default(); let fmt_layer = tracing_subscriber::fmt::layer() + // Logs belong on stderr so a command's stdout stays pipeable. + .with_writer(std::io::stderr) .with_target(console_config.with_target) .with_level(console_config.with_level) .with_thread_ids(console_config.with_thread_ids) @@ -91,7 +118,10 @@ pub fn init(config: &TracingConfig) -> Result> { let registry = registry.with(loki_layer); registry.try_init()?; - Ok(Some(LokiInit { controller, task })) + Ok(Some(LokiWorker { + controller, + handle: tokio::spawn(task), + })) } else { registry.try_init()?; Ok(None) diff --git a/crates/tracing/src/lib.rs b/crates/tracing/src/lib.rs index 7a4d407d..c467d9d6 100644 --- a/crates/tracing/src/lib.rs +++ b/crates/tracing/src/lib.rs @@ -18,4 +18,4 @@ pub mod metrics; pub use config::{ConsoleConfig, LokiConfig, TracingConfig, TracingConfigBuilder}; -pub use init::{LokiInit, init}; +pub use init::{LokiWorker, init}; diff --git a/flake.lock b/flake.lock index 43f361a5..d220f839 100644 --- a/flake.lock +++ b/flake.lock @@ -30,11 +30,11 @@ ] }, "locked": { - "lastModified": 1786935912, - "narHash": "sha256-26qGGBO364d5JbTMrFchrxp7xY8CJp3D5cjkND99t7E=", + "lastModified": 1788165049, + "narHash": "sha256-en4IoUeCqvq9F66YhwOrUFw1nc70OBhrJwrzfalezvY=", "owner": "oxalica", "repo": "rust-overlay", - "rev": "b479967b8ed7aca40ba52cf12f460484c53928a9", + "rev": "d03cd474bd97389dcc2e8cd3b3bb6b8c6e346b1a", "type": "github" }, "original": {