Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix(metrics): use raw metrics vec in connection metrics #14017

Merged
merged 2 commits into from
Dec 18, 2023
Merged

fix(metrics): use raw metrics vec in connection metrics #14017

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7aed370
fix(metrics): use raw metrics vec in connection metrics
wenym1 Dec 15, 2023
79f1348
Merge branch 'main' into yiming/connection-raw-metrics-vec
wenym1 Dec 18, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
76 changes: 37 additions & 39 deletions src/common/src/monitor/connection.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,19 +27,17 @@ use hyper::client::connect::Connection;
use hyper::client::HttpConnector;
use hyper::service::Service;
use pin_project_lite::pin_project;
use prometheus::Registry;
use prometheus::{
register_int_counter_vec_with_registry, register_int_gauge_vec_with_registry, IntCounter,
IntCounterVec, IntGauge, IntGaugeVec, Registry,
};
use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
use tonic::transport::{Channel, Endpoint};
use tracing::{info, warn};

use crate::metrics::{
LabelGuardedIntCounter, LabelGuardedIntCounterVec, LabelGuardedIntGauge,
LabelGuardedIntGaugeVec,
};
use crate::metrics::LabelGuardedIntCounterVec;
use crate::monitor::GLOBAL_METRICS_REGISTRY;
use crate::{
register_guarded_int_counter_vec_with_registry, register_guarded_int_gauge_vec_with_registry,
};
use crate::register_guarded_int_counter_vec_with_registry;

pub trait MonitorAsyncReadWrite {
fn on_read(&mut self, _size: usize) {}
Expand Down Expand Up @@ -276,15 +274,15 @@ where

#[derive(Clone)]
pub struct ConnectionMetrics {
connection_count: LabelGuardedIntGaugeVec<2>,
connection_create_rate: LabelGuardedIntCounterVec<2>,
connection_err_rate: LabelGuardedIntCounterVec<2>,
connection_count: IntGaugeVec,
connection_create_rate: IntCounterVec,
connection_err_rate: IntCounterVec,

read_rate: LabelGuardedIntCounterVec<2>,
reader_count: LabelGuardedIntGaugeVec<2>,
read_rate: IntCounterVec,
reader_count: IntGaugeVec,

write_rate: LabelGuardedIntCounterVec<2>,
writer_count: LabelGuardedIntGaugeVec<2>,
write_rate: IntCounterVec,
writer_count: IntGaugeVec,

io_err_rate: LabelGuardedIntCounterVec<4>,
}
Expand All @@ -295,55 +293,55 @@ pub static GLOBAL_CONNECTION_METRICS: LazyLock<ConnectionMetrics> =
impl ConnectionMetrics {
pub fn new(registry: &Registry) -> Self {
let labels = ["connection_type", "uri"];
let connection_count = register_guarded_int_gauge_vec_with_registry!(
let connection_count = register_int_gauge_vec_with_registry!(
"connection_count",
"The number of current existing connection",
&labels,
registry,
)
.unwrap();

let connection_create_rate = register_guarded_int_counter_vec_with_registry!(
let connection_create_rate = register_int_counter_vec_with_registry!(
"connection_create_rate",
"Rate on creating new connection",
&labels,
registry,
)
.unwrap();

let connection_err_rate = register_guarded_int_counter_vec_with_registry!(
let connection_err_rate = register_int_counter_vec_with_registry!(
"connection_err_rate",
"Error rate on creating new connection",
&labels,
registry,
)
.unwrap();

let read_rate = register_guarded_int_counter_vec_with_registry!(
let read_rate = register_int_counter_vec_with_registry!(
"connection_read_rate",
"Read rate of a connection",
&labels,
registry,
)
.unwrap();

let reader_count = register_guarded_int_gauge_vec_with_registry!(
let reader_count = register_int_gauge_vec_with_registry!(
"connection_reader_count",
"The number of current existing reader",
&labels,
registry,
)
.unwrap();

let write_rate = register_guarded_int_counter_vec_with_registry!(
let write_rate = register_int_counter_vec_with_registry!(
"connection_write_rate",
"Write rate of a connection",
&labels,
registry,
)
.unwrap();

let writer_count = register_guarded_int_gauge_vec_with_registry!(
let writer_count = register_int_gauge_vec_with_registry!(
"connection_writer_count",
"The number of current existing writer",
&labels,
Expand Down Expand Up @@ -521,23 +519,23 @@ impl MonitorNewConnection for MonitorNewConnectionImpl {
let labels = [self.connection_type.as_str(), endpoint.as_str()];
let read_rate = GLOBAL_CONNECTION_METRICS
.read_rate
.with_guarded_label_values(&labels);
.with_label_values(&labels);
let reader_count = GLOBAL_CONNECTION_METRICS
.reader_count
.with_guarded_label_values(&labels);
.with_label_values(&labels);
let write_rate = GLOBAL_CONNECTION_METRICS
.write_rate
.with_guarded_label_values(&labels);
.with_label_values(&labels);
let writer_count = GLOBAL_CONNECTION_METRICS
.writer_count
.with_guarded_label_values(&labels);
.with_label_values(&labels);
let connection_count = GLOBAL_CONNECTION_METRICS
.connection_count
.with_guarded_label_values(&labels);
.with_label_values(&labels);

GLOBAL_CONNECTION_METRICS
.connection_create_rate
.with_guarded_label_values(&labels)
.with_label_values(&labels)
.inc();

MonitorAsyncReadWriteImpl::new(
Expand All @@ -554,7 +552,7 @@ impl MonitorNewConnection for MonitorNewConnectionImpl {
fn on_err(&self, endpoint: String) {
GLOBAL_CONNECTION_METRICS
.connection_err_rate
.with_guarded_label_values(&[self.connection_type.as_str(), endpoint.as_str()])
.with_label_values(&[self.connection_type.as_str(), endpoint.as_str()])
.inc();
}
}
Expand All @@ -566,27 +564,27 @@ pub struct MonitorAsyncReadWriteImpl {
connection_type: String,

unreported_read_rate: u64,
read_rate: LabelGuardedIntCounter<2>,
reader_count_guard: LabelGuardedIntGauge<2>,
read_rate: IntCounter,
reader_count_guard: IntGauge,
is_eof: bool,

unreported_write_rate: u64,
write_rate: LabelGuardedIntCounter<2>,
writer_count_guard: LabelGuardedIntGauge<2>,
write_rate: IntCounter,
writer_count_guard: IntGauge,
is_shutdown: bool,

connection_count_guard: LabelGuardedIntGauge<2>,
connection_count_guard: IntGauge,
}

impl MonitorAsyncReadWriteImpl {
pub fn new(
endpoint: String,
connection_type: String,
read_rate: LabelGuardedIntCounter<2>,
reader_count: LabelGuardedIntGauge<2>,
write_rate: LabelGuardedIntCounter<2>,
writer_count: LabelGuardedIntGauge<2>,
connection_count: LabelGuardedIntGauge<2>,
read_rate: IntCounter,
reader_count: IntGauge,
write_rate: IntCounter,
writer_count: IntGauge,
connection_count: IntGauge,
) -> Self {
reader_count.inc();
writer_count.inc();
Expand Down
Loading