enhancement(internal_logs source): Add option to rate limit internal_logs sources

Author: bruceg

changelog.d/internal_logs_broadcast_rate_limit.enhancement.md

@@ -0,0 +1,6 @@
+Add support for optionally applying rate limiting to the `internal_logs` source controlled by the
+`--internal-logs-source-rate-limit` CLI option and `VECTOR_INTERNAL_LOGS_SOURCE_RATE_LIMIT`
+environment variable. This provides the same rate limiting functionality as was available before
+version 0.51.1 but with a rate limit window separate from the console one.
+
+authors: bruceg

docs/DEVELOPING.md

@@ -264,7 +264,8 @@ to detect common problems.
 
 ### Disabling internal log rate limiting
 
-Vector rate limits its own internal logs by default (10-second windows). During development, you may want to see all log occurrences.
+Vector rate limits the console output of internal logs by default (10-second windows). During
+development, you may want to see all log occurrences.
 
 **Globally** (CLI flag or environment variable):
 
@@ -284,6 +285,10 @@ warn!(message = "Error occurred.", %error, internal_log_rate_limit = false);
 info!(message = "Processing batch.", batch_size, internal_log_rate_secs = 1);
 ```
 
+Note: The `internal_logs` source is _not_ rate limited by default. To enable rate limiting on all
+such sources, set the `--internal-logs-source-rate-limit` CLI flag or
+`VECTOR_INTERNAL_LOGS_SOURCE_RATE_LIMIT` environment variable to an integer number of seconds.
+
 ## Testing
 
 Testing is very important since Vector's primary design principle is reliability.

lib/tracing-limit/src/lib.rs

@@ -13,6 +13,10 @@
 //! - **3rd+ occurrences**: Silent until window expires
 //! - **After window**: Emits a summary of suppressed count, then next event normally
 //!
+//! Note: the suppressed-count summary and the resumption of normal emission are both
+//! triggered by the *next arriving event* after the window has elapsed, not by the
+//! window expiry itself. If the event stops firing, no summary is ever emitted.
+//!
 //! # Rate limit grouping
 //!
 //! Events are rate limited independently based on a combination of:
@@ -163,6 +167,7 @@ where
     /// - 1st occurrence: Emitted normally
     /// - 2nd occurrence: Shows "suppressing" warning
     /// - 3rd+ occurrences: Silent until window expires
+    /// - After window: Summary and next event emitted on next arrival (see module-level note)
     pub fn with_default_limit(mut self, internal_log_rate_limit: u64) -> Self {
         self.internal_log_rate_limit = internal_log_rate_limit;
         self

src/app.rs

@@ -212,6 +212,7 @@ impl Application {
             opts.root.log_format,
             opts.log_level(),
             opts.root.internal_log_rate_limit,
+            opts.root.internal_logs_source_rate_limit,
         );
 
         #[cfg(unix)]
@@ -658,14 +659,20 @@ pub async fn load_configs(
     Ok(config)
 }
 
-pub fn init_logging(color: bool, format: LogFormat, log_level: &str, rate: u64) {
+pub fn init_logging(
+    color: bool,
+    format: LogFormat,
+    log_level: &str,
+    rate: u64,
+    broadcast_rate_limit: Option<NonZeroU64>,
+) {
     let level = get_log_levels(log_level);
     let json = match format {
         LogFormat::Text => false,
         LogFormat::Json => true,
     };
 
-    trace::init(color, json, &level, rate);
+    trace::init(color, json, &level, rate, broadcast_rate_limit);
     debug!(
         message = "Internal log rate limit configured.",
         internal_log_rate_secs = rate,

src/cli.rs

@@ -193,7 +193,8 @@ pub struct RootOpts {
     /// This controls the time window for rate limiting Vector's own internal logs.
     /// Within each time window, the first occurrence of a log is emitted, the second
     /// shows a suppression warning, and subsequent occurrences are silent until the
-    /// window expires.
+    /// window expires. When the window expires and the log fires again, a summary of
+    /// the suppressed count is emitted followed by the log itself.
     ///
     /// Logs are grouped by their location in the code and the `component_id` field, so logs
     /// from different components are rate limited independently.
@@ -210,6 +211,16 @@ pub struct RootOpts {
     )]
     pub internal_log_rate_limit: u64,
 
+    /// Apply a rate limit (in seconds) to the broadcast channel that feeds all `internal_logs`
+    /// sources. When set, the first occurrence of a repeated log is emitted, the second shows a
+    /// suppression warning, and subsequent occurrences are silent until the window expires. When
+    /// the window expires and the log fires again, a summary of the suppressed count is emitted
+    /// followed by the log itself. Unset by default so that `internal_logs` consumers receive
+    /// every log event. This limit is independent of `--internal-log-rate-limit`, which only
+    /// applies to stdout/stderr output.
+    #[arg(long, env = "VECTOR_INTERNAL_LOGS_SOURCE_RATE_LIMIT")]
+    pub internal_logs_source_rate_limit: Option<NonZeroU64>,
+
     /// Set the duration in seconds to wait for graceful shutdown after SIGINT or SIGTERM are
     /// received. After the duration has passed, Vector will force shutdown. To never force
     /// shutdown, use `--no-graceful-shutdown-limit`.

src/sources/internal_logs.rs

@@ -238,7 +238,7 @@ mod tests {
     #[tokio::test]
     #[serial]
     async fn receives_logs() {
-        trace::init(false, false, "debug", 10);
+        trace::init(false, false, "debug", 10, None);
         trace::reset_early_buffer();
 
         assert_source_compliance(&SOURCE_TAGS, run_test()).await;
@@ -363,7 +363,7 @@ mod tests {
     #[tokio::test]
     #[serial]
     async fn registered_extra_span_field_is_captured() {
-        trace::init(false, false, "info", 10);
+        trace::init(false, false, "info", 10, None);
         trace::reset_early_buffer();
 
         let test_id: u8 = rand::random();
@@ -400,7 +400,7 @@ mod tests {
     #[tokio::test]
     #[serial]
     async fn repeated_logs_are_not_rate_limited() {
-        trace::init(false, false, "info", 10);
+        trace::init(false, false, "info", 10, None);
         trace::reset_early_buffer();
 
         let rx = start_source().await;

src/test_util/mod.rs

@@ -128,7 +128,7 @@ pub fn trace_init() {
 
     let levels = std::env::var("VECTOR_LOG").unwrap_or_else(|_| "error".to_string());
 
-    trace::init(color, false, &levels, 10);
+    trace::init(color, false, &levels, 10, None);
 
     // Initialize metrics as well
     vector_lib::metrics::init_test();

src/trace.rs

@@ -57,22 +57,41 @@ fn metrics_layer_enabled() -> bool {
     !matches!(std::env::var("DISABLE_INTERNAL_METRICS_TRACING_INTEGRATION"), Ok(x) if x == "true")
 }
 
-pub fn init(color: bool, json: bool, levels: &str, internal_log_rate_limit: u64) {
+pub fn init(
+    color: bool,
+    json: bool,
+    levels: &str,
+    internal_log_rate_limit: u64,
+    broadcast_rate_limit: Option<std::num::NonZeroU64>,
+) {
     let fmt_filter = tracing_subscriber::filter::Targets::from_str(levels).expect(
         "logging filter targets were not formatted correctly or did not specify a valid level",
     );
 
     let metrics_layer =
         metrics_layer_enabled().then(|| MetricsLayer::new().with_filter(LevelFilter::INFO));
 
-    // BroadcastLayer should NOT be rate limited because it feeds the internal_logs source,
-    // which users rely on to capture ALL internal Vector logs for debugging/monitoring.
-    // Console output (stdout/stderr) has its own separate rate limiting below.
-    let broadcast_layer = BroadcastLayer::new().with_filter(fmt_filter.clone());
+    // The broadcast layer feeds the internal_logs source. By default it is NOT rate limited so
+    // 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`.
+    //
+    // Two separate `Option<Layer>` values are used rather than a single branch because
+    // `RateLimitedLayer<BroadcastLayer<S>>` and `BroadcastLayer<S>` are distinct types.
+    // `tracing_subscriber` implements `Layer` for `Option<L>`, so exactly one of these will be
+    // `Some` and contribute an active layer while the other is a no-op `None`.
+    let rate_limited_broadcast = broadcast_rate_limit.map(|rate| {
+        RateLimitedLayer::new(BroadcastLayer::new())
+            .with_default_limit(rate.get())
+            .with_filter(fmt_filter.clone())
+    });
+    let unlimited_broadcast = broadcast_rate_limit
+        .is_none()
+        .then(|| BroadcastLayer::new().with_filter(fmt_filter.clone()));
 
     let subscriber = tracing_subscriber::registry()
         .with(metrics_layer)
