You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2026/04/08 08:18:30 UTC
(opendal) branch main updated: fix(metrics): fix missing errors total counter (#7357)
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/opendal.git
The following commit(s) were added to refs/heads/main by this push:
new 17fceb6b3 fix(metrics): fix missing errors total counter (#7357)
17fceb6b3 is described below
commit 17fceb6b3ce1b797412b6b99f09f4c573408c957
Author: dentiny <de...@gmail.com>
AuthorDate: Wed Apr 8 01:18:24 2026 -0700
fix(metrics): fix missing errors total counter (#7357)
* fix(metrics): fix missing errors total counter
* simplify
---
core/layers/observe-metrics-common/src/lib.rs | 89 ++++++++++++++-------------
1 file changed, 47 insertions(+), 42 deletions(-)
diff --git a/core/layers/observe-metrics-common/src/lib.rs b/core/layers/observe-metrics-common/src/lib.rs
index ba9c6c781..8d588ca08 100644
--- a/core/layers/observe-metrics-common/src/lib.rs
+++ b/core/layers/observe-metrics-common/src/lib.rs
@@ -1004,6 +1004,7 @@ pub struct MetricsWrapper<R, I: MetricsIntercept> {
start: Instant,
size: u64,
+ completed: bool,
}
impl<R, I: MetricsIntercept> Drop for MetricsWrapper<R, I> {
@@ -1011,6 +1012,13 @@ impl<R, I: MetricsIntercept> Drop for MetricsWrapper<R, I> {
let size = self.size;
let duration = self.start.elapsed();
+ if !self.completed {
+ self.interceptor.observe(
+ self.labels.clone().with_error(ErrorKind::Unexpected),
+ MetricValue::OperationErrorsTotal,
+ );
+ }
+
if self.labels.operation == Operation::Read.into_static()
|| self.labels.operation == Operation::Write.into_static()
{
@@ -1048,8 +1056,17 @@ impl<R, I: MetricsIntercept> MetricsWrapper<R, I> {
labels,
start,
size: 0,
+ completed: false,
}
}
+
+ fn record_error(&mut self, err: &Error) {
+ self.completed = true;
+ self.interceptor.observe(
+ self.labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
+ );
+ }
}
impl<R: oio::Read, I: MetricsIntercept> oio::Read for MetricsWrapper<R, I> {
@@ -1058,14 +1075,12 @@ impl<R: oio::Read, I: MetricsIntercept> oio::Read for MetricsWrapper<R, I> {
.read()
.await
.inspect(|bs| {
+ if bs.is_empty() {
+ self.completed = true;
+ }
self.size += bs.len() as u64;
})
- .inspect_err(|err| {
- self.interceptor.observe(
- self.labels.clone().with_error(err.kind()),
- MetricValue::OperationErrorsTotal,
- );
- })
+ .inspect_err(|err| self.record_error(err))
}
}
@@ -1079,30 +1094,27 @@ impl<R: oio::Write, I: MetricsIntercept> oio::Write for MetricsWrapper<R, I> {
.inspect(|_| {
self.size += size as u64;
})
- .inspect_err(|err| {
- self.interceptor.observe(
- self.labels.clone().with_error(err.kind()),
- MetricValue::OperationErrorsTotal,
- );
- })
+ .inspect_err(|err| self.record_error(err))
}
async fn close(&mut self) -> Result<Metadata> {
- self.inner.close().await.inspect_err(|err| {
- self.interceptor.observe(
- self.labels.clone().with_error(err.kind()),
- MetricValue::OperationErrorsTotal,
- );
- })
+ let result = self
+ .inner
+ .close()
+ .await
+ .inspect_err(|err| self.record_error(err));
+ self.completed = true;
+ result
}
async fn abort(&mut self) -> Result<()> {
- self.inner.abort().await.inspect_err(|err| {
- self.interceptor.observe(
- self.labels.clone().with_error(err.kind()),
- MetricValue::OperationErrorsTotal,
- );
- })
+ let result = self
+ .inner
+ .abort()
+ .await
+ .inspect_err(|err| self.record_error(err));
+ self.completed = true;
+ result
}
}
@@ -1114,14 +1126,11 @@ impl<R: oio::List, I: MetricsIntercept> oio::List for MetricsWrapper<R, I> {
.inspect(|entry| {
if entry.is_some() {
self.size += 1;
+ } else {
+ self.completed = true;
}
})
- .inspect_err(|err| {
- self.interceptor.observe(
- self.labels.clone().with_error(err.kind()),
- MetricValue::OperationErrorsTotal,
- );
- })
+ .inspect_err(|err| self.record_error(err))
}
}
@@ -1133,20 +1142,16 @@ impl<R: oio::Delete, I: MetricsIntercept> oio::Delete for MetricsWrapper<R, I> {
.inspect(|_| {
self.size += 1;
})
- .inspect_err(|err| {
- self.interceptor.observe(
- self.labels.clone().with_error(err.kind()),
- MetricValue::OperationErrorsTotal,
- );
- })
+ .inspect_err(|err| self.record_error(err))
}
async fn close(&mut self) -> Result<()> {
- self.inner.close().await.inspect_err(|err| {
- self.interceptor.observe(
- self.labels.clone().with_error(err.kind()),
- MetricValue::OperationErrorsTotal,
- );
- })
+ let result = self
+ .inner
+ .close()
+ .await
+ .inspect_err(|err| self.record_error(err));
+ self.completed = true;
+ result
}
}