Repository: keephq/keep
Pull request: feat: interval option in alert specification
close https://github.com/keephq/keep/issues/189

Problem statement:
Add "interval" to the alert spec
You can schedule an alert by adding the `-i/--interval` flag to the Keep CLI command.

However, sometimes you want to schedule multiple alerts with different intervals. 
The easiest way to do so will be to add the "interval" property to the main alert spec, so Keep will be able to schedule it without flag.

Hints from issue discussion:

Commit message: feat: add auth0 provider

Patch to review:
diff --git a/examples/alerts/db_disk_space.yml b/examples/alerts/db_disk_space.yml
index 92908ec442..e4fbc004db 100644
--- a/examples/alerts/db_disk_space.yml
+++ b/examples/alerts/db_disk_space.yml
@@ -8,6 +8,8 @@ alert:
   services:
     - db
     - api
+  # Run every 60 seconds
+  interval: 60
   steps:
     - name: db-no-space
       provider:
diff --git a/keep/alert/alert.py b/keep/alert/alert.py
index 438e0f63e4..0d8131b8f4 100644
--- a/keep/alert/alert.py
+++ b/keep/alert/alert.py
@@ -21,6 +21,7 @@ class Alert:
     alert_source: str
     alert_owners: typing.List[str]
     alert_tags: typing.List[str]
+    alert_interval: int
     alert_steps: typing.List[Step]
     alert_actions: typing.List[Action]
     alert_file: str = None
diff --git a/keep/alertmanager/alertcheduler.py b/keep/alertmanager/alertcheduler.py
new file mode 100644
index 0000000000..8de81b1ad7
--- /dev/null
+++ b/keep/alertmanager/alertcheduler.py
@@ -0,0 +1,57 @@
+import logging
+import threading
+import time
+import typing
+
+from keep.alert.alert import Alert
+
+
+class AlertScheduler:
+    def __init__(self, alert_manager):
+        self.logger = logging.getLogger(__name__)
+        self.threads = []
+        self.alert_manager = alert_manager
+        self._stop = False
+
+    def run_alerts(self, alerts: typing.List[Alert]):
+        for alert in alerts:
+            thread = threading.Thread(
+                target=self._run_alerts_with_interval,
+                args=[alert],
+                daemon=True,
+            )
+            thread.start()
+            self.threads.append(thread)
+        # as long as the stop flag is not set, sleep
+        while not self._stop:
+            time.sleep(1)
+
+    def stop(self):
+        self.logger.info("Stopping scheduled alerts")
+        self._stop = True
+        # Now wait for the threads to finish
+        for thread in self.threads:
+            thread.join()
+        self.logger.info("Scheduled alerts stopped")
+
+    def _run_alerts_with_interval(
+        self,
+        alert: Alert,
+    ):
+        """Simple scheduling of alerts with interval
+
+        TODO: Use https://github.com/agronholm/apscheduler
+
+        Args:
+            alert (Alert): The alert to run.
+        """
+        while True and not self._stop:
+            self.logger.info(f"Running alert {alert.alert_id}...")
+            self.alert_manager._run_alert(alert)
+            self.logger.info(f"Alert {alert.alert_id} ran")
+            if alert.alert_interval > 0:
+                self.logger.info(f"Sleeping for {alert.alert_interval} seconds...")
+                time.sleep(alert.alert_interval)
+            else:
+                self.logger.info("Alert will not run again")
+                break
diff --git a/keep/alertmanager/alertmanager.py b/keep/alertmanager/alertmanager.py
index 695413e158..0f5df8edf2 100644
--- a/keep/alertmanager/alertmanager.py
+++ b/keep/alertmanager/alertmanager.py
@@ -1,60 +1,58 @@
 import logging
 import os
+import threading
 import time
 import typing
 
 from keep.alert.alert import Alert
+from keep.alertmanager.alertcheduler import AlertScheduler
 from keep.parser.parser import Parser
 
 
 class AlertManager:
-    def __init__(self):
+    def __init__(self, interval: int = 0):
         self.parser = Parser()
         self.logger = logging.getLogger(__name__)
+        self.scheduler = AlertScheduler(self)
+        self.default_interval = interval
+        self.scheduler_mode = False
 
-    def run(
-        self,
-        alerts_path: str | list[str],
-        providers_file: str = None,
-        interval: int = 0,
-    ):
+    def stop(self):
+        if self.scheduler_mode:
+            self.logger.info("Stopping alert manager")
+            self.scheduler.stop()
+            self.logger.info("Alert manager stopped")
+        else:
+            pass
+
+    def run(self, alerts_path: str | list[str], providers_file: str = None):
         """
         Run alerts from a file or directory.
 
         Args:
-            alert (str): Either a an alert yaml or a directory containing alert yamls or a list of urls to get the alerts from.
+            alert (str): Either an alert yaml or a directory containing alert yamls or a list of URLs to get the alerts from.
             providers_file (str, optional): The path to the providers yaml. Defaults to None.
+            interval (int, optional): The interval in seconds between consecutive runs. Defaults to 0 (run once).
         """