-        .with(broadcast_layer);
+        .with(rate_limited_broadcast)
+        .with(unlimited_broadcast);
 
     #[cfg(feature = "tokio-console")]
     let subscriber = {
@@ -389,3 +408,110 @@ impl tracing::field::Visit for SpanFields {
         self.record(field, format!("{value:?}"));
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::{str::FromStr, time::Duration};
+
+    use serial_test::serial;
+    use tracing::info;
+    use tracing_subscriber::layer::SubscriberExt;
+
+    use futures::StreamExt as _;
+
+    use super::*;
+
+    /// Verifies that `RateLimitedLayer<BroadcastLayer>` — the configuration produced by
+    /// `init` when `broadcast_rate_limit` is `Some` — suppresses repeated identical log
+    /// events within the rate-limit window and emits a summary once the window expires.
+    ///
+    /// All `info!` calls share the same call site (they are inside the loop), so the
+    /// `RateLimitedLayer` treats them as one event for suppression purposes.
+    ///
+    /// The test uses `tracing::subscriber::with_default` rather than `init` to avoid
+    /// touching the process-global subscriber, and `spawn_blocking` so that the
+    /// synchronous `std::thread::sleep` — needed because `tracing-limit` uses
+    /// `std::time::Instant` when compiled as a library dependency — does not block the
+    /// async executor.
+    #[tokio::test]
+    #[serial]
+    async fn broadcast_rate_limits_repeated_messages() {
+        let trace_sub = TraceSubscription::subscribe();
+        // Disable early buffering so events flow directly to the broadcast channel
+        // rather than being held in the startup buffer.
+        stop_early_buffering();
+
+        let filter =
+            tracing_subscriber::filter::Targets::from_str("info").expect("valid filter string");
+        let subscriber = tracing_subscriber::registry().with(
+            RateLimitedLayer::new(BroadcastLayer::new())
+                .with_default_limit(1) // 1-second window
+                .with_filter(filter),
+        );
+
+        tokio::task::spawn_blocking(move || {
+            tracing::subscriber::with_default(subscriber, || {
+                for i in 0..4u32 {
+                    if i == 3 {
+                        // Wait for the 1-second window to expire so that the 4th call
+                        // triggers a summary message before being emitted normally.
+                        std::thread::sleep(Duration::from_millis(1100));
+                    }
+                    info!("Rate limited broadcast message.");
+                }
+            });
+        })
+        .await
+        .expect("blocking task panicked");
+
+        // All sends happened synchronously inside spawn_blocking above, so items are
+        // already in the broadcast ring buffer. We drain the stream until we have the
+        // 4 expected matching messages, with a generous timeout to guard against
+        // unexpected delays on heavily loaded machines.
+        //
+        // Limitation: the "suppressed N times" summary fires on the *next* arriving
+        // event after the window expires, not at window expiry itself.
+        const EXPECTED: usize = 4;
+        let mut stream = trace_sub.into_stream();
+        let messages: Vec<String> = tokio::time::timeout(Duration::from_secs(5), async {
+            let mut collected = Vec::with_capacity(EXPECTED);
+            loop {
+                let event = stream
+                    .next()
+                    .await
+                    .expect("broadcast stream ended unexpectedly");
+                if let Some(msg) = event.get("message") {
+                    let msg = msg.to_string_lossy().into_owned();
+                    if msg.contains("Rate limited broadcast message") {
+                        collected.push(msg);
+                        if collected.len() == EXPECTED {
+                            break;
+                        }
+                    }
+                }
+            }
+            collected
+        })
+        .await
+        .expect("timed out waiting for rate-limited broadcast messages");
+
+        // Expected sequence:
+        //   [0] First occurrence  → emitted normally.
+        //   [1] Second occurrence → suppression warning emitted, original dropped.
+        //   (Third occurrence is silently dropped; nothing appears in broadcast.)
+        //   [2] Fourth occurrence (after window) → summary "suppressed 2 times" emitted.
+        //   [3] Fourth occurrence → emitted normally after the window resets.
+        assert_eq!(messages[0], "Rate limited broadcast message.");
+        assert!(
+            messages[1].contains("is being suppressed to avoid flooding."),
+            "expected suppression warning, got: {}",
+            messages[1]
+        );
+        assert!(
+            messages[2].contains("has been suppressed 2 times."),
+            "expected summary, got: {}",
+            messages[2]
+        );
+        assert_eq!(messages[3], "Rate limited broadcast message.");
+    }
+}

website/cue/reference/cli.cue

@@ -199,6 +199,11 @@ cli: {
 			type:        "integer"
 			env_var:     "VECTOR_INTERNAL_LOG_RATE_LIMIT"
 		}
+		"internal-logs-source-rate-limit": {
+			description: env_vars.VECTOR_INTERNAL_LOGS_SOURCE_RATE_LIMIT.description
+			type:        "integer"
+			env_var:     "VECTOR_INTERNAL_LOGS_SOURCE_RATE_LIMIT"
+		}
 	}
 
 	options: _core_options
