rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2025-12-25 23:49:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: wait for longer anyway&wait metrics

Browse Source
This commit is contained in:
discord9
2025-06-11 15:19:47 +08:00
parent 91f373e66e
commit 22c61432f6
2 changed files with 23 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
src/flow/src/batching_mode/state.rs
View File

@@ -32,6 +32,7 @@ use crate::batching_mode::MIN_REFRESH_DURATION;
use crate::error::{DatatypesSnafu, InternalSnafu, TimeSnafu, UnexpectedSnafu}; use crate::error::{DatatypesSnafu, InternalSnafu, TimeSnafu, UnexpectedSnafu};
use crate::metrics::{ use crate::metrics::{
METRIC_FLOW_BATCHING_ENGINE_QUERY_TIME_RANGE, METRIC_FLOW_BATCHING_ENGINE_QUERY_WINDOW_CNT, METRIC_FLOW_BATCHING_ENGINE_QUERY_TIME_RANGE, METRIC_FLOW_BATCHING_ENGINE_QUERY_WINDOW_CNT,
METRIC_FLOW_BATCHING_ENGINE_WAIT_TIME,
}; };
use crate::{Error, FlowId}; use crate::{Error, FlowId};
@@ -102,8 +103,19 @@ impl TaskState {
}) })
.unwrap_or(last_duration); .unwrap_or(last_duration);
METRIC_FLOW_BATCHING_ENGINE_WAIT_TIME
.with_label_values(&[
flow_id.to_string().as_str(),
time_window_size
.unwrap_or_default()
.as_secs_f64()
.to_string()
.as_str(),
])
.observe(next_duration.as_secs_f64());
// if have dirty time window, execute immediately to clean dirty time window // if have dirty time window, execute immediately to clean dirty time window
if self.dirty_time_windows.windows.is_empty() { /*if self.dirty_time_windows.windows.is_empty() {
self.last_update_time + next_duration self.last_update_time + next_duration
} else { } else {
debug!( debug!(
@@ -113,7 +125,9 @@ impl TaskState {
self.dirty_time_windows.windows self.dirty_time_windows.windows
); );
Instant::now() Instant::now()
} }*/
self.last_update_time + next_duration
} }
} }

7
src/flow/src/metrics.rs
View File

@@ -35,6 +35,13 @@ lazy_static! {
vec![0.0, 5., 10., 20., 40., 80., 160., 320., 640.,] vec![0.0, 5., 10., 20., 40., 80., 160., 320., 640.,]
) )
.unwrap(); .unwrap();
pub static ref METRIC_FLOW_BATCHING_ENGINE_WAIT_TIME: HistogramVec = register_histogram_vec!(
"greptime_flow_batching_engine_wait_time_secs",
"flow batching engine wait time between query(seconds)",
&["flow_id", "time_window_size"],
vec![0.0, 5., 10., 20., 40., 80., 160., 320., 640.,]
)
.unwrap();
pub static ref METRIC_FLOW_BATCHING_ENGINE_SLOW_QUERY: HistogramVec = register_histogram_vec!( pub static ref METRIC_FLOW_BATCHING_ENGINE_SLOW_QUERY: HistogramVec = register_histogram_vec!(
"greptime_flow_batching_engine_slow_query_secs", "greptime_flow_batching_engine_slow_query_secs",
"flow batching engine slow query(seconds)", "flow batching engine slow query(seconds)",