-        self.logger.info(
-            f"Running alert(s) from {alerts_path}", extra={"interval": interval}
-        )
-        # If interval is set, run the alert every INTERVAL seconds until the user stops the process
-        if interval > 0:
+        self.logger.info(f"Running alert(s) from {alerts_path}")
+        alerts = self.get_alerts(alerts_path, providers_file)
+        alerts_errors = []
+        # If at least one alert has an interval, run alerts using the scheduler,
+        #   otherwise, just run it
+        if self.default_interval or any([alert.alert_interval for alert in alerts]):
+            # running alerts in scheduler mode
             self.logger.info(
-                "Running in interval mode. Press Ctrl+C to stop the process."
-            )
-            while True:
-                try:
-                    self._run(alerts_path, providers_file)
-                except Exception:
-                    self.logger.exception("Error running alert in interval mode")
-                self.logger.info(f"Sleeping for {interval} seconds...")
-                time.sleep(interval)
-        # If interval is not set, run the alert once
-        else:
-            errors = self._run(alerts_path, providers_file)
-        # TODO: errors should be part of the Alert/Action/Step class so it'll be distinguishable
-        if any(errors):
-            self.logger.error(
-                f"Alert(s) from {alerts_path} ran with errors",
-                extra={"interval": interval},
+                "Found at least one alert with an interval, running in scheduler mode"
             )
-            raise Exception("Alert(s) ran with errors")
+            self.scheduler_mode = True
+            # This will halt until KeyboardInterrupt
+            self.scheduler.run_alerts(alerts)
+            self.logger.info("Alert(s) scheduled")
         else:
