risingwave_common_metrics/
lib.rs

1// Copyright 2024 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15#![feature(type_alias_impl_trait)]
16#![feature(impl_trait_in_assoc_type)]
17#![feature(register_tool)]
18#![feature(trait_alias)]
19#![register_tool(rw)]
20
21use std::ops::Deref;
22use std::sync::LazyLock;
23
24use hytra::TrAdder;
25use prometheus::core::{Atomic, AtomicU64, GenericCounter, GenericGauge};
26use prometheus::proto::Metric;
27use prometheus::register_int_counter_with_registry;
28use tracing::Subscriber;
29use tracing_subscriber::Layer;
30use tracing_subscriber::layer::Context;
31use tracing_subscriber::registry::LookupSpan;
32
33mod error_metrics;
34mod gauge_ext;
35mod guarded_metrics;
36mod metrics;
37pub mod monitor;
38mod relabeled_metric;
39
40pub use error_metrics::*;
41pub use gauge_ext::*;
42pub use guarded_metrics::*;
43pub use metrics::*;
44pub use relabeled_metric::*;
45
46#[derive(Debug)]
47pub struct TrAdderAtomic(TrAdder<i64>);
48
49impl Atomic for TrAdderAtomic {
50    type T = i64;
51
52    fn new(val: i64) -> Self {
53        let v = TrAdderAtomic(TrAdder::new());
54        v.0.inc(val);
55        v
56    }
57
58    fn set(&self, _val: i64) {
59        panic!("TrAdderAtomic doesn't support set operation.")
60    }
61
62    fn get(&self) -> i64 {
63        self.0.get()
64    }
65
66    fn inc_by(&self, delta: i64) {
67        self.0.inc(delta)
68    }
69
70    fn dec_by(&self, delta: i64) {
71        self.0.inc(-delta)
72    }
73}
74
75pub type TrAdderGauge = GenericGauge<TrAdderAtomic>;
76
77/// [`MetricsLayer`] is a struct used for monitoring the frequency of certain specific logs and
78/// counting them using Prometheus metrics. Currently, it is used to monitor the frequency of retry
79/// occurrences of aws sdk.
80pub struct MetricsLayer {
81    pub aws_sdk_retry_counts: GenericCounter<AtomicU64>,
82}
83
84impl<S> Layer<S> for MetricsLayer
85where
86    S: Subscriber + for<'a> LookupSpan<'a>,
87{
88    fn on_event(&self, _event: &tracing::Event<'_>, _ctx: Context<'_, S>) {
89        // Currently one retry will only generate one debug log,
90        // so we can monitor the number of retry only through the metadata target.
91        // Refer to <https://docs.rs/aws-smithy-client/0.55.3/src/aws_smithy_client/retry.rs.html>
92        self.aws_sdk_retry_counts.inc();
93    }
94}
95
96impl MetricsLayer {
97    #[expect(clippy::new_without_default)]
98    pub fn new() -> Self {
99        static AWS_SDK_RETRY_COUNTS: LazyLock<GenericCounter<AtomicU64>> = LazyLock::new(|| {
100            let registry = crate::monitor::GLOBAL_METRICS_REGISTRY.deref();
101            register_int_counter_with_registry!(
102                "aws_sdk_retry_counts",
103                "Total number of aws sdk retry happens",
104                registry
105            )
106            .unwrap()
107        });
108
109        Self {
110            aws_sdk_retry_counts: AWS_SDK_RETRY_COUNTS.deref().clone(),
111        }
112    }
113}
114
115#[derive(Debug, Default, Clone, Copy, serde::Serialize, serde::Deserialize)]
116pub enum MetricLevel {
117    #[default]
118    Disabled = 0,
119    Critical = 1,
120    Info = 2,
121    Debug = 3,
122}
123
124impl clap::ValueEnum for MetricLevel {
125    fn value_variants<'a>() -> &'a [Self] {
126        &[Self::Disabled, Self::Critical, Self::Info, Self::Debug]
127    }
128
129    fn to_possible_value<'a>(&self) -> ::std::option::Option<clap::builder::PossibleValue> {
130        match self {
131            Self::Disabled => Some(clap::builder::PossibleValue::new("disabled").alias("0")),
132            Self::Critical => Some(clap::builder::PossibleValue::new("critical")),
133            Self::Info => Some(clap::builder::PossibleValue::new("info").alias("1")),
134            Self::Debug => Some(clap::builder::PossibleValue::new("debug")),
135        }
136    }
137}
138
139impl PartialEq<Self> for MetricLevel {
140    fn eq(&self, other: &Self) -> bool {
141        (*self as u8).eq(&(*other as u8))
142    }
143}
144
145impl PartialOrd for MetricLevel {
146    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
147        (*self as u8).partial_cmp(&(*other as u8))
148    }
149}
150
151pub fn get_label<T: std::str::FromStr>(metric: &Metric, label: &str) -> Option<T> {
152    metric
153        .get_label()
154        .iter()
155        .find(|lp| lp.name() == label)
156        .and_then(|lp| lp.value().parse::<T>().ok())
157}
158
159// Must ensure the label exists and can be parsed into `T`
160pub fn get_label_infallible<T: std::str::FromStr>(metric: &Metric, label: &str) -> T {
161    get_label(metric, label).expect("label not found or can't be parsed")
162}
risingwave_common_metrics/lib.rs

risingwave_common_metrics/
lib.rs
