diff --git a/Cargo.lock b/Cargo.lock index b0979a31e..16310c8f4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -255,7 +255,6 @@ dependencies = [ "libcgroups", "libcontainer", "log", - "multi_log", "net_util", "netlink-packet-route", "nix 0.29.0", @@ -270,7 +269,6 @@ dependencies = [ "serde_json", "serial_test", "simple_test_case", - "simplelog", "syslog-tracing", "tempfile", "test-helpers", @@ -1192,15 +1190,6 @@ dependencies = [ "zeroize", ] -[[package]] -name = "deranged" -version = "0.3.11" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b42b6fa04a440b495c8b04d0e71b707c585f83cb9cb28cf8cd0d976c315e31b4" -dependencies = [ - "powerfmt", -] - [[package]] name = "derive_builder" version = "0.20.2" @@ -2474,15 +2463,6 @@ dependencies = [ "windows-sys 0.52.0", ] -[[package]] -name = "multi_log" -version = "0.1.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ad56bb3c7c7c15b4e25de86c9123e886e5f80a0c03ace219b453b081c2bf20d7" -dependencies = [ - "log", -] - [[package]] name = "nanorand" version = "0.7.0" @@ -2640,12 +2620,6 @@ dependencies = [ "serde", ] -[[package]] -name = "num-conv" -version = "0.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "51d515d32fb182ee37cda2ccdcb92950d6a3c2893aa280e540671c2cd0f3b1d9" - [[package]] name = "num-integer" version = "0.1.46" @@ -2695,15 +2669,6 @@ dependencies = [ "syn 2.0.100", ] -[[package]] -name = "num_threads" -version = "0.1.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5c7398b9c8b70908f6371f47ed36737907c87c52af34c268fed0bf0ceb92ead9" -dependencies = [ - "libc", -] - [[package]] name = "object" version = "0.36.7" @@ -2951,12 +2916,6 @@ dependencies = [ "zerovec 0.11.5", ] -[[package]] -name = "powerfmt" -version = "0.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "439ee305def115ba05938db6eb1644ff94165c5ab5e9420d1c1bcedbba909391" - [[package]] name = "ppv-lite86" version = "0.2.21" @@ -3712,17 +3671,6 @@ dependencies = [ "syn 2.0.100", ] -[[package]] -name = "simplelog" -version = "0.12.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "16257adbfaef1ee58b1363bdc0664c9b8e1e30aed86049635fb5f147d065a9c0" -dependencies = [ - "log", - "termcolor", - "time", -] - [[package]] name = "siphasher" version = "0.3.11" @@ -4498,39 +4446,6 @@ dependencies = [ "once_cell", ] -[[package]] -name = "time" -version = "0.3.39" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dad298b01a40a23aac4580b67e3dbedb7cc8402f3592d7f49469de2ea4aecdd8" -dependencies = [ - "deranged", - "itoa", - "libc", - "num-conv", - "num_threads", - "powerfmt", - "serde", - "time-core", - "time-macros", -] - -[[package]] -name = "time-core" -version = "0.1.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "765c97a5b985b7c11d7bc27fa927dc4fe6af3a6dfb021d28deb60d3bf51e76ef" - -[[package]] -name = "time-macros" -version = "0.2.20" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e8093bc3e81c3bc5f7879de09619d06c9a5a5e45ca44dfeeb7225bae38005c5c" -dependencies = [ - "num-conv", - "time-core", -] - [[package]] name = "timezone_provider" version = "0.1.2" @@ -4846,6 +4761,16 @@ dependencies = [ "tracing-core", ] +[[package]] +name = "tracing-serde" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "704b1aeb7be0d0a84fc9828cae51dab5970fee5088f83d1dd7ee6f6246fc6ff1" +dependencies = [ + "serde", + "tracing-core", +] + [[package]] name = "tracing-subscriber" version = "0.3.19" @@ -4856,12 +4781,15 @@ dependencies = [ "nu-ansi-term", "once_cell", "regex", + "serde", + "serde_json", "sharded-slab", "smallvec", "thread_local", "tracing", "tracing-core", "tracing-log", + "tracing-serde", ] [[package]] diff --git a/auraed/Cargo.toml b/auraed/Cargo.toml index 1d49b09bf..02f4c21a4 100644 --- a/auraed/Cargo.toml +++ b/auraed/Cargo.toml @@ -71,7 +71,7 @@ tokio-stream = { version = "0.1.17", features = ["net", "sync"] } tonic = { workspace = true, features = ["tls-ring"] } tonic-health = { workspace = true } tracing = { workspace = true, features = ["log"] } -tracing-subscriber = { version = "0.3", features = ["env-filter", "registry"] } +tracing-subscriber = { version = "0.3", features = ["env-filter", "json", "registry"] } uuid = { workspace = true } validation = { workspace = true, features = ["regex", "tonic"] } validation_macros = { path = "../crates/validation/macros" } @@ -89,10 +89,8 @@ seccompiler = "0.4.0" [dev-dependencies] futures-util = { workspace = true } -multi_log = "0.1.2" pretty_assertions = "1.3.0" serial_test = { workspace = true } -simplelog = "0.12.0" simple_test_case = "1.1.0" test-helpers = { workspace = true } test-helpers-macros = { workspace = true } diff --git a/auraed/src/ebpf/perf_buffer_reader.rs b/auraed/src/ebpf/perf_buffer_reader.rs index 30a9b9fe5..92bdd8390 100644 --- a/auraed/src/ebpf/perf_buffer_reader.rs +++ b/auraed/src/ebpf/perf_buffer_reader.rs @@ -121,6 +121,13 @@ pub trait PerfBufferReader { } }; + // Clear cached readiness after an empty read so the next + // call to `readable_mut` waits for more events. + if events.read == 0 && events.lost == 0 { + guard.clear_ready(); + continue; + } + if events.lost > 0 { error!( "buffer full, dropped {} perf events - this should never happen!", diff --git a/auraed/src/graceful_shutdown.rs b/auraed/src/graceful_shutdown.rs index ef9b504d0..fc5122df8 100644 --- a/auraed/src/graceful_shutdown.rs +++ b/auraed/src/graceful_shutdown.rs @@ -13,7 +13,10 @@ * SPDX-License-Identifier: Apache-2.0 * \* -------------------------------------------------------------------------- */ -use crate::{VmService, cells::CellService, discovery::DiscoveryService}; +use crate::{ + VmService, cells::CellService, discovery::DiscoveryService, + observe::ObserveService, +}; use proto::{ cells::cell_service_server::CellServiceServer, discovery::discovery_service_server::DiscoveryServiceServer, @@ -31,6 +34,7 @@ pub(crate) struct GracefulShutdown { health_reporter: HealthReporter, cell_service: CellService, vm_service: VmService, + observe_service: ObserveService, shutdown_broadcaster: Sender<()>, } @@ -39,12 +43,14 @@ impl GracefulShutdown { health_reporter: HealthReporter, cell_service: CellService, vm_service: VmService, + observe_service: ObserveService, ) -> Self { let (tx, _) = channel(()); Self { health_reporter, cell_service, vm_service, + observe_service, shutdown_broadcaster: tx, } } @@ -82,6 +88,7 @@ impl GracefulShutdown { // health_reporter.set_not_serving::>().await; + self.observe_service.shutdown_log_streams(); self.shutdown_broadcaster.send_replace(()); // wait for all subscribers to drop self.shutdown_broadcaster.closed().await; diff --git a/auraed/src/init/logging.rs b/auraed/src/init/logging.rs index c5edb9819..596ff7a1a 100644 --- a/auraed/src/init/logging.rs +++ b/auraed/src/init/logging.rs @@ -12,6 +12,7 @@ * Copyright 2022 - 2024, the aurae contributors * * SPDX-License-Identifier: Apache-2.0 * \* -------------------------------------------------------------------------- */ +use crate::logging::log_channel::LogChannel; use tracing::{Level, info}; use tracing_subscriber::{ EnvFilter, Layer, layer::SubscriberExt, util::SubscriberInitExt, @@ -32,30 +33,53 @@ pub(crate) enum LoggingError { SyslogError, } -pub(crate) fn init(verbose: bool, container: bool) -> Result<(), LoggingError> { - // The logger will log to stdout. - // - // We hold the opinion that the program is either "verbose" - // or it's not. - // - // Normal mode: Info, Warn, Error - // Verbose mode: Debug, Trace, Info, Warn, Error +// The logger will log to stdout. +// +// We hold the opinion that the program is either "verbose" +// or it's not. +// +// Normal mode: Info, Warn, Error +// Verbose mode: Debug, Trace, Info, Warn, Error +pub(crate) fn init( + verbose: bool, + container: bool, + log_channel: LogChannel, +) -> Result<(), LoggingError> { let tracing_level = if verbose { Level::TRACE } else { Level::INFO }; if container { - init_container_logging(tracing_level) + init_container_logging(tracing_level, log_channel) } else { match std::process::id() { - 1 => init_pid1_logging(tracing_level), - _ => init_daemon_logging(tracing_level), + 1 => init_pid1_logging(tracing_level, log_channel), + _ => init_daemon_logging(tracing_level, log_channel), } } } -fn init_container_logging(tracing_level: Level) -> Result<(), LoggingError> { +/// Build the layer that sends tracing events to the Observe log stream. +fn broadcast_layer( + log_channel: LogChannel, + tracing_level: Level, +) -> impl Layer +where + S: tracing::Subscriber + + for<'a> tracing_subscriber::registry::LookupSpan<'a>, +{ + tracing_subscriber::fmt::layer() + .json() + .with_writer(log_channel) + .with_filter(EnvFilter::new(format!( + "auraed={tracing_level},auraed::observe=off" + ))) +} + +fn init_container_logging( + tracing_level: Level, + log_channel: LogChannel, +) -> Result<(), LoggingError> { info!("initializing container logging"); - // Stdout let stdout_layer = Layer::with_filter( tracing_subscriber::fmt::layer().compact(), EnvFilter::new(format!("auraed={tracing_level}")), @@ -63,15 +87,18 @@ fn init_container_logging(tracing_level: Level) -> Result<(), LoggingError> { tracing_subscriber::registry() .with(stdout_layer) + .with(broadcast_layer(log_channel, tracing_level)) .try_init() .map_err(|e| e.into()) } -/// when we run as a daemon we want to log to stdout and syslog. -fn init_daemon_logging(tracing_level: Level) -> Result<(), LoggingError> { +/// When we run as a daemon, log to stdout and syslog. +fn init_daemon_logging( + tracing_level: Level, + log_channel: LogChannel, +) -> Result<(), LoggingError> { info!("initializing syslog logging"); - // Syslog let syslog_identity = c"auraed"; let syslog_facility = Default::default(); let syslog_options = syslog_tracing::Options::LOG_PID; @@ -84,8 +111,6 @@ fn init_daemon_logging(tracing_level: Level) -> Result<(), LoggingError> { }; let syslog_layer = tracing_subscriber::fmt::layer().with_writer(syslog); - - // Stdout let stdout_layer = Layer::with_filter( tracing_subscriber::fmt::layer().compact(), EnvFilter::new(format!("auraed={tracing_level}")), @@ -94,26 +119,25 @@ fn init_daemon_logging(tracing_level: Level) -> Result<(), LoggingError> { tracing_subscriber::registry() .with(syslog_layer) .with(stdout_layer) + .with(broadcast_layer(log_channel, tracing_level)) .try_init() .map_err(|e| e.into()) } -#[allow(unused)] -fn init_stdout_logging(tracing_level: Level) -> Result<(), LoggingError> { - info!("initializing stdout logging"); - tracing_subscriber::fmt() - .compact() - .with_env_filter(format!("auraed={tracing_level}")) - .finish() - .try_init() - .map_err(|e| e.into()) -} - -fn init_pid1_logging(tracing_level: Level) -> Result<(), LoggingError> { +fn init_pid1_logging( + tracing_level: Level, + log_channel: LogChannel, +) -> Result<(), LoggingError> { info!("initializing pid1 logging"); - tracing_subscriber::fmt() - .compact() - .with_env_filter(format!("auraed={tracing_level}")) + + let stdout_layer = Layer::with_filter( + tracing_subscriber::fmt::layer().compact(), + EnvFilter::new(format!("auraed={tracing_level}")), + ); + + tracing_subscriber::registry() + .with(stdout_layer) + .with(broadcast_layer(log_channel, tracing_level)) .try_init() - .map_err(|e| LoggingError::SetupFailure { source: e }) + .map_err(|e| LoggingError::SetupFailure { source: Box::new(e) }) } diff --git a/auraed/src/init/mod.rs b/auraed/src/init/mod.rs index f8300cbb3..78c5b387e 100644 --- a/auraed/src/init/mod.rs +++ b/auraed/src/init/mod.rs @@ -23,6 +23,7 @@ use self::system_runtimes::{ CellSystemRuntime, ContainerSystemRuntime, DaemonSystemRuntime, Pid1SystemRuntime, SystemRuntime, SystemRuntimeError, }; +use crate::logging::log_channel::LogChannel; use std::fs::File; use std::io::{BufReader, Read}; mod fileio; @@ -79,12 +80,14 @@ pub async fn init( verbose: bool, nested: bool, socket_address: Option, + log_channel: LogChannel, ) -> (Context, SocketStream) { let context = Context::get(nested); let init_result = init_with_runtimes( context, verbose, socket_address, + log_channel, Pid1SystemRuntime {}, CellSystemRuntime {}, ContainerSystemRuntime {}, @@ -98,10 +101,12 @@ pub async fn init( } } +#[allow(clippy::too_many_arguments)] async fn init_with_runtimes( context: Context, verbose: bool, socket_address: Option, + log_channel: LogChannel, pid1_runtime: RPid1, cell_runtime: RCell, container_runtime: RContainer, @@ -114,12 +119,18 @@ where RDaemon: SystemRuntime, { match context { - Context::Pid1 => pid1_runtime.init(verbose, socket_address).await, - Context::Cell => cell_runtime.init(verbose, socket_address).await, + Context::Pid1 => { + pid1_runtime.init(verbose, socket_address, log_channel).await + } + Context::Cell => { + cell_runtime.init(verbose, socket_address, log_channel).await + } Context::Container => { - container_runtime.init(verbose, socket_address).await + container_runtime.init(verbose, socket_address, log_channel).await + } + Context::Daemon => { + daemon_runtime.init(verbose, socket_address, log_channel).await } - Context::Daemon => daemon_runtime.init(verbose, socket_address).await, } } @@ -337,6 +348,7 @@ mod tests { self, _verbose: bool, _socket_address: Option, + _log_channel: LogChannel, ) -> Result { let _ = self.calls.fetch_add(1, Ordering::SeqCst); Err(SystemRuntimeError::Other(anyhow!(self.label))) @@ -382,6 +394,7 @@ mod tests { ctx, false, None, + LogChannel::new("test"), pid1.clone(), cell.clone(), container.clone(), diff --git a/auraed/src/init/system_runtimes/cell_system_runtime.rs b/auraed/src/init/system_runtimes/cell_system_runtime.rs index f501e7be9..d369ef19d 100644 --- a/auraed/src/init/system_runtimes/cell_system_runtime.rs +++ b/auraed/src/init/system_runtimes/cell_system_runtime.rs @@ -20,6 +20,7 @@ use crate::AURAED_RUNTIME; use crate::init::{ BANNER, logging, system_runtimes::create_unix_socket_stream, }; +use crate::logging::log_channel::LogChannel; use tonic::async_trait; use tracing::info; @@ -31,9 +32,10 @@ impl SystemRuntime for CellSystemRuntime { self, verbose: bool, socket_address: Option, + log_channel: LogChannel, ) -> Result { println!("{BANNER}"); - logging::init(verbose, false)?; + logging::init(verbose, false, log_channel)?; info!("Running as a cell"); create_unix_socket_stream( socket_address.map(PathBuf::from).unwrap_or_else(|| { diff --git a/auraed/src/init/system_runtimes/container_system_runtime.rs b/auraed/src/init/system_runtimes/container_system_runtime.rs index 2210faef1..1e9b2de8e 100644 --- a/auraed/src/init/system_runtimes/container_system_runtime.rs +++ b/auraed/src/init/system_runtimes/container_system_runtime.rs @@ -20,6 +20,7 @@ use crate::AURAED_RUNTIME; use crate::init::{ BANNER, logging, system_runtimes::create_unix_socket_stream, }; +use crate::logging::log_channel::LogChannel; use tonic::async_trait; use tracing::info; @@ -31,9 +32,10 @@ impl SystemRuntime for ContainerSystemRuntime { self, verbose: bool, socket_address: Option, + log_channel: LogChannel, ) -> Result { println!("{BANNER}"); - logging::init(verbose, true)?; + logging::init(verbose, true, log_channel)?; info!("Running as a container."); create_unix_socket_stream( socket_address.map(PathBuf::from).unwrap_or_else(|| { diff --git a/auraed/src/init/system_runtimes/daemon_system_runtime.rs b/auraed/src/init/system_runtimes/daemon_system_runtime.rs index 79254f662..35bf8de17 100644 --- a/auraed/src/init/system_runtimes/daemon_system_runtime.rs +++ b/auraed/src/init/system_runtimes/daemon_system_runtime.rs @@ -21,6 +21,7 @@ use crate::init::{ BANNER, logging, system_runtimes::{create_tcp_socket_stream, create_unix_socket_stream}, }; +use crate::logging::log_channel::LogChannel; use tonic::async_trait; use tracing::{info, trace}; @@ -32,9 +33,10 @@ impl SystemRuntime for DaemonSystemRuntime { self, verbose: bool, socket_address: Option, + log_channel: LogChannel, ) -> Result { println!("{BANNER}"); - logging::init(verbose, false)?; + logging::init(verbose, false, log_channel)?; info!("Running as a daemon."); // Running as a daemon supports both TCP and Unix sockets for listening, depending on the diff --git a/auraed/src/init/system_runtimes/mod.rs b/auraed/src/init/system_runtimes/mod.rs index 3dc77b4b0..c8dc0593d 100644 --- a/auraed/src/init/system_runtimes/mod.rs +++ b/auraed/src/init/system_runtimes/mod.rs @@ -28,6 +28,7 @@ use tonic::async_trait; use tracing::{info, trace}; use super::{fs::FsError, logging::LoggingError, network::NetworkError}; +use crate::logging::log_channel::LogChannel; mod cell_system_runtime; mod container_system_runtime; @@ -66,6 +67,7 @@ pub(crate) trait SystemRuntime { self, verbose: bool, socket_address: Option, + log_channel: LogChannel, ) -> Result; } diff --git a/auraed/src/init/system_runtimes/pid1_system_runtime.rs b/auraed/src/init/system_runtimes/pid1_system_runtime.rs index c7cea9dff..903256ad1 100644 --- a/auraed/src/init/system_runtimes/pid1_system_runtime.rs +++ b/auraed/src/init/system_runtimes/pid1_system_runtime.rs @@ -21,6 +21,7 @@ use crate::init::{ power::spawn_thread_power_button_listener, system_runtimes::create_tcp_socket_stream, }; +use crate::logging::log_channel::LogChannel; use nix::{ mount::MsFlags, unistd::{mkdir, symlinkat}, @@ -61,11 +62,12 @@ impl SystemRuntime for Pid1SystemRuntime { self, verbose: bool, socket_address: Option, + log_channel: LogChannel, ) -> Result { println!("{BANNER}"); // Initialize the PID 1 logger - logging::init(verbose, false)?; + logging::init(verbose, false, log_channel)?; info!("Running as pid 1"); trace!("Configure filesystem"); diff --git a/auraed/src/lib.rs b/auraed/src/lib.rs index 4648f3cdd..bf579b0c7 100644 --- a/auraed/src/lib.rs +++ b/auraed/src/lib.rs @@ -171,6 +171,7 @@ pub async fn run( runtime: &AuraedRuntime, context: AuraeContext, socket_stream: T, + log_channel: LogChannel, ) -> Result<(), Box> where T: tokio_stream::Stream> + Send + 'static, @@ -245,10 +246,7 @@ pub async fn run( let (health_reporter, health_service) = tonic_health::server::health_reporter(); - let observe_service = ObserveService::new( - LogChannel::new(String::from("auraed")), - perf_events, - ); + let observe_service = ObserveService::new(log_channel, perf_events); let observe_service_server = ObserveServiceServer::new(observe_service.clone()); @@ -285,6 +283,7 @@ pub async fn run( health_reporter, cell_service, vm_service, + observe_service, ); let graceful_shutdown_signal = graceful_shutdown.subscribe(); @@ -339,11 +338,17 @@ pub async fn run( } let runtime = AURAED_RUNTIME.get_or_init(|| runtime); + let log_channel = LogChannel::new(String::from("auraed")); - let (context, stream) = init::init(verbose, nested, socket).await; + let (context, stream) = + init::init(verbose, nested, socket, log_channel.clone()).await; match stream { - SocketStream::Tcp(stream) => inner(runtime, context, stream).await, - SocketStream::Unix(stream) => inner(runtime, context, stream).await, + SocketStream::Tcp(stream) => { + inner(runtime, context, stream, log_channel).await + } + SocketStream::Unix(stream) => { + inner(runtime, context, stream, log_channel).await + } } } diff --git a/auraed/src/logging/broadcast_writer.rs b/auraed/src/logging/broadcast_writer.rs new file mode 100644 index 000000000..a9fa8a48a --- /dev/null +++ b/auraed/src/logging/broadcast_writer.rs @@ -0,0 +1,202 @@ +/* -------------------------------------------------------------------------- *\ + * | █████╗ ██╗ ██╗██████╗ █████╗ ███████╗ | * + * | ██╔══██╗██║ ██║██╔══██╗██╔══██╗██╔════╝ | * + * | ███████║██║ ██║██████╔╝███████║█████╗ | * + * | ██╔══██║██║ ██║██╔══██╗██╔══██║██╔══╝ | * + * | ██║ ██║╚██████╔╝██║ ██║██║ ██║███████╗ | * + * | ╚═╝ ╚═╝ ╚═════╝ ╚═╝ ╚═╝╚═╝ ╚═╝╚══════╝ | * + * +--------------------------------------------+ * + * * + * Distributed Systems Runtime * + * -------------------------------------------------------------------------- * + * Copyright 2022 - 2024, the aurae contributors * + * SPDX-License-Identifier: Apache-2.0 * +\* -------------------------------------------------------------------------- */ + +//! `tracing_subscriber::fmt::MakeWriter` impl for `LogChannel`. +//! +//! Wiring `LogChannel` directly as the writer type means a layer is built +//! by passing the channel through `with_writer`: +//! +//! ```ignore +//! tracing_subscriber::fmt::layer().with_writer(log_channel.clone()) +//! ``` +//! +//! `MakeWriter::make_writer()` is invoked once per `tracing` event, the +//! returned `BroadcastWriter` accumulates the formatted bytes in a per-event +//! buffer, and `Drop` flushes one `LogItem` into the broadcast. + +use super::{get_timestamp_sec, log_channel::LogChannel}; +use proto::observe::LogItem; +use std::io; +use tokio::sync::broadcast::Sender; +use tracing_subscriber::fmt::MakeWriter; + +impl<'a> MakeWriter<'a> for LogChannel { + type Writer = BroadcastWriter<'a>; + + fn make_writer(&'a self) -> Self::Writer { + BroadcastWriter { + tx: self.sender(), + channel: &self.name, + buf: Vec::new(), + } + } +} + +/// Per-event writer. Accumulates bytes and flushes a single `LogItem` on drop. +#[derive(Debug)] +pub struct BroadcastWriter<'a> { + tx: &'a Sender, + channel: &'a str, + buf: Vec, +} + +impl<'a> io::Write for BroadcastWriter<'a> { + fn write(&mut self, buf: &[u8]) -> io::Result { + self.buf.extend_from_slice(buf); + Ok(buf.len()) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +} + +impl<'a> Drop for BroadcastWriter<'a> { + fn drop(&mut self) { + if self.buf.is_empty() || self.tx.receiver_count() == 0 { + return; + } + // Reuse the buffer's allocation: fmt::Layer emits one trailing '\n' + // per event, which we drop in place, and `String::from_utf8` takes + // ownership of the `Vec` without copying on the (overwhelmingly + // common) valid-UTF-8 path. + let mut buf = std::mem::take(&mut self.buf); + while buf.last() == Some(&b'\n') { + let _ = buf.pop(); + } + let line = String::from_utf8(buf).unwrap_or_else(|e| { + String::from_utf8_lossy(e.as_bytes()).into_owned() + }); + let _ = self.tx.send(LogItem { + channel: self.channel.to_owned(), + line, + timestamp: get_timestamp_sec(), + }); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::Value; + use tokio::sync::broadcast::error::TryRecvError; + use tracing::subscriber::with_default; + use tracing_subscriber::{ + EnvFilter, Layer, layer::SubscriberExt, registry, + }; + + fn build_subscriber( + channel: &LogChannel, + filter: &str, + ) -> impl tracing::Subscriber + Send + Sync { + let layer = tracing_subscriber::fmt::layer() + .json() + .with_writer(channel.clone()) + .with_filter(EnvFilter::new(filter)); + registry().with(layer) + } + + fn drain( + rx: &mut tokio::sync::broadcast::Receiver, + ) -> Vec { + let mut out = Vec::new(); + loop { + match rx.try_recv() { + Ok(item) => out.push(item), + Err(TryRecvError::Empty | TryRecvError::Closed) => break, + Err(TryRecvError::Lagged(_)) => { + panic!("unexpected Lagged in test") + } + } + } + out + } + + fn parse_line(line: &str) -> Value { + serde_json::from_str(line).expect("valid JSON log line") + } + + #[test] + fn emits_exactly_one_log_item_per_event_as_json() { + let channel = LogChannel::new("test-channel"); + let mut rx = channel.subscribe(); + let subscriber = build_subscriber(&channel, "info"); + + let message = "streamed log message"; + let before = get_timestamp_sec(); + with_default(subscriber, || { + tracing::info!(target: "auraed::test", "{message}"); + }); + let after = get_timestamp_sec(); + + let items = drain(&mut rx); + assert_eq!(items.len(), 1, "expected exactly one LogItem"); + let item = &items[0]; + + assert_eq!(item.channel, "test-channel"); + assert!( + item.timestamp >= before && item.timestamp <= after, + "timestamp {} not in [{before}, {after}]", + item.timestamp + ); + let json = parse_line(&item.line); + assert_eq!(json["level"].as_str(), Some("INFO"), "json: {json}"); + assert_eq!( + json["target"].as_str(), + Some("auraed::test"), + "json: {json}" + ); + assert_eq!( + json["fields"]["message"].as_str(), + Some(message), + "json: {json}" + ); + } + + #[test] + fn env_filter_excludes_observe_target() { + let channel = LogChannel::new("test"); + let mut rx = channel.subscribe(); + let subscriber = build_subscriber(&channel, "info,auraed::observe=off"); + + with_default(subscriber, || { + tracing::info!(target: "auraed::observe", "suppressed"); + tracing::info!(target: "auraed::cells", "visible"); + }); + + let items = drain(&mut rx); + assert_eq!(items.len(), 1, "exactly one event should pass the filter"); + + let json = parse_line(&items[0].line); + assert_eq!(json["target"].as_str(), Some("auraed::cells")); + assert_eq!(json["fields"]["message"].as_str(), Some("visible")); + } + + #[test] + fn short_circuits_when_no_subscribers() { + let channel = LogChannel::new("test"); + let subscriber = build_subscriber(&channel, "info"); + assert_eq!(channel.sender().receiver_count(), 0); + + with_default(subscriber, || { + for _ in 0..50 { + tracing::info!("no subscribers"); + } + }); + + let mut rx = channel.subscribe(); + assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); + } +} diff --git a/auraed/src/logging/log_channel.rs b/auraed/src/logging/log_channel.rs index 46099a778..b582c9d85 100644 --- a/auraed/src/logging/log_channel.rs +++ b/auraed/src/logging/log_channel.rs @@ -17,6 +17,13 @@ use super::get_timestamp_sec; use proto::observe::LogItem; use tokio::sync::broadcast::{self, Receiver, Sender}; +/// Capacity of the broadcast ring and each gRPC forwarding queue. +/// +/// This preserves the original effective ring size: Tokio rounded the former +/// capacity of 40 up to 64. Using that exact power of two makes the allocation +/// explicit and lets each forwarding queue absorb one full ring of messages. +pub(crate) const LOG_STREAM_CAPACITY: usize = 64; + /// Abstraction Layer for one log generating entity /// LogChannel provides channels between Log producers and log consumers #[derive(Clone, Debug)] @@ -28,10 +35,9 @@ pub struct LogChannel { impl LogChannel { /// Constructor creating the channel for log communication - pub fn new(name: String) -> LogChannel { - // TODO: decide for a cap. 40 is arbitrary - let (tx, _) = broadcast::channel(40); - LogChannel { name, tx } + pub fn new(name: impl Into) -> LogChannel { + let (tx, _) = broadcast::channel(LOG_STREAM_CAPACITY); + LogChannel { name: name.into(), tx } } /// Getter for consumer channel @@ -39,6 +45,11 @@ impl LogChannel { self.tx.subscribe() } + /// Borrows the producer side for the tracing writer. + pub(crate) fn sender(&self) -> &Sender { + &self.tx + } + /// Wrapper that sends a log line to the channel pub fn send(&self, line: String) { // send returns an Err if there are no receivers. We ignore that. @@ -54,23 +65,10 @@ impl LogChannel { #[cfg(test)] mod tests { use super::*; - use log::Level; - use simplelog::SimpleLogger; - - fn init_logging() { - let logger_simple = SimpleLogger::new( - Level::Trace.to_level_filter(), - simplelog::Config::default(), - ); - - multi_log::MultiLogger::init(vec![logger_simple], Level::Trace) - .expect("failed to initialize logger"); - } #[tokio::test] async fn test_ringbuffer_queue() { - init_logging(); - let channel = LogChannel::new("Test".into()); + let channel = LogChannel::new("Test"); let mut rx = channel.subscribe(); channel.send("hello".into()); diff --git a/auraed/src/logging/mod.rs b/auraed/src/logging/mod.rs index 50320bb74..77f55dc08 100644 --- a/auraed/src/logging/mod.rs +++ b/auraed/src/logging/mod.rs @@ -22,8 +22,8 @@ use std::time::SystemTime; /// LogChannel provides channels between Log producers and log consumers pub mod log_channel; -/// Implements Log trait. Used to add grpc API to log targets for rust internal logging -pub mod stream_logger; +/// `MakeWriter` adapter that pipes formatted `tracing` events into a `LogChannel` broadcast. +pub mod broadcast_writer; /// Get UNIX timestamp in seconds for logging pub fn get_timestamp_sec() -> i64 { diff --git a/auraed/src/logging/stream_logger.rs b/auraed/src/logging/stream_logger.rs deleted file mode 100644 index a816b7d94..000000000 --- a/auraed/src/logging/stream_logger.rs +++ /dev/null @@ -1,56 +0,0 @@ -/* -------------------------------------------------------------------------- *\ - * | █████╗ ██╗ ██╗██████╗ █████╗ ███████╗ | * - * | ██╔══██╗██║ ██║██╔══██╗██╔══██╗██╔════╝ | * - * | ███████║██║ ██║██████╔╝███████║█████╗ | * - * | ██╔══██║██║ ██║██╔══██╗██╔══██║██╔══╝ | * - * | ██║ ██║╚██████╔╝██║ ██║██║ ██║███████╗ | * - * | ╚═╝ ╚═╝ ╚═════╝ ╚═╝ ╚═╝╚═╝ ╚═╝╚══════╝ | * - * +--------------------------------------------+ * - * * - * Distributed Systems Runtime * - * -------------------------------------------------------------------------- * - * Copyright 2022 - 2024, the aurae contributors * - * SPDX-License-Identifier: Apache-2.0 * -\* -------------------------------------------------------------------------- */ - -use log::Log; -use proto::observe::LogItem; -use tokio::sync::broadcast::Sender; - -/// Sends log messages generated in rust code to the logging channel -/// The logging channel is consumed by the observe API -#[derive(Debug)] -pub struct StreamLogger { - /// Channel used to send log messages to grpc API - pub producer: Sender, -} - -impl StreamLogger { - #[allow(unused)] - /// Constructor requires channel between api and logger - pub fn new(producer: Sender) -> StreamLogger { - StreamLogger { producer } - } -} - -impl Log for StreamLogger { - fn enabled(&self, _metadata: &log::Metadata) -> bool { - true - } - - fn log(&self, record: &log::Record) { - // send returns an Err if there are no receivers. We ignore that. - let _ = self.producer.send(LogItem { - channel: "rust-logs".to_string(), - line: format!( - "{}:{} -- {}", - record.level(), - record.target(), - record.args() - ), - timestamp: 0, - }); - } - - fn flush(&self) {} -} diff --git a/auraed/src/observe/observe_service.rs b/auraed/src/observe/observe_service.rs index 9c1a8f36b..8f5bf4ae5 100644 --- a/auraed/src/observe/observe_service.rs +++ b/auraed/src/observe/observe_service.rs @@ -21,7 +21,7 @@ use super::error::ObserveServiceError; use super::observed_event_stream::ObservedEventStream; use super::proc_cache::{ProcCache, ProcfsProcessInfo}; use crate::ebpf::tracepoint::PerfEventBroadcast; -use crate::logging::log_channel::LogChannel; +use crate::logging::log_channel::{LOG_STREAM_CAPACITY, LogChannel}; use aurae_ebpf_shared::{ForkedProcess, ProcessExit, Signal}; use cgroup_cache::CgroupCache; use proto::observe::{ @@ -33,11 +33,11 @@ use proto::observe::{ use std::collections::HashMap; use std::sync::Arc; use std::time::Duration; -use tokio::sync::mpsc; -use tokio::sync::{Mutex, broadcast::Receiver}; +use tokio::sync::{Mutex, broadcast::Receiver, broadcast::error::RecvError}; +use tokio::sync::{mpsc, watch}; use tokio_stream::wrappers::ReceiverStream; use tonic::{Request, Response, Status}; -use tracing::{info, instrument}; +use tracing::{debug, info, instrument}; #[derive(Debug, Clone)] pub struct ObserveService { @@ -47,6 +47,7 @@ pub struct ObserveService { posix_signals: Option>, sub_process_consumer_list: Arc>>>, + log_stream_shutdown: watch::Sender, } type PerfEvents = ( @@ -57,6 +58,7 @@ type PerfEvents = ( impl ObserveService { pub fn new(aurae_logger: LogChannel, perf_events: PerfEvents) -> Self { + let (log_stream_shutdown, _) = watch::channel(false); let proc_cache = match perf_events { (Some(f), Some(e), _) => Some(ProcCache::new( Duration::from_secs(60), @@ -73,9 +75,14 @@ impl ObserveService { proc_cache, posix_signals: perf_events.2, sub_process_consumer_list: Arc::new(Mutex::new(HashMap::new())), + log_stream_shutdown, } } + pub(crate) fn shutdown_log_streams(&self) { + let _ = self.log_stream_shutdown.send_replace(true); + } + pub async fn register_sub_process_channel( &self, pid: i32, @@ -144,6 +151,48 @@ impl ObserveService { ReceiverStream::new(events) } + + /// Bridge a broadcast `LogItem` consumer to a gRPC stream. Lagged + /// consumers skip dropped items and resume with the oldest item still in + /// the broadcast ring. The forwarder exits when the client disconnects or + /// all senders are dropped. + fn spawn_log_forwarder( + &self, + mut consumer: Receiver, + wrap: impl Fn(LogItem) -> R + Send + 'static, + ) -> ReceiverStream> { + let (tx, rx) = mpsc::channel::>(LOG_STREAM_CAPACITY); + let mut shutdown = self.log_stream_shutdown.subscribe(); + + // The broadcast layer's EnvFilter excludes events with target + // `auraed::observe`, so any tracing emitted from inside this task + // does not feed back into the consumer. + let _ignored = tokio::spawn(async move { + loop { + let item = tokio::select! { + _ = shutdown.wait_for(|shutdown| *shutdown) => break, + _ = tx.closed() => break, + item = consumer.recv() => item, + }; + match item { + Ok(item) => { + tokio::select! { + _ = shutdown.wait_for(|shutdown| *shutdown) => break, + result = tx.send(Ok(wrap(item))) => { + if result.is_err() { + break; // gRPC client gone + } + } + } + } + Err(RecvError::Lagged(_)) => continue, + Err(RecvError::Closed) => break, + } + } + }); + + ReceiverStream::new(rx) + } } fn map_get_posix_signals_stream_response( @@ -179,27 +228,10 @@ impl observe_service_server::ObserveService for ObserveService { &self, _request: Request, ) -> Result, Status> { - let (tx, rx) = - mpsc::channel::>(4); - let mut log_consumer = self.get_aurae_daemon_log_stream(); - - // TODO: error handling. Warning: recursively logging if error message is also send to this grpc api endpoint - // .. thus disabled logging here. - let _ignored = tokio::spawn(async move { - // Log consumer will error if: - // the producer is closed (no more logs) - // the receiver is lagging - while let Ok(log_item) = log_consumer.recv().await { - let resp = - GetAuraeDaemonLogStreamResponse { item: Some(log_item) }; - if tx.send(Ok(resp)).await.is_err() { - // receiver is gone - break; - } - } - }); - - Ok(Response::new(ReceiverStream::new(rx))) + let consumer = self.get_aurae_daemon_log_stream(); + Ok(Response::new(self.spawn_log_forwarder(consumer, |item| { + GetAuraeDaemonLogStreamResponse { item: Some(item) } + }))) } type GetSubProcessStreamStream = @@ -215,10 +247,9 @@ impl observe_service_server::ObserveService for ObserveService { ObserveServiceError::InvalidLogChannelType { channel_type } })?; - println!("Requested Channel {channel:?}"); - println!("Requested Process ID {pid}"); + debug!("get_sub_process_stream channel={channel:?} pid={pid}"); - let mut log_consumer = { + let consumer = { let mut consumer_list = self.sub_process_consumer_list.lock().await; consumer_list .get_mut(&pid) @@ -228,29 +259,12 @@ impl observe_service_server::ObserveService for ObserveService { pid, channel_type: channel, })? - .clone() - } - .subscribe(); - - let (tx, rx) = - mpsc::channel::>(4); - - // TODO: error handling. Warning: recursively logging if error message is also send to this grpc api endpoint - // .. thus disabled logging here. - let _ignored = tokio::spawn(async move { - // Log consumer will error if: - // the producer is closed (no more logs) - // the receiver is lagging - while let Ok(log_item) = log_consumer.recv().await { - let resp = GetSubProcessStreamResponse { item: Some(log_item) }; - if tx.send(Ok(resp)).await.is_err() { - // receiver is gone - break; - } - } - }); + .subscribe() + }; - Ok(Response::new(ReceiverStream::new(rx))) + Ok(Response::new(self.spawn_log_forwarder(consumer, |item| { + GetSubProcessStreamResponse { item: Some(item) } + }))) } type GetPosixSignalsStreamStream = @@ -274,7 +288,7 @@ impl observe_service_server::ObserveService for ObserveService { #[cfg(test)] mod tests { - use super::ObserveService; + use super::*; use crate::logging::log_channel::LogChannel; use proto::observe::LogChannelType; @@ -387,4 +401,91 @@ mod tests { svc.sub_process_consumer_list.lock().await.clear(); } + + /// A lagged consumer resumes with the oldest item still in the ring. + #[tokio::test] + async fn lagged_consumer_continues() { + use tokio_stream::StreamExt as _; + + let aurae_logger = LogChannel::new("auraed"); + let svc = ObserveService::new(aurae_logger.clone(), (None, None, None)); + + let mut stream = + ::get_aurae_daemon_log_stream( + &svc, + Request::new(GetAuraeDaemonLogStreamRequest {}), + ) + .await + .expect("handler returned stream") + .into_inner(); + + const BURST: usize = 1024; + for i in 0..BURST { + aurae_logger.send(format!("msg {i:04}")); + } + + let first = tokio::time::timeout( + std::time::Duration::from_secs(2), + stream.next(), + ) + .await + .expect("did not receive first item before timeout") + .expect("stream ended unexpectedly") + .expect("stream item should be Ok"); + + let line = first.item.expect("LogItem present").line; + assert_eq!(line, format!("msg {:04}", BURST - LOG_STREAM_CAPACITY)); + } + + #[tokio::test] + async fn shutdown_ends_log_stream() { + use tokio_stream::StreamExt as _; + + let svc = + ObserveService::new(LogChannel::new("auraed"), (None, None, None)); + let mut stream = + ::get_aurae_daemon_log_stream( + &svc, + Request::new(GetAuraeDaemonLogStreamRequest {}), + ) + .await + .expect("handler returned stream") + .into_inner(); + + svc.shutdown_log_streams(); + + let item = tokio::time::timeout(Duration::from_secs(2), stream.next()) + .await + .expect("stream did not stop before timeout"); + assert!(item.is_none(), "stream should end without an error item"); + } + + #[tokio::test] + async fn shutdown_stops_backpressured_log_forwarder() { + let aurae_logger = LogChannel::new("auraed"); + let svc = ObserveService::new(aurae_logger.clone(), (None, None, None)); + let _stream = + ::get_aurae_daemon_log_stream( + &svc, + Request::new(GetAuraeDaemonLogStreamRequest {}), + ) + .await + .expect("handler returned stream") + .into_inner(); + + for i in 0..=LOG_STREAM_CAPACITY { + aurae_logger.send(format!("msg {i}")); + tokio::task::yield_now().await; + } + + svc.shutdown_log_streams(); + + tokio::time::timeout(Duration::from_secs(2), async { + while aurae_logger.sender().receiver_count() != 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("forwarder remained subscribed after shutdown"); + } }