enhancement(buffers): improve disk buffer lifecycle visibility

changelog.d/disk_buffer_orphan_safety.enhancement.md

@@ -0,0 +1,3 @@
+Vector now detects unreferenced disk buffers and reports disk-buffer lifecycle events during configuration reloads through logs and the `disk_buffer_lifecycle_events_total` counter.
+
+authors: Jansen-w

lib/vector-buffers/src/lib.rs

@@ -31,6 +31,7 @@ pub mod test;
 pub mod topology;
 
 pub(crate) mod variants;
+pub use variants::disk_v2::DiskBufferUsageSnapshot;
 
 /// `disk_v2`'s write-buffer size, re-exported under `test` so the harness can
 /// size payloads against the real value instead of hardcoding it.

lib/vector-buffers/src/topology/channel/sender.rs

@@ -10,7 +10,7 @@ use crate::{
     BufferInstrumentation, Bufferable, WhenFull,
     buffer_usage_data::BufferUsageHandle,
     internal_events::BufferSendDuration,
-    variants::disk_v2::{self, ProductionFilesystem, TryWriteOutcome},
+    variants::disk_v2::{self, DiskBufferUsageSnapshot, ProductionFilesystem, TryWriteOutcome},
 };
 
 /// Adapter for papering over various sender backends.
@@ -160,6 +160,16 @@ where
             Self::DiskV2(_) => None,
         }
     }
+
+    fn usage_snapshot(&self) -> Option<DiskBufferUsageSnapshot> {
+        match self {
+            Self::InMemory(_) => Some(DiskBufferUsageSnapshot {
+                event_count: 0,
+                byte_size: 0,
+            }),
+            Self::DiskV2(writer) => writer.try_lock().ok().map(|writer| writer.usage_snapshot()),
+        }
+    }
 }
 
 enum UsageAccounting {
@@ -277,6 +287,19 @@ impl<T: Bufferable> BufferSender<T> {
     pub fn with_custom_instrumentation(&mut self, instrumentation: impl BufferInstrumentation<T>) {
         self.custom_instrumentation = Some(Arc::new(instrumentation));
     }
+
+    /// Returns the aggregate approximate usage of all disk stages in this buffer topology.
+    ///
+    /// Returns `None` rather than waiting if any disk writer is currently busy.
+    pub fn disk_usage_snapshot(&self) -> Option<DiskBufferUsageSnapshot> {
+        let mut usage = self.base.usage_snapshot()?;
+        if let Some(overflow) = self.overflow.as_ref() {
+            let overflow_usage = overflow.disk_usage_snapshot()?;
+            usage.event_count = usage.event_count.saturating_add(overflow_usage.event_count);
+            usage.byte_size = usage.byte_size.saturating_add(overflow_usage.byte_size);
+        }
+        Some(usage)
+    }
 }
 
 impl<T: Bufferable> BufferSender<T> {
@@ -380,3 +403,45 @@ impl<T: Bufferable> BufferSender<T> {
         Ok(())
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::num::{NonZeroU64, NonZeroUsize};
+
+    use tempfile::tempdir;
+    use tracing::Span;
+
+    use super::SenderAdapter;
+    use crate::{BufferConfig, BufferType, MemoryBufferSize, WhenFull, test::SizedRecord};
+
+    #[tokio::test]
+    async fn busy_disk_stage_makes_diagnostic_snapshot_unavailable() {
+        let config = BufferConfig::Chained(vec![
+            BufferType::Memory {
+                size: MemoryBufferSize::MaxEvents(NonZeroUsize::new(1).unwrap()),
+                when_full: WhenFull::Overflow,
+            },
+            BufferType::DiskV2 {
+                max_size: NonZeroU64::new(268_435_488).unwrap(),
+                when_full: WhenFull::Block,
+            },
+        ]);
+        let data_dir = tempdir().unwrap();
+        let (sender, _receiver) = config
+            .build::<SizedRecord>(
+                Some(data_dir.path().to_path_buf()),
+                "busy".into(),
+                Span::none(),
+            )
+            .await
+            .unwrap();
+        let overflow = sender.overflow.as_ref().unwrap();
+        let writer = match &overflow.base {
+            SenderAdapter::DiskV2(writer) => writer,
+            SenderAdapter::InMemory(_) => unreachable!(),
+        };
+        let _guard = writer.lock().await;
+
+        assert_eq!(sender.disk_usage_snapshot(), None);
+    }
+}

lib/vector-buffers/src/variants/disk_v2/mod.rs

@@ -179,7 +179,7 @@ use std::{
 
 use async_trait::async_trait;
 use snafu::{ResultExt, Snafu};
-use tracing::debug;
+use tracing::{debug, info};
 use vector_common::finalization::Finalizable;
 
 mod backed_archive;
@@ -212,6 +212,15 @@ use crate::{
     },
 };
 
+/// An approximate view of unacknowledged records in a disk buffer.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct DiskBufferUsageSnapshot {
+    /// Number of events that have not been acknowledged.
+    pub event_count: u64,
+    /// Encoded size of records that have not been acknowledged.
+    pub byte_size: u64,
+}
+
 /// Error that occurred when creating/loading a disk buffer.
 #[derive(Debug, Snafu)]
 pub enum BufferError<T>
@@ -396,7 +405,8 @@ where
     usage_handle.set_buffer_limits(Some(max_size.get()), None);
 
     let buffer_path = get_disk_v2_data_dir_path(data_dir, id);
-    let builder = DiskBufferConfigBuilder::from_path(buffer_path).max_buffer_size(max_size.get());
+    let builder =
+        DiskBufferConfigBuilder::from_path(buffer_path.clone()).max_buffer_size(max_size.get());
     // Shrink the data-file size (and the matching record size) so files fill and
     // rotate constantly. That is what reaches the rare recovery paths the bug hides
     // in: reopening a file whose last write was cut short, and reusing a file number
@@ -412,9 +422,21 @@ where
         None => builder,
     };
     let config = builder.build()?;
