enhancement(buffers): improve disk buffer lifecycle visibility

changelog.d/disk_buffer_orphan_safety.enhancement.md

@@ -0,0 +1,7 @@
+Vector now reports disk buffer lifecycle events during startup and configuration reloads. Each event is logged and increments the `topology_disk_buffer_lifecycle_events_total` counter with an `event` tag:
+
+- `removed`: A removed sink still had data in its disk buffer, or Vector could not read how much data it held. The log says whether the sink's current instance keeps draining the buffer in the background or whether the reload waits for it to finish.
+- `replacement_unreferenced`: A changed sink no longer uses its previous disk buffer. If the reload succeeds, the new sink will not read any data left in the old buffer.
+- `unreferenced`: A disk buffer in the data directory is not used by the current configuration. It may still be draining through a detached sink instance. If not, any data it holds is not being processed.
+
+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

@@ -82,6 +82,7 @@ pub enum CounterName {
     StoppedTotal,
     TagCardinalityUntrackedEventsTotal,
     TagValueLimitExceededTotal,
+    TopologyDiskBufferLifecycleEventsTotal,
     ValueLimitReachedTotal,
     WebsocketBytesSentTotal,
     WebsocketMessagesSentTotal,
@@ -465,6 +466,9 @@ impl CounterName {
             Self::StoppedTotal => "stopped_total",
             Self::TagCardinalityUntrackedEventsTotal => "tag_cardinality_untracked_events_total",
             Self::TagValueLimitExceededTotal => "tag_value_limit_exceeded_total",
+            Self::TopologyDiskBufferLifecycleEventsTotal => {
+                "topology_disk_buffer_lifecycle_events_total"
+            }
             Self::ValueLimitReachedTotal => "value_limit_reached_total",
             Self::WebsocketBytesSentTotal => "websocket_bytes_sent_total",
             Self::WebsocketMessagesSentTotal => "websocket_messages_sent_total",

src/internal_events/disk_buffer.rs

@@ -0,0 +1,123 @@
+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. A disk buffer writer was busy writing or waiting for free space, so the buffer's usage could not be read. 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. A disk buffer writer was busy writing or waiting for free space, so the buffer's usage could not be read. 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::TopologyDiskBufferLifecycleEventsTotal,
+            "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) {
+        match self.usage {
+            Some(usage) if usage.event_count > 0 || usage.byte_size > 0 => 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.",
+            ),
+            Some(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. No buffered bytes were observed, but additional writes may still be in flight until the old input is detached.",
+            ),
+            None => 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. A disk buffer writer was busy writing or waiting for free space, so the buffer's usage could not be read. If reload succeeds, any buffered bytes will be left unreferenced.",
+            ),
+        }
+        counter!(
+            CounterName::TopologyDiskBufferLifecycleEventsTotal,
+            "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. If not, any buffered bytes it holds are not being processed.",
+        );
+        counter!(
+            CounterName::TopologyDiskBufferLifecycleEventsTotal,
+            "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;