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 2025/03/26 09:18:04 UTC
(opendal) branch refactor-io-metrics created (now f26d60faa)
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a change to branch refactor-io-metrics
in repository https://gitbox.apache.org/repos/asf/opendal.git
at f26d60faa refactor: Polishing IO metrics by adding more useful metrics
This branch includes the following new commits:
new f26d60faa refactor: Polishing IO metrics by adding more useful metrics
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(opendal) 01/01: refactor: Polishing IO metrics by adding more useful metrics
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch refactor-io-metrics
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit f26d60faaab3de1f5fde83f659c4c6648fd70559
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Wed Mar 26 17:17:19 2025 +0800
refactor: Polishing IO metrics by adding more useful metrics
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/layers/observe/metrics.rs | 1321 +++++++++++++++++++-----------------
core/src/layers/observe/mod.rs | 1 +
core/src/raw/http_util/body.rs | 25 +
3 files changed, 716 insertions(+), 631 deletions(-)
diff --git a/core/src/layers/observe/metrics.rs b/core/src/layers/observe/metrics.rs
index 5cff51021..5d457b009 100644
--- a/core/src/layers/observe/metrics.rs
+++ b/core/src/layers/observe/metrics.rs
@@ -18,13 +18,136 @@
use std::fmt::Debug;
use std::fmt::Formatter;
use std::fmt::Write;
+use std::pin::Pin;
use std::sync::Arc;
+use std::task::ready;
+use std::task::Context;
+use std::task::Poll;
use std::time::Duration;
use std::time::Instant;
+use futures::Stream;
+use futures::StreamExt;
+use http::StatusCode;
+
use crate::raw::*;
use crate::*;
+/// The metric label for the scheme like s3, fs, cos.
+pub static LABEL_SCHEME: &str = "scheme";
+/// The metric label for the namespace like bucket name in s3.
+pub static LABEL_NAMESPACE: &str = "namespace";
+/// The metric label for the root path.
+pub static LABEL_ROOT: &str = "root";
+/// The metric label for the operation like read, write, list.
+pub static LABEL_OPERATION: &str = "operation";
+/// The metric label for the error.
+pub static LABEL_ERROR: &str = "error";
+/// The metrics label for the path. (will be removed)
+pub static LABEL_PATH: &str = "path";
+/// The metric label for the http code.
+pub static LABEL_STATUS_CODE: &str = "status_code";
+
+/// MetricLabels are the labels for the metrics.
+#[derive(Default, Debug, Clone)]
+pub struct MetricLabels {
+ pub scheme: Scheme,
+ pub namespace: Arc<str>,
+ pub root: Arc<str>,
+ pub operation: &'static str,
+
+ /// Only available for `OperationErrorsTotal`
+ pub error: Option<ErrorKind>,
+ /// Only available for `HttpStatusErrorsTotal`
+ pub status_code: Option<StatusCode>,
+}
+
+impl MetricLabels {
+ /// Create a new set of MetricLabels.
+ fn new(info: Arc<AccessorInfo>, op: &'static str) -> Self {
+ MetricLabels {
+ scheme: info.scheme(),
+ namespace: info.name(),
+ root: info.root(),
+ operation: op,
+ ..MetricLabels::default()
+ }
+ }
+
+ /// Add error to the metric labels.
+ fn with_error(mut self, err: ErrorKind) -> Self {
+ self.error = Some(err);
+ self
+ }
+
+ /// Add status code to the metric labels.
+ fn with_status_code(mut self, code: StatusCode) -> Self {
+ self.status_code = Some(code);
+ self
+ }
+}
+
+/// MetricValue is the value the opendal sends to the metrics impls.
+///
+/// Metrcis impls can be `prometheus_client`, `metrics` etc.
+///
+/// Every metrics impls SHOULD implement observe over the MetricValue to make
+/// sure they provides the consistent metrics for users.
+#[non_exhaustive]
+#[derive(Debug, Clone, Copy)]
+pub enum MetricValue {
+ OperationBytes(u64),
+ OperationBytesRate(f64),
+ OperationBytesTotal(u64),
+ OperationCountTotal,
+ OperationDurationSeconds(Duration),
+ OperationErrorsTotal,
+ OperationExecuting(isize),
+ OperationTtfbSeconds(Duration),
+
+ HttpCountTotal,
+ HttpConnectionErrorsTotal,
+ HttpExecuting(isize),
+ HttpRequestBytes(u64),
+ HttpRequestBytesRate(f64),
+ HttpRequestBytesTotal(u64),
+ HttpRequestDurationSeconds(Duration),
+ HttpResponseBytes(u64),
+ HttpResponseBytesRate(f64),
+ HttpResponseBytesTotal(u64),
+ HttpResponseDurationSeconds(Duration),
+ HttpStatusErrorsTotal,
+}
+
+impl MetricValue {
+ /// Returns the related info of this metric value in the format `(name, help)`.
+ pub fn info(&self) -> (&'static str, &'static str) {
+ match self {
+ MetricValue::OperationBytes(_) => ("opendal_operation_bytes", "Current operation size in bytes, represents the size of data being processed in the current operation"),
+ MetricValue::OperationBytesRate(_) => ("opendal_operation_bytes_rate", "Histogram of data processing rates in bytes per second within individual operations"),
+ MetricValue::OperationBytesTotal(_) => ("opendal_operation_bytes_total", "Total number of bytes processed by operations"),
+ MetricValue::OperationCountTotal => ("opendal_operation_count_total","Total number of operations completed"),
+ MetricValue::OperationDurationSeconds(_) => ("opendal_operation_duration_seconds","Duration of operations in seconds, measured from start to completion"),
+ MetricValue::OperationErrorsTotal => ("opendal_operation_errors_total","Total number of failed operations"),
+ MetricValue::OperationExecuting(_) => ("opendal_operation_executing","Number of operations currently being executed"),
+ MetricValue::OperationTtfbSeconds(_) => ("opendal_operation_ttfb_seconds","Time to first byte in seconds for operations"),
+
+ MetricValue::HttpCountTotal => ("opendal_http_count_total","Total number of HTTP requests initiated"),
+ MetricValue::HttpConnectionErrorsTotal => ("opendal_http_connection_errors_total","Total number of HTTP requests that failed before receiving a response (DNS failures, connection refused, timeouts, TLS errors)"),
+ MetricValue::HttpStatusErrorsTotal => ("opendal_http_connection_errors_total","Total number of HTTP requests that received error status codes (non-2xx responses)"),
+ MetricValue::HttpExecuting(_) => ("opendal_http_executing","Number of HTTP requests currently in flight from this client"),
+ MetricValue::HttpRequestBytes(_) => ("opendal_http_request_bytes","Histogram of HTTP request body sizes in bytes"),
+ MetricValue::HttpRequestBytesRate(_) => ("opendal_http_request_bytes_rate","Histogram of HTTP request bytes per second rates"),
+ MetricValue::HttpRequestBytesTotal(_) =>( "opendal_http_request_bytes_total","Total number of bytes sent in HTTP request bodies"),
+ MetricValue::HttpRequestDurationSeconds(_) => ("opendal_http_request_duration_seconds","Histogram of time durations in seconds spent sending HTTP requests, from first byte sent to receiving the first byte"),
+ MetricValue::HttpResponseBytes(_) => ("opendal_http_response_bytes","Histogram of HTTP response body sizes in bytes"),
+ MetricValue::HttpResponseBytesRate(_) => ("opendal_http_response_bytes_rate","Histogram of HTTP response bytes per second rates"),
+ MetricValue::HttpResponseBytesTotal(_) =>( "opendal_http_response_bytes_total","Total number of bytes received in HTTP response bodies"),
+ MetricValue::HttpResponseDurationSeconds(_) => ("opendal_http_response_duration_seconds","Histogram of time durations in seconds spent receiving HTTP responses, from first byte received to last byte received"),
+ }
+ }
+}
+
/// The metric metadata which contains the metric name and help.
pub struct MetricMetadata {
name: &'static str,
@@ -78,23 +201,15 @@ pub static METRIC_HTTP_REQUEST_BYTES: MetricMetadata = MetricMetadata {
help: "Histogram of the bytes transferred during http requests",
};
-/// The metric label for the scheme like s3, fs, cos.
-pub static LABEL_SCHEME: &str = "scheme";
-/// The metric label for the namespace like bucket name in s3.
-pub static LABEL_NAMESPACE: &str = "namespace";
-/// The metric label for the root path.
-pub static LABEL_ROOT: &str = "root";
-/// The metric label for the path used by request.
-pub static LABEL_PATH: &str = "path";
-/// The metric label for the operation like read, write, list.
-pub static LABEL_OPERATION: &str = "operation";
-/// The metric label for the error kind.
-pub static LABEL_ERROR: &str = "error";
-
/// The interceptor for metrics.
///
/// All metrics related libs should implement this trait to observe opendal's internal operations.
pub trait MetricsIntercept: Debug + Clone + Send + Sync + Unpin + 'static {
+ /// Observe the metric value.
+ fn observe(&self, labels: MetricLabels, value: MetricValue) {
+ let _ = (labels, value);
+ }
+
/// Observe the operation duration in seconds.
fn observe_operation_duration_seconds(
&self,
@@ -102,7 +217,9 @@ pub trait MetricsIntercept: Debug + Clone + Send + Sync + Unpin + 'static {
path: &str,
op: Operation,
duration: Duration,
- );
+ ) {
+ let _ = (info, path, op, duration);
+ }
/// Observe the operation bytes happened in IO like read and write.
fn observe_operation_bytes(
@@ -111,7 +228,9 @@ pub trait MetricsIntercept: Debug + Clone + Send + Sync + Unpin + 'static {
path: &str,
op: Operation,
bytes: usize,
- );
+ ) {
+ let _ = (info, path, op, bytes);
+ }
/// Observe the operation errors total.
fn observe_operation_errors_total(
@@ -120,7 +239,9 @@ pub trait MetricsIntercept: Debug + Clone + Send + Sync + Unpin + 'static {
path: &str,
op: Operation,
error: ErrorKind,
- );
+ ) {
+ let _ = (info, path, op, error);
+ }
/// Observe the http request duration in seconds.
fn observe_http_request_duration_seconds(
@@ -183,28 +304,130 @@ pub struct MetricsHttpFetcher<I: MetricsIntercept> {
impl<I: MetricsIntercept> HttpFetch for MetricsHttpFetcher<I> {
async fn fetch(&self, req: http::Request<Buffer>) -> Result<http::Response<HttpBody>> {
- // Extract context from the http request.
- let size = req.body().len();
- let op = req.extensions().get::<Operation>().copied();
+ let labels = MetricLabels::new(
+ self.info.clone(),
+ req.extensions()
+ .get::<Operation>()
+ .copied()
+ .map(Operation::into_static)
+ .unwrap_or("unknown"),
+ );
+
+ self.interceptor
+ .observe(labels.clone(), MetricValue::HttpCountTotal);
+ self.interceptor
+ .observe(labels.clone(), MetricValue::HttpExecuting(1));
let start = Instant::now();
+ let req_size = req.body().len();
+
let res = self.inner.fetch(req).await;
- let duration = start.elapsed();
-
- // We only inject the metrics when the operation exists.
- if let Some(op) = op {
- if res.is_ok() {
- self.interceptor.observe_http_request_duration_seconds(
- self.info.clone(),
- op,
- duration,
- );
+ let req_duration = start.elapsed();
+
+ match res {
+ Err(err) => {
+ self.interceptor
+ .observe(labels.clone(), MetricValue::HttpExecuting(-1));
+ self.interceptor
+ .observe(labels.clone(), MetricValue::HttpConnectionErrorsTotal);
+ Err(err)
+ }
+ Ok(resp) if resp.status().is_client_error() && resp.status().is_server_error() => {
self.interceptor
- .observe_http_request_bytes(self.info.clone(), op, size);
+ .observe(labels.clone(), MetricValue::HttpExecuting(-1));
+
+ self.interceptor.observe(
+ labels.clone().with_status_code(resp.status()),
+ MetricValue::HttpStatusErrorsTotal,
+ );
+ Ok(resp)
+ }
+ Ok(resp) => {
+ self.interceptor.observe(
+ labels.clone(),
+ MetricValue::HttpRequestBytes(req_size as u64),
+ );
+ self.interceptor.observe(
+ labels.clone(),
+ MetricValue::HttpRequestBytesRate(req_size as f64 / req_duration.as_secs_f64()),
+ );
+ self.interceptor.observe(
+ labels.clone(),
+ MetricValue::HttpRequestBytesTotal(req_size as u64),
+ );
+ self.interceptor.observe(
+ labels.clone(),
+ MetricValue::HttpRequestDurationSeconds(req_duration),
+ );
+
+ let (parts, body) = resp.into_parts();
+ let body = body.map_inner(|s| {
+ Box::new(MetricsStream {
+ inner: s,
+ interceptor: self.interceptor.clone(),
+ labels: labels.clone(),
+ size: 0,
+ start: Instant::now(),
+ })
+ });
+
+ Ok(http::Response::from_parts(parts, body))
}
}
+ }
+}
- res
+pub struct MetricsStream<S, I> {
+ inner: S,
+ interceptor: I,
+
+ labels: MetricLabels,
+ size: u64,
+ start: Instant,
+}
+
+impl<S, I> Stream for MetricsStream<S, I>
+where
+ S: Stream<Item = Result<Buffer>> + Send + Sync + Unpin + 'static,
+ I: MetricsIntercept,
+{
+ type Item = Result<Buffer>;
+
+ fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+ match ready!(self.inner.poll_next_unpin(cx)) {
+ Some(Ok(bs)) => {
+ self.size += bs.len() as u64;
+ Poll::Ready(Some(Ok(bs)))
+ }
+ Some(Err(err)) => Poll::Ready(Some(Err(err))),
+ None => {
+ let resp_size = self.size;
+ let resp_duration = self.start.elapsed();
+
+ self.interceptor.observe(
+ self.labels.clone(),
+ MetricValue::HttpResponseBytes(resp_size),
+ );
+ self.interceptor.observe(
+ self.labels.clone(),
+ MetricValue::HttpResponseBytesRate(
+ resp_size as f64 / resp_duration.as_secs_f64(),
+ ),
+ );
+ self.interceptor.observe(
+ self.labels.clone(),
+ MetricValue::HttpResponseBytesTotal(resp_size),
+ );
+ self.interceptor.observe(
+ self.labels.clone(),
+ MetricValue::HttpResponseDurationSeconds(resp_duration),
+ );
+ self.interceptor
+ .observe(self.labels.clone(), MetricValue::HttpExecuting(-1));
+
+ Poll::Ready(None)
+ }
+ }
}
}
@@ -239,507 +462,519 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
}
async fn create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
- let op = Operation::CreateDir;
+ let labels = MetricLabels::new(self.info.clone(), Operation::CreateDir.into_static());
let start = Instant::now();
- self.inner()
+
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationExecuting(1));
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationCountTotal);
+
+ let res = self
+ .inner()
.create_dir(path, args)
.await
.inspect(|_| {
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- path,
- op,
- start.elapsed(),
+ self.interceptor.observe(
+ labels.clone(),
+ MetricValue::OperationDurationSeconds(start.elapsed()),
);
})
- .inspect_err(move |err| {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- path,
- op,
- err.kind(),
+ .inspect_err(|err| {
+ self.interceptor.observe(
+ labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
);
- })
+ });
+
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationExecuting(-1));
+ res
}
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
- let op = Operation::ReaderStart;
+ let labels = MetricLabels::new(self.info.clone(), Operation::ReaderRead.into_static());
let start = Instant::now();
+
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationExecuting(1));
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationCountTotal);
+
let (rp, reader) = self
.inner
.read(path, args)
.await
.inspect(|_| {
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- path,
- op,
- start.elapsed(),
+ self.interceptor.observe(
+ labels.clone(),
+ MetricValue::OperationTtfbSeconds(start.elapsed()),
);
})
.inspect_err(|err| {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- path,
- op,
- err.kind(),
+ self.interceptor.observe(
+ labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
);
})?;
Ok((
rp,
- MetricsWrapper::new(
- reader,
- self.interceptor.clone(),
- self.info.clone(),
- path.to_string(),
- ),
+ MetricsWrapper::new(reader, self.interceptor.clone(), labels, start),
))
}
async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
- let op = Operation::WriterStart;
+ let labels = MetricLabels::new(self.info.clone(), Operation::WriterStart.into_static());
let start = Instant::now();
- let (rp, writer) = self
- .inner
- .write(path, args)
- .await
- .inspect(|_| {
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- path,
- op,
- start.elapsed(),
- );
- })
- .inspect_err(|err| {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- path,
- op,
- err.kind(),
- );
- })?;
+
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationExecuting(1));
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationCountTotal);
+
+ let (rp, writer) = self.inner.write(path, args).await.inspect_err(|err| {
+ self.interceptor.observe(
+ labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
+ );
+ })?;
Ok((
rp,
- MetricsWrapper::new(
- writer,
- self.interceptor.clone(),
- self.info.clone(),
- path.to_string(),
- ),
+ MetricsWrapper::new(writer, self.interceptor.clone(), labels, start),
))
}
async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
- let op = Operation::Copy;
+ let labels = MetricLabels::new(self.info.clone(), Operation::Copy.into_static());
let start = Instant::now();
- self.inner()
+
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationExecuting(1));
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationCountTotal);
+
+ let res = self
+ .inner()
.copy(from, to, args)
.await
.inspect(|_| {
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- from,
- op,
- start.elapsed(),
+ self.interceptor.observe(
+ labels.clone(),
+ MetricValue::OperationDurationSeconds(start.elapsed()),
);
})
- .inspect_err(move |err| {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- from,
- op,
- err.kind(),
+ .inspect_err(|err| {
+ self.interceptor.observe(
+ labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
);
- })
+ });
+
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationExecuting(-1));
+ res
}
async fn rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
- let op = Operation::Rename;
+ let labels = MetricLabels::new(self.info.clone(), Operation::Rename.into_static());
let start = Instant::now();
- self.inner()
+
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationExecuting(1));
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationCountTotal);
+
+ let res = self
+ .inner()
.rename(from, to, args)
.await
.inspect(|_| {
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- from,
- op,
- start.elapsed(),
+ self.interceptor.observe(
+ labels.clone(),
+ MetricValue::OperationDurationSeconds(start.elapsed()),
);
})
- .inspect_err(move |err| {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- from,
- op,
- err.kind(),
+ .inspect_err(|err| {
+ self.interceptor.observe(
+ labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
);
- })
+ });
+
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationExecuting(-1));
+ res
}
async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
- let op = Operation::Stat;
+ let labels = MetricLabels::new(self.info.clone(), Operation::Stat.into_static());
let start = Instant::now();
- self.inner()
+
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationExecuting(1));
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationCountTotal);
+
+ let res = self
+ .inner()
.stat(path, args)
.await
.inspect(|_| {
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- path,
- op,
- start.elapsed(),
+ self.interceptor.observe(
+ labels.clone(),
+ MetricValue::OperationDurationSeconds(start.elapsed()),
);
})
- .inspect_err(move |err| {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- path,
- op,
- err.kind(),
+ .inspect_err(|err| {
+ self.interceptor.observe(
+ labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
);
- })
+ });
+
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationExecuting(-1));
+ res
}
async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
- let op = Operation::DeleterStart;
+ let labels = MetricLabels::new(self.info.clone(), Operation::DeleterStart.into_static());
let start = Instant::now();
- let (rp, writer) = self
+
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationExecuting(1));
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationCountTotal);
+
+ let (rp, deleter) = self
.inner
.delete()
.await
.inspect(|_| {
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- "",
- op,
- start.elapsed(),
+ self.interceptor.observe(
+ labels.clone(),
+ MetricValue::OperationDurationSeconds(start.elapsed()),
);
})
.inspect_err(|err| {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- "",
- op,
- err.kind(),
+ self.interceptor.observe(
+ labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
);
})?;
Ok((
rp,
- MetricsWrapper::new(
- writer,
- self.interceptor.clone(),
- self.info.clone(),
- "".to_string(),
- ),
+ MetricsWrapper::new(deleter, self.interceptor.clone(), labels, start),
))
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
- let op = Operation::ListerStart;
+ let labels = MetricLabels::new(self.info.clone(), Operation::ListerNext.into_static());
let start = Instant::now();
+
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationExecuting(1));
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationCountTotal);
+
let (rp, lister) = self
.inner
.list(path, args)
.await
.inspect(|_| {
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- path,
- op,
- start.elapsed(),
+ self.interceptor.observe(
+ labels.clone(),
+ MetricValue::OperationDurationSeconds(start.elapsed()),
);
})
.inspect_err(|err| {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- path,
- op,
- err.kind(),
+ self.interceptor.observe(
+ labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
);
})?;
Ok((
rp,
- MetricsWrapper::new(
- lister,
- self.interceptor.clone(),
- self.info.clone(),
- path.to_string(),
- ),
+ MetricsWrapper::new(lister, self.interceptor.clone(), labels, start),
))
}
async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
- let op = Operation::Presign;
+ let labels = MetricLabels::new(self.info.clone(), Operation::Presign.into_static());
let start = Instant::now();
- self.inner()
+
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationExecuting(1));
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationCountTotal);
+
+ let res = self
+ .inner()
.presign(path, args)
.await
.inspect(|_| {
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- path,
- op,
- start.elapsed(),
+ self.interceptor.observe(
+ labels.clone(),
+ MetricValue::OperationDurationSeconds(start.elapsed()),
);
})
- .inspect_err(move |err| {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- path,
- op,
- err.kind(),
+ .inspect_err(|err| {
+ self.interceptor.observe(
+ labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
);
- })
+ });
+
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationExecuting(-1));
+ res
}
fn blocking_create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
- let op = Operation::CreateDir;
+ let labels = MetricLabels::new(self.info.clone(), Operation::CreateDir.into_static());
let start = Instant::now();
- self.inner()
+
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationExecuting(1));
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationCountTotal);
+
+ let res = self
+ .inner()
.blocking_create_dir(path, args)
.inspect(|_| {
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- path,
- op,
- start.elapsed(),
+ self.interceptor.observe(
+ labels.clone(),
+ MetricValue::OperationDurationSeconds(start.elapsed()),
);
})
- .inspect_err(move |err| {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- path,
- op,
- err.kind(),
+ .inspect_err(|err| {
+ self.interceptor.observe(
+ labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
);
- })
+ });
+
+ res
}
fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
- let op = Operation::ReaderStart;
+ let labels = MetricLabels::new(self.info.clone(), Operation::ReaderStart.into_static());
let start = Instant::now();
+
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationExecuting(1));
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationCountTotal);
+
let (rp, reader) = self
.inner
.blocking_read(path, args)
.inspect(|_| {
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- path,
- op,
- start.elapsed(),
+ self.interceptor.observe(
+ labels.clone(),
+ MetricValue::OperationTtfbSeconds(start.elapsed()),
);
})
.inspect_err(|err| {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- path,
- op,
- err.kind(),
+ self.interceptor.observe(
+ labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
);
})?;
Ok((
rp,
- MetricsWrapper::new(
- reader,
- self.interceptor.clone(),
- self.info.clone(),
- path.to_string(),
- ),
+ MetricsWrapper::new(reader, self.interceptor.clone(), labels, start),
))
}
fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
- let op = Operation::WriterStart;
+ let labels = MetricLabels::new(self.info.clone(), Operation::WriterWrite.into_static());
let start = Instant::now();
- let (rp, writer) = self
- .inner
- .blocking_write(path, args)
- .inspect(|_| {
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- path,
- op,
- start.elapsed(),
- );
- })
- .inspect_err(|err| {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- path,
- op,
- err.kind(),
- );
- })?;
+
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationExecuting(1));
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationCountTotal);
+
+ let (rp, writer) = self.inner.blocking_write(path, args).inspect_err(|err| {
+ self.interceptor.observe(
+ labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
+ );
+ })?;
Ok((
rp,
- MetricsWrapper::new(
- writer,
- self.interceptor.clone(),
- self.info.clone(),
- path.to_string(),
- ),
+ MetricsWrapper::new(writer, self.interceptor.clone(), labels, start),
))
}
fn blocking_copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
- let op = Operation::Copy;
+ let labels = MetricLabels::new(self.info.clone(), Operation::Copy.into_static());
let start = Instant::now();
- self.inner()
+
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationExecuting(1));
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationCountTotal);
+
+ let res = self
+ .inner()
.blocking_copy(from, to, args)
.inspect(|_| {
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- from,
- op,
- start.elapsed(),
+ self.interceptor.observe(
+ labels.clone(),
+ MetricValue::OperationDurationSeconds(start.elapsed()),
);
})
- .inspect_err(move |err| {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- from,
- op,
- err.kind(),
+ .inspect_err(|err| {
+ self.interceptor.observe(
+ labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
);
- })
+ });
+
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationExecuting(-1));
+ res
}
fn blocking_rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
- let op = Operation::Rename;
+ let labels = MetricLabels::new(self.info.clone(), Operation::Rename.into_static());
let start = Instant::now();
- self.inner()
+ let res = self
+ .inner()
.blocking_rename(from, to, args)
.inspect(|_| {
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- from,
- op,
- start.elapsed(),
+ self.interceptor.observe(
+ labels.clone(),
+ MetricValue::OperationDurationSeconds(start.elapsed()),
);
})
.inspect_err(|err| {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- from,
- op,
- err.kind(),
+ self.interceptor.observe(
+ labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
);
- })
+ });
+
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationExecuting(-1));
+ res
}
fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
- let op = Operation::Stat;
+ let labels = MetricLabels::new(self.info.clone(), Operation::Stat.into_static());
let start = Instant::now();
- self.inner()
+
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationExecuting(1));
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationCountTotal);
+
+ let res = self
+ .inner()
.blocking_stat(path, args)
.inspect(|_| {
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- path,
- op,
- start.elapsed(),
+ self.interceptor.observe(
+ labels.clone(),
+ MetricValue::OperationDurationSeconds(start.elapsed()),
);
})
- .inspect_err(move |err| {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- path,
- op,
- err.kind(),
+ .inspect_err(|err| {
+ self.interceptor.observe(
+ labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
);
- })
+ });
+
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationExecuting(-1));
+ res
}
fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
- let op = Operation::DeleterStart;
+ let labels = MetricLabels::new(self.info.clone(), Operation::DeleterStart.into_static());
let start = Instant::now();
- let (rp, writer) = self
+
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationExecuting(1));
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationCountTotal);
+
+ let (rp, deleter) = self
.inner
.blocking_delete()
.inspect(|_| {
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- "",
- op,
- start.elapsed(),
+ self.interceptor.observe(
+ labels.clone(),
+ MetricValue::OperationDurationSeconds(start.elapsed()),
);
})
.inspect_err(|err| {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- "",
- op,
- err.kind(),
+ self.interceptor.observe(
+ labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
);
})?;
Ok((
rp,
- MetricsWrapper::new(
- writer,
- self.interceptor.clone(),
- self.info.clone(),
- "".to_string(),
- ),
+ MetricsWrapper::new(deleter, self.interceptor.clone(), labels, start),
))
}
fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
- let op = Operation::ListerStart;
+ let labels = MetricLabels::new(self.info.clone(), Operation::ListerNext.into_static());
let start = Instant::now();
+
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationExecuting(1));
+ self.interceptor
+ .observe(labels.clone(), MetricValue::OperationCountTotal);
+
let (rp, lister) = self
.inner
.blocking_list(path, args)
.inspect(|_| {
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- path,
- op,
- start.elapsed(),
+ self.interceptor.observe(
+ labels.clone(),
+ MetricValue::OperationDurationSeconds(start.elapsed()),
);
})
.inspect_err(|err| {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- path,
- op,
- err.kind(),
+ self.interceptor.observe(
+ labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
);
})?;
Ok((
rp,
- MetricsWrapper::new(
- lister,
- self.interceptor.clone(),
- self.info.clone(),
- path.to_string(),
- ),
+ MetricsWrapper::new(lister, self.interceptor.clone(), labels, start),
))
}
}
@@ -747,398 +982,222 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
pub struct MetricsWrapper<R, I: MetricsIntercept> {
inner: R,
interceptor: I,
- info: Arc<AccessorInfo>,
+ labels: MetricLabels,
+
+ start: Instant,
+ size: u64,
+}
+
+impl<R, I: MetricsIntercept> Drop for MetricsWrapper<R, I> {
+ fn drop(&mut self) {
+ let size = self.size;
+ let duration = self.start.elapsed();
- path: String,
+ self.interceptor
+ .observe(self.labels.clone(), MetricValue::OperationBytes(self.size));
+ self.interceptor.observe(
+ self.labels.clone(),
+ MetricValue::OperationBytesRate(size as f64 / duration.as_secs_f64()),
+ );
+ self.interceptor
+ .observe(self.labels.clone(), MetricValue::OperationBytesTotal(size));
+ self.interceptor.observe(
+ self.labels.clone(),
+ MetricValue::OperationDurationSeconds(duration),
+ );
+ self.interceptor
+ .observe(self.labels.clone(), MetricValue::OperationExecuting(-1));
+ }
}
impl<R, I: MetricsIntercept> MetricsWrapper<R, I> {
- fn new(inner: R, interceptor: I, info: Arc<AccessorInfo>, path: String) -> Self {
+ fn new(inner: R, interceptor: I, labels: MetricLabels, start: Instant) -> Self {
Self {
inner,
interceptor,
- info,
- path,
+ labels,
+ start,
+ size: 0,
}
}
}
impl<R: oio::Read, I: MetricsIntercept> oio::Read for MetricsWrapper<R, I> {
async fn read(&mut self) -> Result<Buffer> {
- let op = Operation::ReaderRead;
-
- let start = Instant::now();
-
- let res = match self.inner.read().await {
- Ok(bs) => {
- self.interceptor.observe_operation_bytes(
- self.info.clone(),
- &self.path,
- op,
- bs.len(),
- );
- Ok(bs)
- }
- Err(err) => {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- &self.path,
- op,
- err.kind(),
+ self.inner
+ .read()
+ .await
+ .inspect(|bs| {
+ self.size += bs.len() as u64;
+ })
+ .inspect_err(|err| {
+ self.interceptor.observe(
+ self.labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
);
- Err(err)
- }
- };
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- &self.path,
- op,
- start.elapsed(),
- );
- res
+ })
}
}
impl<R: oio::BlockingRead, I: MetricsIntercept> oio::BlockingRead for MetricsWrapper<R, I> {
fn read(&mut self) -> Result<Buffer> {
- let op = Operation::ReaderRead;
-
- let start = Instant::now();
-
- let res = match self.inner.read() {
- Ok(bs) => {
- self.interceptor.observe_operation_bytes(
- self.info.clone(),
- &self.path,
- op,
- bs.len(),
- );
- Ok(bs)
- }
- Err(err) => {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- &self.path,
- op,
- err.kind(),
+ self.inner
+ .read()
+ .inspect(|bs| {
+ self.size += bs.len() as u64;
+ })
+ .inspect_err(|err| {
+ self.interceptor.observe(
+ self.labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
);
- Err(err)
- }
- };
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- &self.path,
- op,
- start.elapsed(),
- );
- res
+ })
}
}
impl<R: oio::Write, I: MetricsIntercept> oio::Write for MetricsWrapper<R, I> {
async fn write(&mut self, bs: Buffer) -> Result<()> {
- let op = Operation::WriterWrite;
-
- let start = Instant::now();
let size = bs.len();
- let res = match self.inner.write(bs).await {
- Ok(()) => {
- self.interceptor
- .observe_operation_bytes(self.info.clone(), &self.path, op, size);
- Ok(())
- }
- Err(err) => {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- &self.path,
- op,
- err.kind(),
+ self.inner
+ .write(bs)
+ .await
+ .inspect(|_| {
+ self.size += size as u64;
+ })
+ .inspect_err(|err| {
+ self.interceptor.observe(
+ self.labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
);
- Err(err)
- }
- };
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- &self.path,
- op,
- start.elapsed(),
- );
- res
+ })
}
async fn close(&mut self) -> Result<Metadata> {
- let op = Operation::WriterClose;
-
- let start = Instant::now();
-
- let res = match self.inner.close().await {
- Ok(meta) => Ok(meta),
- Err(err) => {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- &self.path,
- op,
- err.kind(),
- );
- Err(err)
- }
- };
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- &self.path,
- op,
- start.elapsed(),
- );
- res
+ self.inner.close().await.inspect_err(|err| {
+ self.interceptor.observe(
+ self.labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
+ );
+ })
}
async fn abort(&mut self) -> Result<()> {
- let op = Operation::WriterAbort;
-
- let start = Instant::now();
-
- let res = match self.inner.abort().await {
- Ok(()) => Ok(()),
- Err(err) => {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- &self.path,
- op,
- err.kind(),
- );
- Err(err)
- }
- };
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- &self.path,
- op,
- start.elapsed(),
- );
- res
+ self.inner.abort().await.inspect_err(|err| {
+ self.interceptor.observe(
+ self.labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
+ );
+ })
}
}
impl<R: oio::BlockingWrite, I: MetricsIntercept> oio::BlockingWrite for MetricsWrapper<R, I> {
fn write(&mut self, bs: Buffer) -> Result<()> {
- let op = Operation::WriterWrite;
-
- let start = Instant::now();
let size = bs.len();
- let res = match self.inner.write(bs) {
- Ok(()) => {
- self.interceptor
- .observe_operation_bytes(self.info.clone(), &self.path, op, size);
- Ok(())
- }
- Err(err) => {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- &self.path,
- op,
- err.kind(),
+ self.inner
+ .write(bs)
+ .inspect(|_| {
+ self.size += size as u64;
+ })
+ .inspect_err(|err| {
+ self.interceptor.observe(
+ self.labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
);
- Err(err)
- }
- };
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- &self.path,
- op,
- start.elapsed(),
- );
- res
+ })
}
fn close(&mut self) -> Result<Metadata> {
- let op = Operation::WriterClose;
-
- let start = Instant::now();
-
- let res = match self.inner.close() {
- Ok(meta) => Ok(meta),
- Err(err) => {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- &self.path,
- op,
- err.kind(),
- );
- Err(err)
- }
- };
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- &self.path,
- op,
- start.elapsed(),
- );
- res
+ self.inner.close().inspect_err(|err| {
+ self.interceptor.observe(
+ self.labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
+ );
+ })
}
}
impl<R: oio::List, I: MetricsIntercept> oio::List for MetricsWrapper<R, I> {
async fn next(&mut self) -> Result<Option<oio::Entry>> {
- let op = Operation::ListerNext;
-
- let start = Instant::now();
-
- let res = match self.inner.next().await {
- Ok(entry) => Ok(entry),
- Err(err) => {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- &self.path,
- op,
- err.kind(),
+ self.inner
+ .next()
+ .await
+ .inspect(|_| {
+ self.size += 1;
+ })
+ .inspect_err(|err| {
+ self.interceptor.observe(
+ self.labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
);
- Err(err)
- }
- };
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- &self.path,
- op,
- start.elapsed(),
- );
- res
+ })
}
}
impl<R: oio::BlockingList, I: MetricsIntercept> oio::BlockingList for MetricsWrapper<R, I> {
fn next(&mut self) -> Result<Option<oio::Entry>> {
- let op = Operation::ListerNext;
-
- let start = Instant::now();
-
- let res = match self.inner.next() {
- Ok(entry) => Ok(entry),
- Err(err) => {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- &self.path,
- op,
- err.kind(),
+ self.inner
+ .next()
+ .inspect(|_| {
+ self.size += 1;
+ })
+ .inspect_err(|err| {
+ self.interceptor.observe(
+ self.labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
);
- Err(err)
- }
- };
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- &self.path,
- op,
- start.elapsed(),
- );
- res
+ })
}
}
impl<R: oio::Delete, I: MetricsIntercept> oio::Delete for MetricsWrapper<R, I> {
fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
- let op = Operation::DeleterDelete;
-
- let start = Instant::now();
-
- let res = match self.inner.delete(path, args) {
- Ok(entry) => Ok(entry),
- Err(err) => {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- &self.path,
- op,
- err.kind(),
+ self.inner
+ .delete(path, args)
+ .inspect(|_| {
+ self.size += 1;
+ })
+ .inspect_err(|err| {
+ self.interceptor.observe(
+ self.labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
);
- Err(err)
- }
- };
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- &self.path,
- op,
- start.elapsed(),
- );
- res
+ })
}
async fn flush(&mut self) -> Result<usize> {
- let op = Operation::DeleterFlush;
-
- let start = Instant::now();
-
- let res = match self.inner.flush().await {
- Ok(entry) => Ok(entry),
- Err(err) => {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- &self.path,
- op,
- err.kind(),
- );
- Err(err)
- }
- };
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- &self.path,
- op,
- start.elapsed(),
- );
- res
+ self.inner.flush().await.inspect_err(|err| {
+ self.interceptor.observe(
+ self.labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
+ );
+ })
}
}
impl<R: oio::BlockingDelete, I: MetricsIntercept> oio::BlockingDelete for MetricsWrapper<R, I> {
fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
- let op = Operation::DeleterDelete;
-
- let start = Instant::now();
-
- let res = match self.inner.delete(path, args) {
- Ok(entry) => Ok(entry),
- Err(err) => {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- &self.path,
- op,
- err.kind(),
+ self.inner
+ .delete(path, args)
+ .inspect(|_| {
+ self.size += 1;
+ })
+ .inspect_err(|err| {
+ self.interceptor.observe(
+ self.labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
);
- Err(err)
- }
- };
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- &self.path,
- op,
- start.elapsed(),
- );
- res
+ })
}
fn flush(&mut self) -> Result<usize> {
- let op = Operation::DeleterFlush;
-
- let start = Instant::now();
-
- let res = match self.inner.flush() {
- Ok(entry) => Ok(entry),
- Err(err) => {
- self.interceptor.observe_operation_errors_total(
- self.info.clone(),
- &self.path,
- op,
- err.kind(),
- );
- Err(err)
- }
- };
- self.interceptor.observe_operation_duration_seconds(
- self.info.clone(),
- &self.path,
- op,
- start.elapsed(),
- );
- res
+ self.inner.flush().inspect_err(|err| {
+ self.interceptor.observe(
+ self.labels.clone().with_error(err.kind()),
+ MetricValue::OperationErrorsTotal,
+ );
+ })
}
}
diff --git a/core/src/layers/observe/mod.rs b/core/src/layers/observe/mod.rs
index 96aca4b8d..399eed448 100644
--- a/core/src/layers/observe/mod.rs
+++ b/core/src/layers/observe/mod.rs
@@ -43,6 +43,7 @@ pub use metrics::LABEL_OPERATION;
pub use metrics::LABEL_PATH;
pub use metrics::LABEL_ROOT;
pub use metrics::LABEL_SCHEME;
+pub use metrics::LABEL_STATUS_CODE;
pub use metrics::METRIC_HTTP_REQUEST_BYTES;
pub use metrics::METRIC_HTTP_REQUEST_DURATION_SECONDS;
pub use metrics::METRIC_OPERATION_BYTES;
diff --git a/core/src/raw/http_util/body.rs b/core/src/raw/http_util/body.rs
index 268915f88..34b1a22e5 100644
--- a/core/src/raw/http_util/body.rs
+++ b/core/src/raw/http_util/body.rs
@@ -74,6 +74,31 @@ impl HttpBody {
}
}
+ /// Map the inner stream.
+ #[cfg(not(target_arch = "wasm32"))]
+ pub(crate) fn map_inner(
+ mut self,
+ f: impl FnOnce(
+ Box<dyn Stream<Item = Result<Buffer>> + Send + Sync + Unpin + 'static>,
+ )
+ -> Box<dyn Stream<Item = Result<Buffer>> + Send + Sync + Unpin + 'static>,
+ ) -> Self {
+ self.stream = f(self.stream);
+ self
+ }
+
+ /// Map the inner stream.
+ #[cfg(target_arch = "wasm32")]
+ pub(crate) fn map_inner(
+ mut self,
+ f: impl FnOnce(
+ Box<dyn Stream<Item = Result<Buffer>> + Unpin + 'static>,
+ ) -> Box<dyn Stream<Item = Result<Buffer>> + Unpin + 'static>,
+ ) -> Self {
+ self.stream = f(self.stream);
+ self
+ }
+
/// Check if the consumed data is equal to the expected content length.
#[inline]
fn check(&self) -> Result<()> {