fix(internal_logs source): report broadcast lag drops in component_discarded_events_total (#25218)

Author: thomasqueirozb

changelog.d/24220_internal_logs_drops.fix.md

@@ -0,0 +1,3 @@
+Fixed the `internal_logs` source silently dropping events under high load. Events dropped when the underlying broadcast receiver lags now increment the standard `component_discarded_events_total{intentional="false"}` metric so the loss is observable.
+
+authors: thomasqueirozb

src/sources/internal_logs.rs

@@ -158,7 +158,9 @@ async fn run(
 
     // Note: This loop, or anything called within it, MUST NOT generate
     // any logs that don't break the loop, as that could cause an
-    // infinite loop since it receives all such logs.
+    // infinite loop since it receives all such logs. Broadcast lag is
+    // reported by `TraceSubscription::into_stream` itself, after an event
+    // has been received, so its log cannot lag the receiver again.
     while let Some(mut log) = rx.next().await {
         // TODO: Should this actually be in memory size?
         let byte_size = log.estimated_json_encoded_size_of().get();
@@ -502,4 +504,99 @@ mod tests {
 
         assert_eq!(definitions, Some(expected_definition))
     }
+
+    // Verify that broadcast lag is surfaced via `component_discarded_events_total` rather
+    // than being silently swallowed.
+    //
+    // Strategy: run inside a single-threaded tokio runtime (the default for `#[tokio::test]`).
+    // While the current task holds the CPU without yielding, no other tokio tasks are scheduled.
+    // We flood the broadcast channel (capacity 99) with more events than it can hold. The source
+    // task cannot poll between emits, so the broadcast overflows and records a lag count. After
+    // we yield, the source observes the lag and emits `ComponentEventsDropped`.
+    #[tokio::test]
+    #[serial]
+    async fn broadcast_lag_increments_discarded_metric() {
+        trace::init(false, false, "error", 10, None);
+        vector_lib::metrics::init_test();
+        trace::reset_early_buffer();
+
+        // The downstream receiver is kept alive but never polled, so the source does not stop
+        // on a closed output and no events are consumed downstream before we check the metric.
+        let (tx, _rx) = SourceSender::new_test();
+        let source = InternalLogsConfig::default()
+            .build(SourceContext::new_test(tx, None))
+            .await
+            .unwrap();
+        tokio::spawn(source);
+
+        // Yield so the source task subscribes and starts polling the broadcast, then stop early
+        // buffering so new events go to the live broadcast.
+        tokio::task::yield_now().await;
+        trace::stop_early_buffering();
+        tokio::task::yield_now().await;
+
+        let controller =
+            vector_lib::metrics::Controller::get().expect("metrics controller must be initialized");
+        controller.reset();
+
+        // Emit more events than the broadcast capacity (99) without yielding. In a
+        // single-threaded runtime this guarantees the source task cannot poll between emits,
+        // so the broadcast overflows and accumulates a lag count.
+        for i in 0usize..200 {
+            error!(message = "Broadcast lag test.", i);
+        }
+
+        // Yield enough times for the source task to observe the lag and emit
+        // `ComponentEventsDropped`.
+        for _ in 0..20 {
+            tokio::task::yield_now().await;
+        }
+
+        let discarded_any = controller
+            .capture_metrics()
+            .into_iter()
+            .any(|m| m.name() == "component_discarded_events_total");
+
+        assert!(
+            discarded_any,
+            "expected component_discarded_events_total to be emitted when broadcast lags"
+        );
+    }
+
+    // After a lag, the broadcast receiver points at the oldest slot of a full buffer. Any log
+    // emitted before the next receive, such as the `ComponentEventsDropped` error, overwrites
+    // that slot and lags the receiver again. Verify the source still delivers the newest event of
+    // the burst instead of repeating that cycle.
+    #[tokio::test]
+    #[serial]
+    async fn recovers_after_broadcast_lag() {
+        trace::init(false, false, "error", 10, None);
+        trace::reset_early_buffer();
+
+        let mut rx = start_source().await;
+
+        // Overflow the broadcast (capacity 99) without yielding, so the source lags on its next
+        // receive. The last event is the newest one retained by the broadcast.
+        for i in 0usize..200 {
+            error!(message = "Broadcast lag test.", i);
+        }
+        error!(message = "Last event after lag.");
+
+        let last = Value::from("Last event after lag.");
+        let delivered = tokio::time::timeout(Duration::from_secs(5), async {
+            while let Some(event) = rx.next().await {
+                if event.as_log().get(event_path!("message")) == Some(&last) {
+                    return true;
+                }
+            }
+            false
+        })
+        .await
+        .unwrap_or(false);
+
+        assert!(
+            delivered,
+            "source did not deliver events after broadcast lag"
+        );
+    }
 }

src/trace.rs

@@ -16,7 +16,7 @@ use tokio::sync::{
     broadcast::{self, Receiver, Sender},
     oneshot,
 };
-use tokio_stream::wrappers::BroadcastStream;
+use tokio_stream::wrappers::{BroadcastStream, errors::BroadcastStreamRecvError};
 use tracing::{Event, Subscriber};
 use tracing_limit::RateLimitedLayer;
 use tracing_subscriber::{
@@ -27,8 +27,11 @@ use tracing_subscriber::{
     util::SubscriberInitExt,
 };
 pub use tracing_tower::{InstrumentableService, InstrumentedService};
-use vector_lib::SpanField;
-use vector_lib::lookup::event_path;
+use vector_lib::{
+    SpanField,
+    internal_event::{ComponentEventsDropped, UNINTENTIONAL},
+    lookup::event_path,
+};
 use vrl::value::Value;
 
 use crate::event::LogEvent;
@@ -308,10 +311,34 @@ impl TraceSubscription {
     }
 
     /// Converts this subscription into a raw stream of log events.
+    ///
+    /// Events dropped because the underlying broadcast receiver lagged are reported via
+    /// `ComponentEventsDropped` when the next event is received, not when the lag is detected.
+    /// After a lag, the receiver points at the oldest slot of a full buffer, so any log emitted
+    /// before the next receive (including the `ComponentEventsDropped` error itself) overwrites
+    /// that slot and lags the receiver again, indefinitely. Once an event has been received, that
+    /// slot is consumed and the drop can be logged safely.
     pub fn into_stream(self) -> impl Stream<Item = LogEvent> + Unpin {
-        // We ignore errors because the only error we get is when the broadcast receiver lags, and there's nothing we
-        // can actually do about that so there's no reason to force callers to even deal with it.
-        BroadcastStream::new(self.trace_rx).filter_map(|event| ready(event.ok()))
+        BroadcastStream::new(self.trace_rx)
+            .scan(0u64, |dropped, event| {
+                ready(Some(match event {
+                    Ok(log) => {
+                        let count = std::mem::take(dropped);
+                        if count > 0 {
+                            emit!(ComponentEventsDropped::<UNINTENTIONAL> {
+                                count: usize::try_from(count).unwrap_or(usize::MAX),
+                                reason: "Internal logs broadcast receiver lagged.",
+                            });
+                        }
+                        Some(log)
+                    }
+                    Err(BroadcastStreamRecvError::Lagged(n)) => {
+                        *dropped += n;
+                        None
+                    }
+                }))
+            })
+            .filter_map(ready)
     }
 }