From c19b649da1e164ed07acf86e5a68fd9cd55c8b07 Mon Sep 17 00:00:00 2001 From: Taiming Liu Date: Thu, 24 Apr 2025 22:30:34 -0700 Subject: [PATCH] feat: delay for scheduled pipelines (#6616) Co-authored-by: sai nikhil kethe --- src/common/migration/pipeline_func.rs | 1 + src/config/src/meta/pipeline/components.rs | 2 + src/service/alerts/derived_streams.rs | 18 +++- src/service/alerts/scheduler/handlers.rs | 42 +++++---- .../components/pipeline/NodeForm/Query.vue | 9 ++ .../pipeline/NodeForm/ScheduledPipeline.vue | 86 ++++++++++++++++++- web/src/locales/languages/en.json | 3 +- 7 files changed, 141 insertions(+), 20 deletions(-) diff --git a/src/common/migration/pipeline_func.rs b/src/common/migration/pipeline_func.rs index 8bfa7a4dcd..7a2ba60707 100644 --- a/src/common/migration/pipeline_func.rs +++ b/src/common/migration/pipeline_func.rs @@ -107,6 +107,7 @@ async fn migrate_pipelines() -> Result<(), anyhow::Error> { query_condition: old_derived_stream.query_condition, trigger_condition: old_derived_stream.trigger_condition, tz_offset: old_derived_stream.tz_offset, + delay: None, }; let pipeline_source = PipelineSource::Scheduled(new_derived_stream.clone()); diff --git a/src/config/src/meta/pipeline/components.rs b/src/config/src/meta/pipeline/components.rs index 8081b5f4b9..dcde9d22c3 100644 --- a/src/config/src/meta/pipeline/components.rs +++ b/src/config/src/meta/pipeline/components.rs @@ -52,6 +52,8 @@ pub struct DerivedStream { /// The negative secs means the Western Hemisphere #[serde(default)] pub tz_offset: i32, + #[serde(skip_serializing_if = "Option::is_none")] + pub delay: Option, } #[derive(Debug, Clone, Serialize, Deserialize)] diff --git a/src/service/alerts/derived_streams.rs b/src/service/alerts/derived_streams.rs index 5eb795cbbd..bd944bc4bc 100644 --- a/src/service/alerts/derived_streams.rs +++ b/src/service/alerts/derived_streams.rs @@ -143,7 +143,7 @@ pub async fn save( } // Save the trigger to db - let next_run_at = Utc::now().timestamp_micros(); + let next_run_at = get_next_run_at(&derived_stream)?; let trigger = db::scheduler::Trigger { org: derived_stream.org_id.to_string(), module: db::scheduler::TriggerModule::DerivedStream, @@ -220,3 +220,19 @@ impl DerivedStreamExt for DerivedStream { .await } } + +pub(super) fn get_next_run_at(derived_stream: &DerivedStream) -> Result { + let delay_in_mins = derived_stream.delay.unwrap_or_default(); + // validate & parse delay value + if delay_in_mins < 0 { + return Err(anyhow::anyhow!( + "Invalid delay value. Value must be non-negative" + )); + } + + let delay = chrono::Duration::minutes(delay_in_mins as _); + Ok(chrono::Utc::now() + .checked_add_signed(delay) + .ok_or(anyhow::anyhow!("DateTime arithmetic overflow"))? + .timestamp_micros()) +} diff --git a/src/service/alerts/scheduler/handlers.rs b/src/service/alerts/scheduler/handlers.rs index 9e3cd54f1b..eeaf92556c 100644 --- a/src/service/alerts/scheduler/handlers.rs +++ b/src/service/alerts/scheduler/handlers.rs @@ -844,7 +844,7 @@ async fn handle_derived_stream_triggers( "Pipeline associated with trigger not found: {}/{}/{}/{}. Checking after 5 mins.", org_id, stream_type, pipeline_name, pipeline_id ); - // Check after 7 days if the pipeline is created + // Check after 5 mins if the pipeline is created new_trigger.next_run_at += Duration::try_minutes(5) .unwrap() .num_microseconds() @@ -882,7 +882,7 @@ async fn handle_derived_stream_triggers( }; if !pipeline.enabled { - // Pipeline not enabled, check after 7 days + // Pipeline not enabled, check again in 5 mins let msg = format!( "Pipeline associated with trigger not enabled: {}/{}/{}/{}. Checking after 5 mins.", org_id, stream_type, pipeline_name, pipeline_id @@ -973,9 +973,15 @@ async fn handle_derived_stream_triggers( // in case the range [start_time, end_time] is greater than querying period, it needs to // evaluate and ingest 1 period at a time. - let suppossed_to_be_run_at = trigger.next_run_at; - let delay = current_time - suppossed_to_be_run_at; // delay is in microseconds - let mut final_end_time = suppossed_to_be_run_at; + let user_defined_delay = derived_stream + .delay + .and_then(|delay_in_mins| { + chrono::Duration::try_minutes(delay_in_mins as _).and_then(|td| td.num_microseconds()) + }) + .unwrap_or_default(); + let supposed_to_be_run_at = trigger.next_run_at - user_defined_delay; + let delay = current_time - user_defined_delay - supposed_to_be_run_at; // delay is in microseconds + let mut final_end_time = supposed_to_be_run_at; // let now = Utc::now().timestamp_micros(); let period_num_microseconds = Duration::try_minutes(derived_stream.trigger_condition.period) .unwrap() @@ -996,20 +1002,19 @@ async fn handle_derived_stream_triggers( // query for the period from 5:15pm to 5:20pm. But, if the suppossed to be run at is // 5:10pm, then we need ingest data for the period from 5:05pm to 5:15pm. // Which is to cover the skipped period from 5:05pm to 5:15pm. - if delay >= period_num_microseconds { + let end = if delay >= period_num_microseconds { // final_end_time is the last multiple of given frequency after the "suppossed to be run // at" timestamp let frequency_count = delay / period_num_microseconds; - final_end_time = suppossed_to_be_run_at + (frequency_count * period_num_microseconds); - ( - Some(t0), - std::cmp::min(suppossed_to_be_run_at, t0 + period_num_microseconds), - ) + final_end_time = supposed_to_be_run_at + (frequency_count * period_num_microseconds); + std::cmp::min(supposed_to_be_run_at, t0 + period_num_microseconds) } else { - (Some(t0), suppossed_to_be_run_at) - } + supposed_to_be_run_at + }; + + (Some(t0), end) } else { - (None, suppossed_to_be_run_at) + (None, supposed_to_be_run_at) }; // In case the scheduler background job (watch_timeout) updates the trigger retries @@ -1025,7 +1030,8 @@ async fn handle_derived_stream_triggers( // Go to the next nun at, but use the same trigger start time new_trigger.next_run_at = derived_stream .trigger_condition - .get_aligned_next_trigger_time(false, derived_stream.tz_offset, false)?; + .get_aligned_next_trigger_time(false, derived_stream.tz_offset, false)? + + user_defined_delay; // Start over next time new_trigger.retries = 0; db::scheduler::update_trigger(new_trigger).await?; @@ -1295,7 +1301,8 @@ async fn handle_derived_stream_triggers( // Store the last used derived stream period end time if let Some(start_time) = start { new_trigger.data = json::to_string(&ScheduledTriggerData { - period_end_time: Some(start_time), // updated start_time as end_time + // updated start_time as end_time + period_end_time: Some(start_time + user_defined_delay), tolerance: 0, last_satisfied_at: None, }) @@ -1312,7 +1319,8 @@ async fn handle_derived_stream_triggers( // Go to the next nun at, but use the same trigger start time new_trigger.next_run_at = derived_stream .trigger_condition - .get_aligned_next_trigger_time(false, derived_stream.tz_offset, false)?; + .get_aligned_next_trigger_time(false, derived_stream.tz_offset, false)? + + user_defined_delay; // If the trigger didn't fail, we need to reset the `retries` count. // Only cumulative failures should be used to check with `max_retries` diff --git a/web/src/components/pipeline/NodeForm/Query.vue b/web/src/components/pipeline/NodeForm/Query.vue index d3bc0ba051..bd34f3ae3f 100644 --- a/web/src/components/pipeline/NodeForm/Query.vue +++ b/web/src/components/pipeline/NodeForm/Query.vue @@ -42,6 +42,7 @@ along with this program. If not, see . v-model:trigger="streamRoute.trigger_condition" v-model:sql="streamRoute.query_condition.sql" v-model:promql="streamRoute.query_condition.promql" + v-model:delay="streamRoute.delay" v-model:promql_condition=" streamRoute.query_condition.promql_condition " @@ -56,6 +57,7 @@ along with this program. If not, see . @update:fullscreen="updateFullscreenMode" @update:stream_type="updateStreamType" @expandLog="toggleExpandLog" + @update:delay="updateDelay" class="q-mt-sm" /> @@ -127,6 +129,7 @@ interface StreamRoute { cron: string; timezone: any; }; + delay: number; context_attributes: any; description: string; enabled: boolean; @@ -240,6 +243,7 @@ const getDefaultStreamRoute: any = () => { frequency: frequency <= 15 ? 15 : frequency, timezone: "UTC", }, + delay: 0, context_attributes: [ { key: "", @@ -407,6 +411,7 @@ const saveQueryData = async () => { silence: 0, timezone: formData.trigger_condition.timezone, }, + delay: formData.delay, }; if (formData.trigger_condition.frequency_type === "cron") { @@ -536,6 +541,10 @@ const updateQueryType = (val: string) => { const toggleExpandLog = (index: number) => { expandedLogs.value = []; }; + +const updateDelay = (val: any) => { + streamRoute.value.delay = parseInt(val); +};