opendal_layer_prometheus_client/
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 opendal_core::OperationContext;
23use opendal_core::raw::*;
24use opendal_layer_observe_metrics_common as observe;
25use prometheus_client::encoding::EncodeLabel;
26use prometheus_client::encoding::EncodeLabelSet;
27use prometheus_client::encoding::LabelSetEncoder;
28use prometheus_client::metrics::counter::Counter;
29use prometheus_client::metrics::family::Family;
30use prometheus_client::metrics::family::MetricConstructor;
31use prometheus_client::metrics::gauge::Gauge;
32use prometheus_client::metrics::histogram::Histogram;
33use prometheus_client::registry::Metric;
34use prometheus_client::registry::Registry;
35use prometheus_client::registry::Unit;
36
37/// `PrometheusClientLayer` records OpenDAL operation and HTTP fetch metrics with
38/// [prometheus-client](https://docs.rs/prometheus-client).
39///
40/// # Prometheus Metrics
41///
42/// This layer registers operation metrics and HTTP fetch metrics. Please see the documentation of [`observe`] module for metric details.
43/// 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/).
44///
45/// # Examples
46///
47/// ```no_run
48/// # use log::info;
49/// # use opendal_core::services;
50/// # use opendal_core::Operator;
51/// # use opendal_core::Result;
52/// # use opendal_layer_prometheus_client::PrometheusClientLayer;
53/// #
54/// # #[tokio::main]
55/// # async fn main() -> Result<()> {
56/// let mut registry = prometheus_client::registry::Registry::default();
57///
58/// let op = Operator::new(services::Memory::default())?
59///     .layer(PrometheusClientLayer::builder().register(&mut registry));
60///
61/// // Write data into object test.
62/// op.write("test", "Hello, World!").await?;
63/// // Read data from the object.
64/// let bs = op.read("test").await?;
65/// info!("content: {}", String::from_utf8_lossy(&bs.to_bytes()));
66///
67/// // Get object metadata.
68/// let meta = op.stat("test").await?;
69/// info!("meta: {:?}", meta);
70///
71/// // Export prometheus metrics.
72/// let mut buf = String::new();
73/// prometheus_client::encoding::text::encode(&mut buf, &registry).unwrap();
74/// println!("## Prometheus Metrics");
75/// println!("{}", buf);
76/// # Ok(())
77/// # }
78/// ```
79#[derive(Clone, Debug)]
80pub struct PrometheusClientLayer {
81    interceptor: PrometheusClientInterceptor,
82}
83
84impl PrometheusClientLayer {
85    /// Create a [`PrometheusClientLayerBuilder`] to set the configuration of metrics.
86    pub fn builder() -> PrometheusClientLayerBuilder {
87        PrometheusClientLayerBuilder::default()
88    }
89}
90
91impl Layer for PrometheusClientLayer {
92    fn apply_service(&self, inner: Servicer) -> Servicer {
93        observe::MetricsLayer::new(self.interceptor.clone()).apply_service(inner)
94    }
95
96    // Reuse the same interceptor because this layer registers both operation and HTTP metrics.
97    fn apply_context(&self, srv: Servicer, inner: OperationContext) -> OperationContext {
98        observe::MetricsLayer::new(self.interceptor.clone()).apply_context(srv, inner)
99    }
100}
101
102/// [`PrometheusClientLayerBuilder`] is a config builder to build a [`PrometheusClientLayer`].
103pub struct PrometheusClientLayerBuilder {
104    bytes_buckets: Vec<f64>,
105    bytes_rate_buckets: Vec<f64>,
106    entries_buckets: Vec<f64>,
107    entries_rate_buckets: Vec<f64>,
108    duration_seconds_buckets: Vec<f64>,
109    ttfb_buckets: Vec<f64>,
110    disable_label_root: bool,
111}
112
113impl Default for PrometheusClientLayerBuilder {
114    fn default() -> Self {
115        Self {
116            bytes_buckets: observe::DEFAULT_BYTES_BUCKETS.to_vec(),
117            bytes_rate_buckets: observe::DEFAULT_BYTES_RATE_BUCKETS.to_vec(),
118            entries_buckets: observe::DEFAULT_ENTRIES_BUCKETS.to_vec(),
119            entries_rate_buckets: observe::DEFAULT_ENTRIES_RATE_BUCKETS.to_vec(),
120            duration_seconds_buckets: observe::DEFAULT_DURATION_SECONDS_BUCKETS.to_vec(),
121            ttfb_buckets: observe::DEFAULT_TTFB_BUCKETS.to_vec(),
122            disable_label_root: false,
123        }
124    }
125}
126
127impl PrometheusClientLayerBuilder {
128    /// Set buckets for bytes related histogram like `operation_bytes`.
129    pub fn bytes_buckets(mut self, buckets: Vec<f64>) -> Self {
130        if !buckets.is_empty() {
131            self.bytes_buckets = buckets;
132        }
133        self
134    }
135
136    /// Set buckets for bytes rate related histogram like `operation_bytes_rate`.
137    pub fn bytes_rate_buckets(mut self, buckets: Vec<f64>) -> Self {
138        if !buckets.is_empty() {
139            self.bytes_rate_buckets = buckets;
140        }
141        self
142    }
143
144    /// Set buckets for entries related histogram like `operation_entries`.
145    pub fn entries_buckets(mut self, buckets: Vec<f64>) -> Self {
146        if !buckets.is_empty() {
147            self.entries_buckets = buckets;
148        }
149        self
150    }
151
152    /// Set buckets for entries rate related histogram like `operation_entries_rate`.
153    pub fn entries_rate_buckets(mut self, buckets: Vec<f64>) -> Self {
154        if !buckets.is_empty() {
155            self.entries_rate_buckets = buckets;
156        }
157        self
158    }
159
160    /// Set buckets for duration seconds related histogram like `operation_duration_seconds`.
161    pub fn duration_seconds_buckets(mut self, buckets: Vec<f64>) -> Self {
162        if !buckets.is_empty() {
163            self.duration_seconds_buckets = buckets;
164        }
165        self
166    }
167
168    /// Set buckets for ttfb related histogram like `operation_ttfb_seconds`.
169    pub fn ttfb_buckets(mut self, buckets: Vec<f64>) -> Self {
170        if !buckets.is_empty() {
171            self.ttfb_buckets = buckets;
172        }
173        self
174    }
175
176    /// The 'root' label might have risks of being high cardinality, users can choose to disable it
177    /// when they found it's not useful for their metrics.
178    pub fn disable_label_root(mut self, disable: bool) -> Self {
179        self.disable_label_root = disable;
180        self
181    }
182
183    /// Register the metrics into the registry and return a [`PrometheusClientLayer`].
184    ///
185    /// # Example
186    ///
187    /// ```no_run
188    /// # use opendal_core::services;
189    /// # use opendal_core::Operator;
190    /// # use opendal_core::Result;
191    /// # use opendal_layer_prometheus_client::PrometheusClientLayer;
192    /// #
193    /// # #[tokio::main]
194    /// # async fn main() -> Result<()> {
195    /// // Pick a builder and configure it.
196    /// let builder = services::Memory::default();
197    /// let mut registry = prometheus_client::registry::Registry::default();
198    ///
199    /// let _ = Operator::new(builder)?
200    ///     .layer(PrometheusClientLayer::builder().register(&mut registry));
201    /// # Ok(())
202    /// # }
203    /// ```
204    pub fn register(self, registry: &mut Registry) -> PrometheusClientLayer {
205        let operation_bytes =
206            Family::<OperationLabels, Histogram, _>::new_with_constructor(HistogramConstructor {
207                buckets: self.bytes_buckets.clone(),
208            });
209        let operation_bytes_rate =
210            Family::<OperationLabels, Histogram, _>::new_with_constructor(HistogramConstructor {
211                buckets: self.bytes_rate_buckets.clone(),
212            });
213        let operation_entries =
214            Family::<OperationLabels, Histogram, _>::new_with_constructor(HistogramConstructor {
215                buckets: self.entries_buckets.clone(),
216            });
217        let operation_entries_rate =
218            Family::<OperationLabels, Histogram, _>::new_with_constructor(HistogramConstructor {
219                buckets: self.entries_rate_buckets.clone(),
220            });
221        let operation_duration_seconds =
222            Family::<OperationLabels, Histogram, _>::new_with_constructor(HistogramConstructor {
223                buckets: self.duration_seconds_buckets.clone(),
224            });
225        let operation_errors_total = Family::<OperationLabels, Counter>::default();
226        let operation_executing = Family::<OperationLabels, Gauge>::default();
227        let operation_ttfb_seconds =
228            Family::<OperationLabels, Histogram, _>::new_with_constructor(HistogramConstructor {
229                buckets: self.ttfb_buckets.clone(),
230            });
231
232        let http_executing = Family::<OperationLabels, Gauge>::default();
233        let http_request_bytes =
234            Family::<OperationLabels, Histogram, _>::new_with_constructor(HistogramConstructor {
235                buckets: self.bytes_buckets.clone(),
236            });
237        let http_request_bytes_rate =
238            Family::<OperationLabels, Histogram, _>::new_with_constructor(HistogramConstructor {
239                buckets: self.bytes_rate_buckets.clone(),
240            });
241        let http_request_duration_seconds =
242            Family::<OperationLabels, Histogram, _>::new_with_constructor(HistogramConstructor {
243                buckets: self.duration_seconds_buckets.clone(),
244            });
245        let http_response_bytes =
246            Family::<OperationLabels, Histogram, _>::new_with_constructor(HistogramConstructor {
247                buckets: self.bytes_buckets.clone(),
248            });
249        let http_response_bytes_rate =
250            Family::<OperationLabels, Histogram, _>::new_with_constructor(HistogramConstructor {
251                buckets: self.bytes_rate_buckets.clone(),
252            });
253        let http_response_duration_seconds =
254            Family::<OperationLabels, Histogram, _>::new_with_constructor(HistogramConstructor {
255                buckets: self.duration_seconds_buckets.clone(),
256            });
257        let http_connection_errors_total = Family::<OperationLabels, Counter>::default();
258        let http_status_errors_total = Family::<OperationLabels, Counter>::default();
259
260        register_metric(
261            registry,
262            operation_bytes.clone(),
263            observe::MetricValue::OperationBytes(0),
264        );
265        register_metric(
266            registry,
267            operation_bytes_rate.clone(),
268            observe::MetricValue::OperationBytesRate(0.0),
269        );
270        register_metric(
271            registry,
272            operation_entries.clone(),
273            observe::MetricValue::OperationEntries(0),
274        );
275        register_metric(
276            registry,
277            operation_entries_rate.clone(),
278            observe::MetricValue::OperationEntriesRate(0.0),
279        );
280        register_metric(
281            registry,
282            operation_duration_seconds.clone(),
283            observe::MetricValue::OperationDurationSeconds(Duration::default()),
284        );
285        register_metric(
286            registry,
287            operation_errors_total.clone(),
288            observe::MetricValue::OperationErrorsTotal,
289        );
290        register_metric(
291            registry,
292            operation_executing.clone(),
293            observe::MetricValue::OperationExecuting(0),
294        );
295        register_metric(
296            registry,
297            operation_ttfb_seconds.clone(),
298            observe::MetricValue::OperationTtfbSeconds(Duration::default()),
299        );
300
301        register_metric(
302            registry,
303            http_executing.clone(),
304            observe::MetricValue::HttpExecuting(0),
305        );
306        register_metric(
307            registry,
308            http_request_bytes.clone(),
309            observe::MetricValue::HttpRequestBytes(0),
310        );
311        register_metric(
312            registry,
313            http_request_bytes_rate.clone(),
314            observe::MetricValue::HttpRequestBytesRate(0.0),
315        );
316        register_metric(
317            registry,
318            http_request_duration_seconds.clone(),
319            observe::MetricValue::HttpRequestDurationSeconds(Duration::default()),
320        );
321        register_metric(
322            registry,
323            http_response_bytes.clone(),
324            observe::MetricValue::HttpResponseBytes(0),
325        );
326        register_metric(
327            registry,
328            http_response_bytes_rate.clone(),
329            observe::MetricValue::HttpResponseBytesRate(0.0),
330        );
331        register_metric(
332            registry,
333            http_response_duration_seconds.clone(),
334            observe::MetricValue::HttpResponseDurationSeconds(Duration::default()),
335        );
336        register_metric(
337            registry,
338            http_connection_errors_total.clone(),
339            observe::MetricValue::HttpConnectionErrorsTotal,
340        );
341        register_metric(
342            registry,
343            http_status_errors_total.clone(),
344            observe::MetricValue::HttpStatusErrorsTotal,
345        );
346
347        PrometheusClientLayer {
348            interceptor: PrometheusClientInterceptor {
349                operation_bytes,
350                operation_bytes_rate,
351                operation_entries,
352                operation_entries_rate,
353                operation_duration_seconds,
354                operation_errors_total,
355                operation_executing,
356                operation_ttfb_seconds,
357
358                http_executing,
359                http_request_bytes,
360                http_request_bytes_rate,
361                http_request_duration_seconds,
362                http_response_bytes,
363                http_response_bytes_rate,
364                http_response_duration_seconds,
365                http_connection_errors_total,
366                http_status_errors_total,
367
368                disable_label_root: self.disable_label_root,
369            },
370        }
371    }
372}
373
374#[derive(Clone)]
375struct HistogramConstructor {
376    buckets: Vec<f64>,
377}
378
379impl MetricConstructor<Histogram> for HistogramConstructor {
380    fn new_metric(&self) -> Histogram {
381        Histogram::new(self.buckets.iter().cloned())
382    }
383}
384
385#[doc(hidden)]
386#[derive(Clone, Debug)]
387pub struct PrometheusClientInterceptor {
388    operation_bytes: Family<OperationLabels, Histogram, HistogramConstructor>,
389    operation_bytes_rate: Family<OperationLabels, Histogram, HistogramConstructor>,
390    operation_entries: Family<OperationLabels, Histogram, HistogramConstructor>,
391    operation_entries_rate: Family<OperationLabels, Histogram, HistogramConstructor>,
392    operation_duration_seconds: Family<OperationLabels, Histogram, HistogramConstructor>,
393    operation_errors_total: Family<OperationLabels, Counter>,
394    operation_executing: Family<OperationLabels, Gauge>,
395    operation_ttfb_seconds: Family<OperationLabels, Histogram, HistogramConstructor>,
396
397    http_executing: Family<OperationLabels, Gauge>,
398    http_request_bytes: Family<OperationLabels, Histogram, HistogramConstructor>,
399    http_request_bytes_rate: Family<OperationLabels, Histogram, HistogramConstructor>,
400    http_request_duration_seconds: Family<OperationLabels, Histogram, HistogramConstructor>,
401    http_response_bytes: Family<OperationLabels, Histogram, HistogramConstructor>,
402    http_response_bytes_rate: Family<OperationLabels, Histogram, HistogramConstructor>,
403    http_response_duration_seconds: Family<OperationLabels, Histogram, HistogramConstructor>,
404    http_connection_errors_total: Family<OperationLabels, Counter>,
405    http_status_errors_total: Family<OperationLabels, Counter>,
406
407    disable_label_root: bool,
408}
409
410impl observe::MetricsIntercept for PrometheusClientInterceptor {
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) => self
418                .operation_bytes
419                .get_or_create(&labels)
420                .observe(v as f64),
421            observe::MetricValue::OperationBytesRate(v) => {
422                self.operation_bytes_rate.get_or_create(&labels).observe(v)
423            }
424            observe::MetricValue::OperationEntries(v) => self
425                .operation_entries
426                .get_or_create(&labels)
427                .observe(v as f64),
428            observe::MetricValue::OperationEntriesRate(v) => self
429                .operation_entries_rate
430                .get_or_create(&labels)
431                .observe(v),
432            observe::MetricValue::OperationDurationSeconds(v) => self
433                .operation_duration_seconds
434                .get_or_create(&labels)
435                .observe(v.as_secs_f64()),
436            observe::MetricValue::OperationErrorsTotal => {
437                self.operation_errors_total.get_or_create(&labels).inc();
438            }
439            observe::MetricValue::OperationExecuting(v) => {
440                self.operation_executing
441                    .get_or_create(&labels)
442                    .inc_by(v as i64);
443            }
444            observe::MetricValue::OperationTtfbSeconds(v) => self
445                .operation_ttfb_seconds
446                .get_or_create(&labels)
447                .observe(v.as_secs_f64()),
448
449            observe::MetricValue::HttpExecuting(v) => {
450                self.http_executing.get_or_create(&labels).inc_by(v as i64);
451            }
452            observe::MetricValue::HttpRequestBytes(v) => self
453                .http_request_bytes
454                .get_or_create(&labels)
455                .observe(v as f64),
456            observe::MetricValue::HttpRequestBytesRate(v) => self
457                .http_request_bytes_rate
458                .get_or_create(&labels)
459                .observe(v),
460            observe::MetricValue::HttpRequestDurationSeconds(v) => self
461                .http_request_duration_seconds
462                .get_or_create(&labels)
463                .observe(v.as_secs_f64()),
464            observe::MetricValue::HttpResponseBytes(v) => self
465                .http_response_bytes
466                .get_or_create(&labels)
467                .observe(v as f64),
468            observe::MetricValue::HttpResponseBytesRate(v) => self
469                .http_response_bytes_rate
470                .get_or_create(&labels)
471                .observe(v),
472            observe::MetricValue::HttpResponseDurationSeconds(v) => self
473                .http_response_duration_seconds
474                .get_or_create(&labels)
475                .observe(v.as_secs_f64()),
476            observe::MetricValue::HttpConnectionErrorsTotal => {
477                self.http_connection_errors_total
478                    .get_or_create(&labels)
479                    .inc();
480            }
481            observe::MetricValue::HttpStatusErrorsTotal => {
482                self.http_status_errors_total.get_or_create(&labels).inc();
483            }
484            _ => {}
485        };
486    }
487}
488
489#[derive(Clone, Debug, PartialEq, Eq, Hash)]
490struct OperationLabels {
491    labels: observe::MetricLabels,
492    disable_label_root: bool,
493}
494
495impl EncodeLabelSet for OperationLabels {
496    fn encode(&self, encoder: &mut LabelSetEncoder<'_>) -> std::fmt::Result {
497        (observe::LABEL_SCHEME, self.labels.scheme).encode(encoder.encode_label())?;
498        (observe::LABEL_NAMESPACE, self.labels.namespace.as_ref())
499            .encode(encoder.encode_label())?;
500        if !self.disable_label_root {
501            (observe::LABEL_ROOT, self.labels.root.as_ref()).encode(encoder.encode_label())?;
502        }
503        (observe::LABEL_OPERATION, self.labels.operation).encode(encoder.encode_label())?;
504
505        if let Some(error) = &self.labels.error {
506            (observe::LABEL_ERROR, error.into_static()).encode(encoder.encode_label())?;
507        }
508        if let Some(code) = &self.labels.status_code {
509            (observe::LABEL_STATUS_CODE, code.as_str()).encode(encoder.encode_label())?;
510        }
511        if let Some(service_operation) = self.labels.service_operation {
512            (observe::LABEL_SERVICE_OPERATION, service_operation).encode(encoder.encode_label())?;
513        }
514        Ok(())
515    }
516}
517
518fn register_metric(registry: &mut Registry, metric: impl Metric, value: observe::MetricValue) {
519    let ((name, unit), help) = (value.name_with_unit(), value.help());
520
521    if let Some(unit) = unit {
522        registry.register_with_unit(name, help, Unit::Other(unit.to_string()), metric);
523    } else {
524        registry.register(name, help, metric);
525    }
526}
opendal_layer_prometheus_client/lib.rs

opendal_layer_prometheus_client/
lib.rs
