fix(internal_logs source): remove rate limit (#24218)

* fix(internal_logs source): remove rate limit

* add changelog

* Debug commit

* Fix validated - Revert "Debug commit"

This reverts commit c7b9ec9.

* add unit test

* fix check-events

Author: pront

changelog.d/internal_logs_rate_limiting.fix.md

@@ -0,0 +1,4 @@
+The `internal_logs` source now captures all internal Vector logs without rate limiting. Previously, repeated log messages were silently
+dropped.
+
+authors: pront

src/sources/internal_logs.rs

@@ -213,6 +213,8 @@ mod tests {
     use vector_lib::{event::Value, lookup::OwnedTargetPath};
     use vrl::value::kind::Collection;
 
+    use serial_test::serial;
+
     use super::*;
     use crate::{
         event::Event,
@@ -234,6 +236,7 @@ mod tests {
     // cases because `consume_early_buffer` (called within the
     // `start_source` helper) panics when called more than once.
     #[tokio::test]
+    #[serial]
     async fn receives_logs() {
         trace::init(false, false, "debug", 10);
         trace::reset_early_buffer();
@@ -341,6 +344,43 @@ mod tests {
         rx
     }
 
+    // NOTE: This test requires #[serial] because it directly interacts with global tracing state.
+    // This is a pre-existing limitation around tracing initialization in tests.
+    #[tokio::test]
+    #[serial]
+    async fn repeated_logs_are_not_rate_limited() {
+        trace::init(false, false, "info", 10);
+        trace::reset_early_buffer();
+
+        let rx = start_source().await;
+
+        // Generate 20 identical log messages with the same component_id
+        for _ in 0..20 {
+            info!(component_id = "test", "Repeated test message.");
+        }
+
+        sleep(Duration::from_millis(50)).await;
+        let events = collect_ready(rx).await;
+
+        // Filter to only our test messages
+        let test_events: Vec<_> = events
+            .iter()
+            .filter(|e| {
+                e.as_log()
+                    .get("message")
+                    .map(|m| m.to_string_lossy() == "Repeated test message.")
+                    .unwrap_or(false)
+            })
+            .collect();
+
+        // We should receive all 20 messages, no rate limiting.
+        assert_eq!(
+            test_events.len(),
+            20,
+            "internal_logs source should capture all repeated messages without rate limiting"
+        );
+    }
+
     #[test]
     fn output_schema_definition_vector_namespace() {
         let config = InternalLogsConfig::default();

src/trace.rs

@@ -64,9 +64,10 @@ pub fn init(color: bool, json: bool, levels: &str, internal_log_rate_limit: u64)
     let metrics_layer =
         metrics_layer_enabled().then(|| MetricsLayer::new().with_filter(LevelFilter::INFO));
 
-    let broadcast_layer = RateLimitedLayer::new(BroadcastLayer::new())
-        .with_default_limit(internal_log_rate_limit)
-        .with_filter(fmt_filter.clone());
+    // 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());
 
     let subscriber = tracing_subscriber::registry()
         .with(metrics_layer)