enhancement(buffers): improve disk buffer lifecycle visibility

changelog.d/disk_buffer_orphan_safety.enhancement.md

@@ -0,0 +1,3 @@
+Vector now detects orphaned disk buffers and provides lifecycle visibility when disk-buffered sinks continue draining in the background during configuration reloads.
+
+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

@@ -14,7 +14,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.
@@ -164,6 +164,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 {
@@ -278,6 +288,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> {
@@ -381,3 +404,45 @@ impl<T: Bufferable> BufferSender<T> {
         Ok(())
     }
 }
+
+#[cfg(test)]
+mod usage_snapshot_tests {
+    use std::num::{NonZeroU64, NonZeroUsize};
+
+    use temp_dir::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::new().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/ledger.rs

@@ -564,6 +564,11 @@ where
             .increment_dropped_event_count_and_byte_size(0, total_record_size, false);
     }
 
+    /// Removes an unread data file tail that could not be interpreted as a record.
+    pub fn track_abandoned_tail_bytes(&self, total_size: u64) {
+        self.decrement_total_buffer_size(total_size);
+    }
+
     /// Marks the writer as finished.
     ///
     /// If the writer was not yet marked done, `false` is returned.  Otherwise, `true` is returned,

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 unread records in a disk buffer.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct DiskBufferUsageSnapshot {
+    /// Number of unread events in the buffer.
+    pub event_count: u64,
+    /// Encoded size of unread records in the buffer.
+    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(),
+            accounted_events = usage.event_count,
+            accounted_bytes = usage.byte_size,
+            message = "Opened disk buffer with remaining accounted data; resuming processing.",
+        );
+    }
+
+    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/reader.rs

@@ -598,8 +598,7 @@ where
                 abandoned_tail_bytes,
                 "Data file tail was abandoned after bad read. Adjusting buffer size to compensate.",
             );
-            self.ledger
-                .decrement_total_buffer_size(abandoned_tail_bytes);
+            self.ledger.track_abandoned_tail_bytes(abandoned_tail_bytes);
         }
 
         Ok(())

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

@@ -913,6 +913,18 @@ where
         }
     }
 
+    /// Gets the current published usage of the buffer.
+    ///
+    /// Writes are included after they are flushed and reads after they are acknowledged, matching
+    /// the ledger's existing published-data semantics. The two counters are sampled separately, so
+    /// the result is approximate when reads and writes are being published concurrently.
+    pub fn usage_snapshot(&self) -> super::DiskBufferUsageSnapshot {
+        super::DiskBufferUsageSnapshot {
+            event_count: self.ledger.get_total_records(),
+            byte_size: self.ledger.get_total_buffer_size(),
+        }
+    }
+
     fn get_next_record_id(&mut self) -> u64 {
         self.next_record_id + self.unflushed_events
     }

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;