wcy-fdu commented on code in PR #1930: URL: https://github.com/apache/incubator-opendal/pull/1930#discussion_r1165794254
########## core/src/layers/prometheus.rs: ########## @@ -0,0 +1,1240 @@ +// 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 registry = prometheus::default_registry(); +/// +/// let op = Operator::new(builder) +/// .expect("must init") +/// .layer(PrometheusLayer::with_registry(registry.clone())) +/// .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, Clone)] +pub struct PrometheusLayer { + registry: Registry, +} + +impl PrometheusLayer { + /// create PrometheusLayer by incoming registry. + pub fn with_registry(registry: Registry) -> Self { + Self { registry } + } +} + +impl Default for PrometheusLayer { + fn default() -> Self { + Self { + registry: prometheus::Registry::default(), + } + } +} +impl<A: Accessor> Layer<A> for PrometheusLayer { + type LayeredAccessor = PrometheusAccessor<A>; + + fn layer(&self, inner: A) -> Self::LayeredAccessor { + let meta = inner.info(); + let scheme = meta.scheme(); + + PrometheusAccessor { + inner, + stats: Arc::new(PrometheusMetrics::new(self.registry.clone())), + scheme: scheme.to_string(), + } + } +} +/// [`PrometheusMetrics`] provide the performance and IO metrics. +#[derive(Debug)] +pub struct PrometheusMetrics { + // metadata + pub requests_total_metadata: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_metadata: HistogramVec, + + // create + pub requests_total_create: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_create: HistogramVec, + + /// read + pub requests_total_read: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_read: HistogramVec, + pub bytes_total_read: HistogramVec, + + // write + pub requests_total_write: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_write: HistogramVec, + pub bytes_total_write: HistogramVec, + + // stat + pub requests_total_stat: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_stat: HistogramVec, + + // delete + pub requests_total_delete: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_delete: HistogramVec, + + // list + pub requests_total_list: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_list: HistogramVec, + + // scan + pub requests_total_scan: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_scan: HistogramVec, + + // presign + pub requests_total_presign: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_presign: HistogramVec, + + // batch + pub requests_total_batch: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_batch: HistogramVec, + + // blocking create + pub requests_total_blocking_create: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_blocking_create: HistogramVec, + + // blocking read + pub requests_total_blocking_read: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_blocking_read: HistogramVec, + pub bytes_total_blocking_read: HistogramVec, + + // blocking write + pub requests_total_blocking_write: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_blocking_write: HistogramVec, + pub bytes_total_blocking_write: HistogramVec, + + // blocking stat + pub requests_total_blocking_stat: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_blocking_stat: HistogramVec, + + // blocking delete + pub requests_total_blocking_delete: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_blocking_delete: HistogramVec, + + // blocking list + pub requests_total_blocking_list: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_blocking_list: HistogramVec, + + // blocking scan + pub requests_total_blocking_scan: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_blocking_scan: HistogramVec, +} + +impl PrometheusMetrics { + /// new with prometheus register. + pub fn new(registry: Registry) -> Self { + // metadata Review Comment: Question: is there meta cache in opendal? If yes, metadata latency is meaningful. 🤔 ########## core/src/layers/prometheus.rs: ########## @@ -0,0 +1,1240 @@ +// 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 registry = prometheus::default_registry(); +/// +/// let op = Operator::new(builder) +/// .expect("must init") +/// .layer(PrometheusLayer::with_registry(registry.clone())) +/// .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, Clone)] +pub struct PrometheusLayer { + registry: Registry, +} + +impl PrometheusLayer { + /// create PrometheusLayer by incoming registry. + pub fn with_registry(registry: Registry) -> Self { + Self { registry } + } +} + +impl Default for PrometheusLayer { + fn default() -> Self { + Self { + registry: prometheus::Registry::default(), + } + } +} +impl<A: Accessor> Layer<A> for PrometheusLayer { + type LayeredAccessor = PrometheusAccessor<A>; + + fn layer(&self, inner: A) -> Self::LayeredAccessor { + let meta = inner.info(); + let scheme = meta.scheme(); + + PrometheusAccessor { + inner, + stats: Arc::new(PrometheusMetrics::new(self.registry.clone())), + scheme: scheme.to_string(), + } + } +} +/// [`PrometheusMetrics`] provide the performance and IO metrics. +#[derive(Debug)] +pub struct PrometheusMetrics { + // metadata + pub requests_total_metadata: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_metadata: HistogramVec, + + // create + pub requests_total_create: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_create: HistogramVec, + + /// read + pub requests_total_read: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_read: HistogramVec, + pub bytes_total_read: HistogramVec, + + // write + pub requests_total_write: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_write: HistogramVec, + pub bytes_total_write: HistogramVec, + + // stat + pub requests_total_stat: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_stat: HistogramVec, + + // delete + pub requests_total_delete: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_delete: HistogramVec, + + // list + pub requests_total_list: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_list: HistogramVec, + + // scan + pub requests_total_scan: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_scan: HistogramVec, + + // presign + pub requests_total_presign: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_presign: HistogramVec, + + // batch + pub requests_total_batch: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_batch: HistogramVec, + + // blocking create + pub requests_total_blocking_create: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_blocking_create: HistogramVec, + + // blocking read + pub requests_total_blocking_read: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_blocking_read: HistogramVec, + pub bytes_total_blocking_read: HistogramVec, + + // blocking write + pub requests_total_blocking_write: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_blocking_write: HistogramVec, + pub bytes_total_blocking_write: HistogramVec, + + // blocking stat + pub requests_total_blocking_stat: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_blocking_stat: HistogramVec, + + // blocking delete + pub requests_total_blocking_delete: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_blocking_delete: HistogramVec, + + // blocking list + pub requests_total_blocking_list: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_blocking_list: HistogramVec, + + // blocking scan + pub requests_total_blocking_scan: GenericCounterVec<AtomicU64>, + pub requests_duration_seconds_blocking_scan: HistogramVec, +} + +impl PrometheusMetrics { + /// new with prometheus register. + pub fn new(registry: Registry) -> Self { + // metadata Review Comment: Question: is there meta cache in OpenDAL? If yes, metadata latency is meaningful. 🤔 -- 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
