From 21d6caf005222230057aeb0d62ff5bc0672269d1 Mon Sep 17 00:00:00 2001 From: Tommy McCormick Date: Sat, 1 Aug 2026 12:02:27 -0400 Subject: [PATCH 1/2] fix(ebpf): rearm perf buffer readiness after draining Auraed polls each per-CPU eBPF perf buffer through Tokio AsyncFd. After a buffer became readable, the reader drained it but left AsyncFd readiness cached. Empty reads then completed immediately and the reader tasks consumed the Tokio worker pool. This prevented gRPC startup, cell requests, and signal handling from making progress. Clear readiness after an empty read so the task waits for the next kernel notification. Keep readiness set after a non-empty read so busy buffers continue draining. --- auraed/src/ebpf/perf_buffer_reader.rs | 7 +++++++ 1 file changed, 7 insertions(+) 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!", From e9eaa6ff1f66fd7628882ee4c57f5169df39d61e Mon Sep 17 00:00:00 2001 From: Tommy McCormick Date: Sun, 3 May 2026 13:28:12 -0400 Subject: [PATCH 2/2] fix(observe): stream daemon logs through tracing Use one LogChannel for the tracing subscriber and ObserveService. Stop log forwarders during graceful shutdown. --- Cargo.lock | 98 ++------- auraed/Cargo.toml | 4 +- auraed/src/graceful_shutdown.rs | 9 +- auraed/src/init/logging.rs | 92 +++++--- auraed/src/init/mod.rs | 21 +- .../system_runtimes/cell_system_runtime.rs | 4 +- .../container_system_runtime.rs | 4 +- .../system_runtimes/daemon_system_runtime.rs | 4 +- auraed/src/init/system_runtimes/mod.rs | 2 + .../system_runtimes/pid1_system_runtime.rs | 4 +- auraed/src/lib.rs | 19 +- auraed/src/logging/broadcast_writer.rs | 202 +++++++++++++++++ auraed/src/logging/log_channel.rs | 34 ++- auraed/src/logging/mod.rs | 4 +- auraed/src/logging/stream_logger.rs | 56 ----- auraed/src/observe/observe_service.rs | 203 +++++++++++++----- 16 files changed, 495 insertions(+), 265 deletions(-) create mode 100644 auraed/src/logging/broadcast_writer.rs delete mode 100644 auraed/src/logging/stream_logger.rs 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/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"); + } }