-            self.logger.info(
-                f"Alert(s) from {alerts_path} ran successfully",
-                extra={"interval": interval},
-            )
+            # running alerts in the regular mode
+            alerts_errors = self._run_alerts(alerts)
+
+        return alerts_errors
 
     def get_alerts(
         self, alert_path: str | tuple[str], providers_file: str = None
@@ -67,6 +65,10 @@ def get_alerts(
             alerts.extend(self._get_alerts_from_directory(alert_path, providers_file))
         else:
             alerts = self.parser.parse(alert_path, providers_file)
+
+        # override the default interval if it is not set
+        for alert in alerts:
+            alert.alert_interval = alert.alert_interval or self.default_interval
         return alerts
 
     def _run(self, alert_path: str | tuple[str], providers_file: str = None):
@@ -101,44 +103,40 @@ def _get_alerts_from_directory(
                     )
         return alerts
 
+    def _run_alert(self, alert: Alert):
+        self.logger.info(f"Running alert {alert.alert_id}")
+        errors = []
+        try:
+            errors = alert.run()
+        except Exception as e:
+            self.logger.error(
+                f"Error running alert {alert.alert_id}", extra={"exception": e}
+            )
+            if alert.on_failure:
+                self.logger.info(
+                    f"Running on_failure action for alert {alert.alert_id}"
+                )
+                # Adding the exception message to the provider context so it'll be available for the action
+                message = f"Alert `{alert.alert_id}` failed with exception: `{str(e)}`"
+                alert.on_failure.provider_context = {"message": message}
+                alert.on_failure.run()
+            raise
+        if any(errors):
+            self.logger.info(msg=f"Alert {alert.alert_id} ran with errors")
+        else:
+            self.logger.info(f"Alert {alert.alert_id} ran successfully")
+        return errors
+
     def _run_alerts(self, alerts: typing.List[Alert]):
         alerts_errors = []
         for alert in alerts:
-            # otherwise any(errors) might throw an exception
-            errors = []
-            self.logger.info(f"Running alert {alert.alert_id}")
             try:
-                errors = alert.run()
+                errors = self._run_alert(alert)
+                alerts_errors.append(errors)
             except Exception as e:
                 self.logger.error(
                     f"Error running alert {alert.alert_id}", extra={"exception": e}
                 )
-                if alert.on_failure:
-                    self.logger.info(
-                        f"Running on_failure action for alert {alert.alert_id}"
-                    )
-                    # Adding the exception message to the provider context so it'll be available for the action
-                    message = (
-                        f"Alert `{alert.alert_id}` failed with exception: `{str(e)}`"
-                    )
-                    alert.on_failure.provider_context = {"message": message}
-                    alert.on_failure.run()
                 raise
-            if any(errors):
-                self.logger.info(msg=f"Alert {alert.alert_id} ran with errors")
-            else:
-                self.logger.info(f"Alert {alert.alert_id} ran successfully")
-            alerts_errors.extend(errors)
-        return alerts_errors
 
-    def run_step(self, alert_id: str, step: str):
-        self.logger.info(f"Running step {step} of alert {alert.alert_id}")
-        try:
-            alert = self.get_alerts(alert_id)
-            alert.run_step(step)
-        except Exception as e:
-            self.logger.error(
-                f"Error running step {step} of alert {alert.alert_id}",
-                extra={"exception": e},
-            )
-        self.logger.info(f"Step {step} of alert {alert.alert_id} ran successfully")
+        return alerts_errors
diff --git a/keep/cli/cli.py b/keep/cli/cli.py
index 6b64a22771..b918f0cc32 100644
--- a/keep/cli/cli.py
+++ b/keep/cli/cli.py
@@ -192,11 +192,13 @@ def run(
 ):
     """Run the alert."""
     logger.debug(f"Running alert in {alerts_directory or alert_url}")
-    alert_manager = AlertManager()
+    alert_manager = AlertManager(interval)
     try:
-        alert_manager.run(
-            alerts_directory or alert_url, providers_file, interval=interval
-        )
+        alert_manager.run(alerts_directory or alert_url, providers_file)
+    except KeyboardInterrupt:
+        logger.info("Keep stopped by user, stopping the scheduler")
+        alert_manager.stop()
+        logger.info("Scheduler stopped")
     except Exception as e:
         logger.error(f"Error running alert {alerts_directory or alert_url}: {e}")
         if info.verbose:
diff --git a/keep/parser/parser.py b/keep/parser/parser.py
index 14f04ccd25..defc397548 100644
--- a/keep/parser/parser.py
+++ b/keep/parser/parser.py
@@ -100,12 +100,14 @@ def _parse_alert(self, alert: dict, alert_source: str) -> Alert:
         alert_tags = self._parse_tags(alert)
         alert_steps = self._parse_steps(alert)
         alert_actions = self._parse_actions(alert)
+        alert_interval = self._parse_interval(alert)
         on_failure_action = self._get_on_failure_action(alert)
         alert = Alert(
             alert_id=alert_id,
             alert_source=alert_source,
             alert_owners=alert_owners,
             alert_tags=alert_tags,
+            alert_interval=alert_interval,
             alert_steps=alert_steps,
             alert_actions=alert_actions,
             on_failure=on_failure_action,
@@ -198,6 +200,10 @@ def _parse_tags(self, alert) -> typing.List[str]:
         alert_tags = alert.get("tags", [])
         return alert_tags
 
+    def _parse_interval(self, alert) -> int:
+        alert_interval = alert.get("interval", 0)
+        return alert_interval
+
     def _parse_steps(self, alert) -> typing.List[Step]:
         self.logger.debug("Parsing steps")
         alert_steps = alert.get("steps", [])


Code review comments:
diff_hunk: @@ -1,60 +1,58 @@
 import logging
 import os
+import threading
 import time
 import typing
 
 from keep.alert.alert import Alert
+from keep.alertmanager.alertcheduler import AlertScheduler
 from keep.parser.parser import Parser
 
 
 class AlertManager:
-    def __init__(self):
+    def __init__(self, interval: int = 0):
         self.parser = Parser()
         self.logger = logging.getLogger(__name__)
+        self.scheduler = AlertScheduler(self)
+        self.default_interval = interval
+        self.scheduler_mode = False
 
-    def run(
-        self,
-        alerts_path: str | list[str],
-        providers_file: str = None,
-        interval: int = 0,
-    ):
+    def stop(self):
+        if self.scheduler_mode:
+            self.logger.info("Stopping alert manager")
+            self.scheduler.stop()
+            self.logger.info("Alert manager stopped")
+        else:
+            pass
+
+    def run(self, alerts_path: str | list[str], providers_file: str = None):
         """
         Run alerts from a file or directory.
 
         Args:
-            alert (str): Either a an alert yaml or a directory containing alert yamls or a list of urls to get the alerts from.
+            alert (str): Either an alert yaml or a directory containing alert yamls or a list of URLs to get the alerts from.
             providers_file (str, optional): The path to the providers yaml. Defaults to None.
+            interval (int, optional): The interval in seconds between consecutive runs. Defaults to 0 (run once).
path: keep/alertmanager/alertmanager.py
text: @user1:
it's not a parameter for this function

diff_hunk: @@ -1,60 +1,58 @@
 import logging
 import os
+import threading
 import time
 import typing
 
 from keep.alert.alert import Alert
+from keep.alertmanager.alertcheduler import AlertScheduler
 from keep.parser.parser import Parser
 
 
 class AlertManager:
-    def __init__(self):
+    def __init__(self, interval: int = 0):
         self.parser = Parser()
         self.logger = logging.getLogger(__name__)
+        self.scheduler = AlertScheduler(self)
+        self.default_interval = interval
+        self.scheduler_mode = False
 
-    def run(
-        self,
-        alerts_path: str | list[str],
-        providers_file: str = None,
-        interval: int = 0,
-    ):
+    def stop(self):
+        if self.scheduler_mode:
+            self.logger.info("Stopping alert manager")
+            self.scheduler.stop()
+            self.logger.info("Alert manager stopped")
+        else:
+            pass
+
+    def run(self, alerts_path: str | list[str], providers_file: str = None):
         """
         Run alerts from a file or directory.
 
         Args:
-            alert (str): Either a an alert yaml or a directory containing alert yamls or a list of urls to get the alerts from.
+            alert (str): Either an alert yaml or a directory containing alert yamls or a list of URLs to get the alerts from.
             providers_file (str, optional): The path to the providers yaml. Defaults to None.
+            interval (int, optional): The interval in seconds between consecutive runs. Defaults to 0 (run once).
path: keep/alertmanager/alertmanager.py
text: @user1:
```suggestion
```