wcy-fdu commented on code in PR #1930: URL: https://github.com/apache/incubator-opendal/pull/1930#discussion_r1165769114
########## core/src/layers/prometheus.rs: ########## @@ -0,0 +1,1207 @@ +// 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::Debug; +use std::fmt::Formatter; +use std::io; +use std::sync::Arc; +use std::task::Context; +use std::task::Poll; + +use async_trait::async_trait; +use bytes::Bytes; +use futures::FutureExt; +use futures::TryFutureExt; +use log::debug; + +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::Registry; +use prometheus::{core::AtomicU64, HistogramVec}; + +use crate::ops::*; +use crate::raw::Accessor; +use crate::raw::*; +use crate::*; +/// Add [prometheus](https://docs.rs/prometheus) for every operations. +/// +/// # Examples +/// +/// ``` +/// use log::debug; +/// use log::info; +/// use opendal::services; +/// use opendal::Operator; +/// use opendal::Result; +/// +/// use opendal::layers::PrometheusLayer; +/// use prometheus::Encoder; +/// +/// /// Visit [`opendal::services`] for more service related config. +/// /// Visit [`opendal::Object`] for more object level APIs. +/// #[tokio::main] +/// async fn main() -> Result<()> { +/// // Pick a builder and configure it. +/// let builder = services::Memory::default(); +/// +/// let op = Operator::new(builder) +/// .expect("must init") +/// .layer(PrometheusLayer) +/// .finish(); +/// debug!("operator: {op:?}"); +/// +/// // Write data into object test. +/// op.write("test", "Hello, World!").await?; +/// // Read data from object. +/// let bs = op.read("test").await?; +/// info!("content: {}", String::from_utf8_lossy(&bs)); +/// +/// // Get object metadata. +/// let meta = op.stat("test").await?; +/// info!("meta: {:?}", meta); +/// +/// // Export prometheus metrics. +/// let mut buffer = Vec::<u8>::new(); +/// let encoder = prometheus::TextEncoder::new(); +/// encoder.encode(&prometheus::gather(), &mut buffer).unwrap(); +/// println!("## Prometheus Metrics"); +/// println!("{}", String::from_utf8(buffer.clone()).unwrap()); +/// Ok(()) +/// } +/// ``` +#[derive(Debug, Copy, Clone)] +pub struct PrometheusLayer; + +impl<A: Accessor> Layer<A> for PrometheusLayer { + type LayeredAccessor = PrometheusAccessor<A>; + + fn layer(&self, inner: A) -> Self::LayeredAccessor { + let mut meta = inner.info(); + let schema = meta.scheme(); + let registry = match meta.prometheus_register() { + Some(register) => register, + None => prometheus::default_registry(), + }; + + PrometheusAccessor { + inner, + stats: Arc::new(PrometheusMetrics::new(registry.clone())), + schema: schema.to_string(), + } + } +} +/// [`PrometheusMetrics`] provide the performance and IO metrics. +#[derive(Debug)] +pub struct PrometheusMetrics { + // metadata + pub opendal_requests_total_metadata: GenericCounterVec<AtomicU64>, + pub opendal_requests_duration_seconds_metadata: HistogramVec, + + // create + pub opendal_requests_total_create: GenericCounterVec<AtomicU64>, + pub opendal_requests_duration_seconds_create: HistogramVec, + + /// read + pub opendal_requests_total_read: GenericCounterVec<AtomicU64>, + pub opendal_requests_duration_seconds_read: HistogramVec, + pub opendal_bytes_total_read: HistogramVec, + + // write + pub opendal_requests_total_write: GenericCounterVec<AtomicU64>, + pub opendal_requests_duration_seconds_write: HistogramVec, + pub opendal_bytes_total_write: HistogramVec, + + // stat + pub opendal_requests_total_stat: GenericCounterVec<AtomicU64>, + pub opendal_requests_duration_seconds_stat: HistogramVec, + + // delete + pub opendal_requests_total_delete: GenericCounterVec<AtomicU64>, + pub opendal_requests_duration_seconds_delete: HistogramVec, + + // list + pub opendal_requests_total_list: GenericCounterVec<AtomicU64>, + pub opendal_requests_duration_seconds_list: HistogramVec, + + // scan + pub opendal_requests_total_scan: GenericCounterVec<AtomicU64>, + pub opendal_requests_duration_seconds_scan: HistogramVec, + + // presign + pub opendal_requests_total_presign: GenericCounterVec<AtomicU64>, + pub opendal_requests_duration_seconds_presign: HistogramVec, + + // batch + pub opendal_requests_total_batch: GenericCounterVec<AtomicU64>, + pub opendal_requests_duration_seconds_batch: HistogramVec, + + // blocking create + pub opendal_requests_total_blocking_create: GenericCounterVec<AtomicU64>, + pub opendal_requests_duration_seconds_blocking_create: HistogramVec, + + // blocking read + pub opendal_requests_total_blocking_read: GenericCounterVec<AtomicU64>, + pub opendal_requests_duration_seconds_blocking_read: HistogramVec, + pub opendal_bytes_total_blocking_read: HistogramVec, + + // blocking write + pub opendal_requests_total_blocking_write: GenericCounterVec<AtomicU64>, + pub opendal_requests_duration_seconds_blocking_write: HistogramVec, + pub opendal_bytes_total_blocking_write: HistogramVec, + + // blocking stat + pub opendal_requests_total_blocking_stat: GenericCounterVec<AtomicU64>, + pub opendal_requests_duration_seconds_blocking_stat: HistogramVec, + + // blocking delete + pub opendal_requests_total_blocking_delete: GenericCounterVec<AtomicU64>, + pub opendal_requests_duration_seconds_blocking_delete: HistogramVec, + + // blocking list + pub opendal_requests_total_blocking_list: GenericCounterVec<AtomicU64>, + pub opendal_requests_duration_seconds_blocking_list: HistogramVec, + + // blocking scan + pub opendal_requests_total_blocking_scan: GenericCounterVec<AtomicU64>, + pub opendal_requests_duration_seconds_blocking_scan: HistogramVec, +} + +impl PrometheusMetrics { + /// new with prometheus register. + pub fn new(registry: Registry) -> Self { + // metadata + let opendal_requests_total_metadata = register_int_counter_vec_with_registry!( + "opendal_requests_total_metadata", + "Total times of metadata be called", + &["schema"], + registry + ) + .unwrap(); + let opts = histogram_opts!( + "opendal_requests_duration_seconds_metadata", + "Histogram of the time spent on getting metadata", + exponential_buckets(0.01, 2.0, 16).unwrap() + ); + + let opendal_requests_duration_seconds_metadata = + register_histogram_vec_with_registry!(opts, &["schema"], registry).unwrap(); + + // create + let opendal_requests_total_create = register_int_counter_vec_with_registry!( + "opendal_requests_total_create", + "Total times of create be called", + &["schema"], + registry + ) + .unwrap(); + let opts = histogram_opts!( + "opendal_requests_duration_seconds_create", + "Histogram of the time spent on creating", + exponential_buckets(0.01, 2.0, 16).unwrap() + ); + + let opendal_requests_duration_seconds_create = + register_histogram_vec_with_registry!(opts, &["schema"], registry).unwrap(); + + // read + let opendal_requests_total_read = register_int_counter_vec_with_registry!( + "opendal_requests_total_read", + "Total times of read be called", + &["schema"], + registry + ) + .unwrap(); + let opts = histogram_opts!( + "opendal_requests_duration_seconds_read", + "Histogram of the time spent on reading", + exponential_buckets(0.01, 2.0, 16).unwrap() + ); + let opendal_requests_duration_seconds_read = + register_histogram_vec_with_registry!(opts, &["schema"], registry).unwrap(); + + let opts = histogram_opts!( + "opendal_bytes_total_read", + "read size", + exponential_buckets(0.01, 2.0, 16).unwrap() + ); + let opendal_bytes_total_read = + register_histogram_vec_with_registry!(opts, &["schema"], registry).unwrap(); + + // write + let opendal_requests_total_write = register_int_counter_vec_with_registry!( + "opendal_requests_total_write", + "Total times of write be called", + &["schema"], + registry + ) + .unwrap(); + let opts = histogram_opts!( + "opendal_requests_duration_seconds_write", + "Histogram of the time spent on writing", + exponential_buckets(0.01, 2.0, 16).unwrap() + ); + let opendal_requests_duration_seconds_write = + register_histogram_vec_with_registry!(opts, &["schema"], registry).unwrap(); + + let opts = histogram_opts!( + "opendal_bytes_total_write", + "write size", + exponential_buckets(0.01, 2.0, 16).unwrap() + ); + let opendal_bytes_total_write = + register_histogram_vec_with_registry!(opts, &["schema"], registry).unwrap(); + + // stat + let opendal_requests_total_stat = register_int_counter_vec_with_registry!( + "opendal_requests_total_stat", + "Total times of stat be called", + &["schema"], + registry + ) + .unwrap(); + let opts = histogram_opts!( + "opendal_requests_duration_seconds_stat", + "stat letency", + exponential_buckets(0.01, 2.0, 16).unwrap() + ); + + let opendal_requests_duration_seconds_stat = + register_histogram_vec_with_registry!(opts, &["schema"], registry).unwrap(); + + // delete + let opendal_requests_total_delete = register_int_counter_vec_with_registry!( + "opendal_requests_total_delete", + "Total times of delete be called", + &["schema"], + registry + ) + .unwrap(); + let opts = histogram_opts!( + "opendal_requests_duration_seconds_delete", + "delete letency", + exponential_buckets(0.01, 2.0, 16).unwrap() + ); + + let opendal_requests_duration_seconds_delete = + register_histogram_vec_with_registry!(opts, &["schema"], registry).unwrap(); + + // list + let opendal_requests_total_list = register_int_counter_vec_with_registry!( + "opendal_requests_total_list", + "Total times of list be called", + &["schema"], + registry + ) + .unwrap(); + let opts = histogram_opts!( + "opendal_requests_duration_seconds_list", + "list letency", + exponential_buckets(0.01, 2.0, 16).unwrap() + ); + + let opendal_requests_duration_seconds_list = + register_histogram_vec_with_registry!(opts, &["schema"], registry).unwrap(); + + // scan + let opendal_requests_total_scan = register_int_counter_vec_with_registry!( + "opendal_requests_total_scan", + "Total times of scan be called", + &["schema"], + registry + ) + .unwrap(); + let opts = histogram_opts!( + "opendal_requests_duration_seconds_scan", + "scan letency", + exponential_buckets(0.01, 2.0, 16).unwrap() + ); + + let opendal_requests_duration_seconds_scan = + register_histogram_vec_with_registry!(opts, &["schema"], registry).unwrap(); + + // presign + let opendal_requests_total_presign = register_int_counter_vec_with_registry!( + "opendal_requests_total_presign", + "Total times of presign be called", + &["schema"], + registry + ) + .unwrap(); + let opts = histogram_opts!( + "opendal_requests_duration_seconds_presign", + "presign letency", + exponential_buckets(0.01, 2.0, 16).unwrap() + ); + + let opendal_requests_duration_seconds_presign = + register_histogram_vec_with_registry!(opts, &["schema"], registry).unwrap(); + + // batch + let opendal_requests_total_batch = register_int_counter_vec_with_registry!( + "opendal_requests_total_batch", + "Total times of batch be called", + &["schema"], + registry + ) + .unwrap(); + let opts = histogram_opts!( + "opendal_requests_duration_seconds_batch", + "batch letency", + exponential_buckets(0.01, 2.0, 16).unwrap() + ); + + let opendal_requests_duration_seconds_batch = + register_histogram_vec_with_registry!(opts, &["schema"], registry).unwrap(); + + // blocking_create + let opendal_requests_total_blocking_create = register_int_counter_vec_with_registry!( + "opendal_requests_total_blocking_create", + "Total times of blocking_create be called", + &["schema"], + registry + ) + .unwrap(); + let opts = histogram_opts!( + "opendal_requests_duration_seconds_blocking_create", + "blocking create letency", + exponential_buckets(0.01, 2.0, 16).unwrap() + ); + + let opendal_requests_duration_seconds_blocking_create = + register_histogram_vec_with_registry!(opts, &["schema"], registry).unwrap(); + + // blocking_read + let opendal_requests_total_blocking_read = register_int_counter_vec_with_registry!( + "opendal_requests_total_blocking_read", + "Total times of blocking_read be called", + &["schema"], + registry + ) + .unwrap(); + let opts = histogram_opts!( + "opendal_requests_duration_seconds_blocking_read", + "Histogram of the time spent on blocking_reading", + exponential_buckets(0.01, 2.0, 16).unwrap() + ); + let opendal_requests_duration_seconds_blocking_read = + register_histogram_vec_with_registry!(opts, &["schema"], registry).unwrap(); + + let opts = histogram_opts!( + "opendal_bytes_total_blocking_read", + "blocking_read size", + exponential_buckets(0.01, 2.0, 16).unwrap() + ); + let opendal_bytes_total_blocking_read = + register_histogram_vec_with_registry!(opts, &["schema"], registry).unwrap(); + + // blocking_write + let opendal_requests_total_blocking_write = register_int_counter_vec_with_registry!( + "opendal_requests_total_blocking_write", + "Total times of blocking_write be called", + &["schema"], + registry + ) + .unwrap(); + let opts = histogram_opts!( + "opendal_requests_duration_seconds_blocking_write", + "Histogram of the time spent on blocking_writing", + exponential_buckets(0.01, 2.0, 16).unwrap() + ); + let opendal_requests_duration_seconds_blocking_write = + register_histogram_vec_with_registry!(opts, &["schema"], registry).unwrap(); + + let opts = histogram_opts!( + "opendal_bytes_total_blocking_write", + "total size by blocking_write", + exponential_buckets(0.01, 2.0, 16).unwrap() + ); + let opendal_bytes_total_blocking_write = + register_histogram_vec_with_registry!(opts, &["schema"], registry).unwrap(); + + // blocking_stat + let opendal_requests_total_blocking_stat = register_int_counter_vec_with_registry!( + "opendal_requests_total_blocking_stat", + "Total times of blocking_stat be called", + &["schema"], + registry + ) + .unwrap(); + let opts = histogram_opts!( + "opendal_requests_duration_seconds_blocking_stat", + "blocking_stat letency", + exponential_buckets(0.01, 2.0, 16).unwrap() + ); + + let opendal_requests_duration_seconds_blocking_stat = + register_histogram_vec_with_registry!(opts, &["schema"], registry).unwrap(); + + // blocking_delete + let opendal_requests_total_blocking_delete = register_int_counter_vec_with_registry!( + "opendal_requests_total_blocking_delete", + "Total times of blocking_delete be called", + &["schema"], + registry + ) + .unwrap(); + let opts = histogram_opts!( + "opendal_requests_duration_seconds_blocking_delete", + "blocking_delete letency", + exponential_buckets(0.01, 2.0, 16).unwrap() + ); + + let opendal_requests_duration_seconds_blocking_delete = + register_histogram_vec_with_registry!(opts, &["schema"], registry).unwrap(); + + // blocking_list + let opendal_requests_total_blocking_list = register_int_counter_vec_with_registry!( + "opendal_requests_total_blocking_list", + "Total times of blocking_list be called", + &["schema"], + registry + ) + .unwrap(); + let opts = histogram_opts!( + "opendal_requests_duration_seconds_blocking_list", + "blocking_list letency", + exponential_buckets(0.01, 2.0, 16).unwrap() + ); + + let opendal_requests_duration_seconds_blocking_list = + register_histogram_vec_with_registry!(opts, &["schema"], registry).unwrap(); + + // blocking_scan + let opendal_requests_total_blocking_scan = register_int_counter_vec_with_registry!( + "opendal_requests_total_blocking_scan", + "Total times of blocking_scan be called", + &["schema"], + registry + ) + .unwrap(); + + let opts = histogram_opts!( + "opendal_requests_duration_seconds_blocking_scan", Review Comment: Is this OK? `requests_total_read{operation="read",scheme="memory"} 1` Notice that `metadata` is not included in `Operation`, so I just missed this one temporarily. You can let me add it as well if you want. 😄 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@opendal.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