-    Buffer::from_config(config, usage_handle)
+    let (writer, reader) = Buffer::from_config(config, usage_handle)
         .await
-        .map_err(Into::into)
+        .map_err(Box::<dyn Error + Send + Sync>::from)?;
+    let usage = writer.usage_snapshot();
+    if usage.event_count > 0 || usage.byte_size > 0 {
+        info!(
+            buffer_id = id,
+            buffer_dir = buffer_path.to_string_lossy().as_ref(),
+            buffered_events = usage.event_count,
+            buffered_bytes = usage.byte_size,
+            message = "Opened disk buffer containing buffered bytes.",
+        );
+    }
+
+    Ok((writer, reader))
 }
 
 pub(crate) fn get_disk_v2_data_dir_path(base_dir: &Path, buffer_id: &str) -> PathBuf {

lib/vector-buffers/src/variants/disk_v2/tests/basic.rs

@@ -236,16 +236,19 @@ async fn initial_size_correct_with_multievents() {
             drop(ledger);
 
             // Reopen the buffer with a full reader to verify the persisted data.
-            let (writer, mut reader, ledger, usage) =
+            let (mut writer, mut reader, ledger, usage) =
                 create_default_buffer_v2_with_usage::<_, MultiEventRecord>(&data_dir).await;
-            drop(writer);
+            let disk_usage = writer.usage_snapshot();
+            writer.close();
 
             // Make sure our usage data agrees with our expected event count and byte size:
             let snapshot = usage.snapshot();
             assert_eq!(expected_events as u64, snapshot.received_event_count);
             assert_eq!(expected_bytes as u64, snapshot.received_byte_size);
             assert_eq!(expected_events as u64, ledger.get_total_records());
             assert_eq!(expected_bytes, total_bytes_written);
+            assert_eq!(expected_events as u64, disk_usage.event_count);
+            assert_eq!(expected_bytes as u64, disk_usage.byte_size);
 
             // Make sure we can read all of the records we wrote, and recalculate some of these
             // values from the source:
@@ -260,6 +263,9 @@ async fn initial_size_correct_with_multievents() {
 
             assert_eq!(expected_events, total_record_events);
             assert_eq!(expected_records, total_records_read);
+            let disk_usage = writer.usage_snapshot();
+            assert_eq!(0, disk_usage.event_count);
+            assert_eq!(0, disk_usage.byte_size);
         }
     })
     .await;