@@ -678,12 +683,31 @@ cli: {
 			}
 		}
 		VECTOR_INTERNAL_LOG_RATE_LIMIT: {
-			description: "Set the internal log rate limit. This limits Vector from emitting identical logs more than once over the given number of seconds."
+			description: """
+				Set the internal log rate limit in seconds. Within each time window, the first occurrence of a
+				log is emitted, the second shows a suppression warning, and subsequent occurrences are silent
+				until the window expires. When the window expires and the log fires again, a summary of the
+				suppressed count is emitted followed by the log itself.
+				"""
 			type: uint: {
 				default: 10
 				unit:    null
 			}
 		}
+		VECTOR_INTERNAL_LOGS_SOURCE_RATE_LIMIT: {
+			description: """
+				Apply a rate limit (in seconds) to the broadcast channel that feeds all `internal_logs` sources.
+				When set, the first occurrence of a repeated log is emitted, the second shows a suppression
+				warning, and subsequent occurrences are silent until the window expires. When the window expires
+				and the log fires again, a summary of the suppressed count is emitted followed by the log itself.
+				Unset by default so that `internal_logs` consumers receive every log event. This limit is
+				independent of `VECTOR_INTERNAL_LOG_RATE_LIMIT`, which only applies to stdout/stderr output.
+				"""
+			type: uint: {
+				default: null
+				unit:    "seconds"
+			}
+		}
 		VECTOR_GRACEFUL_SHUTDOWN_LIMIT_SECS: {
 			description: "Set the duration in seconds to wait for graceful shutdown after SIGINT or SIGTERM are received. After the duration has passed, Vector will force shutdown. To never force shutdown, use `--no-graceful-shutdown-limit`."
 			type: uint: {