You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by ko...@apache.org on 2025/06/08 11:49:06 UTC
(opendal) branch add-fastmetrics-layer created (now 575710fb7)
This is an automated email from the ASF dual-hosted git repository.
koushiro pushed a change to branch add-fastmetrics-layer
in repository https://gitbox.apache.org/repos/asf/opendal.git
at 575710fb7 feat(layers): add fastmetrics layer
This branch includes the following new commits:
new 575710fb7 feat(layers): add fastmetrics layer
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: feat(layers): add fastmetrics layer
Posted by ko...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
koushiro pushed a commit to branch add-fastmetrics-layer
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 575710fb7888f2345a427a371a98ad95c2ce9d61
Author: koushiro <ko...@gmail.com>
AuthorDate: Sun Jun 8 19:48:54 2025 +0800
feat(layers): add fastmetrics layer
---
core/Cargo.lock | 15 ++
core/Cargo.toml | 4 +
core/src/layers/fastmetrics.rs | 459 +++++++++++++++++++++++++++++++++++++++++
core/src/layers/mod.rs | 7 +
4 files changed, 485 insertions(+)
diff --git a/core/Cargo.lock b/core/Cargo.lock
index eb7d85be3..f6393c455 100644
--- a/core/Cargo.lock
+++ b/core/Cargo.lock
@@ -2908,6 +2908,20 @@ dependencies = [
"web-time",
]
+[[package]]
+name = "fastmetrics"
+version = "0.3.0-rc.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "774a5f2be60bec7269d30012dae25a1ead785244a8f38e06e2dd5163b595d083"
+dependencies = [
+ "cfg-if",
+ "dtoa",
+ "foldhash",
+ "itoa",
+ "parking_lot 0.12.3",
+ "paste",
+]
+
[[package]]
name = "fastrace"
version = "0.7.9"
@@ -5356,6 +5370,7 @@ dependencies = [
"divan",
"dotenvy",
"etcd-client",
+ "fastmetrics",
"fastrace",
"fastrace-jaeger",
"flume",
diff --git a/core/Cargo.toml b/core/Cargo.toml
index 74671155f..78baea1ec 100644
--- a/core/Cargo.toml
+++ b/core/Cargo.toml
@@ -91,6 +91,8 @@ layers-mime-guess = ["dep:mime_guess"]
layers-prometheus = ["dep:prometheus"]
# Enable layers prometheus support, with prometheus-client crate
layers-prometheus-client = ["dep:prometheus-client"]
+# Enable layers fastmetrics support.
+layers-fastmetrics = ["dep:fastmetrics"]
# Enable layers fastrace support.
layers-fastrace = ["dep:fastrace"]
# Enable layers tracing support.
@@ -397,6 +399,8 @@ opentelemetry = { version = "0.30.0", optional = true }
prometheus = { version = "0.13", features = ["process"], optional = true }
# for layers-prometheus-client
prometheus-client = { version = "0.23.1", optional = true }
+# for fastmetrics
+fastmetrics = { version = "0.3.0-rc.4", optional = true }
# for layers-tracing
tracing = { version = "0.1", optional = true }
# for layers-dtrace
diff --git a/core/src/layers/fastmetrics.rs b/core/src/layers/fastmetrics.rs
new file mode 100644
index 000000000..47199320c
--- /dev/null
+++ b/core/src/layers/fastmetrics.rs
@@ -0,0 +1,459 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::{fmt, time::Duration};
+
+use fastmetrics::encoder::EncodeLabelSet;
+use fastmetrics::encoder::LabelSetEncoder;
+use fastmetrics::metrics::counter::Counter;
+use fastmetrics::metrics::family::Family;
+use fastmetrics::metrics::family::MetricFactory;
+use fastmetrics::metrics::gauge::Gauge;
+use fastmetrics::metrics::histogram::Histogram;
+use fastmetrics::registry::with_global_registry_mut;
+use fastmetrics::registry::Register;
+use fastmetrics::registry::Registry;
+use fastmetrics::registry::RegistryError;
+
+use crate::layers::observe;
+use crate::raw::*;
+use crate::*;
+
+/// Add [fastmetrics](https://docs.rs/fastmetrics/) for every operation.
+///
+/// # Examples
+///
+/// ```no_run
+/// # use opendal::layers::FastmetricsLayer;
+/// # use opendal::services;
+/// # use opendal::Operator;
+/// # use opendal::Result;
+///
+/// # fn main() -> Result<()> {
+/// let mut registry = fastmetrics::registry::Registry::default();
+/// let _ = Operator::new(services::Memory::default())?
+/// .layer(FastmetricsLayer::builder().register(&mut registry)?)
+/// .finish();
+/// # Ok(())
+/// # }
+/// ```
+#[derive(Clone, Debug)]
+pub struct FastmetricsLayer {
+ interceptor: FastmetricsInterceptor,
+}
+
+impl FastmetricsLayer {
+ /// Create a [`FastmetricsLayerBuilder`] to set the configuration of metrics.
+ pub fn builder() -> FastmetricsLayerBuilder {
+ FastmetricsLayerBuilder::default()
+ }
+}
+
+impl<A: Access> Layer<A> for FastmetricsLayer {
+ type LayeredAccess = observe::MetricsAccessor<A, FastmetricsInterceptor>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ observe::MetricsLayer::new(self.interceptor.clone()).layer(inner)
+ }
+}
+
+/// [`FastmetricsLayerBuilder`] is a config builder to build a [`FastmetricsLayer`].
+pub struct FastmetricsLayerBuilder {
+ bytes_buckets: Vec<f64>,
+ bytes_rate_buckets: Vec<f64>,
+ entries_buckets: Vec<f64>,
+ entries_rate_buckets: Vec<f64>,
+ duration_seconds_buckets: Vec<f64>,
+ ttfb_buckets: Vec<f64>,
+ disable_label_root: bool,
+}
+
+impl Default for FastmetricsLayerBuilder {
+ fn default() -> Self {
+ Self {
+ bytes_buckets: observe::DEFAULT_BYTES_BUCKETS.to_vec(),
+ bytes_rate_buckets: observe::DEFAULT_BYTES_RATE_BUCKETS.to_vec(),
+ entries_buckets: observe::DEFAULT_ENTRIES_BUCKETS.to_vec(),
+ entries_rate_buckets: observe::DEFAULT_ENTRIES_RATE_BUCKETS.to_vec(),
+ duration_seconds_buckets: observe::DEFAULT_DURATION_SECONDS_BUCKETS.to_vec(),
+ ttfb_buckets: observe::DEFAULT_TTFB_BUCKETS.to_vec(),
+ disable_label_root: false,
+ }
+ }
+}
+
+impl FastmetricsLayerBuilder {
+ /// Set buckets for bytes related histogram like `operation_bytes`.
+ pub fn bytes_buckets(mut self, buckets: Vec<f64>) -> Self {
+ if !buckets.is_empty() {
+ self.bytes_buckets = buckets;
+ }
+ self
+ }
+
+ /// Set buckets for bytes rate related histogram like `operation_bytes_rate`.
+ pub fn bytes_rate_buckets(mut self, buckets: Vec<f64>) -> Self {
+ if !buckets.is_empty() {
+ self.bytes_rate_buckets = buckets;
+ }
+ self
+ }
+
+ /// Set buckets for entries related histogram like `operation_entries`.
+ pub fn entries_buckets(mut self, buckets: Vec<f64>) -> Self {
+ if !buckets.is_empty() {
+ self.entries_buckets = buckets;
+ }
+ self
+ }
+
+ /// Set buckets for entries rate related histogram like `operation_entries_rate`.
+ pub fn entries_rate_buckets(mut self, buckets: Vec<f64>) -> Self {
+ if !buckets.is_empty() {
+ self.entries_rate_buckets = buckets;
+ }
+ self
+ }
+
+ /// Set buckets for duration seconds related histogram like `operation_duration_seconds`.
+ pub fn duration_seconds_buckets(mut self, buckets: Vec<f64>) -> Self {
+ if !buckets.is_empty() {
+ self.duration_seconds_buckets = buckets;
+ }
+ self
+ }
+
+ /// Set buckets for ttfb related histogram like `operation_ttfb_seconds`.
+ pub fn ttfb_buckets(mut self, buckets: Vec<f64>) -> Self {
+ if !buckets.is_empty() {
+ self.ttfb_buckets = buckets;
+ }
+ self
+ }
+
+ /// The 'root' label might have risks of being high cardinality; users can choose to disable it
+ /// when they found it's not useful for their metrics.
+ pub fn disable_label_root(mut self, disable: bool) -> Self {
+ self.disable_label_root = disable;
+ self
+ }
+
+ /// Register the metrics into the registry and return a [`FastmetricsLayer`].
+ ///
+ /// # Examples
+ ///
+ /// ```no_run
+ /// # use log::debug;
+ /// # use opendal::layers::FastmetricsLayer;
+ /// # use opendal::services;
+ /// # use opendal::Operator;
+ /// # use opendal::Result;
+ ///
+ /// # #[tokio::main]
+ /// # async fn main() -> Result<()> {
+ /// let mut registry = fastmetrics::registry::Registry::default();
+ ///
+ /// // Pick a builder and configure it.
+ /// let builder = services::Memory::default();
+ /// let _ = Operator::new(builder)?
+ /// .layer(FastmetricsLayer::builder().register(&mut registry)?)
+ /// .finish();
+ /// # Ok(())
+ /// # }
+ /// ```
+ pub fn register(self, registry: &mut Registry) -> Result<FastmetricsLayer> {
+ let operation_bytes = Family::<OperationLabels, Histogram, _>::new(HistogramFactory {
+ buckets: self.bytes_buckets.clone(),
+ });
+ let operation_bytes_rate = Family::<OperationLabels, Histogram, _>::new(HistogramFactory {
+ buckets: self.bytes_rate_buckets.clone(),
+ });
+ let operation_entries = Family::<OperationLabels, Histogram, _>::new(HistogramFactory {
+ buckets: self.entries_buckets.clone(),
+ });
+ let operation_entries_rate =
+ Family::<OperationLabels, Histogram, _>::new(HistogramFactory {
+ buckets: self.entries_rate_buckets.clone(),
+ });
+ let operation_duration_seconds =
+ Family::<OperationLabels, Histogram, _>::new(HistogramFactory {
+ buckets: self.duration_seconds_buckets.clone(),
+ });
+ let operation_errors_total = Family::<OperationLabels, Counter>::default();
+ let operation_executing = Family::<OperationLabels, Gauge>::default();
+ let operation_ttfb_seconds =
+ Family::<OperationLabels, Histogram, _>::new(HistogramFactory {
+ buckets: self.ttfb_buckets.clone(),
+ });
+
+ let http_executing = Family::<OperationLabels, Gauge>::default();
+ let http_request_bytes = Family::<OperationLabels, Histogram, _>::new(HistogramFactory {
+ buckets: self.bytes_buckets.clone(),
+ });
+ let http_request_bytes_rate =
+ Family::<OperationLabels, Histogram, _>::new(HistogramFactory {
+ buckets: self.bytes_rate_buckets.clone(),
+ });
+ let http_request_duration_seconds =
+ Family::<OperationLabels, Histogram, _>::new(HistogramFactory {
+ buckets: self.duration_seconds_buckets.clone(),
+ });
+ let http_response_bytes = Family::<OperationLabels, Histogram, _>::new(HistogramFactory {
+ buckets: self.bytes_buckets.clone(),
+ });
+ let http_response_bytes_rate =
+ Family::<OperationLabels, Histogram, _>::new(HistogramFactory {
+ buckets: self.bytes_rate_buckets.clone(),
+ });
+ let http_response_duration_seconds =
+ Family::<OperationLabels, Histogram, _>::new(HistogramFactory {
+ buckets: self.duration_seconds_buckets.clone(),
+ });
+ let http_connection_errors_total = Family::<OperationLabels, Counter>::default();
+ let http_status_errors_total = Family::<OperationLabels, Counter>::default();
+
+ let interceptor = FastmetricsInterceptor {
+ operation_bytes,
+ operation_bytes_rate,
+ operation_entries,
+ operation_entries_rate,
+ operation_duration_seconds,
+ operation_errors_total,
+ operation_executing,
+ operation_ttfb_seconds,
+
+ http_executing,
+ http_request_bytes,
+ http_request_bytes_rate,
+ http_request_duration_seconds,
+ http_response_bytes,
+ http_response_bytes_rate,
+ http_response_duration_seconds,
+ http_connection_errors_total,
+ http_status_errors_total,
+
+ disable_label_root: self.disable_label_root,
+ };
+ interceptor
+ .register(registry)
+ .map_err(|err| Error::new(ErrorKind::Unexpected, err.to_string()).set_source(err))?;
+
+ Ok(FastmetricsLayer { interceptor })
+ }
+
+ /// Register the metrics into the global registry and return a [`FastmetricsLayer`].
+ ///
+ /// # Examples
+ ///
+ /// ```no_run
+ /// # use opendal::layers::FastmetricsLayer;
+ /// # use opendal::services;
+ /// # use opendal::Operator;
+ /// # use opendal::Result;
+ ///
+ /// # fn main() -> Result<()> {
+ /// // Pick a builder and configure it.
+ /// let builder = services::Memory::default();
+ /// let _ = Operator::new(builder)?
+ /// .layer(FastmetricsLayer::builder().register_global()?)
+ /// .finish();
+ /// # Ok(())
+ /// # }
+ /// ```
+ pub fn register_global(self) -> Result<FastmetricsLayer> {
+ with_global_registry_mut(|registry| self.register(registry))
+ }
+}
+
+#[derive(Clone)]
+struct HistogramFactory {
+ buckets: Vec<f64>,
+}
+
+impl MetricFactory<Histogram> for HistogramFactory {
+ fn new_metric(&self) -> Histogram {
+ Histogram::new(self.buckets.iter().cloned())
+ }
+}
+
+#[derive(Clone, Debug)]
+pub struct FastmetricsInterceptor {
+ operation_bytes: Family<OperationLabels, Histogram, HistogramFactory>,
+ operation_bytes_rate: Family<OperationLabels, Histogram, HistogramFactory>,
+ operation_entries: Family<OperationLabels, Histogram, HistogramFactory>,
+ operation_entries_rate: Family<OperationLabels, Histogram, HistogramFactory>,
+ operation_duration_seconds: Family<OperationLabels, Histogram, HistogramFactory>,
+ operation_errors_total: Family<OperationLabels, Counter>,
+ operation_executing: Family<OperationLabels, Gauge>,
+ operation_ttfb_seconds: Family<OperationLabels, Histogram, HistogramFactory>,
+
+ http_executing: Family<OperationLabels, Gauge>,
+ http_request_bytes: Family<OperationLabels, Histogram, HistogramFactory>,
+ http_request_bytes_rate: Family<OperationLabels, Histogram, HistogramFactory>,
+ http_request_duration_seconds: Family<OperationLabels, Histogram, HistogramFactory>,
+ http_response_bytes: Family<OperationLabels, Histogram, HistogramFactory>,
+ http_response_bytes_rate: Family<OperationLabels, Histogram, HistogramFactory>,
+ http_response_duration_seconds: Family<OperationLabels, Histogram, HistogramFactory>,
+ http_connection_errors_total: Family<OperationLabels, Counter>,
+ http_status_errors_total: Family<OperationLabels, Counter>,
+
+ disable_label_root: bool,
+}
+
+impl Register for FastmetricsInterceptor {
+ fn register(&self, registry: &mut Registry) -> Result<(), RegistryError> {
+ macro_rules! register_metrics {
+ ($($field:ident => $value:expr),* $(,)?) => {
+ $(
+ {
+ let ((name, unit), help) = ($value.name_with_unit(), $value.help());
+ registry.register_metric(name, help, unit, self.$field.clone())?;
+ }
+ )*
+ };
+ }
+
+ register_metrics! {
+ // Operation metrics
+ operation_bytes => observe::MetricValue::OperationBytes(0),
+ operation_bytes_rate => observe::MetricValue::OperationBytesRate(0.0),
+ operation_entries => observe::MetricValue::OperationEntries(0),
+ operation_entries_rate => observe::MetricValue::OperationEntriesRate(0.0),
+ operation_duration_seconds => observe::MetricValue::OperationDurationSeconds(Duration::default()),
+ operation_errors_total => observe::MetricValue::OperationErrorsTotal,
+ operation_executing => observe::MetricValue::OperationExecuting(0),
+ operation_ttfb_seconds => observe::MetricValue::OperationTtfbSeconds(Duration::default()),
+
+ // HTTP metrics
+ http_executing => observe::MetricValue::HttpExecuting(0),
+ http_request_bytes => observe::MetricValue::HttpRequestBytes(0),
+ http_request_bytes_rate => observe::MetricValue::HttpRequestBytesRate(0.0),
+ http_request_duration_seconds => observe::MetricValue::HttpRequestDurationSeconds(Duration::default()),
+ http_response_bytes => observe::MetricValue::HttpResponseBytes(0),
+ http_response_bytes_rate => observe::MetricValue::HttpResponseBytesRate(0.0),
+ http_response_duration_seconds => observe::MetricValue::HttpResponseDurationSeconds(Duration::default()),
+ http_connection_errors_total => observe::MetricValue::HttpConnectionErrorsTotal,
+ http_status_errors_total => observe::MetricValue::HttpStatusErrorsTotal,
+ }
+
+ Ok(())
+ }
+}
+
+impl observe::MetricsIntercept for FastmetricsInterceptor {
+ fn observe(&self, labels: observe::MetricLabels, value: observe::MetricValue) {
+ let labels = OperationLabels {
+ labels,
+ disable_label_root: self.disable_label_root,
+ };
+ match value {
+ observe::MetricValue::OperationBytes(v) => {
+ self.operation_bytes
+ .with_or_new(&labels, |hist| hist.observe(v as f64));
+ }
+ observe::MetricValue::OperationBytesRate(v) => {
+ self.operation_bytes_rate
+ .with_or_new(&labels, |hist| hist.observe(v));
+ }
+ observe::MetricValue::OperationEntries(v) => {
+ self.operation_entries
+ .with_or_new(&labels, |hist| hist.observe(v as f64));
+ }
+ observe::MetricValue::OperationEntriesRate(v) => {
+ self.operation_entries_rate
+ .with_or_new(&labels, |hist| hist.observe(v));
+ }
+ observe::MetricValue::OperationDurationSeconds(v) => {
+ self.operation_duration_seconds
+ .with_or_new(&labels, |hist| hist.observe(v.as_secs_f64()));
+ }
+ observe::MetricValue::OperationErrorsTotal => {
+ self.operation_errors_total
+ .with_or_new(&labels, |counter| counter.inc());
+ }
+ observe::MetricValue::OperationExecuting(v) => {
+ self.operation_executing
+ .with_or_new(&labels, |gauge| gauge.inc_by(v as i64));
+ }
+ observe::MetricValue::OperationTtfbSeconds(v) => {
+ self.operation_ttfb_seconds
+ .with_or_new(&labels, |hist| hist.observe(v.as_secs_f64()));
+ }
+
+ observe::MetricValue::HttpExecuting(v) => {
+ self.http_executing
+ .with_or_new(&labels, |gauge| gauge.inc_by(v as i64));
+ }
+ observe::MetricValue::HttpRequestBytes(v) => {
+ self.http_request_bytes
+ .with_or_new(&labels, |hist| hist.observe(v as f64));
+ }
+ observe::MetricValue::HttpRequestBytesRate(v) => {
+ self.http_request_bytes_rate
+ .with_or_new(&labels, |hist| hist.observe(v));
+ }
+ observe::MetricValue::HttpRequestDurationSeconds(v) => {
+ self.http_request_duration_seconds
+ .with_or_new(&labels, |hist| hist.observe(v.as_secs_f64()));
+ }
+ observe::MetricValue::HttpResponseBytes(v) => {
+ self.http_response_bytes
+ .with_or_new(&labels, |hist| hist.observe(v as f64));
+ }
+ observe::MetricValue::HttpResponseBytesRate(v) => {
+ self.http_response_bytes_rate
+ .with_or_new(&labels, |hist| hist.observe(v));
+ }
+ observe::MetricValue::HttpResponseDurationSeconds(v) => {
+ self.http_response_duration_seconds
+ .with_or_new(&labels, |hist| hist.observe(v.as_secs_f64()));
+ }
+ observe::MetricValue::HttpConnectionErrorsTotal => {
+ self.http_connection_errors_total
+ .with_or_new(&labels, |counter| counter.inc());
+ }
+ observe::MetricValue::HttpStatusErrorsTotal => {
+ self.http_status_errors_total
+ .with_or_new(&labels, |counter| counter.inc());
+ }
+ };
+ }
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+struct OperationLabels {
+ labels: observe::MetricLabels,
+ disable_label_root: bool,
+}
+
+impl EncodeLabelSet for OperationLabels {
+ fn encode(&self, encoder: &mut dyn LabelSetEncoder) -> fmt::Result {
+ encoder.encode(&(observe::LABEL_SCHEME, self.labels.scheme.into_static()))?;
+ encoder.encode(&(observe::LABEL_NAMESPACE, self.labels.namespace.as_ref()))?;
+ if !self.disable_label_root {
+ encoder.encode(&(observe::LABEL_ROOT, self.labels.root.as_ref()))?;
+ }
+ encoder.encode(&(observe::LABEL_OPERATION, self.labels.operation))?;
+ if let Some(error) = &self.labels.error {
+ encoder.encode(&(observe::LABEL_ERROR, error.into_static()))?;
+ }
+ if let Some(code) = &self.labels.status_code {
+ encoder.encode(&(observe::LABEL_STATUS_CODE, code.as_str()))?;
+ }
+ Ok(())
+ }
+}
diff --git a/core/src/layers/mod.rs b/core/src/layers/mod.rs
index a4adf9161..dc8d79c29 100644
--- a/core/src/layers/mod.rs
+++ b/core/src/layers/mod.rs
@@ -68,6 +68,13 @@ pub use self::prometheus_client::PrometheusClientLayer;
#[cfg(feature = "layers-prometheus-client")]
pub use self::prometheus_client::PrometheusClientLayerBuilder;
+#[cfg(feature = "layers-fastmetrics")]
+mod fastmetrics;
+#[cfg(feature = "layers-fastmetrics")]
+pub use self::fastmetrics::FastmetricsLayer;
+#[cfg(feature = "layers-fastmetrics")]
+pub use self::fastmetrics::FastmetricsLayerBuilder;
+
mod retry;
pub use self::retry::RetryInterceptor;
pub use self::retry::RetryLayer;