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

Author: taimingl

src/service/alerts/derived_streams.rs

@@ -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

@@ -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`