fix: pipeline next_run_at with delay (openobserve#6872)

Author: taimingl

src/config/src/meta/alerts/mod.rs

@@ -70,6 +70,7 @@ impl TriggerCondition {
         freq_in_secs: bool,
         timezone_offset: i32,
         apply_silence: bool,
+        start_from: Option<i64>,
     ) -> Result<i64, anyhow::Error> {
         let frequency = if freq_in_secs {
             self.frequency
@@ -84,11 +85,16 @@ impl TriggerCondition {
             }
             _ => 0,
         };
+        let start_utc = start_from.map_or(Ok(Utc::now()), |from| {
+            chrono::DateTime::<Utc>::from_timestamp_micros(from).ok_or(anyhow::anyhow!(
+                "Error converting start_from value to timestamp"
+            ))
+        })?;
         let timezone_offset = FixedOffset::east_opt(timezone_offset * 60).unwrap();
         if self.frequency_type == FrequencyType::Cron {
             let schedule = Schedule::from_str(&self.cron)?;
             if apply_silence {
-                let silence = Utc::now() + Duration::try_minutes(self.silence).unwrap();
+                let silence = start_utc + Duration::try_minutes(self.silence).unwrap();
                 let silence = silence.with_timezone(&timezone_offset);
                 // Check for the cron timestamp after the silence period
                 Ok(schedule.after(&silence).next().unwrap().timestamp_micros() + tolerance)
@@ -110,14 +116,14 @@ impl TriggerCondition {
             // should use the max of (frequency, silence) as the next_run_at.
             // Silence period is in minutes, and the frequency is in seconds.
             let delta = std::cmp::max(frequency, self.silence * 60);
-            Ok(Utc::now().timestamp_micros()
+            Ok(start_utc.timestamp_micros()
                 + Duration::try_seconds(delta)
                     .unwrap()
                     .num_microseconds()
                     .unwrap()
                 + tolerance)
         } else {
-            Ok(Utc::now().timestamp_micros()
+            Ok(start_utc.timestamp_micros()
                 + Duration::try_seconds(frequency)
                     .unwrap()
                     .num_microseconds()
@@ -183,9 +189,14 @@ impl TriggerCondition {
         freq_in_secs: bool,
         timezone_offset: i32,
         apply_silence: bool,
+        start_from: Option<i64>,
     ) -> Result<i64, anyhow::Error> {
-        let next_run_at =
-            self.get_next_trigger_time_non_aligned(freq_in_secs, timezone_offset, apply_silence)?;
+        let next_run_at = self.get_next_trigger_time_non_aligned(
+            freq_in_secs,
+            timezone_offset,
+            apply_silence,
+            start_from,
+        )?;
         // Cron frequency is handled by the cron library, so we don't need to align it
         if self.frequency_type != FrequencyType::Cron {
             // `align_time` expects frequency in seconds, so convert if necessary
@@ -205,11 +216,22 @@ impl TriggerCondition {
         freq_in_secs: bool,
         timezone_offset: i32,
         apply_silence: bool,
+        start_from: Option<i64>,
     ) -> Result<i64, anyhow::Error> {
         if self.align_time {
-            self.get_aligned_next_trigger_time(freq_in_secs, timezone_offset, apply_silence)
+            self.get_aligned_next_trigger_time(
+                freq_in_secs,
+                timezone_offset,
+                apply_silence,
+                start_from,
+            )
         } else {
-            self.get_next_trigger_time_non_aligned(freq_in_secs, timezone_offset, apply_silence)
+            self.get_next_trigger_time_non_aligned(
+                freq_in_secs,
+                timezone_offset,
+                apply_silence,
+                start_from,
+            )
         }
     }
 }
@@ -595,7 +617,7 @@ mod test {
             ..Default::default()
         };
         let result = condition
-            .get_next_trigger_time_non_aligned(true, 0, false)
+            .get_next_trigger_time_non_aligned(true, 0, false, None)
             .unwrap();
         let dt = DateTime::from_timestamp_micros(result).unwrap();
         let after_5_minutes = Utc::now() + Duration::minutes(5);
@@ -609,7 +631,7 @@ mod test {
             ..Default::default()
         };
         let result = condition
-            .get_next_trigger_time_non_aligned(true, 0, false)
+            .get_next_trigger_time_non_aligned(true, 0, false, None)
             .unwrap();
         let dt = DateTime::from_timestamp_micros(result).unwrap();
         assert_eq!(dt.minute() % 5, 0);
@@ -622,7 +644,7 @@ mod test {
             ..Default::default()
         };
         let result = condition
-            .get_next_trigger_time_non_aligned(true, 0, true)
+            .get_next_trigger_time_non_aligned(true, 0, true, None)
             .unwrap();
         let dt = DateTime::from_timestamp_micros(result).unwrap();
         // The next trigger should be after the silence period
@@ -637,7 +659,7 @@ mod test {
             ..Default::default()
         };
         let result = condition
-            .get_next_trigger_time_non_aligned(true, 0, false)
+            .get_next_trigger_time_non_aligned(true, 0, false, None)
             .unwrap();
         let dt = DateTime::from_timestamp_micros(result).unwrap();
         // The next trigger should be within the tolerance range
@@ -659,7 +681,7 @@ mod test {
 
         // Mock the current time for testing
         let result = condition
-            .get_aligned_next_trigger_time(true, 0, false)
+            .get_aligned_next_trigger_time(true, 0, false, None)
             .unwrap();
         let dt = DateTime::from_timestamp_micros(result).unwrap();
 
@@ -682,7 +704,7 @@ mod test {
         // For cron expressions, the time should not be aligned by our function
         // as it's handled by the cron library
         let result = condition
-            .get_aligned_next_trigger_time(true, 0, false)
+            .get_aligned_next_trigger_time(true, 0, false, None)
             .unwrap();
         let dt = DateTime::from_timestamp_micros(result).unwrap();
 

src/service/alerts/derived_streams.rs

@@ -143,24 +143,38 @@ pub async fn save(
     }
 
     // Save the trigger to db
-    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,
-        module_key: trigger_module_key,
-        next_run_at,
-        is_realtime: false,
-        is_silenced: false,
-        ..Default::default()
-    };
-
-    match db::scheduler::get(&trigger.org, trigger.module.clone(), &trigger.module_key).await {
-        Ok(_) => db::scheduler::update_trigger(trigger)
-            .await
-            .map_err(|_| anyhow::anyhow!("Trigger already exists, but failed to update")),
-        Err(_) => db::scheduler::push(trigger)
-            .await
-            .map_err(|e| anyhow::anyhow!("Error save DerivedStream trigger: {}", e)),
+    match db::scheduler::get(
+        &derived_stream.org_id,
+        db::scheduler::TriggerModule::DerivedStream,
+        &trigger_module_key,
+    )
+    .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,
+                module_key: trigger_module_key,
+                next_run_at,
+                is_realtime: false,
+                is_silenced: false,
+                ..Default::default()
+            };
+            db::scheduler::push(trigger)
+                .await
+                .map_err(|e| anyhow::anyhow!("Error save DerivedStream trigger: {}", e))
+        }
     }
 }
 
@@ -213,8 +227,10 @@ impl DerivedStreamExt for DerivedStream {
     }
 }
 
-pub(super) fn get_next_run_at(derived_stream: &DerivedStream) -> Result<i64, anyhow::Error> {
-    let delay_in_mins = derived_stream.delay.unwrap_or_default();
+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!(
@@ -223,7 +239,12 @@ pub(super) fn get_next_run_at(derived_stream: &DerivedStream) -> Result<i64, any
     }
 
     let delay = chrono::Duration::minutes(delay_in_mins as _);
-    Ok(chrono::Utc::now()
+    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())