From af0b2f4e4696c05f0d23af1bb05e5ef14095dcfb Mon Sep 17 00:00:00 2001 From: koushiro Date: Fri, 30 Aug 2024 13:08:31 +0800 Subject: [PATCH 01/11] refactor(layers/prometheus): provide consistent APIs --- core/src/layers/observe/mod.rs | 35 ++++ core/src/layers/prometheus.rs | 306 ++++++++++++++++++++++----------- 2 files changed, 244 insertions(+), 97 deletions(-) diff --git a/core/src/layers/observe/mod.rs b/core/src/layers/observe/mod.rs index 57e26f3c4b5..d9b7cb70326 100644 --- a/core/src/layers/observe/mod.rs +++ b/core/src/layers/observe/mod.rs @@ -20,6 +20,7 @@ //! This module offers essential components to facilitate the implementation of observability in OpenDAL. mod metrics; + pub use metrics::MetricMetadata; pub use metrics::MetricsAccessor; pub use metrics::MetricsIntercept; @@ -33,3 +34,37 @@ pub use metrics::LABEL_SCHEME; pub use metrics::METRIC_OPERATION_BYTES; pub use metrics::METRIC_OPERATION_DURATION_SECONDS; pub use metrics::METRIC_OPERATION_ERRORS_TOTAL; + +pub(crate) fn path_label_value(path: &str, path_level: usize) -> Option<&str> { + if path.is_empty() { + return None; + } + + if path_level > 0 { + let label_value = path + .char_indices() + .filter(|&(_, c)| c == '/') + .nth(path_level - 1) + .map_or(path, |(i, _)| &path[..i]); + Some(label_value) + } else { + None + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_path_label_value() { + let path = "abc/def/ghi"; + assert_eq!(path_label_value(path, 0), None); + assert_eq!(path_label_value(path, 1), Some("abc")); + assert_eq!(path_label_value(path, 2), Some("abc/def")); + assert_eq!(path_label_value(path, 3), Some("abc/def/ghi")); + assert_eq!(path_label_value(path, usize::MAX), Some("abc/def/ghi")); + + assert_eq!(path_label_value("", 1), None); + } +} diff --git a/core/src/layers/prometheus.rs b/core/src/layers/prometheus.rs index adbc436ceab..8949901e15a 100644 --- a/core/src/layers/prometheus.rs +++ b/core/src/layers/prometheus.rs @@ -20,12 +20,10 @@ use std::time::Duration; use prometheus::core::AtomicU64; use prometheus::core::GenericCounterVec; -use prometheus::exponential_buckets; use prometheus::histogram_opts; -use prometheus::register_histogram_vec_with_registry; -use prometheus::register_int_counter_vec_with_registry; use prometheus::HistogramVec; use prometheus::Registry; +use prometheus::{exponential_buckets, Opts}; use crate::layers::observe; use crate::raw::Access; @@ -53,26 +51,24 @@ use crate::*; /// /// # Examples /// -/// ```no_build -/// use log::debug; -/// use log::info; -/// use opendal::layers::PrometheusLayer; -/// use opendal::services; -/// use opendal::Operator; -/// use opendal::Result; -/// use prometheus::Encoder; +/// ```no_run +/// # use log::debug; +/// # use log::info; +/// # use opendal::layers::PrometheusLayer; +/// # use opendal::services; +/// # use opendal::Operator; +/// # use opendal::Result; +/// # use prometheus::Encoder; /// -/// /// Visit [`opendal::services`] for more service related config. -/// /// Visit [`opendal::Operator`] for more operator level APIs. -/// #[tokio::main] -/// async fn main() -> Result<()> { +/// # #[tokio::main] +/// # async fn main() -> Result<()> { /// // Pick a builder and configure it. /// let builder = services::Memory::default(); /// let registry = prometheus::default_registry(); /// /// let op = Operator::new(builder) /// .expect("must init") -/// .layer(PrometheusLayer::new(registry.clone())) +/// .layer(PrometheusLayer::new(registry)) /// .finish(); /// debug!("operator: {op:?}"); /// @@ -93,47 +89,123 @@ use crate::*; /// println!("## Prometheus Metrics"); /// println!("{}", String::from_utf8(buffer.clone()).unwrap()); /// Ok(()) -/// } +/// # } /// ``` -#[derive(Clone, Debug)] +#[derive(Clone, Debug, Default)] pub struct PrometheusLayer { - registry: Registry, - operation_duration_seconds_buckets: Vec, - operation_bytes_buckets: Vec, - path_label_level: usize, + interceptor: PrometheusInterceptor, } impl PrometheusLayer { /// Create a [`PrometheusLayer`] while registering itself to this registry. - pub fn new(registry: Registry) -> Self { - Self { - registry, - operation_duration_seconds_buckets: exponential_buckets(0.01, 2.0, 16).unwrap(), - operation_bytes_buckets: exponential_buckets(1.0, 2.0, 16).unwrap(), - path_label_level: 0, - } + pub fn new(registry: &Registry) -> Self { + let interceptor = PrometheusInterceptor::default(); + Self { interceptor }.register(registry) + } + + /// Register the metrics into the registry. + /// + /// # Example + /// + /// ```no_run + /// # use log::debug; + /// # use opendal::layers::PrometheusLayer; + /// # use opendal::services; + /// # use opendal::Operator; + /// # use opendal::Result; + /// # + /// # #[tokio::main] + /// # async fn main() -> Result<()> { + /// // Pick a builder and configure it. + /// let builder = services::Memory::default(); + /// let registry = prometheus::default_registry(); + /// + /// let op = Operator::new(builder) + /// .expect("must init") + /// .layer(PrometheusLayer::default().register(registry)) + /// .finish(); + /// debug!("operator: {op:?}"); + /// + /// Ok(()) + /// # } + /// ``` + pub fn register(self, registry: &Registry) -> Self { + self.interceptor.register(registry); + self } /// Set buckets for `operation_duration_seconds` histogram. /// - /// You could call the [`linear_buckets`](https://docs.rs/prometheus/latest/prometheus/fn.linear_buckets.html) - /// or [`exponential_buckets`](https://docs.rs/prometheus/latest/prometheus/fn.exponential_buckets.html) - /// to generate the buckets. + /// # Example + /// + /// ```no_run + /// # use log::debug; + /// # use opendal::layers::PrometheusLayer; + /// # use opendal::services; + /// # use opendal::Operator; + /// # use opendal::Result; + /// # + /// # #[tokio::main] + /// # async fn main() -> Result<()> { + /// // Pick a builder and configure it. + /// let builder = services::Memory::default(); + /// let registry = prometheus::default_registry(); + /// + /// let buckets = prometheus::exponential_buckets(0.01, 2.0, 16).unwrap(); + /// let op = Operator::new(builder) + /// .expect("must init") + /// .layer( + /// PrometheusLayer::default() + /// .operation_duration_seconds_buckets(buckets) + /// .register(registry) + /// ) + /// .finish(); + /// debug!("operator: {op:?}"); + /// + /// Ok(()) + /// # } + /// ``` pub fn operation_duration_seconds_buckets(mut self, buckets: Vec) -> Self { if !buckets.is_empty() { - self.operation_duration_seconds_buckets = buckets; + self.interceptor = self.interceptor.with_operation_duration_seconds(buckets); } self } /// Set buckets for `operation_bytes` histogram. /// - /// You could call the [`linear_buckets`](https://docs.rs/prometheus/latest/prometheus/fn.linear_buckets.html) - /// or [`exponential_buckets`](https://docs.rs/prometheus/latest/prometheus/fn.exponential_buckets.html) - /// to generate the buckets. + /// # Example + /// + /// ```no_run + /// # use log::debug; + /// # use opendal::layers::PrometheusLayer; + /// # use opendal::services; + /// # use opendal::Operator; + /// # use opendal::Result; + /// # + /// # #[tokio::main] + /// # async fn main() -> Result<()> { + /// // Pick a builder and configure it. + /// let builder = services::Memory::default(); + /// let registry = prometheus::default_registry(); + /// + /// let buckets = prometheus::exponential_buckets(1.0, 2.0, 16).unwrap(); + /// let op = Operator::new(builder) + /// .expect("must init") + /// .layer( + /// PrometheusLayer::default() + /// .operation_bytes_buckets(buckets) + /// .register(registry) + /// ) + /// .finish(); + /// debug!("operator: {op:?}"); + /// + /// Ok(()) + /// # } + /// ``` pub fn operation_bytes_buckets(mut self, buckets: Vec) -> Self { if !buckets.is_empty() { - self.operation_bytes_buckets = buckets; + self.interceptor = self.interceptor.with_operation_bytes(buckets); } self } @@ -143,8 +215,37 @@ impl PrometheusLayer { /// - level = 0: we will ignore the path label. /// - level > 0: the path label will be the path split by "/" and get the last n level, /// if n=1 and input path is "abc/def/ghi", and then we will get "abc/" as the path label. + /// + /// # Example + /// + /// ```no_run + /// # use log::debug; + /// # use opendal::layers::PrometheusLayer; + /// # use opendal::services; + /// # use opendal::Operator; + /// # use opendal::Result; + /// # + /// # #[tokio::main] + /// # async fn main() -> Result<()> { + /// // Pick a builder and configure it. + /// let builder = services::Memory::default(); + /// let registry = prometheus::default_registry(); + /// + /// let op = Operator::new(builder) + /// .expect("must init") + /// .layer( + /// PrometheusLayer::default() + /// .enable_path_label(1) + /// .register(registry) + /// ) + /// .finish(); + /// debug!("operator: {op:?}"); + /// + /// Ok(()) + /// # } + /// ``` pub fn enable_path_label(mut self, level: usize) -> Self { - self.path_label_level = level; + self.interceptor = self.interceptor.with_path_label_level(level); self } } @@ -153,13 +254,7 @@ impl Layer for PrometheusLayer { type LayeredAccess = observe::MetricsAccessor; fn layer(&self, inner: A) -> Self::LayeredAccess { - let interceptor = PrometheusInterceptor::register( - self.registry.clone(), - self.operation_duration_seconds_buckets.clone(), - self.operation_bytes_buckets.clone(), - self.path_label_level, - ); - observe::MetricsLayer::new(interceptor).layer(inner) + observe::MetricsLayer::new(self.interceptor.clone()).layer(inner) } } @@ -171,43 +266,41 @@ pub struct PrometheusInterceptor { path_label_level: usize, } -impl PrometheusInterceptor { - fn register( - registry: Registry, - operation_duration_seconds_buckets: Vec, - operation_bytes_buckets: Vec, - path_label_level: usize, - ) -> Self { +impl Default for PrometheusInterceptor { + fn default() -> Self { + let operation_duration_seconds_buckets = exponential_buckets(0.01, 2.0, 16).unwrap(); + let operation_bytes_buckets = exponential_buckets(1.0, 2.0, 16).unwrap(); + let path_label_level = 0; + let labels = OperationLabels::names(false, path_label_level); - let operation_duration_seconds = register_histogram_vec_with_registry!( + let operation_duration_seconds = HistogramVec::new( histogram_opts!( observe::METRIC_OPERATION_DURATION_SECONDS.name(), observe::METRIC_OPERATION_DURATION_SECONDS.help(), operation_duration_seconds_buckets ), &labels, - registry ) - .unwrap(); - let operation_bytes = register_histogram_vec_with_registry!( + .unwrap(); + let operation_bytes = HistogramVec::new( histogram_opts!( observe::METRIC_OPERATION_BYTES.name(), observe::METRIC_OPERATION_BYTES.help(), operation_bytes_buckets ), &labels, - registry ) - .unwrap(); + .unwrap(); - let labels = OperationLabels::names(true, path_label_level); - let operation_errors_total = register_int_counter_vec_with_registry!( - observe::METRIC_OPERATION_ERRORS_TOTAL.name(), - observe::METRIC_OPERATION_ERRORS_TOTAL.help(), + let labels = OperationLabels::names(false, path_label_level); + let operation_errors_total = GenericCounterVec::new( + Opts::new( + observe::METRIC_OPERATION_ERRORS_TOTAL.name(), + observe::METRIC_OPERATION_ERRORS_TOTAL.help(), + ), &labels, - registry ) - .unwrap(); + .unwrap(); Self { operation_duration_seconds, @@ -218,6 +311,53 @@ impl PrometheusInterceptor { } } +impl PrometheusInterceptor { + fn with_operation_duration_seconds(mut self, buckets: Vec) -> Self { + let labels = OperationLabels::names(false, self.path_label_level); + self.operation_duration_seconds = HistogramVec::new( + histogram_opts!( + observe::METRIC_OPERATION_DURATION_SECONDS.name(), + observe::METRIC_OPERATION_DURATION_SECONDS.help(), + buckets + ), + &labels, + ) + .unwrap(); + self + } + + fn with_operation_bytes(mut self, buckets: Vec) -> Self { + let labels = OperationLabels::names(false, self.path_label_level); + self.operation_bytes = HistogramVec::new( + histogram_opts!( + observe::METRIC_OPERATION_BYTES.name(), + observe::METRIC_OPERATION_BYTES.help(), + buckets + ), + &labels, + ) + .unwrap(); + self + } + + fn with_path_label_level(mut self, level: usize) -> Self { + self.path_label_level = level; + self + } + + fn register(&self, registry: &Registry) { + registry + .register(Box::new(self.operation_duration_seconds.clone())) + .unwrap(); + registry + .register(Box::new(self.operation_bytes.clone())) + .unwrap(); + registry + .register(Box::new(self.operation_errors_total.clone())) + .unwrap(); + } +} + impl observe::MetricsIntercept for PrometheusInterceptor { fn observe_operation_duration_seconds( &self, @@ -236,7 +376,7 @@ impl observe::MetricsIntercept for PrometheusInterceptor { error: None, path, } - .into_values(self.path_label_level); + .into_values(self.path_label_level); self.operation_duration_seconds .with_label_values(&labels) @@ -260,7 +400,7 @@ impl observe::MetricsIntercept for PrometheusInterceptor { error: None, path, } - .into_values(self.path_label_level); + .into_values(self.path_label_level); self.operation_bytes .with_label_values(&labels) @@ -284,7 +424,7 @@ impl observe::MetricsIntercept for PrometheusInterceptor { error: Some(error), path, } - .into_values(self.path_label_level); + .into_values(self.path_label_level); self.operation_errors_total.with_label_values(&labels).inc(); } @@ -337,8 +477,8 @@ impl<'a> OperationLabels<'a> { self.op.into_static(), ]); - if path_label_level > 0 { - labels.push(get_path_label(self.path, path_label_level)); + if let Some(path) = observe::path_label_value(self.path, path_label_level) { + labels.push(path); } if let Some(error) = self.error { @@ -348,31 +488,3 @@ impl<'a> OperationLabels<'a> { labels } } - -fn get_path_label(path: &str, path_level: usize) -> &str { - if path_level > 0 { - return path - .char_indices() - .filter(|&(_, c)| c == '/') - .nth(path_level - 1) - .map_or(path, |(i, _)| &path[..i]); - } - "" -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_get_path_label() { - let path = "abc/def/ghi"; - assert_eq!(get_path_label(path, 0), ""); - assert_eq!(get_path_label(path, 1), "abc"); - assert_eq!(get_path_label(path, 2), "abc/def"); - assert_eq!(get_path_label(path, 3), "abc/def/ghi"); - assert_eq!(get_path_label(path, usize::MAX), "abc/def/ghi"); - - assert_eq!(get_path_label("", 0), ""); - } -} From af4dd4de6c1b89d90b5948e1756292ce18c392b7 Mon Sep 17 00:00:00 2001 From: koushiro Date: Fri, 30 Aug 2024 13:11:38 +0800 Subject: [PATCH 02/11] fix fmt --- core/src/layers/prometheus.rs | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/core/src/layers/prometheus.rs b/core/src/layers/prometheus.rs index 8949901e15a..dff2954e9d9 100644 --- a/core/src/layers/prometheus.rs +++ b/core/src/layers/prometheus.rs @@ -281,7 +281,7 @@ impl Default for PrometheusInterceptor { ), &labels, ) - .unwrap(); + .unwrap(); let operation_bytes = HistogramVec::new( histogram_opts!( observe::METRIC_OPERATION_BYTES.name(), @@ -290,7 +290,7 @@ impl Default for PrometheusInterceptor { ), &labels, ) - .unwrap(); + .unwrap(); let labels = OperationLabels::names(false, path_label_level); let operation_errors_total = GenericCounterVec::new( @@ -300,7 +300,7 @@ impl Default for PrometheusInterceptor { ), &labels, ) - .unwrap(); + .unwrap(); Self { operation_duration_seconds, @@ -322,7 +322,7 @@ impl PrometheusInterceptor { ), &labels, ) - .unwrap(); + .unwrap(); self } @@ -336,7 +336,7 @@ impl PrometheusInterceptor { ), &labels, ) - .unwrap(); + .unwrap(); self } @@ -376,7 +376,7 @@ impl observe::MetricsIntercept for PrometheusInterceptor { error: None, path, } - .into_values(self.path_label_level); + .into_values(self.path_label_level); self.operation_duration_seconds .with_label_values(&labels) @@ -400,7 +400,7 @@ impl observe::MetricsIntercept for PrometheusInterceptor { error: None, path, } - .into_values(self.path_label_level); + .into_values(self.path_label_level); self.operation_bytes .with_label_values(&labels) @@ -424,7 +424,7 @@ impl observe::MetricsIntercept for PrometheusInterceptor { error: Some(error), path, } - .into_values(self.path_label_level); + .into_values(self.path_label_level); self.operation_errors_total.with_label_values(&labels).inc(); } From 8a4754bb7d41bf2034fac8d27afb2fd220ca3abb Mon Sep 17 00:00:00 2001 From: koushiro Date: Fri, 30 Aug 2024 13:18:34 +0800 Subject: [PATCH 03/11] refactor(layers/prometheus-client): provide consistent APIs --- core/src/layers/prometheus_client.rs | 346 ++++++++++++++++++++------- 1 file changed, 254 insertions(+), 92 deletions(-) diff --git a/core/src/layers/prometheus_client.rs b/core/src/layers/prometheus_client.rs index 9c443ec013d..70692bf3e51 100644 --- a/core/src/layers/prometheus_client.rs +++ b/core/src/layers/prometheus_client.rs @@ -16,7 +16,6 @@ // under the License. use std::fmt; -use std::fmt::Debug; use std::sync::Arc; use std::time::Duration; @@ -25,6 +24,7 @@ use prometheus_client::encoding::EncodeLabelSet; use prometheus_client::encoding::LabelSetEncoder; use prometheus_client::metrics::counter::Counter; use prometheus_client::metrics::family::Family; +use prometheus_client::metrics::family::MetricConstructor; use prometheus_client::metrics::histogram; use prometheus_client::metrics::histogram::Histogram; use prometheus_client::registry::Registry; @@ -38,19 +38,16 @@ use crate::*; /// /// # Examples /// -/// ```no_build -/// use log::debug; -/// use log::info; -/// use opendal::layers::PrometheusClientLayer; -/// use opendal::layers::PrometheusClientInterceptor; -/// use opendal::services; -/// use opendal::Operator; -/// use opendal::Result; +/// ```no_run +/// # use log::debug; +/// # use log::info; +/// # use opendal::layers::PrometheusClientLayer; +/// # use opendal::services; +/// # use opendal::Operator; +/// # use opendal::Result; /// -/// /// Visit [`opendal::services`] for more service related config. -/// /// Visit [`opendal::Operator`] for more operator level APIs. -/// #[tokio::main] -/// async fn main() -> Result<()> { +/// # #[tokio::main] +/// # async fn main() -> Result<()> { /// // Pick a builder and configure it. /// let builder = services::Memory::default(); /// let mut registry = prometheus_client::registry::Registry::default(); @@ -77,52 +74,164 @@ use crate::*; /// println!("## Prometheus Metrics"); /// println!("{}", buf); /// Ok(()) -/// } +/// # } /// ``` -#[derive(Clone, Debug)] -pub struct PrometheusClientLayer(observe::MetricsLayer); +#[derive(Clone, Debug, Default)] +pub struct PrometheusClientLayer { + interceptor: PrometheusClientInterceptor, +} impl PrometheusClientLayer { /// Create a new [`PrometheusClientLayer`]. pub fn new(registry: &mut Registry) -> Self { - let operation_duration_seconds = Family::::new_with_constructor(|| { - let buckets = histogram::exponential_buckets(0.01, 2.0, 16); - Histogram::new(buckets) - }); - registry.register_with_unit( - observe::METRIC_OPERATION_DURATION_SECONDS.name(), - observe::METRIC_OPERATION_DURATION_SECONDS.help(), - Unit::Seconds, - operation_duration_seconds.clone(), - ); + let interceptor = PrometheusClientInterceptor::default(); + Self { interceptor }.register(registry) + } - let operation_bytes = Family::::new_with_constructor(|| { - let buckets = histogram::exponential_buckets(1.0, 2.0, 16); - Histogram::new(buckets) - }); - registry.register_with_unit( - observe::METRIC_OPERATION_BYTES.name(), - observe::METRIC_OPERATION_BYTES.help(), - Unit::Bytes, - operation_bytes.clone(), - ); + /// Register the metrics into the registry. + /// + /// # Examples + /// + /// ```no_run + /// # use log::debug; + /// # use opendal::layers::PrometheusClientLayer; + /// # use opendal::services; + /// # use opendal::Operator; + /// # use opendal::Result; + /// + /// # #[tokio::main] + /// # async fn main() -> Result<()> { + /// // Pick a builder and configure it. + /// let builder = services::Memory::default(); + /// let mut registry = prometheus_client::registry::Registry::default(); + /// + /// let op = Operator::new(builder) + /// .expect("must init") + /// .layer(PrometheusClientLayer::default().register(&mut registry)) + /// .finish(); + /// debug!("operator: {op:?}"); + /// + /// Ok(()) + /// # } + /// ``` + pub fn register(self, registry: &mut Registry) -> Self { + self.interceptor.register(registry); + self + } - let operation_errors_total = - Family::::new_with_constructor(|| { - Counter::default() - }); - registry.register( - observe::METRIC_OPERATION_ERRORS_TOTAL.name(), - observe::METRIC_OPERATION_ERRORS_TOTAL.help(), - operation_errors_total.clone(), - ); + /// Set buckets for `operation_duration_seconds` histogram. + /// + /// # Examples + /// + /// ```no_run + /// # use log::debug; + /// # use opendal::layers::PrometheusClientLayer; + /// # use opendal::services; + /// # use opendal::Operator; + /// # use opendal::Result; + /// + /// # #[tokio::main] + /// # async fn main() -> Result<()> { + /// // Pick a builder and configure it. + /// let builder = services::Memory::default(); + /// let mut registry = prometheus_client::registry::Registry::default(); + /// + /// let buckets = prometheus_client::metrics::histogram::exponential_buckets(0.01, 2.0, 16).collect(); + /// let op = Operator::new(builder) + /// .expect("must init") + /// .layer( + /// PrometheusClientLayer::default() + /// .operation_duration_seconds_buckets(buckets) + /// .register(&mut registry) + /// ) + /// .finish(); + /// debug!("operator: {op:?}"); + /// + /// Ok(()) + /// # } + /// ``` + pub fn operation_duration_seconds_buckets(mut self, buckets: Vec) -> Self { + if !buckets.is_empty() { + self.interceptor = self.interceptor.with_operation_duration_seconds(buckets); + } + self + } - let interceptor = PrometheusClientInterceptor { - operation_duration_seconds, - operation_bytes, - operation_errors_total, - }; - Self(observe::MetricsLayer::new(interceptor)) + /// Set buckets for `operation_bytes` histogram. + /// + /// # Examples + /// + /// ```no_run + /// # use log::debug; + /// # use opendal::layers::PrometheusClientLayer; + /// # use opendal::services; + /// # use opendal::Operator; + /// # use opendal::Result; + /// + /// # #[tokio::main] + /// # async fn main() -> Result<()> { + /// // Pick a builder and configure it. + /// let builder = services::Memory::default(); + /// let mut registry = prometheus_client::registry::Registry::default(); + /// + /// let buckets = prometheus_client::metrics::histogram::exponential_buckets(1.0, 2.0, 16).collect(); + /// let op = Operator::new(builder) + /// .expect("must init") + /// .layer( + /// PrometheusClientLayer::default() + /// .operation_bytes_buckets(buckets) + /// .register(&mut registry) + /// ) + /// .finish(); + /// debug!("operator: {op:?}"); + /// + /// Ok(()) + /// # } + /// ``` + pub fn operation_bytes_buckets(mut self, buckets: Vec) -> Self { + if !buckets.is_empty() { + self.interceptor = self.interceptor.with_operation_bytes(buckets); + } + self + } + + /// Set the level of path label. + /// + /// - level = 0: we will ignore the path label. + /// - level > 0: the path label will be the path split by "/" and get the last n level, + /// if n=1 and input path is "abc/def/ghi", and then we will get "abc/" as the path label. + /// + /// # Examples + /// + /// ```no_run + /// # use log::debug; + /// # use opendal::layers::PrometheusClientLayer; + /// # use opendal::services; + /// # use opendal::Operator; + /// # use opendal::Result; + /// + /// # #[tokio::main] + /// # async fn main() -> Result<()> { + /// // Pick a builder and configure it. + /// let builder = services::Memory::default(); + /// let mut registry = prometheus_client::registry::Registry::default(); + /// + /// let op = Operator::new(builder) + /// .expect("must init") + /// .layer( + /// PrometheusClientLayer::default() + /// .enable_path_label(1) + /// .register(&mut registry) + /// ) + /// .finish(); + /// debug!("operator: {op:?}"); + /// + /// Ok(()) + /// # } + /// ``` + pub fn enable_path_label(mut self, level: usize) -> Self { + self.interceptor = self.interceptor.with_path_label_level(level); + self } } @@ -130,15 +239,86 @@ impl Layer for PrometheusClientLayer { type LayeredAccess = observe::MetricsAccessor; fn layer(&self, inner: A) -> Self::LayeredAccess { - self.0.layer(inner) + observe::MetricsLayer::new(self.interceptor.clone()).layer(inner) } } -#[derive(Debug, Clone)] +#[derive(Clone, Debug)] pub struct PrometheusClientInterceptor { - operation_duration_seconds: Family, - operation_bytes: Family, - operation_errors_total: Family, + operation_duration_seconds: Family, + operation_bytes: Family, + operation_errors_total: Family, + path_label_level: usize, +} + +#[derive(Clone)] +struct HistogramConstructor { + buckets: Vec, +} + +impl MetricConstructor for HistogramConstructor { + fn new_metric(&self) -> Histogram { + Histogram::new(self.buckets.iter().cloned()) + } +} + +impl Default for PrometheusClientInterceptor { + fn default() -> Self { + let operation_duration_seconds = + Family::::new_with_constructor(HistogramConstructor { + buckets: histogram::exponential_buckets(0.01, 2.0, 16).collect(), + }); + let operation_bytes = + Family::::new_with_constructor(HistogramConstructor { + buckets: histogram::exponential_buckets(1.0, 2.0, 16).collect(), + }); + let operation_errors_total = Family::::default(); + Self { + operation_duration_seconds, + operation_bytes, + operation_errors_total, + path_label_level: 0, + } + } +} + +impl PrometheusClientInterceptor { + fn with_operation_duration_seconds(mut self, buckets: Vec) -> Self { + self.operation_duration_seconds = + Family::::new_with_constructor(HistogramConstructor { buckets }); + self + } + + fn with_operation_bytes(mut self, buckets: Vec) -> Self { + self.operation_bytes = + Family::::new_with_constructor(HistogramConstructor { buckets }); + self + } + + fn with_path_label_level(mut self, level: usize) -> Self { + self.path_label_level = level; + self + } + + fn register(&self, registry: &mut Registry) { + registry.register_with_unit( + "operation_duration", + observe::METRIC_OPERATION_DURATION_SECONDS.help(), + Unit::Seconds, + self.operation_duration_seconds.clone(), + ); + registry.register_with_unit( + "operation", + observe::METRIC_OPERATION_BYTES.help(), + Unit::Bytes, + self.operation_bytes.clone(), + ); + registry.register( + "operation_errors", + observe::METRIC_OPERATION_ERRORS_TOTAL.help(), + self.operation_errors_total.clone(), + ); + } } impl observe::MetricsIntercept for PrometheusClientInterceptor { @@ -147,7 +327,7 @@ impl observe::MetricsIntercept for PrometheusClientInterceptor { scheme: Scheme, namespace: Arc, root: Arc, - _: &str, + path: &str, op: Operation, duration: Duration, ) { @@ -156,8 +336,9 @@ impl observe::MetricsIntercept for PrometheusClientInterceptor { scheme, namespace, root, - path: None, - op, + operation: op, + path: observe::path_label_value(path, self.path_label_level).map(Into::into), + error: None, }) .observe(duration.as_secs_f64()) } @@ -167,7 +348,7 @@ impl observe::MetricsIntercept for PrometheusClientInterceptor { scheme: Scheme, namespace: Arc, root: Arc, - _: &str, + path: &str, op: Operation, bytes: usize, ) { @@ -176,8 +357,9 @@ impl observe::MetricsIntercept for PrometheusClientInterceptor { scheme, namespace, root, - path: None, - op, + operation: op, + path: observe::path_label_value(path, self.path_label_level).map(Into::into), + error: None, }) .observe(bytes as f64) } @@ -187,18 +369,18 @@ impl observe::MetricsIntercept for PrometheusClientInterceptor { scheme: Scheme, namespace: Arc, root: Arc, - _: &str, + path: &str, op: Operation, error: ErrorKind, ) { self.operation_errors_total - .get_or_create(&OperationErrorsTotalLabels { + .get_or_create(&OperationLabels { scheme, namespace, root, - path: None, - op, - error: error.into_static(), + operation: op, + path: observe::path_label_value(path, self.path_label_level).map(Into::into), + error: Some(error.into_static()), }) .inc(); } @@ -209,8 +391,9 @@ struct OperationLabels { scheme: Scheme, namespace: Arc, root: Arc, + operation: Operation, path: Option, - op: Operation, + error: Option<&'static str>, } impl EncodeLabelSet for OperationLabels { @@ -218,34 +401,13 @@ impl EncodeLabelSet for OperationLabels { (observe::LABEL_SCHEME, self.scheme.into_static()).encode(encoder.encode_label())?; (observe::LABEL_NAMESPACE, self.namespace.as_str()).encode(encoder.encode_label())?; (observe::LABEL_ROOT, self.root.as_str()).encode(encoder.encode_label())?; + (observe::LABEL_OPERATION, self.operation.into_static()).encode(encoder.encode_label())?; if let Some(path) = &self.path { (observe::LABEL_PATH, path.as_str()).encode(encoder.encode_label())?; } - (observe::LABEL_OPERATION, self.op.into_static()).encode(encoder.encode_label())?; - Ok(()) - } -} - -#[derive(Clone, Debug, Eq, PartialEq, Hash)] -struct OperationErrorsTotalLabels { - scheme: Scheme, - namespace: Arc, - root: Arc, - path: Option, - op: Operation, - error: &'static str, -} - -impl EncodeLabelSet for OperationErrorsTotalLabels { - fn encode(&self, mut encoder: LabelSetEncoder) -> Result<(), fmt::Error> { - (observe::LABEL_SCHEME, self.scheme.into_static()).encode(encoder.encode_label())?; - (observe::LABEL_NAMESPACE, self.namespace.as_str()).encode(encoder.encode_label())?; - (observe::LABEL_ROOT, self.root.as_str()).encode(encoder.encode_label())?; - if let Some(path) = &self.path { - (observe::LABEL_PATH, path.as_str()).encode(encoder.encode_label())?; + if let Some(error) = self.error { + (observe::LABEL_ERROR, error).encode(encoder.encode_label())?; } - (observe::LABEL_OPERATION, self.op.into_static()).encode(encoder.encode_label())?; - (observe::LABEL_ERROR, self.error).encode(encoder.encode_label())?; Ok(()) } } From 4136ab11d2fa674529bb4c701d74509aa9e4e88b Mon Sep 17 00:00:00 2001 From: koushiro Date: Fri, 30 Aug 2024 15:13:20 +0800 Subject: [PATCH 04/11] add opendal_ prefix --- core/src/layers/prometheus_client.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/core/src/layers/prometheus_client.rs b/core/src/layers/prometheus_client.rs index 70692bf3e51..d56ad592afa 100644 --- a/core/src/layers/prometheus_client.rs +++ b/core/src/layers/prometheus_client.rs @@ -302,19 +302,19 @@ impl PrometheusClientInterceptor { fn register(&self, registry: &mut Registry) { registry.register_with_unit( - "operation_duration", + "opendal_operation_duration", observe::METRIC_OPERATION_DURATION_SECONDS.help(), Unit::Seconds, self.operation_duration_seconds.clone(), ); registry.register_with_unit( - "operation", + "opendal_operation", observe::METRIC_OPERATION_BYTES.help(), Unit::Bytes, self.operation_bytes.clone(), ); registry.register( - "operation_errors", + "opendal_operation_errors", observe::METRIC_OPERATION_ERRORS_TOTAL.help(), self.operation_errors_total.clone(), ); From 1d8c4e36a7776d7962ea89eaf6fa63ec5209540c Mon Sep 17 00:00:00 2001 From: koushiro Date: Tue, 3 Sep 2024 15:01:36 +0800 Subject: [PATCH 05/11] update --- core/src/layers/observe/mod.rs | 7 +- core/src/layers/prometheus.rs | 266 +++++++-------------------- core/src/layers/prometheus_client.rs | 221 +++++++++++----------- 3 files changed, 179 insertions(+), 315 deletions(-) diff --git a/core/src/layers/observe/mod.rs b/core/src/layers/observe/mod.rs index d9b7cb70326..6f4586d155b 100644 --- a/core/src/layers/observe/mod.rs +++ b/core/src/layers/observe/mod.rs @@ -35,7 +35,12 @@ pub use metrics::METRIC_OPERATION_BYTES; pub use metrics::METRIC_OPERATION_DURATION_SECONDS; pub use metrics::METRIC_OPERATION_ERRORS_TOTAL; -pub(crate) fn path_label_value(path: &str, path_level: usize) -> Option<&str> { +/// Return the path label value according to the given `path` and `level`. +/// +/// - level = 0: return `None`, which means we ignore the path label. +/// - level > 0: the path label will be the path split by "/" and get the last n level, +/// if n=1 and input path is "abc/def/ghi", and then we'll use "abc/" as the path label. +pub fn path_label_value(path: &str, path_level: usize) -> Option<&str> { if path.is_empty() { return None; } diff --git a/core/src/layers/prometheus.rs b/core/src/layers/prometheus.rs index dff2954e9d9..92055ff6bcd 100644 --- a/core/src/layers/prometheus.rs +++ b/core/src/layers/prometheus.rs @@ -20,10 +20,12 @@ use std::time::Duration; use prometheus::core::AtomicU64; use prometheus::core::GenericCounterVec; +use prometheus::exponential_buckets; use prometheus::histogram_opts; +use prometheus::register_histogram_vec_with_registry; +use prometheus::register_int_counter_vec_with_registry; use prometheus::HistogramVec; use prometheus::Registry; -use prometheus::{exponential_buckets, Opts}; use crate::layers::observe; use crate::raw::Access; @@ -51,24 +53,26 @@ use crate::*; /// /// # Examples /// -/// ```no_run -/// # use log::debug; -/// # use log::info; -/// # use opendal::layers::PrometheusLayer; -/// # use opendal::services; -/// # use opendal::Operator; -/// # use opendal::Result; -/// # use prometheus::Encoder; +/// ```no_build +/// use log::debug; +/// use log::info; +/// use opendal::layers::PrometheusLayer; +/// use opendal::services; +/// use opendal::Operator; +/// use opendal::Result; +/// use prometheus::Encoder; /// -/// # #[tokio::main] -/// # async fn main() -> Result<()> { +/// /// Visit [`opendal::services`] for more service related config. +/// /// Visit [`opendal::Operator`] for more operator level APIs. +/// #[tokio::main] +/// async fn main() -> Result<()> { /// // Pick a builder and configure it. /// let builder = services::Memory::default(); /// let registry = prometheus::default_registry(); /// /// let op = Operator::new(builder) /// .expect("must init") -/// .layer(PrometheusLayer::new(registry)) +/// .layer(PrometheusLayer::new(registry.clone())) /// .finish(); /// debug!("operator: {op:?}"); /// @@ -89,123 +93,47 @@ use crate::*; /// println!("## Prometheus Metrics"); /// println!("{}", String::from_utf8(buffer.clone()).unwrap()); /// Ok(()) -/// # } +/// } /// ``` -#[derive(Clone, Debug, Default)] +#[derive(Clone, Debug)] pub struct PrometheusLayer { - interceptor: PrometheusInterceptor, + registry: Registry, + operation_duration_seconds_buckets: Vec, + operation_bytes_buckets: Vec, + path_label_level: usize, } impl PrometheusLayer { /// Create a [`PrometheusLayer`] while registering itself to this registry. - pub fn new(registry: &Registry) -> Self { - let interceptor = PrometheusInterceptor::default(); - Self { interceptor }.register(registry) - } - - /// Register the metrics into the registry. - /// - /// # Example - /// - /// ```no_run - /// # use log::debug; - /// # use opendal::layers::PrometheusLayer; - /// # use opendal::services; - /// # use opendal::Operator; - /// # use opendal::Result; - /// # - /// # #[tokio::main] - /// # async fn main() -> Result<()> { - /// // Pick a builder and configure it. - /// let builder = services::Memory::default(); - /// let registry = prometheus::default_registry(); - /// - /// let op = Operator::new(builder) - /// .expect("must init") - /// .layer(PrometheusLayer::default().register(registry)) - /// .finish(); - /// debug!("operator: {op:?}"); - /// - /// Ok(()) - /// # } - /// ``` - pub fn register(self, registry: &Registry) -> Self { - self.interceptor.register(registry); - self + pub fn new(registry: Registry) -> Self { + Self { + registry, + operation_duration_seconds_buckets: exponential_buckets(0.01, 2.0, 16).unwrap(), + operation_bytes_buckets: exponential_buckets(1.0, 2.0, 16).unwrap(), + path_label_level: 0, + } } /// Set buckets for `operation_duration_seconds` histogram. /// - /// # Example - /// - /// ```no_run - /// # use log::debug; - /// # use opendal::layers::PrometheusLayer; - /// # use opendal::services; - /// # use opendal::Operator; - /// # use opendal::Result; - /// # - /// # #[tokio::main] - /// # async fn main() -> Result<()> { - /// // Pick a builder and configure it. - /// let builder = services::Memory::default(); - /// let registry = prometheus::default_registry(); - /// - /// let buckets = prometheus::exponential_buckets(0.01, 2.0, 16).unwrap(); - /// let op = Operator::new(builder) - /// .expect("must init") - /// .layer( - /// PrometheusLayer::default() - /// .operation_duration_seconds_buckets(buckets) - /// .register(registry) - /// ) - /// .finish(); - /// debug!("operator: {op:?}"); - /// - /// Ok(()) - /// # } - /// ``` + /// You could call the [`linear_buckets`](https://docs.rs/prometheus/latest/prometheus/fn.linear_buckets.html) + /// or [`exponential_buckets`](https://docs.rs/prometheus/latest/prometheus/fn.exponential_buckets.html) + /// to generate the buckets. pub fn operation_duration_seconds_buckets(mut self, buckets: Vec) -> Self { if !buckets.is_empty() { - self.interceptor = self.interceptor.with_operation_duration_seconds(buckets); + self.operation_duration_seconds_buckets = buckets; } self } /// Set buckets for `operation_bytes` histogram. /// - /// # Example - /// - /// ```no_run - /// # use log::debug; - /// # use opendal::layers::PrometheusLayer; - /// # use opendal::services; - /// # use opendal::Operator; - /// # use opendal::Result; - /// # - /// # #[tokio::main] - /// # async fn main() -> Result<()> { - /// // Pick a builder and configure it. - /// let builder = services::Memory::default(); - /// let registry = prometheus::default_registry(); - /// - /// let buckets = prometheus::exponential_buckets(1.0, 2.0, 16).unwrap(); - /// let op = Operator::new(builder) - /// .expect("must init") - /// .layer( - /// PrometheusLayer::default() - /// .operation_bytes_buckets(buckets) - /// .register(registry) - /// ) - /// .finish(); - /// debug!("operator: {op:?}"); - /// - /// Ok(()) - /// # } - /// ``` + /// You could call the [`linear_buckets`](https://docs.rs/prometheus/latest/prometheus/fn.linear_buckets.html) + /// or [`exponential_buckets`](https://docs.rs/prometheus/latest/prometheus/fn.exponential_buckets.html) + /// to generate the buckets. pub fn operation_bytes_buckets(mut self, buckets: Vec) -> Self { if !buckets.is_empty() { - self.interceptor = self.interceptor.with_operation_bytes(buckets); + self.operation_bytes_buckets = buckets; } self } @@ -215,37 +143,8 @@ impl PrometheusLayer { /// - level = 0: we will ignore the path label. /// - level > 0: the path label will be the path split by "/" and get the last n level, /// if n=1 and input path is "abc/def/ghi", and then we will get "abc/" as the path label. - /// - /// # Example - /// - /// ```no_run - /// # use log::debug; - /// # use opendal::layers::PrometheusLayer; - /// # use opendal::services; - /// # use opendal::Operator; - /// # use opendal::Result; - /// # - /// # #[tokio::main] - /// # async fn main() -> Result<()> { - /// // Pick a builder and configure it. - /// let builder = services::Memory::default(); - /// let registry = prometheus::default_registry(); - /// - /// let op = Operator::new(builder) - /// .expect("must init") - /// .layer( - /// PrometheusLayer::default() - /// .enable_path_label(1) - /// .register(registry) - /// ) - /// .finish(); - /// debug!("operator: {op:?}"); - /// - /// Ok(()) - /// # } - /// ``` pub fn enable_path_label(mut self, level: usize) -> Self { - self.interceptor = self.interceptor.with_path_label_level(level); + self.path_label_level = level; self } } @@ -254,7 +153,13 @@ impl Layer for PrometheusLayer { type LayeredAccess = observe::MetricsAccessor; fn layer(&self, inner: A) -> Self::LayeredAccess { - observe::MetricsLayer::new(self.interceptor.clone()).layer(inner) + let interceptor = PrometheusInterceptor::register( + self.registry.clone(), + self.operation_duration_seconds_buckets.clone(), + self.operation_bytes_buckets.clone(), + self.path_label_level, + ); + observe::MetricsLayer::new(interceptor).layer(inner) } } @@ -266,39 +171,41 @@ pub struct PrometheusInterceptor { path_label_level: usize, } -impl Default for PrometheusInterceptor { - fn default() -> Self { - let operation_duration_seconds_buckets = exponential_buckets(0.01, 2.0, 16).unwrap(); - let operation_bytes_buckets = exponential_buckets(1.0, 2.0, 16).unwrap(); - let path_label_level = 0; - +impl PrometheusInterceptor { + fn register( + registry: Registry, + operation_duration_seconds_buckets: Vec, + operation_bytes_buckets: Vec, + path_label_level: usize, + ) -> Self { let labels = OperationLabels::names(false, path_label_level); - let operation_duration_seconds = HistogramVec::new( + let operation_duration_seconds = register_histogram_vec_with_registry!( histogram_opts!( observe::METRIC_OPERATION_DURATION_SECONDS.name(), observe::METRIC_OPERATION_DURATION_SECONDS.help(), operation_duration_seconds_buckets ), &labels, + registry ) .unwrap(); - let operation_bytes = HistogramVec::new( + let operation_bytes = register_histogram_vec_with_registry!( histogram_opts!( observe::METRIC_OPERATION_BYTES.name(), observe::METRIC_OPERATION_BYTES.help(), operation_bytes_buckets ), &labels, + registry ) .unwrap(); - let labels = OperationLabels::names(false, path_label_level); - let operation_errors_total = GenericCounterVec::new( - Opts::new( - observe::METRIC_OPERATION_ERRORS_TOTAL.name(), - observe::METRIC_OPERATION_ERRORS_TOTAL.help(), - ), + let labels = OperationLabels::names(true, path_label_level); + let operation_errors_total = register_int_counter_vec_with_registry!( + observe::METRIC_OPERATION_ERRORS_TOTAL.name(), + observe::METRIC_OPERATION_ERRORS_TOTAL.help(), &labels, + registry ) .unwrap(); @@ -311,53 +218,6 @@ impl Default for PrometheusInterceptor { } } -impl PrometheusInterceptor { - fn with_operation_duration_seconds(mut self, buckets: Vec) -> Self { - let labels = OperationLabels::names(false, self.path_label_level); - self.operation_duration_seconds = HistogramVec::new( - histogram_opts!( - observe::METRIC_OPERATION_DURATION_SECONDS.name(), - observe::METRIC_OPERATION_DURATION_SECONDS.help(), - buckets - ), - &labels, - ) - .unwrap(); - self - } - - fn with_operation_bytes(mut self, buckets: Vec) -> Self { - let labels = OperationLabels::names(false, self.path_label_level); - self.operation_bytes = HistogramVec::new( - histogram_opts!( - observe::METRIC_OPERATION_BYTES.name(), - observe::METRIC_OPERATION_BYTES.help(), - buckets - ), - &labels, - ) - .unwrap(); - self - } - - fn with_path_label_level(mut self, level: usize) -> Self { - self.path_label_level = level; - self - } - - fn register(&self, registry: &Registry) { - registry - .register(Box::new(self.operation_duration_seconds.clone())) - .unwrap(); - registry - .register(Box::new(self.operation_bytes.clone())) - .unwrap(); - registry - .register(Box::new(self.operation_errors_total.clone())) - .unwrap(); - } -} - impl observe::MetricsIntercept for PrometheusInterceptor { fn observe_operation_duration_seconds( &self, @@ -477,8 +337,8 @@ impl<'a> OperationLabels<'a> { self.op.into_static(), ]); - if let Some(path) = observe::path_label_value(self.path, path_label_level) { - labels.push(path); + if let Some(path_label) = observe::path_label_value(self.path, path_label_level) { + labels.push(path_label); } if let Some(error) = self.error { diff --git a/core/src/layers/prometheus_client.rs b/core/src/layers/prometheus_client.rs index d56ad592afa..dd8fba85ca3 100644 --- a/core/src/layers/prometheus_client.rs +++ b/core/src/layers/prometheus_client.rs @@ -25,10 +25,9 @@ use prometheus_client::encoding::LabelSetEncoder; use prometheus_client::metrics::counter::Counter; use prometheus_client::metrics::family::Family; use prometheus_client::metrics::family::MetricConstructor; -use prometheus_client::metrics::histogram; +use prometheus_client::metrics::histogram::exponential_buckets; use prometheus_client::metrics::histogram::Histogram; use prometheus_client::registry::Registry; -use prometheus_client::registry::Unit; use crate::layers::observe; use crate::raw::*; @@ -76,7 +75,7 @@ use crate::*; /// Ok(()) /// # } /// ``` -#[derive(Clone, Debug, Default)] +#[derive(Clone, Debug)] pub struct PrometheusClientLayer { interceptor: PrometheusClientInterceptor, } @@ -84,11 +83,64 @@ pub struct PrometheusClientLayer { impl PrometheusClientLayer { /// Create a new [`PrometheusClientLayer`]. pub fn new(registry: &mut Registry) -> Self { - let interceptor = PrometheusClientInterceptor::default(); - Self { interceptor }.register(registry) + PrometheusClientLayerBuilder::default().register(registry) + } + + /// Create a [`PrometheusClientLayerBuilder`]. + pub fn builder() -> PrometheusClientLayerBuilder { + PrometheusClientLayerBuilder::default() + } +} + +impl Layer for PrometheusClientLayer { + type LayeredAccess = observe::MetricsAccessor; + + fn layer(&self, inner: A) -> Self::LayeredAccess { + observe::MetricsLayer::new(self.interceptor.clone()).layer(inner) + } +} + +#[derive(Clone)] +struct HistogramConstructor { + buckets: Vec, +} + +impl MetricConstructor for HistogramConstructor { + fn new_metric(&self) -> Histogram { + Histogram::new(self.buckets.iter().cloned()) } +} + +/// [`PrometheusClientLayerBuilder`] is a config builder to build a [`PrometheusClientLayer`]. +pub struct PrometheusClientLayerBuilder { + operation_duration_seconds: Family, + operation_bytes: Family, + operation_errors_total: Family, + path_label_level: usize, +} - /// Register the metrics into the registry. +impl Default for PrometheusClientLayerBuilder { + fn default() -> Self { + let operation_duration_seconds = + Family::::new_with_constructor(HistogramConstructor { + buckets: exponential_buckets(0.01, 2.0, 16).collect(), + }); + let operation_bytes = + Family::::new_with_constructor(HistogramConstructor { + buckets: exponential_buckets(1.0, 2.0, 16).collect(), + }); + let operation_errors_total = Family::::default(); + Self { + operation_duration_seconds, + operation_bytes, + operation_errors_total, + path_label_level: 0, + } + } +} + +impl PrometheusClientLayerBuilder { + /// Set buckets for `operation_duration_seconds` histogram. /// /// # Examples /// @@ -105,21 +157,29 @@ impl PrometheusClientLayer { /// let builder = services::Memory::default(); /// let mut registry = prometheus_client::registry::Registry::default(); /// + /// let buckets = prometheus_client::metrics::histogram::exponential_buckets(0.01, 2.0, 16).collect(); /// let op = Operator::new(builder) /// .expect("must init") - /// .layer(PrometheusClientLayer::default().register(&mut registry)) + /// .layer( + /// PrometheusClientLayer::builder() + /// .operation_duration_seconds_buckets(buckets) + /// .register(&mut registry) + /// ) /// .finish(); /// debug!("operator: {op:?}"); /// /// Ok(()) /// # } /// ``` - pub fn register(self, registry: &mut Registry) -> Self { - self.interceptor.register(registry); + pub fn operation_duration_seconds_buckets(mut self, buckets: Vec) -> Self { + if !buckets.is_empty() { + self.operation_duration_seconds = + Family::new_with_constructor(HistogramConstructor { buckets }); + } self } - /// Set buckets for `operation_duration_seconds` histogram. + /// Set buckets for `operation_bytes` histogram. /// /// # Examples /// @@ -136,12 +196,12 @@ impl PrometheusClientLayer { /// let builder = services::Memory::default(); /// let mut registry = prometheus_client::registry::Registry::default(); /// - /// let buckets = prometheus_client::metrics::histogram::exponential_buckets(0.01, 2.0, 16).collect(); + /// let buckets = prometheus_client::metrics::histogram::exponential_buckets(1.0, 2.0, 16).collect(); /// let op = Operator::new(builder) /// .expect("must init") /// .layer( - /// PrometheusClientLayer::default() - /// .operation_duration_seconds_buckets(buckets) + /// PrometheusClientLayer::builder() + /// .operation_bytes_buckets(buckets) /// .register(&mut registry) /// ) /// .finish(); @@ -150,14 +210,18 @@ impl PrometheusClientLayer { /// Ok(()) /// # } /// ``` - pub fn operation_duration_seconds_buckets(mut self, buckets: Vec) -> Self { + pub fn operation_bytes_buckets(mut self, buckets: Vec) -> Self { if !buckets.is_empty() { - self.interceptor = self.interceptor.with_operation_duration_seconds(buckets); + self.operation_bytes = Family::new_with_constructor(HistogramConstructor { buckets }); } self } - /// Set buckets for `operation_bytes` histogram. + /// Set the level of path label. + /// + /// - level = 0: we will ignore the path label. + /// - level > 0: the path label will be the path split by "/" and get the last n level, + /// if n=1 and input path is "abc/def/ghi", and then we will get "abc/" as the path label. /// /// # Examples /// @@ -174,12 +238,11 @@ impl PrometheusClientLayer { /// let builder = services::Memory::default(); /// let mut registry = prometheus_client::registry::Registry::default(); /// - /// let buckets = prometheus_client::metrics::histogram::exponential_buckets(1.0, 2.0, 16).collect(); /// let op = Operator::new(builder) /// .expect("must init") /// .layer( - /// PrometheusClientLayer::default() - /// .operation_bytes_buckets(buckets) + /// PrometheusClientLayer::builder() + /// .enable_path_label(1) /// .register(&mut registry) /// ) /// .finish(); @@ -188,18 +251,12 @@ impl PrometheusClientLayer { /// Ok(()) /// # } /// ``` - pub fn operation_bytes_buckets(mut self, buckets: Vec) -> Self { - if !buckets.is_empty() { - self.interceptor = self.interceptor.with_operation_bytes(buckets); - } + pub fn enable_path_label(mut self, level: usize) -> Self { + self.path_label_level = level; self } - /// Set the level of path label. - /// - /// - level = 0: we will ignore the path label. - /// - level > 0: the path label will be the path split by "/" and get the last n level, - /// if n=1 and input path is "abc/def/ghi", and then we will get "abc/" as the path label. + /// Register the metrics into the registry and return a [`PrometheusClientLayer`]. /// /// # Examples /// @@ -218,109 +275,51 @@ impl PrometheusClientLayer { /// /// let op = Operator::new(builder) /// .expect("must init") - /// .layer( - /// PrometheusClientLayer::default() - /// .enable_path_label(1) - /// .register(&mut registry) - /// ) + /// .layer(PrometheusClientLayer::builder().register(&mut registry)) /// .finish(); /// debug!("operator: {op:?}"); /// /// Ok(()) /// # } /// ``` - pub fn enable_path_label(mut self, level: usize) -> Self { - self.interceptor = self.interceptor.with_path_label_level(level); - self - } -} - -impl Layer for PrometheusClientLayer { - type LayeredAccess = observe::MetricsAccessor; - - fn layer(&self, inner: A) -> Self::LayeredAccess { - observe::MetricsLayer::new(self.interceptor.clone()).layer(inner) - } -} - -#[derive(Clone, Debug)] -pub struct PrometheusClientInterceptor { - operation_duration_seconds: Family, - operation_bytes: Family, - operation_errors_total: Family, - path_label_level: usize, -} - -#[derive(Clone)] -struct HistogramConstructor { - buckets: Vec, -} - -impl MetricConstructor for HistogramConstructor { - fn new_metric(&self) -> Histogram { - Histogram::new(self.buckets.iter().cloned()) - } -} - -impl Default for PrometheusClientInterceptor { - fn default() -> Self { - let operation_duration_seconds = - Family::::new_with_constructor(HistogramConstructor { - buckets: histogram::exponential_buckets(0.01, 2.0, 16).collect(), - }); - let operation_bytes = - Family::::new_with_constructor(HistogramConstructor { - buckets: histogram::exponential_buckets(1.0, 2.0, 16).collect(), - }); - let operation_errors_total = Family::::default(); - Self { - operation_duration_seconds, - operation_bytes, - operation_errors_total, - path_label_level: 0, - } - } -} - -impl PrometheusClientInterceptor { - fn with_operation_duration_seconds(mut self, buckets: Vec) -> Self { - self.operation_duration_seconds = - Family::::new_with_constructor(HistogramConstructor { buckets }); - self - } - - fn with_operation_bytes(mut self, buckets: Vec) -> Self { - self.operation_bytes = - Family::::new_with_constructor(HistogramConstructor { buckets }); - self - } - - fn with_path_label_level(mut self, level: usize) -> Self { - self.path_label_level = level; - self - } - - fn register(&self, registry: &mut Registry) { - registry.register_with_unit( - "opendal_operation_duration", + pub fn register(self, registry: &mut Registry) -> PrometheusClientLayer { + registry.register( + observe::METRIC_OPERATION_DURATION_SECONDS.name(), observe::METRIC_OPERATION_DURATION_SECONDS.help(), - Unit::Seconds, self.operation_duration_seconds.clone(), ); - registry.register_with_unit( - "opendal_operation", + registry.register( + observe::METRIC_OPERATION_BYTES.name(), observe::METRIC_OPERATION_BYTES.help(), - Unit::Bytes, self.operation_bytes.clone(), ); + // `prometheus-client` will automatically add `_total` suffix into the name of counter + // metrics, so we can't use `METRIC_OPERATION_ERRORS_TOTAL.name()` here. registry.register( "opendal_operation_errors", observe::METRIC_OPERATION_ERRORS_TOTAL.help(), self.operation_errors_total.clone(), ); + + PrometheusClientLayer { + interceptor: PrometheusClientInterceptor { + operation_duration_seconds: self.operation_duration_seconds, + operation_bytes: self.operation_bytes, + operation_errors_total: self.operation_errors_total, + path_label_level: self.path_label_level, + }, + } } } +#[derive(Clone, Debug)] +pub struct PrometheusClientInterceptor { + operation_duration_seconds: Family, + operation_bytes: Family, + operation_errors_total: Family, + path_label_level: usize, +} + impl observe::MetricsIntercept for PrometheusClientInterceptor { fn observe_operation_duration_seconds( &self, From 91607d5cb0ed5f0325855ac44de46147094bb1a9 Mon Sep 17 00:00:00 2001 From: koushiro Date: Tue, 3 Sep 2024 15:05:22 +0800 Subject: [PATCH 06/11] update fmt --- core/src/layers/prometheus_client.rs | 21 ++++++++++----------- 1 file changed, 10 insertions(+), 11 deletions(-) diff --git a/core/src/layers/prometheus_client.rs b/core/src/layers/prometheus_client.rs index dd8fba85ca3..67614f18a4c 100644 --- a/core/src/layers/prometheus_client.rs +++ b/core/src/layers/prometheus_client.rs @@ -121,14 +121,12 @@ pub struct PrometheusClientLayerBuilder { impl Default for PrometheusClientLayerBuilder { fn default() -> Self { - let operation_duration_seconds = - Family::::new_with_constructor(HistogramConstructor { - buckets: exponential_buckets(0.01, 2.0, 16).collect(), - }); - let operation_bytes = - Family::::new_with_constructor(HistogramConstructor { - buckets: exponential_buckets(1.0, 2.0, 16).collect(), - }); + let operation_duration_seconds = Family::new_with_constructor(HistogramConstructor { + buckets: exponential_buckets(0.01, 2.0, 16).collect(), + }); + let operation_bytes = Family::new_with_constructor(HistogramConstructor { + buckets: exponential_buckets(1.0, 2.0, 16).collect(), + }); let operation_errors_total = Family::::default(); Self { operation_duration_seconds, @@ -173,8 +171,8 @@ impl PrometheusClientLayerBuilder { /// ``` pub fn operation_duration_seconds_buckets(mut self, buckets: Vec) -> Self { if !buckets.is_empty() { - self.operation_duration_seconds = - Family::new_with_constructor(HistogramConstructor { buckets }); + let constructor = HistogramConstructor { buckets }; + self.operation_duration_seconds = Family::<_, _, _>::new_with_constructor(constructor); } self } @@ -212,7 +210,8 @@ impl PrometheusClientLayerBuilder { /// ``` pub fn operation_bytes_buckets(mut self, buckets: Vec) -> Self { if !buckets.is_empty() { - self.operation_bytes = Family::new_with_constructor(HistogramConstructor { buckets }); + let constructor = HistogramConstructor { buckets }; + self.operation_bytes = Family::<_, _, _>::new_with_constructor(constructor); } self } From 9bd88f3f24fd95db0a60d4c2fd12b17e14f02f34 Mon Sep 17 00:00:00 2001 From: koushiro Date: Tue, 3 Sep 2024 15:17:18 +0800 Subject: [PATCH 07/11] improve builder --- core/src/layers/prometheus_client.rs | 69 ++++++++++++++-------------- 1 file changed, 35 insertions(+), 34 deletions(-) diff --git a/core/src/layers/prometheus_client.rs b/core/src/layers/prometheus_client.rs index 67614f18a4c..3a31c475762 100644 --- a/core/src/layers/prometheus_client.rs +++ b/core/src/layers/prometheus_client.rs @@ -100,38 +100,20 @@ impl Layer for PrometheusClientLayer { } } -#[derive(Clone)] -struct HistogramConstructor { - buckets: Vec, -} - -impl MetricConstructor for HistogramConstructor { - fn new_metric(&self) -> Histogram { - Histogram::new(self.buckets.iter().cloned()) - } -} - /// [`PrometheusClientLayerBuilder`] is a config builder to build a [`PrometheusClientLayer`]. pub struct PrometheusClientLayerBuilder { - operation_duration_seconds: Family, - operation_bytes: Family, - operation_errors_total: Family, + operation_duration_seconds_buckets: Vec, + operation_bytes_buckets: Vec, path_label_level: usize, } impl Default for PrometheusClientLayerBuilder { fn default() -> Self { - let operation_duration_seconds = Family::new_with_constructor(HistogramConstructor { - buckets: exponential_buckets(0.01, 2.0, 16).collect(), - }); - let operation_bytes = Family::new_with_constructor(HistogramConstructor { - buckets: exponential_buckets(1.0, 2.0, 16).collect(), - }); - let operation_errors_total = Family::::default(); + let operation_duration_seconds_buckets = exponential_buckets(0.01, 2.0, 16).collect(); + let operation_bytes_buckets = exponential_buckets(1.0, 2.0, 16).collect(); Self { - operation_duration_seconds, - operation_bytes, - operation_errors_total, + operation_duration_seconds_buckets, + operation_bytes_buckets, path_label_level: 0, } } @@ -171,8 +153,7 @@ impl PrometheusClientLayerBuilder { /// ``` pub fn operation_duration_seconds_buckets(mut self, buckets: Vec) -> Self { if !buckets.is_empty() { - let constructor = HistogramConstructor { buckets }; - self.operation_duration_seconds = Family::<_, _, _>::new_with_constructor(constructor); + self.operation_duration_seconds_buckets = buckets; } self } @@ -210,8 +191,7 @@ impl PrometheusClientLayerBuilder { /// ``` pub fn operation_bytes_buckets(mut self, buckets: Vec) -> Self { if !buckets.is_empty() { - let constructor = HistogramConstructor { buckets }; - self.operation_bytes = Family::<_, _, _>::new_with_constructor(constructor); + self.operation_bytes_buckets = buckets; } self } @@ -282,35 +262,56 @@ impl PrometheusClientLayerBuilder { /// # } /// ``` pub fn register(self, registry: &mut Registry) -> PrometheusClientLayer { + let operation_duration_seconds = + Family::::new_with_constructor(HistogramConstructor { + buckets: self.operation_duration_seconds_buckets, + }); + let operation_bytes = + Family::::new_with_constructor(HistogramConstructor { + buckets: self.operation_bytes_buckets, + }); + let operation_errors_total = Family::::default(); + registry.register( observe::METRIC_OPERATION_DURATION_SECONDS.name(), observe::METRIC_OPERATION_DURATION_SECONDS.help(), - self.operation_duration_seconds.clone(), + operation_duration_seconds.clone(), ); registry.register( observe::METRIC_OPERATION_BYTES.name(), observe::METRIC_OPERATION_BYTES.help(), - self.operation_bytes.clone(), + operation_bytes.clone(), ); // `prometheus-client` will automatically add `_total` suffix into the name of counter // metrics, so we can't use `METRIC_OPERATION_ERRORS_TOTAL.name()` here. registry.register( "opendal_operation_errors", observe::METRIC_OPERATION_ERRORS_TOTAL.help(), - self.operation_errors_total.clone(), + operation_errors_total.clone(), ); PrometheusClientLayer { interceptor: PrometheusClientInterceptor { - operation_duration_seconds: self.operation_duration_seconds, - operation_bytes: self.operation_bytes, - operation_errors_total: self.operation_errors_total, + operation_duration_seconds, + operation_bytes, + operation_errors_total, path_label_level: self.path_label_level, }, } } } +#[derive(Clone)] +struct HistogramConstructor { + buckets: Vec, +} + +impl MetricConstructor for HistogramConstructor { + fn new_metric(&self) -> Histogram { + Histogram::new(self.buckets.iter().cloned()) + } +} + #[derive(Clone, Debug)] pub struct PrometheusClientInterceptor { operation_duration_seconds: Family, From 97998daefb96bbc1a9abf36678acc5b39666e0ab Mon Sep 17 00:00:00 2001 From: koushiro Date: Tue, 3 Sep 2024 17:44:39 +0800 Subject: [PATCH 08/11] update doc --- core/src/layers/prometheus_client.rs | 147 ++++++--------------------- 1 file changed, 33 insertions(+), 114 deletions(-) diff --git a/core/src/layers/prometheus_client.rs b/core/src/layers/prometheus_client.rs index 3a31c475762..75f62c329df 100644 --- a/core/src/layers/prometheus_client.rs +++ b/core/src/layers/prometheus_client.rs @@ -87,6 +87,39 @@ impl PrometheusClientLayer { } /// Create a [`PrometheusClientLayerBuilder`]. + /// + /// # Examples + /// + /// ```no_run + /// # use log::debug; + /// # use opendal::layers::PrometheusClientLayer; + /// # use opendal::services; + /// # use opendal::Operator; + /// # use opendal::Result; + /// + /// # #[tokio::main] + /// # async fn main() -> Result<()> { + /// // Pick a builder and configure it. + /// let builder = services::Memory::default(); + /// let mut registry = prometheus_client::registry::Registry::default(); + /// + /// let duration_seconds_buckets = prometheus_client::metrics::histogram::exponential_buckets(0.01, 2.0, 16).collect(); + /// let bytes_buckets = prometheus_client::metrics::histogram::exponential_buckets(1.0, 2.0, 16).collect(); + /// let op = Operator::new(builder) + /// .expect("must init") + /// .layer( + /// PrometheusClientLayer::builder() + /// .operation_duration_seconds_buckets(duration_seconds_buckets) + /// .operation_bytes_buckets(bytes_buckets) + /// .enable_path_label(1) + /// .register(&mut registry) + /// ) + /// .finish(); + /// debug!("operator: {op:?}"); + /// + /// Ok(()) + /// # } + /// ``` pub fn builder() -> PrometheusClientLayerBuilder { PrometheusClientLayerBuilder::default() } @@ -121,36 +154,6 @@ impl Default for PrometheusClientLayerBuilder { impl PrometheusClientLayerBuilder { /// Set buckets for `operation_duration_seconds` histogram. - /// - /// # Examples - /// - /// ```no_run - /// # use log::debug; - /// # use opendal::layers::PrometheusClientLayer; - /// # use opendal::services; - /// # use opendal::Operator; - /// # use opendal::Result; - /// - /// # #[tokio::main] - /// # async fn main() -> Result<()> { - /// // Pick a builder and configure it. - /// let builder = services::Memory::default(); - /// let mut registry = prometheus_client::registry::Registry::default(); - /// - /// let buckets = prometheus_client::metrics::histogram::exponential_buckets(0.01, 2.0, 16).collect(); - /// let op = Operator::new(builder) - /// .expect("must init") - /// .layer( - /// PrometheusClientLayer::builder() - /// .operation_duration_seconds_buckets(buckets) - /// .register(&mut registry) - /// ) - /// .finish(); - /// debug!("operator: {op:?}"); - /// - /// Ok(()) - /// # } - /// ``` pub fn operation_duration_seconds_buckets(mut self, buckets: Vec) -> Self { if !buckets.is_empty() { self.operation_duration_seconds_buckets = buckets; @@ -159,36 +162,6 @@ impl PrometheusClientLayerBuilder { } /// Set buckets for `operation_bytes` histogram. - /// - /// # Examples - /// - /// ```no_run - /// # use log::debug; - /// # use opendal::layers::PrometheusClientLayer; - /// # use opendal::services; - /// # use opendal::Operator; - /// # use opendal::Result; - /// - /// # #[tokio::main] - /// # async fn main() -> Result<()> { - /// // Pick a builder and configure it. - /// let builder = services::Memory::default(); - /// let mut registry = prometheus_client::registry::Registry::default(); - /// - /// let buckets = prometheus_client::metrics::histogram::exponential_buckets(1.0, 2.0, 16).collect(); - /// let op = Operator::new(builder) - /// .expect("must init") - /// .layer( - /// PrometheusClientLayer::builder() - /// .operation_bytes_buckets(buckets) - /// .register(&mut registry) - /// ) - /// .finish(); - /// debug!("operator: {op:?}"); - /// - /// Ok(()) - /// # } - /// ``` pub fn operation_bytes_buckets(mut self, buckets: Vec) -> Self { if !buckets.is_empty() { self.operation_bytes_buckets = buckets; @@ -201,66 +174,12 @@ impl PrometheusClientLayerBuilder { /// - level = 0: we will ignore the path label. /// - level > 0: the path label will be the path split by "/" and get the last n level, /// if n=1 and input path is "abc/def/ghi", and then we will get "abc/" as the path label. - /// - /// # Examples - /// - /// ```no_run - /// # use log::debug; - /// # use opendal::layers::PrometheusClientLayer; - /// # use opendal::services; - /// # use opendal::Operator; - /// # use opendal::Result; - /// - /// # #[tokio::main] - /// # async fn main() -> Result<()> { - /// // Pick a builder and configure it. - /// let builder = services::Memory::default(); - /// let mut registry = prometheus_client::registry::Registry::default(); - /// - /// let op = Operator::new(builder) - /// .expect("must init") - /// .layer( - /// PrometheusClientLayer::builder() - /// .enable_path_label(1) - /// .register(&mut registry) - /// ) - /// .finish(); - /// debug!("operator: {op:?}"); - /// - /// Ok(()) - /// # } - /// ``` pub fn enable_path_label(mut self, level: usize) -> Self { self.path_label_level = level; self } /// Register the metrics into the registry and return a [`PrometheusClientLayer`]. - /// - /// # Examples - /// - /// ```no_run - /// # use log::debug; - /// # use opendal::layers::PrometheusClientLayer; - /// # use opendal::services; - /// # use opendal::Operator; - /// # use opendal::Result; - /// - /// # #[tokio::main] - /// # async fn main() -> Result<()> { - /// // Pick a builder and configure it. - /// let builder = services::Memory::default(); - /// let mut registry = prometheus_client::registry::Registry::default(); - /// - /// let op = Operator::new(builder) - /// .expect("must init") - /// .layer(PrometheusClientLayer::builder().register(&mut registry)) - /// .finish(); - /// debug!("operator: {op:?}"); - /// - /// Ok(()) - /// # } - /// ``` pub fn register(self, registry: &mut Registry) -> PrometheusClientLayer { let operation_duration_seconds = Family::::new_with_constructor(HistogramConstructor { From 968e548fecabfaeb82bab0cd306d938f49185403 Mon Sep 17 00:00:00 2001 From: koushiro Date: Tue, 3 Sep 2024 18:14:52 +0800 Subject: [PATCH 09/11] update doc --- core/src/layers/prometheus_client.rs | 29 ++++++++++++++++++++++++++-- 1 file changed, 27 insertions(+), 2 deletions(-) diff --git a/core/src/layers/prometheus_client.rs b/core/src/layers/prometheus_client.rs index 75f62c329df..eb21585553e 100644 --- a/core/src/layers/prometheus_client.rs +++ b/core/src/layers/prometheus_client.rs @@ -81,12 +81,37 @@ pub struct PrometheusClientLayer { } impl PrometheusClientLayer { - /// Create a new [`PrometheusClientLayer`]. + /// Create a new [`PrometheusClientLayer`] and register its metrics to the given registry. + /// + /// # Examples + /// + /// ```no_run + /// # use log::debug; + /// # use opendal::layers::PrometheusClientLayer; + /// # use opendal::services; + /// # use opendal::Operator; + /// # use opendal::Result; + /// + /// # #[tokio::main] + /// # async fn main() -> Result<()> { + /// // Pick a builder and configure it. + /// let builder = services::Memory::default(); + /// let mut registry = prometheus_client::registry::Registry::default(); + /// + /// let op = Operator::new(builder) + /// .expect("must init") + /// .layer(PrometheusClientLayer::new(&mut registry)) + /// .finish(); + /// debug!("operator: {op:?}"); + /// + /// Ok(()) + /// # } + /// ``` pub fn new(registry: &mut Registry) -> Self { PrometheusClientLayerBuilder::default().register(registry) } - /// Create a [`PrometheusClientLayerBuilder`]. + /// Create a [`PrometheusClientLayerBuilder`] to modify the default metric configuration. /// /// # Examples /// From 8bc68011ac55b9e4d3dfcafa26208e8a013cfe74 Mon Sep 17 00:00:00 2001 From: koushiro Date: Wed, 4 Sep 2024 01:44:01 +0800 Subject: [PATCH 10/11] update builder --- core/src/layers/prometheus.rs | 12 +- core/src/layers/prometheus_client.rs | 183 ++++++++++++++++++++------- 2 files changed, 145 insertions(+), 50 deletions(-) diff --git a/core/src/layers/prometheus.rs b/core/src/layers/prometheus.rs index 85bb29edcdd..458afd36b96 100644 --- a/core/src/layers/prometheus.rs +++ b/core/src/layers/prometheus.rs @@ -319,7 +319,7 @@ impl PrometheusLayerBuilder { ), &labels, ) - .map_err(parse_prometheus_error)?; + .map_err(parse_prometheus_error)?; let operation_bytes = HistogramVec::new( histogram_opts!( observe::METRIC_OPERATION_BYTES.name(), @@ -328,7 +328,7 @@ impl PrometheusLayerBuilder { ), &labels, ) - .map_err(parse_prometheus_error)?; + .map_err(parse_prometheus_error)?; let labels = OperationLabels::names(true, self.path_label_level); let operation_errors_total = GenericCounterVec::new( @@ -338,7 +338,7 @@ impl PrometheusLayerBuilder { ), &labels, ) - .map_err(parse_prometheus_error)?; + .map_err(parse_prometheus_error)?; registry .register(Box::new(operation_duration_seconds.clone())) @@ -425,7 +425,7 @@ impl observe::MetricsIntercept for PrometheusInterceptor { error: None, path, } - .into_values(self.path_label_level); + .into_values(self.path_label_level); self.operation_duration_seconds .with_label_values(&labels) @@ -449,7 +449,7 @@ impl observe::MetricsIntercept for PrometheusInterceptor { error: None, path, } - .into_values(self.path_label_level); + .into_values(self.path_label_level); self.operation_bytes .with_label_values(&labels) @@ -473,7 +473,7 @@ impl observe::MetricsIntercept for PrometheusInterceptor { error: Some(error), path, } - .into_values(self.path_label_level); + .into_values(self.path_label_level); self.operation_errors_total.with_label_values(&labels).inc(); } diff --git a/core/src/layers/prometheus_client.rs b/core/src/layers/prometheus_client.rs index eb21585553e..38c4120d496 100644 --- a/core/src/layers/prometheus_client.rs +++ b/core/src/layers/prometheus_client.rs @@ -35,6 +35,11 @@ use crate::*; /// Add [prometheus-client](https://docs.rs/prometheus-client) for every operation. /// +/// # Prometheus Metrics +/// +/// We provide several metrics, please see the documentation of [`observe`] module. +/// For a more detailed explanation of these metrics and how they are used, please refer to the [Prometheus documentation](https://prometheus.io/docs/introduction/overview/). +/// /// # Examples /// /// ```no_run @@ -51,9 +56,8 @@ use crate::*; /// let builder = services::Memory::default(); /// let mut registry = prometheus_client::registry::Registry::default(); /// -/// let op = Operator::new(builder) -/// .expect("must init") -/// .layer(PrometheusClientLayer::new(&mut registry)) +/// let op = Operator::new(builder)? +/// .layer(PrometheusClientLayer::builder().register(&mut registry)) /// .finish(); /// debug!("operator: {op:?}"); /// @@ -81,37 +85,13 @@ pub struct PrometheusClientLayer { } impl PrometheusClientLayer { - /// Create a new [`PrometheusClientLayer`] and register its metrics to the given registry. - /// - /// # Examples - /// - /// ```no_run - /// # use log::debug; - /// # use opendal::layers::PrometheusClientLayer; - /// # use opendal::services; - /// # use opendal::Operator; - /// # use opendal::Result; - /// - /// # #[tokio::main] - /// # async fn main() -> Result<()> { - /// // Pick a builder and configure it. - /// let builder = services::Memory::default(); - /// let mut registry = prometheus_client::registry::Registry::default(); + /// Create a [`PrometheusClientLayerBuilder`] to set the configuration of metrics. /// - /// let op = Operator::new(builder) - /// .expect("must init") - /// .layer(PrometheusClientLayer::new(&mut registry)) - /// .finish(); - /// debug!("operator: {op:?}"); + /// # Default Configuration /// - /// Ok(()) - /// # } - /// ``` - pub fn new(registry: &mut Registry) -> Self { - PrometheusClientLayerBuilder::default().register(registry) - } - - /// Create a [`PrometheusClientLayerBuilder`] to modify the default metric configuration. + /// - `operation_duration_seconds_buckets`: `exponential_buckets(0.01, 2.0, 16)` + /// - `operation_bytes_buckets`: `exponential_buckets(1.0, 2.0, 16)` + /// - `path_label`: `0` /// /// # Examples /// @@ -130,13 +110,12 @@ impl PrometheusClientLayer { /// /// let duration_seconds_buckets = prometheus_client::metrics::histogram::exponential_buckets(0.01, 2.0, 16).collect(); /// let bytes_buckets = prometheus_client::metrics::histogram::exponential_buckets(1.0, 2.0, 16).collect(); - /// let op = Operator::new(builder) - /// .expect("must init") + /// let op = Operator::new(builder)? /// .layer( /// PrometheusClientLayer::builder() /// .operation_duration_seconds_buckets(duration_seconds_buckets) /// .operation_bytes_buckets(bytes_buckets) - /// .enable_path_label(1) + /// .path_label(0) /// .register(&mut registry) /// ) /// .finish(); @@ -146,7 +125,13 @@ impl PrometheusClientLayer { /// # } /// ``` pub fn builder() -> PrometheusClientLayerBuilder { - PrometheusClientLayerBuilder::default() + let operation_duration_seconds_buckets = exponential_buckets(0.01, 2.0, 16).collect(); + let operation_bytes_buckets = exponential_buckets(1.0, 2.0, 16).collect(); + PrometheusClientLayerBuilder::new( + operation_duration_seconds_buckets, + operation_bytes_buckets, + 0, + ) } } @@ -165,20 +150,49 @@ pub struct PrometheusClientLayerBuilder { path_label_level: usize, } -impl Default for PrometheusClientLayerBuilder { - fn default() -> Self { - let operation_duration_seconds_buckets = exponential_buckets(0.01, 2.0, 16).collect(); - let operation_bytes_buckets = exponential_buckets(1.0, 2.0, 16).collect(); +impl PrometheusClientLayerBuilder { + fn new( + operation_duration_seconds_buckets: Vec, + operation_bytes_buckets: Vec, + path_label_level: usize, + ) -> Self { Self { operation_duration_seconds_buckets, operation_bytes_buckets, - path_label_level: 0, + path_label_level, } } -} -impl PrometheusClientLayerBuilder { /// Set buckets for `operation_duration_seconds` histogram. + /// + /// # Examples + /// + /// ```no_run + /// # use log::debug; + /// # use opendal::layers::PrometheusClientLayer; + /// # use opendal::services; + /// # use opendal::Operator; + /// # use opendal::Result; + /// + /// # #[tokio::main] + /// # async fn main() -> Result<()> { + /// // Pick a builder and configure it. + /// let builder = services::Memory::default(); + /// let mut registry = prometheus_client::registry::Registry::default(); + /// + /// let buckets = prometheus_client::metrics::histogram::exponential_buckets(0.01, 2.0, 16).collect(); + /// let op = Operator::new(builder)? + /// .layer( + /// PrometheusClientLayer::builder() + /// .operation_duration_seconds_buckets(buckets) + /// .register(&mut registry) + /// ) + /// .finish(); + /// debug!("operator: {op:?}"); + /// + /// Ok(()) + /// # } + /// ``` pub fn operation_duration_seconds_buckets(mut self, buckets: Vec) -> Self { if !buckets.is_empty() { self.operation_duration_seconds_buckets = buckets; @@ -187,6 +201,35 @@ impl PrometheusClientLayerBuilder { } /// Set buckets for `operation_bytes` histogram. + /// + /// # Examples + /// + /// ```no_run + /// # use log::debug; + /// # use opendal::layers::PrometheusClientLayer; + /// # use opendal::services; + /// # use opendal::Operator; + /// # use opendal::Result; + /// + /// # #[tokio::main] + /// # async fn main() -> Result<()> { + /// // Pick a builder and configure it. + /// let builder = services::Memory::default(); + /// let mut registry = prometheus_client::registry::Registry::default(); + /// + /// let buckets = prometheus_client::metrics::histogram::exponential_buckets(1.0, 2.0, 16).collect(); + /// let op = Operator::new(builder)? + /// .layer( + /// PrometheusClientLayer::builder() + /// .operation_bytes_buckets(buckets) + /// .register(&mut registry) + /// ) + /// .finish(); + /// debug!("operator: {op:?}"); + /// + /// Ok(()) + /// # } + /// ``` pub fn operation_bytes_buckets(mut self, buckets: Vec) -> Self { if !buckets.is_empty() { self.operation_bytes_buckets = buckets; @@ -199,12 +242,64 @@ impl PrometheusClientLayerBuilder { /// - level = 0: we will ignore the path label. /// - level > 0: the path label will be the path split by "/" and get the last n level, /// if n=1 and input path is "abc/def/ghi", and then we will get "abc/" as the path label. - pub fn enable_path_label(mut self, level: usize) -> Self { + /// + /// # Examples + /// + /// ```no_run + /// # use log::debug; + /// # use opendal::layers::PrometheusClientLayer; + /// # use opendal::services; + /// # use opendal::Operator; + /// # use opendal::Result; + /// + /// # #[tokio::main] + /// # async fn main() -> Result<()> { + /// // Pick a builder and configure it. + /// let builder = services::Memory::default(); + /// let mut registry = prometheus_client::registry::Registry::default(); + /// + /// let op = Operator::new(builder)? + /// .layer( + /// PrometheusClientLayer::builder() + /// .path_label(1) + /// .register(&mut registry) + /// ) + /// .finish(); + /// debug!("operator: {op:?}"); + /// + /// Ok(()) + /// # } + /// ``` + pub fn path_label(mut self, level: usize) -> Self { self.path_label_level = level; self } /// Register the metrics into the registry and return a [`PrometheusClientLayer`]. + /// + /// # Examples + /// + /// ```no_run + /// # use log::debug; + /// # use opendal::layers::PrometheusClientLayer; + /// # use opendal::services; + /// # use opendal::Operator; + /// # use opendal::Result; + /// + /// # #[tokio::main] + /// # async fn main() -> Result<()> { + /// // Pick a builder and configure it. + /// let builder = services::Memory::default(); + /// let mut registry = prometheus_client::registry::Registry::default(); + /// + /// let op = Operator::new(builder)? + /// .layer(PrometheusClientLayer::builder().register(&mut registry)) + /// .finish(); + /// debug!("operator: {op:?}"); + /// + /// Ok(()) + /// # } + /// ``` pub fn register(self, registry: &mut Registry) -> PrometheusClientLayer { let operation_duration_seconds = Family::::new_with_constructor(HistogramConstructor { From bddf0c2436742bc3a4b1a0ed1cdad10fc7bb16e7 Mon Sep 17 00:00:00 2001 From: koushiro Date: Wed, 4 Sep 2024 01:47:18 +0800 Subject: [PATCH 11/11] adjust fmt --- core/src/layers/prometheus_client.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/core/src/layers/prometheus_client.rs b/core/src/layers/prometheus_client.rs index 38c4120d496..dc61d910d42 100644 --- a/core/src/layers/prometheus_client.rs +++ b/core/src/layers/prometheus_client.rs @@ -302,11 +302,11 @@ impl PrometheusClientLayerBuilder { /// ``` pub fn register(self, registry: &mut Registry) -> PrometheusClientLayer { let operation_duration_seconds = - Family::::new_with_constructor(HistogramConstructor { + Family::::new_with_constructor(HistogramConstructor { buckets: self.operation_duration_seconds_buckets, }); let operation_bytes = - Family::::new_with_constructor(HistogramConstructor { + Family::::new_with_constructor(HistogramConstructor { buckets: self.operation_bytes_buckets, }); let operation_errors_total = Family::::default();