diff --git a/changelog.d/12399_internal_logs_level.enhancement.md b/changelog.d/12399_internal_logs_level.enhancement.md new file mode 100644 index 0000000000000..76585cbe5574a --- /dev/null +++ b/changelog.d/12399_internal_logs_level.enhancement.md @@ -0,0 +1,8 @@ +Add a `level` option to the `internal_logs` source that controls the maximum verbosity of log +events delivered to the source, decoupling it from the console log level. Previously, the source +only received the log events selected by `VECTOR_LOG`, `--verbose`, and `--quiet`; it now receives +log events at the configured `level` (defaulting to `info`) regardless of those startup options, +so, for example, logs can still be collected by the source when console logging is disabled with +`VECTOR_LOG=off`. + +authors: dekelpilli diff --git a/src/sources/internal_logs.rs b/src/sources/internal_logs.rs index b01097b8efa3d..0b5ccfa988979 100644 --- a/src/sources/internal_logs.rs +++ b/src/sources/internal_logs.rs @@ -1,10 +1,11 @@ use chrono::Utc; use futures::{StreamExt, stream}; +use tracing_subscriber::filter::LevelFilter; use vector_lib::{ codecs::BytesDeserializerConfig, config::{LegacyKey, LogNamespace, log_schema}, configurable::configurable_component, - lookup::{OwnedValuePath, lookup_v2::OptionalValuePath, owned_value_path, path}, + lookup::{OwnedValuePath, event_path, lookup_v2::OptionalValuePath, owned_value_path, path}, schema::Definition, }; use vrl::value::Kind; @@ -12,7 +13,7 @@ use vrl::value::Kind; use crate::{ SourceSender, config::{DataType, SourceConfig, SourceContext, SourceOutput}, - event::{EstimatedJsonEncodedSizeOf, Event}, + event::{EstimatedJsonEncodedSizeOf, Event, LogEvent}, internal_events::{InternalLogsBytesReceived, InternalLogsEventsReceived, StreamClosedError}, shutdown::ShutdownSignal, trace::TraceSubscription, @@ -43,12 +44,87 @@ pub struct InternalLogsConfig { #[serde(default = "default_pid_key")] pid_key: OptionalValuePath, + /// The maximum verbosity of log events to expose. + /// + /// Log events at this severity level and above are delivered to this source, + /// independently of the console log level Vector was started with (`VECTOR_LOG`, + /// `--verbose`, and `--quiet`). + /// + /// This setting takes effect once the configuration has been loaded. The few events emitted + /// before that, during early startup, are captured at the console log level (with a floor of + /// `info`), so exposing `debug` or `trace` events from early startup additionally requires + /// starting Vector with a verbose console log level (for example, `VECTOR_LOG=debug`). + #[serde(default = "default_level")] + level: LogLevel, + /// The namespace to use for logs. This overrides the global setting. #[configurable(metadata(docs::hidden))] #[serde(default)] log_namespace: Option, } +/// A log verbosity level. +#[configurable_component] +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +#[serde(rename_all = "lowercase")] +pub enum LogLevel { + /// Expose all log events. + Trace, + + /// Expose log events at the `DEBUG` level and above. + Debug, + + /// Expose log events at the `INFO` level and above. + Info, + + /// Expose log events at the `WARN` level and above. + Warn, + + /// Expose only log events at the `ERROR` level. + Error, +} + +impl LogLevel { + /// Ranks levels by verbosity, such that more verbose levels are greater. + const fn verbosity(self) -> u8 { + match self { + Self::Error => 0, + Self::Warn => 1, + Self::Info => 2, + Self::Debug => 3, + Self::Trace => 4, + } + } + + /// Parses the level recorded on internal log events at `metadata.level`. + fn from_event_level(level: &[u8]) -> Option { + match level { + b"TRACE" => Some(Self::Trace), + b"DEBUG" => Some(Self::Debug), + b"INFO" => Some(Self::Info), + b"WARN" => Some(Self::Warn), + b"ERROR" => Some(Self::Error), + _ => None, + } + } +} + +impl From for LevelFilter { + fn from(level: LogLevel) -> Self { + match level { + LogLevel::Trace => LevelFilter::TRACE, + LogLevel::Debug => LevelFilter::DEBUG, + LogLevel::Info => LevelFilter::INFO, + LogLevel::Warn => LevelFilter::WARN, + LogLevel::Error => LevelFilter::ERROR, + } + } +} + +const fn default_level() -> LogLevel { + LogLevel::Info +} + fn default_pid_key() -> OptionalValuePath { OptionalValuePath::from(owned_value_path!("pid")) } @@ -60,6 +136,7 @@ impl Default for InternalLogsConfig { InternalLogsConfig { host_key: None, pid_key: default_pid_key(), + level: default_level(), log_namespace: None, } } @@ -109,18 +186,30 @@ impl SourceConfig for InternalLogsConfig { .path; let pid_key = self.pid_key.clone().path; + // The broadcast channel feeding all `internal_logs` sources is filtered at the most + // verbose level registered by any running source; events below this source's own level + // are dropped from its stream in `run`. The registration lives as long as the source, so + // that removing or reconfiguring the source through a reload lowers the level again. + let broadcast_registration = crate::trace::register_broadcast_level(self.level.into()); + let subscription = TraceSubscription::subscribe(); + let level = self.level; let log_namespace = cx.log_namespace(self.log_namespace); - Ok(Box::pin(run( - host_key, - pid_key, - subscription, - cx.out, - cx.shutdown, - log_namespace, - ))) + Ok(Box::pin(async move { + let _broadcast_registration = broadcast_registration; + run( + host_key, + pid_key, + level, + subscription, + cx.out, + cx.shutdown, + log_namespace, + ) + .await + })) } fn outputs(&self, global_log_namespace: LogNamespace) -> Vec { @@ -138,9 +227,19 @@ impl SourceConfig for InternalLogsConfig { } } +/// Checks whether a log event is at `max_level` or above in severity. Events whose level is +/// missing or unrecognised are passed through. +fn within_max_level(log: &LogEvent, max_level: LogLevel) -> bool { + log.get(event_path!("metadata", "level")) + .and_then(|value| value.as_bytes()) + .and_then(|bytes| LogLevel::from_event_level(bytes)) + .is_none_or(|level| level.verbosity() <= max_level.verbosity()) +} + async fn run( host_key: Option, pid_key: Option, + max_level: LogLevel, mut subscription: TraceSubscription, mut out: SourceSender, shutdown: ShutdownSignal, @@ -160,6 +259,12 @@ async fn run( // any logs that don't break the loop, as that could cause an // infinite loop since it receives all such logs. while let Some(mut log) = rx.next().await { + // The broadcast channel is shared by all `internal_logs` sources and carries events for + // the most verbose of them, so events below this source's level must be dropped here. + if !within_max_level(&log, max_level) { + continue; + } + // TODO: Should this actually be in memory size? let byte_size = log.estimated_json_encoded_size_of().get(); let json_byte_size = log.estimated_json_encoded_size_of(); @@ -342,9 +447,15 @@ mod tests { } async fn start_source() -> impl Stream + Unpin { + start_source_with_config(InternalLogsConfig::default()).await + } + + async fn start_source_with_config( + config: InternalLogsConfig, + ) -> impl Stream + Unpin { let (tx, rx) = SourceSender::new_test(); - let source = InternalLogsConfig::default() + let source = config .build(SourceContext::new_test(tx, None)) .await .unwrap(); @@ -354,6 +465,62 @@ mod tests { rx } + // The console log level is set to `debug` here, so during the startup window debug events + // are captured into the early buffer, and the source's default `level` of `info` must drop + // the buffered debug event from its output. Once the source is running, the broadcast level + // falls to the source's `info`, keeping later debug events out as well. + #[tokio::test] + #[serial] + async fn drops_events_below_configured_level() { + trace::init(false, false, "debug", 10, None); + trace::reset_early_buffer(); + + let test_id: u8 = rand::random(); + debug!(message = "Buffered debug message.", %test_id); + + let rx = start_source().await; + + debug!(message = "Debug message.", %test_id); + info!(message = "Info message.", %test_id); + + sleep(Duration::from_millis(50)).await; + let mut events = collect_ready(rx).await; + let test_id = Value::from(test_id.to_string()); + events.retain(|event| event.as_log().get("test_id") == Some(&test_id)); + + assert_eq!(events.len(), 1); + assert_eq!(events[0].as_log()["message"], "Info message.".into()); + } + + // The console log level is set to `info` here, so trace events would normally not even be + // generated, but a source configured with `level: trace` must raise the broadcast level and + // receive them. + #[tokio::test] + #[serial] + async fn receives_events_below_console_level_when_configured() { + trace::init(false, false, "info", 10, None); + trace::reset_early_buffer(); + + let config = InternalLogsConfig { + level: LogLevel::Trace, + ..Default::default() + }; + let rx = start_source_with_config(config).await; + + let test_id: u8 = rand::random(); + trace!(message = "Trace message.", %test_id); + info!(message = "Info message.", %test_id); + + sleep(Duration::from_millis(50)).await; + let mut events = collect_ready(rx).await; + let test_id = Value::from(test_id.to_string()); + events.retain(|event| event.as_log().get("test_id") == Some(&test_id)); + + assert_eq!(events.len(), 2); + assert_eq!(events[0].as_log()["message"], "Trace message.".into()); + assert_eq!(events[1].as_log()["message"], "Info message.".into()); + } + // Register a span field through the same macro downstream crates would use, then verify // that emitting a log inside a span carrying that field captures it onto the log event. // This is the regression check for `register_extra_span_field!` extending the diff --git a/src/trace.rs b/src/trace.rs index be7dc1cb07fb7..43028c0a09ae3 100644 --- a/src/trace.rs +++ b/src/trace.rs @@ -5,7 +5,7 @@ use std::{ str::FromStr, sync::{ LazyLock, Mutex, MutexGuard, OnceLock, - atomic::{AtomicBool, Ordering}, + atomic::{AtomicBool, AtomicUsize, Ordering}, }, }; @@ -16,12 +16,12 @@ use tokio::sync::{ oneshot, }; use tokio_stream::wrappers::BroadcastStream; -use tracing::{Event, Subscriber}; +use tracing::{Event, Metadata, Subscriber}; use tracing_limit::RateLimitedLayer; use tracing_subscriber::{ - Layer, + Layer, Registry, filter::LevelFilter, - layer::{Context, SubscriberExt}, + layer::{Context, Filter, SubscriberExt}, registry::LookupSpan, util::SubscriberInitExt, }; @@ -53,6 +53,134 @@ static SUBSCRIBERS: Mutex>>>> = /// has been initialized. static SENDER: OnceLock> = OnceLock::new(); +/// BROADCAST_MAX_LEVEL caches the maximum verbosity of log events delivered to the broadcast +/// channel that feeds `internal_logs` sources, encoded as a usize (see [`level_filter_to_usize`]). +/// It is recomputed by [`recompute_broadcast_max_level`] whenever one of its inputs changes: the +/// levels registered by running `internal_logs` sources (via [`register_broadcast_level`]) and, +/// while early buffering is active, [`STARTUP_BROADCAST_LEVEL`]. This keeps the log events +/// delivered to `internal_logs` sources decoupled from the console log level (`VECTOR_LOG`, +/// `--verbose`, `--quiet`). +static BROADCAST_MAX_LEVEL: AtomicUsize = AtomicUsize::new(DEFAULT_BROADCAST_MAX_LEVEL); + +/// The default broadcast verbosity (3 = `INFO`). +const DEFAULT_BROADCAST_MAX_LEVEL: usize = 3; + +/// STARTUP_BROADCAST_LEVEL holds the verbosity of log events captured while early buffering is +/// active, before the configuration has been loaded and thus before the `level` of any +/// `internal_logs` source is known. It is seeded by [`init`] from the console log level, with a +/// floor of `INFO` so that an `internal_logs` source with the default `level` receives its +/// startup log events even when the console is quieter. +static STARTUP_BROADCAST_LEVEL: AtomicUsize = AtomicUsize::new(DEFAULT_BROADCAST_MAX_LEVEL); + +/// BROADCAST_LEVEL_REGISTRY counts the running `internal_logs` sources registered at each encoded +/// level, updated by [`register_broadcast_level`] and [`BroadcastLevelRegistration`]'s `Drop`. +static BROADCAST_LEVEL_REGISTRY: Mutex<[usize; 6]> = Mutex::new([0; 6]); + +/// Encodes a [`LevelFilter`] as a usize such that greater values are more verbose, allowing +/// levels to be compared numerically and stored atomically. +const fn level_filter_to_usize(level: LevelFilter) -> usize { + match level.into_level() { + None => 0, + Some(tracing::Level::ERROR) => 1, + Some(tracing::Level::WARN) => 2, + Some(tracing::Level::INFO) => 3, + Some(tracing::Level::DEBUG) => 4, + Some(tracing::Level::TRACE) => 5, + } +} + +const fn usize_to_level_filter(level: usize) -> LevelFilter { + match level { + 0 => LevelFilter::OFF, + 1 => LevelFilter::ERROR, + 2 => LevelFilter::WARN, + 3 => LevelFilter::INFO, + 4 => LevelFilter::DEBUG, + _ => LevelFilter::TRACE, + } +} + +fn broadcast_max_level() -> LevelFilter { + usize_to_level_filter(BROADCAST_MAX_LEVEL.load(Ordering::Relaxed)) +} + +fn get_broadcast_level_registry() -> MutexGuard<'static, [usize; 6]> { + BROADCAST_LEVEL_REGISTRY + .lock() + .expect("Couldn't acquire lock on broadcast level registry") +} + +/// Recomputes [`BROADCAST_MAX_LEVEL`] from the levels registered by running `internal_logs` +/// sources and, while early buffering is active, [`STARTUP_BROADCAST_LEVEL`]. With no registered +/// sources and early buffering stopped, the level is `OFF` and the broadcast layer is disabled +/// entirely. +/// +/// Rebuilding the callsite interest cache prompts `tracing` to re-evaluate which callsites are +/// enabled (including the global max level hint), so a change takes effect even though the +/// subscriber was installed earlier. +fn recompute_broadcast_max_level(registry: &[usize; 6]) { + let registered = registry + .iter() + .rposition(|&sources| sources > 0) + .unwrap_or(0); + let startup = if SHOULD_BUFFER.load(Ordering::Acquire) { + STARTUP_BROADCAST_LEVEL.load(Ordering::Relaxed) + } else { + 0 + }; + let new = registered.max(startup); + let old = BROADCAST_MAX_LEVEL.swap(new, Ordering::SeqCst); + if new != old { + tracing_core::callsite::rebuild_interest_cache(); + } +} + +/// Registers a verbosity level of log events to deliver to `internal_logs` sources, for the +/// lifetime of the returned registration. +/// +/// The broadcast channel is shared by all `internal_logs` sources, so it carries events for the +/// most verbose of them, and each source is responsible for dropping events below its own +/// configured level. Dropping the registration — when its source is stopped or removed by a +/// configuration reload — recomputes the level, so the process does not keep paying for a +/// verbosity that no running source asks for. +pub fn register_broadcast_level(level: LevelFilter) -> BroadcastLevelRegistration { + let level = level_filter_to_usize(level); + let mut registry = get_broadcast_level_registry(); + registry[level] += 1; + recompute_broadcast_max_level(®istry); + BroadcastLevelRegistration { level } +} + +/// A running `internal_logs` source's interest in log events at a given level, created by +/// [`register_broadcast_level`]. Dropping it deregisters the level and recomputes the verbosity +/// of the broadcast channel. +#[derive(Debug)] +pub struct BroadcastLevelRegistration { + level: usize, +} + +impl Drop for BroadcastLevelRegistration { + fn drop(&mut self) { + let mut registry = get_broadcast_level_registry(); + registry[self.level] = registry[self.level].saturating_sub(1); + recompute_broadcast_max_level(®istry); + } +} + +/// A level filter for the broadcast layer that reads [`BROADCAST_MAX_LEVEL`] on every check, +/// so that it can be raised at runtime, after the subscriber has been installed. +struct BroadcastFilter; + +impl Filter for BroadcastFilter { + fn enabled(&self, metadata: &Metadata<'_>, _cx: &Context<'_, S>) -> bool { + broadcast_max_level() >= *metadata.level() + } + + fn max_level_hint(&self) -> Option { + Some(broadcast_max_level()) + } +} + fn metrics_layer_enabled() -> bool { !matches!(std::env::var("DISABLE_INTERNAL_METRICS_TRACING_INTEGRATION"), Ok(x) if x == "true") } @@ -68,6 +196,20 @@ pub fn init( "logging filter targets were not formatted correctly or did not specify a valid level", ); + // Seed the verbosity of the early-buffering window: the `level` of any `internal_logs` + // sources is not known until the configuration has been loaded, so startup log events are + // captured at the console log level, with a floor of `INFO` to match the default + // `internal_logs` level. Once early buffering stops, only registered source levels apply. + let startup_level = Filter::::max_level_hint(&fmt_filter) + .map_or(level_filter_to_usize(LevelFilter::TRACE), |hint| { + level_filter_to_usize(hint) + }); + STARTUP_BROADCAST_LEVEL.store( + startup_level.max(DEFAULT_BROADCAST_MAX_LEVEL), + Ordering::Relaxed, + ); + recompute_broadcast_max_level(&get_broadcast_level_registry()); + let metrics_layer = metrics_layer_enabled().then(|| MetricsLayer::new().with_filter(LevelFilter::INFO)); @@ -75,6 +217,11 @@ pub fn init( // that users can capture ALL internal Vector logs for debugging/monitoring. Rate limiting can // be opted into by passing `Some(rate)` for `broadcast_rate_limit`. // + // It is filtered by `BroadcastFilter` rather than the console filter so that the log events + // delivered to `internal_logs` sources are decoupled from the console log level: each such + // source declares its own `level` (defaulting to `INFO`) and registers it for its lifetime + // when it is built. + // // Two separate `Option` values are used rather than a single branch because // `RateLimitedLayer>` and `BroadcastLayer` are distinct types. // `tracing_subscriber` implements `Layer` for `Option`, which allows the unused layer to be @@ -82,11 +229,11 @@ pub fn init( let rate_limited_broadcast = broadcast_rate_limit.map(|rate| { RateLimitedLayer::new(BroadcastLayer::new()) .with_default_limit(rate.get()) - .with_filter(fmt_filter.clone()) + .with_filter(BroadcastFilter) }); let unlimited_broadcast = broadcast_rate_limit .is_none() - .then(|| BroadcastLayer::new().with_filter(fmt_filter.clone())); + .then(|| BroadcastLayer::new().with_filter(BroadcastFilter)); debug_assert_ne!( rate_limited_broadcast.is_some(), unlimited_broadcast.is_some() @@ -141,9 +288,16 @@ pub fn init( } } +/// Resets the early-buffering machinery to its process-start state, so that each test can +/// exercise the full startup lifecycle regardless of which tests ran in the same process before +/// it. #[cfg(test)] pub fn reset_early_buffer() -> Option> { - get_early_buffer().replace(Vec::new()) + *get_trace_subscriber_list() = Some(Vec::new()); + SHOULD_BUFFER.store(true, Ordering::Release); + let events = get_early_buffer().replace(Vec::new()); + recompute_broadcast_max_level(&get_broadcast_level_registry()); + events } /// Gets a mutable reference to the early buffer. @@ -264,6 +418,10 @@ pub fn stop_early_buffering() { _ = subscriber_tx.send(buffered_events.clone()); } } + + // The startup level only applies while early buffering is active; from here on, the broadcast + // level reflects only the levels registered by running `internal_logs` sources. + recompute_broadcast_max_level(&get_broadcast_level_registry()); } /// A subscription to the log events flowing in via `tracing`, in the Vector native format. @@ -518,4 +676,38 @@ mod tests { ); assert_eq!(messages[3], "Rate limited broadcast message."); } + + /// The broadcast level must track the levels of the registered `internal_logs` sources — + /// dropping down again as registrations are dropped, and all the way to `OFF` when none + /// remain — plus the startup level while early buffering is active. + #[test] + #[serial] + fn broadcast_level_follows_registrations() { + // Simulate startup being complete, so that only registrations count. + SHOULD_BUFFER.store(false, Ordering::Release); + recompute_broadcast_max_level(&get_broadcast_level_registry()); + assert_eq!(broadcast_max_level(), LevelFilter::OFF); + + let info_registration = register_broadcast_level(LevelFilter::INFO); + assert_eq!(broadcast_max_level(), LevelFilter::INFO); + + let trace_registration = register_broadcast_level(LevelFilter::TRACE); + assert_eq!(broadcast_max_level(), LevelFilter::TRACE); + + drop(trace_registration); + assert_eq!(broadcast_max_level(), LevelFilter::INFO); + + drop(info_registration); + assert_eq!(broadcast_max_level(), LevelFilter::OFF); + + // While early buffering is active, the startup level applies even with no registrations. + STARTUP_BROADCAST_LEVEL.store(level_filter_to_usize(LevelFilter::DEBUG), Ordering::Relaxed); + SHOULD_BUFFER.store(true, Ordering::Release); + recompute_broadcast_max_level(&get_broadcast_level_registry()); + assert_eq!(broadcast_max_level(), LevelFilter::DEBUG); + + // Leave the process-global state as a fresh test expects it. + STARTUP_BROADCAST_LEVEL.store(DEFAULT_BROADCAST_MAX_LEVEL, Ordering::Relaxed); + recompute_broadcast_max_level(&get_broadcast_level_registry()); + } } diff --git a/website/cue/reference/components/sources/generated/internal_logs.cue b/website/cue/reference/components/sources/generated/internal_logs.cue index f8bbda59b7ef9..211ed697fe3c7 100644 --- a/website/cue/reference/components/sources/generated/internal_logs.cue +++ b/website/cue/reference/components/sources/generated/internal_logs.cue @@ -14,6 +14,31 @@ generated: components: sources: internal_logs: configuration: { required: false type: string: {} } + level: { + description: """ + The maximum verbosity of log events to expose. + + Log events at this severity level and above are delivered to this source, + independently of the console log level Vector was started with (`VECTOR_LOG`, + `--verbose`, and `--quiet`). + + This setting takes effect once the configuration has been loaded. The few events emitted + before that, during early startup, are captured at the console log level (with a floor of + `info`), so exposing `debug` or `trace` events from early startup additionally requires + starting Vector with a verbose console log level (for example, `VECTOR_LOG=debug`). + """ + required: false + type: string: { + default: "info" + enum: { + debug: "Expose log events at the `DEBUG` level and above." + error: "Expose only log events at the `ERROR` level." + info: "Expose log events at the `INFO` level and above." + trace: "Expose all log events." + warn: "Expose log events at the `WARN` level and above." + } + } + } pid_key: { description: """ Overrides the name of the log field used to add the current process ID to each event. diff --git a/website/cue/reference/components/sources/internal_logs.cue b/website/cue/reference/components/sources/internal_logs.cue index f477bbbbf6d4c..678531b43ce57 100644 --- a/website/cue/reference/components/sources/internal_logs.cue +++ b/website/cue/reference/components/sources/internal_logs.cue @@ -123,11 +123,19 @@ components: sources: internal_logs: { how_it_works: { limited_logs: { - title: "Logs are limited by startup options" + title: "Logs are limited by the `level` option" body: """ - At startup, the selection of log messages generated by Vector is determined by a - combination of the `VECTOR_LOG` environment variable and the `--quiet` and `--verbose` - command-line options. The `internal_logs` source only receives logs that are generated by these options. + The selection of log messages delivered to this source is determined by its `level` + option, which defaults to `info`. It is independent of the console log level set at + startup by the `VECTOR_LOG` environment variable and the `--quiet` and `--verbose` + command-line options, so, for example, logs can be collected by this source even when + console logging is disabled entirely with `VECTOR_LOG=off`. + + The one exception is early startup: the handful of log events emitted before the + configuration has been loaded are captured at the console log level, with a floor of + `info`. Exposing `debug` or `trace` events from early startup therefore additionally + requires starting Vector with a verbose console log level (for example, + `VECTOR_LOG=debug`). """ } }