starred/openobserve
1
0
Fork 0
mirror of https://github.com/openobserve/openobserve.git synced 2026-11-06 00:09:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: scheduled pipeline align query time instead of next_run_at (#7007)

Browse Source
This commit is contained in:
Taiming Liu
2025-06-03 12:37:09 +08:00
committed by GitHub
parent 36c71ee30b
commit 9eb21ceecd
2 changed files with 30 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/service/alerts/derived_streams.rs
+1 -28
View File
@@ -142,6 +142,7 @@ pub async fn save(
};
}
let next_run_at = chrono::Utc::now().timestamp_micros();
// Save the trigger to db
match db::scheduler::get(
&derived_stream.org_id,
@@ -151,17 +152,12 @@ pub async fn save(
.await
{
Ok(mut existing_trigger) => {
let next_run_at = get_next_run_at(
derived_stream.delay.unwrap_or_default(),
Some(existing_trigger.next_run_at),
)?;
existing_trigger.next_run_at = next_run_at;
db::scheduler::update_trigger(existing_trigger)
.await
.map_err(|_| anyhow::anyhow!("Trigger already exists, but failed to update"))
}
Err(_) => {
let next_run_at = get_next_run_at(derived_stream.delay.unwrap_or_default(), None)?;
let trigger = db::scheduler::Trigger {
org: derived_stream.org_id.to_string(),
module: db::scheduler::TriggerModule::DerivedStream,
@@ -226,26 +222,3 @@ impl DerivedStreamExt for DerivedStream {
.await
}
}
fn get_next_run_at(
delay_in_mins: i32,
previous_next_run_at: Option<i64>,
) -> Result<i64, anyhow::Error> {
// 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 _);
let supposed_next_run_at =
previous_next_run_at.map_or(Ok(chrono::Utc::now()), |prev_next_run_at| {
chrono::DateTime::<chrono::Utc>::from_timestamp_micros(prev_next_run_at)
.ok_or(anyhow::anyhow!("Invalid previous next run at timestamp"))
})?;
Ok(supposed_next_run_at
.checked_add_signed(delay)
.ok_or(anyhow::anyhow!("DateTime arithmetic overflow"))?
.timestamp_micros())
}
src/service/alerts/scheduler/handlers.rs
+29 -41
View File
@@ -1014,7 +1014,7 @@ async fn handle_derived_stream_triggers(
chrono::Duration::try_minutes(delay_in_mins as _).and_then(|td| td.num_microseconds())
})
.unwrap_or_default();
let mut supposed_to_be_run_at = trigger.next_run_at - user_defined_delay;
let supposed_to_be_run_at = trigger.next_run_at - user_defined_delay;
let period_num_microseconds = Duration::try_minutes(derived_stream.trigger_condition.period)
.unwrap()
.num_microseconds()
@@ -1023,24 +1023,6 @@ async fn handle_derived_stream_triggers(
let (mut start, mut end) = if derived_stream.start_at.is_some() && trigger.data.is_empty() {
(derived_stream.start_at, supposed_to_be_run_at)
} else if let Some(t0) = start_time {
// Don't use only the period_num_microseconds, because, then the delay is lets say 10 secs
// The following code will make a separate query to cover the delay period of 10 secs which
// is unnecessary. Hence, we need to check how big the delay is.
// Note: For pipeline, period and frequency both have the same value.
// For derived stream, period is in minutes, so we need to convert it to seconds for
// align_time
if derived_stream.trigger_condition.align_time {
let aligned_curr_time = TriggerCondition::align_time(
supposed_to_be_run_at,
derived_stream.tz_offset,
derived_stream.trigger_condition.period * 60,
);
// conditionally modify supposed_to_be_run_at
if aligned_curr_time > t0 {
supposed_to_be_run_at = aligned_curr_time;
}
}
// If the delay is equal to or greater than the frequency, we need to ingest data one by
// one If the delay is less than the frequency, we need to ingest data for
// the "next run at" period, For example, if the current time is 5:19pm,
@@ -1057,6 +1039,18 @@ async fn handle_derived_stream_triggers(
} else {
(None, supposed_to_be_run_at)
};
// For derived stream, period is in minutes, so we need to convert it to seconds for align_time
if derived_stream.trigger_condition.align_time {
let aligned_curr_time = TriggerCondition::align_time(
end,
derived_stream.tz_offset,
derived_stream.trigger_condition.period * 60,
);
// conditionally modify supposed_to_be_run_at
if start.is_none_or(|t0| t0 < aligned_curr_time) {
end = aligned_curr_time;
}
}
// In case the scheduler background job (watch_timeout) updates the trigger retries
// (not through this handler), we need to skip to the next run at but with the same
@@ -1069,12 +1063,14 @@ async fn handle_derived_stream_triggers(
new_trigger.module_key
);
// Go to the next nun at, but use the same trigger start time
new_trigger.next_run_at = derived_stream.trigger_condition.get_next_trigger_time(
false,
derived_stream.tz_offset,
false,
Some(end + user_defined_delay),
)?;
new_trigger.next_run_at = derived_stream
.trigger_condition
.get_next_trigger_time_non_aligned(
false,
derived_stream.tz_offset,
false,
Some(trigger.next_run_at),
)?;
// Start over next time
new_trigger.retries = 0;
db::scheduler::update_trigger(new_trigger).await?;
@@ -1298,11 +1294,6 @@ async fn handle_derived_stream_triggers(
} else {
// SUCCESS: move the time range forward by frequency and continue
start = Some(trigger_results.end_time);
// There could still be some data to be processed for the current period
// so we need to move the end time forward by the period length or the
// remaining time whichever is smaller
let _end = period_num_microseconds + 1;
end += _end;
trigger_data_stream.query_took = trigger_results.query_took;
}
} else {
@@ -1316,11 +1307,6 @@ async fn handle_derived_stream_triggers(
// move the time range forward by frequency and continue
start = Some(trigger_results.end_time);
// There could still be some data to be processed for the current period
// so we need to move the end time forward by the period length or the remaining
// time whichever is smaller
let _end = period_num_microseconds + 1;
end += _end;
}
}
};
@@ -1345,12 +1331,14 @@ async fn handle_derived_stream_triggers(
&& new_trigger.retries < max_retries)
{
// Go to the next nun at, but use the same trigger start time
new_trigger.next_run_at = derived_stream.trigger_condition.get_next_trigger_time(
false,
derived_stream.tz_offset,
false,
Some(end + user_defined_delay),
)?;
new_trigger.next_run_at = derived_stream
.trigger_condition
.get_next_trigger_time_non_aligned(
false,
derived_stream.tz_offset,
false,
Some(trigger.next_run_at),
)?;
// If the trigger didn't fail, we need to reset the `retries` count.
// Only cumulative failures should be used to check with `max_retries`