lib/vector-buffers/src/variants/disk_v2/writer.rs

@@ -25,6 +25,7 @@ use snafu::{ResultExt, Snafu};
 use tokio::io::{AsyncWrite, AsyncWriteExt};
 
 use super::{
+    DiskBufferUsageSnapshot,
     common::{DiskBufferConfig, create_crc32c_hasher},
     io::Filesystem,
     ledger::Ledger,
@@ -912,6 +913,18 @@ where
         }
     }
 
+    /// Gets the current approximate usage of the buffer.
+    ///
+    /// Includes both usage published to the ledger and writes buffered locally by this writer. The
+    /// two ledger counters are sampled separately, so the result is approximate when reads are
+    /// being acknowledged concurrently.
+    pub fn usage_snapshot(&self) -> DiskBufferUsageSnapshot {
+        DiskBufferUsageSnapshot {
+            event_count: self.ledger.get_total_records() + self.unflushed_events,
+            byte_size: self.ledger.get_total_buffer_size() + self.unflushed_bytes,
+        }
+    }
+
     fn get_next_record_id(&mut self) -> u64 {
         self.next_record_id + self.unflushed_events
     }

lib/vector-common/src/internal_event/metric_name.rs

@@ -40,6 +40,7 @@ pub enum CounterName {
     ContainersWatchedTotal,
     DecoderBomRemovalsTotal,
     DecoderMalformedReplacementWarningsTotal,
+    DiskBufferLifecycleEventsTotal,
     DorisBytesLoadedTotal,
     DorisRowsFilteredTotal,
     DorisRowsLoadedTotal,
@@ -421,6 +422,7 @@ impl CounterName {
             Self::DecoderMalformedReplacementWarningsTotal => {
                 "decoder_malformed_replacement_warnings_total"
             }
+            Self::DiskBufferLifecycleEventsTotal => "disk_buffer_lifecycle_events_total",
             Self::DorisBytesLoadedTotal => "doris_bytes_loaded_total",
             Self::DorisRowsFilteredTotal => "doris_rows_filtered_total",
             Self::DorisRowsLoadedTotal => "doris_rows_loaded_total",

src/internal_events/disk_buffer.rs

@@ -0,0 +1,116 @@
+use std::path::{Path, PathBuf};
+
+use vector_lib::{
+    NamedInternalEvent,
+    buffers::DiskBufferUsageSnapshot,
+    config::ComponentKey,
+    counter,
+    internal_event::{CounterName, InternalEvent},
+};
+
+#[derive(Debug, NamedInternalEvent)]
+pub struct TopologyDiskBufferRemoved<'a> {
+    pub component_id: &'a ComponentKey,
+    pub buffer_dirs: &'a [PathBuf],
+    pub usage: Option<DiskBufferUsageSnapshot>,
+    pub drain_in_background: bool,
+}
+
+impl InternalEvent for TopologyDiskBufferRemoved<'_> {
+    fn emit(self) {
+        let message = match (self.usage.is_some(), self.drain_in_background) {
+            (true, true) => {
+                "Removing sink with attached disk buffer containing buffered bytes. Its current instance will continue draining in the background until the buffer is empty or Vector shuts down."
+            }
+            (true, false) => {
+                "Removing sink with attached disk buffer containing buffered bytes. Configuration reload will wait for its current instance to drain the buffer."
+            }
+            (false, true) => {
+                "Removing sink with attached disk buffer whose buffered-byte count is unavailable. Its current instance will continue draining in the background until the buffer is empty or Vector shuts down."
+            }
+            (false, false) => {
+                "Removing sink with attached disk buffer whose buffered-byte count is unavailable. Configuration reload will wait for its current instance to drain the buffer."
+            }
+        };
+        if let Some(usage) = self.usage {
+            info!(
+                component_id = %self.component_id,
+                buffer_dirs = ?self.buffer_dirs,
+                buffered_events = usage.event_count,
+                buffered_bytes = usage.byte_size,
+                drain_in_background = self.drain_in_background,
+                message,
+            );
+        } else {
+            info!(
+                component_id = %self.component_id,
+                buffer_dirs = ?self.buffer_dirs,
+                drain_in_background = self.drain_in_background,
+                usage_unavailable = true,
+                message,
+            );
+        }
+        counter!(
+            CounterName::DiskBufferLifecycleEventsTotal,
+            "event" => "removed",
+        )
+        .increment(1);
+    }
+}
+
+#[derive(Debug, NamedInternalEvent)]
+pub struct TopologyDiskBufferReplacementUnreferenced<'a> {
+    pub component_id: &'a ComponentKey,
+    pub buffer_dirs: &'a [PathBuf],
+    pub new_buffer_dirs: &'a [PathBuf],
+    pub usage: Option<DiskBufferUsageSnapshot>,
+}
+
+impl InternalEvent for TopologyDiskBufferReplacementUnreferenced<'_> {
+    fn emit(self) {
+        if let Some(usage) = self.usage {
+            warn!(
+                component_id = %self.component_id,
+                buffer_dirs = ?self.buffer_dirs,
+                new_buffer_dirs = ?self.new_buffer_dirs,
+                buffered_events = usage.event_count,
+                buffered_bytes = usage.byte_size,
+                message = "Replacement sink does not use the previous disk buffer. If reload succeeds, its buffered bytes will be left unreferenced.",
+            );
+        } else {
+            warn!(
+                component_id = %self.component_id,
+                buffer_dirs = ?self.buffer_dirs,
+                new_buffer_dirs = ?self.new_buffer_dirs,
+                usage_unavailable = true,
+                message = "Replacement sink does not use the previous disk buffer, and its buffered-byte count is unavailable. If reload succeeds, any buffered bytes will be left unreferenced.",
+            );
+        }
+        counter!(
+            CounterName::DiskBufferLifecycleEventsTotal,
+            "event" => "replacement_unreferenced",
+        )
+        .increment(1);
+    }
+}
+
+#[derive(Debug, NamedInternalEvent)]
+pub struct TopologyDiskBufferUnreferenced<'a> {
+    pub buffer_id: &'a Path,
+    pub buffer_dir: &'a Path,
+}
+
+impl InternalEvent for TopologyDiskBufferUnreferenced<'_> {
+    fn emit(self) {
+        warn!(
+            buffer_id = self.buffer_id.to_string_lossy().as_ref(),
+            buffer_dir = self.buffer_dir.to_string_lossy().as_ref(),
+            message = "Found disk buffer not referenced by the current configuration. It may still be draining through a detached sink instance; otherwise its buffered bytes are not being processed.",
+        );
+        counter!(
+            CounterName::DiskBufferLifecycleEventsTotal,
+            "event" => "unreferenced",
+        )
+        .increment(1);
+    }
+}

src/internal_events/mod.rs

@@ -43,6 +43,7 @@ mod datadog_traces;
 mod dedupe;
 #[cfg(feature = "sources-demo_logs")]
 mod demo_logs;
+mod disk_buffer;
 #[cfg(all(unix, feature = "sources-dnstap"))]
 mod dnstap;
 #[cfg(feature = "sources-docker_logs")]
@@ -203,6 +204,7 @@ pub(crate) use self::datadog_traces::*;
 pub(crate) use self::dedupe::*;
 #[cfg(feature = "sources-demo_logs")]
 pub(crate) use self::demo_logs::*;
+pub(crate) use self::disk_buffer::*;
 #[cfg(all(unix, feature = "sources-dnstap"))]
 pub(crate) use self::dnstap::*;
 #[cfg(feature = "sources-docker_logs")]

src/topology/mod.rs

@@ -12,6 +12,7 @@ pub mod schema;
 
 pub mod builder;
 mod controller;
+mod orphaned_disk_buffers;
 mod ready_arrays;
 mod running;
 mod task;