Reorganize object store metrics #5821

rdettai · 2025-06-26T15:14:08Z

Description

Closes #5799

Rationalize object store metrics. Be more exhaustive in request and error recording.

Important: This is a breaking change as some metrics are renamed.

How was this PR tested?

TODO: add tests to the metrics wrappers

quickwit/quickwit-storage/src/metrics.rs

quickwit/quickwit-storage/src/object_storage/metrics_wrappers.rs

quickwit/quickwit-storage/src/metrics.rs

quickwit/quickwit-storage/src/object_storage/azure_blob_storage.rs

quickwit/quickwit-storage/src/object_storage/metrics_wrappers.rs

rdettai · 2025-07-15T14:26:50Z

quickwit/quickwit-storage/src/object_storage/metrics_wrappers.rs

+            ActionLabel::DeleteObject => "delete_object",
+            ActionLabel::DeleteObjects => "delete_objects",
+            ActionLabel::GetObject => "get_object",
+            ActionLabel::HeadObject => "head_object",


Azure: get_properties
S3: head_object
Opendal: stat

- Use explicit label values - Track download at the copy level Unified label values for object store actions.

rdettai · 2025-07-15T15:15:00Z

quickwit/quickwit-storage/src/object_storage/metrics_wrappers.rs

+/// This is a fork of `tokio::io::copy_buf` that enables tracking the number of
+/// bytes transferred. This estimate should be accurate as long as the network
+/// is the bottleneck.
+mod copy_buf {
+
+    use std::future::Future;
+    use std::io;
+    use std::pin::Pin;
+    use std::task::{Context, Poll, ready};
+
+    use tokio::io::{AsyncBufRead, AsyncWrite};
+
+    #[derive(Debug)]
+    #[must_use = "futures do nothing unless you `.await` or poll them"]
+    pub struct CopyBuf<'a, R: ?Sized, W: ?Sized> {
+        pub reader: &'a mut R,
+        pub writer: &'a mut W,
+        pub amt: u64,
+    }
+
+    impl<R, W> Future for CopyBuf<'_, R, W>
+    where
+        R: AsyncBufRead + Unpin + ?Sized,
+        W: AsyncWrite + Unpin + ?Sized,
+    {
+        type Output = io::Result<u64>;
+
+        fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+            loop {
+                let me = &mut *self;
+                let buffer = ready!(Pin::new(&mut *me.reader).poll_fill_buf(cx))?;
+                if buffer.is_empty() {
+                    ready!(Pin::new(&mut self.writer).poll_flush(cx))?;
+                    return Poll::Ready(Ok(self.amt));
+                }
+
+                let i = ready!(Pin::new(&mut *me.writer).poll_write(cx, buffer))?;
+                if i == 0 {
+                    return Poll::Ready(Err(std::io::ErrorKind::WriteZero.into()));
+                }
+                self.amt += i as u64;
+                Pin::new(&mut *self.reader).consume(i);
+            }
+        }
+    }
+}


It was actually pretty simple to have an estimate for the number of bytes already downloaded when the error occurred, so I added it.

Reorganize object store metrics

716c43f

rdettai self-assigned this Jun 26, 2025

rdettai requested a review from guilload June 26, 2025 15:14

rdettai force-pushed the revamp-storage-metrics branch from 68c7373 to 8bdb431 Compare July 2, 2025 10:08

Fix feature flags and license headers

88528bf

rdettai force-pushed the revamp-storage-metrics branch from 8bdb431 to 88528bf Compare July 2, 2025 11:55