opendal_layer_fastmetrics/
lib.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18#![doc = include_str!("../README.md")]
19#![cfg_attr(docsrs, feature(doc_cfg))]
20#![cfg_attr(docsrs, doc(auto_cfg))]
21#![deny(missing_docs)]
22use fastmetrics::encoder::EncodeLabelSet;
23use fastmetrics::encoder::LabelSetEncoder;
24use fastmetrics::metrics::counter::Counter;
25use fastmetrics::metrics::family::Family;
26use fastmetrics::metrics::gauge::Gauge;
27use fastmetrics::metrics::histogram::Histogram;
28use fastmetrics::raw::LabelSetSchema;
29use fastmetrics::registry::Register;
30use fastmetrics::registry::Registry;
31use fastmetrics::registry::with_global_registry_mut;
32use opendal_core::raw::*;
33use opendal_core::*;
34use opendal_layer_observe_metrics_common as observe;
35
36/// `FastmetricsLayer` records OpenDAL operation and HTTP fetch metrics with
37/// [fastmetrics](https://docs.rs/fastmetrics/).
38///
39/// # Fastmetrics Metrics
40///
41/// This layer registers operation metrics and HTTP fetch metrics. Please see the documentation of
42/// the [`observe`] module for metric details.
43///
44/// # Examples
45///
46/// ## Basic Usage
47///
48/// ```no_run
49/// # use fastmetrics::format::text::encode;
50/// # use fastmetrics::format::text::TextProfile;
51/// # use log::info;
52/// # use opendal_core::services;
53/// # use opendal_core::Operator;
54/// # use opendal_core::Result;
55/// # use opendal_layer_fastmetrics::FastmetricsLayer;
56/// #
57/// # #[tokio::main]
58/// # async fn main() -> Result<()> {
59/// let mut registry = fastmetrics::registry::Registry::default();
60/// let op = Operator::new(services::Memory::default())?
61///     .layer(FastmetricsLayer::builder().register(&mut registry)?);
62///
63/// // Write data into object test.
64/// op.write("test", "Hello, World!").await?;
65///
66/// // Read data from the object.
67/// let bs = op.read("test").await?;
68/// info!("content: {}", String::from_utf8_lossy(&bs.to_bytes()));
69///
70/// // Get object metadata.
71/// let meta = op.stat("test").await?;
72/// info!("meta: {:?}", meta);
73///
74/// // Export prometheus metrics.
75/// let mut output = String::new();
76/// encode(&mut output, &registry, TextProfile::PrometheusV0_0_4).unwrap();
77/// println!("{}", output);
78/// # Ok(())
79/// # }
80/// ```
81/// ## Global Instance
82///
83/// `FastmetricsLayer` needs to be registered before instantiation.
84///
85/// If there are multiple operators in an application that need the `FastmetricsLayer`, we could
86/// instantiate it once and pass it to multiple operators. But we cannot directly call
87/// `.layer(FastmetricsLayer::builder().register(&mut registry)?)` for different services, because
88/// registering the same metrics multiple times to the same registry will cause register errors.
89/// Therefore, we can provide a global instance for the `FastmetricsLayer`.
90///
91/// ```no_run
92/// # use std::sync::OnceLock;
93/// #
94/// # use fastmetrics::format::text::encode;
95/// # use fastmetrics::format::text::TextProfile;
96/// # use fastmetrics::registry::with_global_registry;
97/// # use log::info;
98/// # use opendal_core::services;
99/// # use opendal_core::Operator;
100/// # use opendal_core::Result;
101/// # use opendal_layer_fastmetrics::FastmetricsLayer;
102/// #
103/// fn global_fastmetrics_layer() -> &'static FastmetricsLayer {
104///     static GLOBAL: OnceLock<FastmetricsLayer> = OnceLock::new();
105///     GLOBAL.get_or_init(|| {
106///         FastmetricsLayer::builder()
107///             .register_global()
108///             .expect("Failed to register with the global registry")
109///     })
110/// }
111///
112/// # #[tokio::main]
113/// # async fn main() -> Result<()> {
114/// let op = Operator::new(services::Memory::default())?
115///     .layer(global_fastmetrics_layer().clone());
116///
117/// // Write data into object test.
118/// op.write("test", "Hello, World!").await?;
119///
120/// // Read data from the object.
121/// let bs = op.read("test").await?;
122/// info!("content: {}", String::from_utf8_lossy(&bs.to_bytes()));
123///
124/// // Get object metadata.
125/// let meta = op.stat("test").await?;
126/// info!("meta: {:?}", meta);
127///
128/// // Export prometheus metrics.
129/// let mut output = String::new();
130/// with_global_registry(|reg| encode(&mut output, &reg, TextProfile::PrometheusV0_0_4).unwrap());
131/// println!("{}", output);
132/// # Ok(())
133/// # }
134/// ```
135#[derive(Clone, Debug)]
136pub struct FastmetricsLayer {
137    interceptor: FastmetricsInterceptor,
138}
139
140impl FastmetricsLayer {
141    /// Create a [`FastmetricsLayerBuilder`] to set the configuration of metrics.
142    pub fn builder() -> FastmetricsLayerBuilder {
143        FastmetricsLayerBuilder::default()
144    }
145}
146
147impl Layer for FastmetricsLayer {
148    fn apply_service(&self, inner: Servicer) -> Servicer {
149        observe::MetricsLayer::new(self.interceptor.clone()).apply_service(inner)
150    }
151
152    // Reuse the same interceptor because this layer registers both operation and HTTP metrics.
153    fn apply_context(&self, srv: Servicer, inner: OperationContext) -> OperationContext {
154        observe::MetricsLayer::new(self.interceptor.clone()).apply_context(srv, inner)
155    }
156}
157
158/// [`FastmetricsLayerBuilder`] is a config builder to build a [`FastmetricsLayer`].
159pub struct FastmetricsLayerBuilder {
160    bytes_buckets: Vec<f64>,
161    bytes_rate_buckets: Vec<f64>,
162    entries_buckets: Vec<f64>,
163    entries_rate_buckets: Vec<f64>,
164    duration_seconds_buckets: Vec<f64>,
165    ttfb_buckets: Vec<f64>,
166    disable_label_root: bool,
167}
168
169impl Default for FastmetricsLayerBuilder {
170    fn default() -> Self {
171        Self {
172            bytes_buckets: observe::DEFAULT_BYTES_BUCKETS.to_vec(),
173            bytes_rate_buckets: observe::DEFAULT_BYTES_RATE_BUCKETS.to_vec(),
174            entries_buckets: observe::DEFAULT_ENTRIES_BUCKETS.to_vec(),
175            entries_rate_buckets: observe::DEFAULT_ENTRIES_RATE_BUCKETS.to_vec(),
176            duration_seconds_buckets: observe::DEFAULT_DURATION_SECONDS_BUCKETS.to_vec(),
177            ttfb_buckets: observe::DEFAULT_TTFB_BUCKETS.to_vec(),
178            disable_label_root: false,
179        }
180    }
181}
182
183impl FastmetricsLayerBuilder {
184    /// Set buckets for bytes related histogram like `operation_bytes`.
185    pub fn bytes_buckets(mut self, buckets: Vec<f64>) -> Self {
186        if !buckets.is_empty() {
187            self.bytes_buckets = buckets;
188        }
189        self
190    }
191
192    /// Set buckets for bytes rate related histogram like `operation_bytes_rate`.
193    pub fn bytes_rate_buckets(mut self, buckets: Vec<f64>) -> Self {
194        if !buckets.is_empty() {
195            self.bytes_rate_buckets = buckets;
196        }
197        self
198    }
199
200    /// Set buckets for entries related histogram like `operation_entries`.
201    pub fn entries_buckets(mut self, buckets: Vec<f64>) -> Self {
202        if !buckets.is_empty() {
203            self.entries_buckets = buckets;
204        }
205        self
206    }
207
208    /// Set buckets for entries rate related histogram like `operation_entries_rate`.
209    pub fn entries_rate_buckets(mut self, buckets: Vec<f64>) -> Self {
210        if !buckets.is_empty() {
211            self.entries_rate_buckets = buckets;
212        }
213        self
214    }
215
216    /// Set buckets for duration seconds related histogram like `operation_duration_seconds`.
217    pub fn duration_seconds_buckets(mut self, buckets: Vec<f64>) -> Self {
218        if !buckets.is_empty() {
219            self.duration_seconds_buckets = buckets;
220        }
221        self
222    }
223
224    /// Set buckets for ttfb related histogram like `operation_ttfb_seconds`.
225    pub fn ttfb_buckets(mut self, buckets: Vec<f64>) -> Self {
226        if !buckets.is_empty() {
227            self.ttfb_buckets = buckets;
228        }
229        self
230    }
231
232    /// The 'root' label might have risks of being high cardinality; users can choose to disable it
233    /// when they found it's not useful for their metrics.
234    pub fn disable_label_root(mut self, disable: bool) -> Self {
235        self.disable_label_root = disable;
236        self
237    }
238
239    /// Register the metrics into the registry and return a [`FastmetricsLayer`].
240    ///
241    /// # Example
242    ///
243    /// ```no_run
244    /// # use opendal_core::services;
245    /// # use opendal_core::Operator;
246    /// # use opendal_core::Result;
247    /// # use opendal_layer_fastmetrics::FastmetricsLayer;
248    /// #
249    /// # fn main() -> Result<()> {
250    /// let mut registry = fastmetrics::registry::Registry::default();
251    ///
252    /// // Pick a builder and configure it.
253    /// let builder = services::Memory::default();
254    /// let _ = Operator::new(builder)?
255    ///     .layer(FastmetricsLayer::builder().register(&mut registry)?);
256    /// # Ok(())
257    /// # }
258    /// ```
259    pub fn register(self, registry: &mut Registry) -> Result<FastmetricsLayer> {
260        let Self {
261            bytes_buckets,
262            bytes_rate_buckets,
263            entries_buckets,
264            entries_rate_buckets,
265            duration_seconds_buckets,
266            ttfb_buckets,
267            disable_label_root,
268        } = self;
269
270        let new_hist_family = |buckets: Vec<f64>| -> Family<OperationLabels, Histogram> {
271            Family::new(move || Histogram::new(buckets.iter().copied()))
272        };
273
274        let operation_bytes = new_hist_family(bytes_buckets.clone());
275        let operation_bytes_rate = new_hist_family(bytes_rate_buckets.clone());
276        let operation_entries = new_hist_family(entries_buckets);
277        let operation_entries_rate = new_hist_family(entries_rate_buckets);
278        let operation_duration_seconds = new_hist_family(duration_seconds_buckets.clone());
279        let operation_errors_total = Family::default();
280        let operation_executing = Family::default();
281        let operation_ttfb_seconds = new_hist_family(ttfb_buckets);
282
283        let http_executing = Family::default();
284        let http_request_bytes = new_hist_family(bytes_buckets.clone());
285        let http_request_bytes_rate = new_hist_family(bytes_rate_buckets.clone());
286        let http_request_duration_seconds = new_hist_family(duration_seconds_buckets.clone());
287        let http_response_bytes = new_hist_family(bytes_buckets);
288        let http_response_bytes_rate = new_hist_family(bytes_rate_buckets);
289        let http_response_duration_seconds = new_hist_family(duration_seconds_buckets);
290        let http_connection_errors_total = Family::default();
291        let http_status_errors_total = Family::default();
292
293        let interceptor = FastmetricsInterceptor {
294            operation_bytes,
295            operation_bytes_rate,
296            operation_entries,
297            operation_entries_rate,
298            operation_duration_seconds,
299            operation_errors_total,
300            operation_executing,
301            operation_ttfb_seconds,
302
303            http_executing,
304            http_request_bytes,
305            http_request_bytes_rate,
306            http_request_duration_seconds,
307            http_response_bytes,
308            http_response_bytes_rate,
309            http_response_duration_seconds,
310            http_connection_errors_total,
311            http_status_errors_total,
312
313            disable_label_root,
314        };
315        interceptor
316            .register(registry)
317            .map_err(|err| Error::new(ErrorKind::Unexpected, err.to_string()).set_source(err))?;
318
319        Ok(FastmetricsLayer { interceptor })
320    }
321
322    /// Register the metrics into the global registry and return a [`FastmetricsLayer`].
323    ///
324    /// # Example
325    ///
326    /// ```no_run
327    /// # use opendal_core::services;
328    /// # use opendal_core::Operator;
329    /// # use opendal_core::Result;
330    /// # use opendal_layer_fastmetrics::FastmetricsLayer;
331    /// #
332    /// # fn main() -> Result<()> {
333    /// // Pick a builder and configure it.
334    /// let builder = services::Memory::default();
335    /// let _ = Operator::new(builder)?
336    ///     .layer(FastmetricsLayer::builder().register_global()?);
337    /// # Ok(())
338    /// # }
339    /// ```
340    pub fn register_global(self) -> Result<FastmetricsLayer> {
341        with_global_registry_mut(|registry| self.register(registry))
342    }
343}
344
345#[doc(hidden)]
346#[derive(Clone, Debug)]
347pub struct FastmetricsInterceptor {
348    operation_bytes: Family<OperationLabels, Histogram>,
349    operation_bytes_rate: Family<OperationLabels, Histogram>,
350    operation_entries: Family<OperationLabels, Histogram>,
351    operation_entries_rate: Family<OperationLabels, Histogram>,
352    operation_duration_seconds: Family<OperationLabels, Histogram>,
353    operation_errors_total: Family<OperationLabels, Counter>,
354    operation_executing: Family<OperationLabels, Gauge>,
355    operation_ttfb_seconds: Family<OperationLabels, Histogram>,
356
357    http_executing: Family<OperationLabels, Gauge>,
358    http_request_bytes: Family<OperationLabels, Histogram>,
359    http_request_bytes_rate: Family<OperationLabels, Histogram>,
360    http_request_duration_seconds: Family<OperationLabels, Histogram>,
361    http_response_bytes: Family<OperationLabels, Histogram>,
362    http_response_bytes_rate: Family<OperationLabels, Histogram>,
363    http_response_duration_seconds: Family<OperationLabels, Histogram>,
364    http_connection_errors_total: Family<OperationLabels, Counter>,
365    http_status_errors_total: Family<OperationLabels, Counter>,
366
367    disable_label_root: bool,
368}
369
370impl Register for FastmetricsInterceptor {
371    fn register(&self, registry: &mut Registry) -> fastmetrics::error::Result<()> {
372        macro_rules! register_metrics {
373            ($($field:ident => $value:expr),* $(,)?) => {
374                $(
375                    {
376                        let ((name, unit), help) = ($value.name_with_unit(), $value.help());
377                        registry.register_metric(name, help, unit, self.$field.clone())?;
378                    }
379                )*
380            };
381        }
382
383        register_metrics! {
384            // Operation metrics
385            operation_bytes => observe::MetricValue::OperationBytes(0),
386            operation_bytes_rate => observe::MetricValue::OperationBytesRate(0.0),
387            operation_entries => observe::MetricValue::OperationEntries(0),
388            operation_entries_rate => observe::MetricValue::OperationEntriesRate(0.0),
389            operation_duration_seconds => observe::MetricValue::OperationDurationSeconds(Duration::default()),
390            operation_errors_total => observe::MetricValue::OperationErrorsTotal,
391            operation_executing => observe::MetricValue::OperationExecuting(0),
392            operation_ttfb_seconds => observe::MetricValue::OperationTtfbSeconds(Duration::default()),
393
394            // HTTP metrics
395            http_executing => observe::MetricValue::HttpExecuting(0),
396            http_request_bytes => observe::MetricValue::HttpRequestBytes(0),
397            http_request_bytes_rate => observe::MetricValue::HttpRequestBytesRate(0.0),
398            http_request_duration_seconds => observe::MetricValue::HttpRequestDurationSeconds(Duration::default()),
399            http_response_bytes => observe::MetricValue::HttpResponseBytes(0),
400            http_response_bytes_rate => observe::MetricValue::HttpResponseBytesRate(0.0),
401            http_response_duration_seconds => observe::MetricValue::HttpResponseDurationSeconds(Duration::default()),
402            http_connection_errors_total => observe::MetricValue::HttpConnectionErrorsTotal,
403            http_status_errors_total => observe::MetricValue::HttpStatusErrorsTotal,
404        }
405
406        Ok(())
407    }
408}
409
410impl observe::MetricsIntercept for FastmetricsInterceptor {
411    fn observe(&self, labels: observe::MetricLabels, value: observe::MetricValue) {
412        let labels = OperationLabels {
413            labels,
414            disable_label_root: self.disable_label_root,
415        };
416        match value {
417            observe::MetricValue::OperationBytes(v) => {
418                self.operation_bytes
419                    .with_or_new(&labels, |hist| hist.observe(v as f64));
420            }
421            observe::MetricValue::OperationBytesRate(v) => {
422                self.operation_bytes_rate
423                    .with_or_new(&labels, |hist| hist.observe(v));
424            }
425            observe::MetricValue::OperationEntries(v) => {
426                self.operation_entries
427                    .with_or_new(&labels, |hist| hist.observe(v as f64));
428            }
429            observe::MetricValue::OperationEntriesRate(v) => {
430                self.operation_entries_rate
431                    .with_or_new(&labels, |hist| hist.observe(v));
432            }
433            observe::MetricValue::OperationDurationSeconds(v) => {
434                self.operation_duration_seconds
435                    .with_or_new(&labels, |hist| hist.observe(v.as_secs_f64()));
436            }
437            observe::MetricValue::OperationErrorsTotal => {
438                self.operation_errors_total
439                    .with_or_new(&labels, |counter| counter.inc());
440            }
441            observe::MetricValue::OperationExecuting(v) => {
442                self.operation_executing
443                    .with_or_new(&labels, |gauge| gauge.inc_by(v as i64));
444            }
445            observe::MetricValue::OperationTtfbSeconds(v) => {
446                self.operation_ttfb_seconds
447                    .with_or_new(&labels, |hist| hist.observe(v.as_secs_f64()));
448            }
449
450            observe::MetricValue::HttpExecuting(v) => {
451                self.http_executing
452                    .with_or_new(&labels, |gauge| gauge.inc_by(v as i64));
453            }
454            observe::MetricValue::HttpRequestBytes(v) => {
455                self.http_request_bytes
456                    .with_or_new(&labels, |hist| hist.observe(v as f64));
457            }
458            observe::MetricValue::HttpRequestBytesRate(v) => {
459                self.http_request_bytes_rate
460                    .with_or_new(&labels, |hist| hist.observe(v));
461            }
462            observe::MetricValue::HttpRequestDurationSeconds(v) => {
463                self.http_request_duration_seconds
464                    .with_or_new(&labels, |hist| hist.observe(v.as_secs_f64()));
465            }
466            observe::MetricValue::HttpResponseBytes(v) => {
467                self.http_response_bytes
468                    .with_or_new(&labels, |hist| hist.observe(v as f64));
469            }
470            observe::MetricValue::HttpResponseBytesRate(v) => {
471                self.http_response_bytes_rate
472                    .with_or_new(&labels, |hist| hist.observe(v));
473            }
474            observe::MetricValue::HttpResponseDurationSeconds(v) => {
475                self.http_response_duration_seconds
476                    .with_or_new(&labels, |hist| hist.observe(v.as_secs_f64()));
477            }
478            observe::MetricValue::HttpConnectionErrorsTotal => {
479                self.http_connection_errors_total
480                    .with_or_new(&labels, |counter| counter.inc());
481            }
482            observe::MetricValue::HttpStatusErrorsTotal => {
483                self.http_status_errors_total
484                    .with_or_new(&labels, |counter| counter.inc());
485            }
486            _ => {}
487        };
488    }
489}
490
491#[derive(Clone, Debug, PartialEq, Eq, Hash)]
492struct OperationLabels {
493    labels: observe::MetricLabels,
494    disable_label_root: bool,
495}
496
497impl LabelSetSchema for OperationLabels {
498    fn names() -> Option<&'static [&'static str]> {
499        static NAMES: &[&str] = &[
500            observe::LABEL_SCHEME,
501            observe::LABEL_NAMESPACE,
502            observe::LABEL_ROOT,
503            observe::LABEL_OPERATION,
504            observe::LABEL_ERROR,
505            observe::LABEL_STATUS_CODE,
506            observe::LABEL_SERVICE_OPERATION,
507        ];
508        Some(NAMES)
509    }
510}
511
512impl EncodeLabelSet for OperationLabels {
513    fn encode(&self, encoder: &mut dyn LabelSetEncoder) -> fastmetrics::error::Result<()> {
514        encoder.encode(&(observe::LABEL_SCHEME, self.labels.scheme))?;
515        encoder.encode(&(observe::LABEL_NAMESPACE, self.labels.namespace.as_ref()))?;
516        if !self.disable_label_root {
517            encoder.encode(&(observe::LABEL_ROOT, self.labels.root.as_ref()))?;
518        }
519        encoder.encode(&(observe::LABEL_OPERATION, self.labels.operation))?;
520        if let Some(error) = &self.labels.error {
521            encoder.encode(&(observe::LABEL_ERROR, error.into_static()))?;
522        }
523        if let Some(code) = &self.labels.status_code {
524            encoder.encode(&(observe::LABEL_STATUS_CODE, code.as_str()))?;
525        }
526        if let Some(service_operation) = self.labels.service_operation {
527            encoder.encode(&(observe::LABEL_SERVICE_OPERATION, service_operation))?;
528        }
529        Ok(())
530    }
531}
opendal_layer_fastmetrics/lib.rs

opendal_layer_fastmetrics/
lib.rs
