feat: time-based alert retention policy

docs/deployment/configuration.mdx

@@ -349,6 +349,20 @@ These endpoints are rate-limited according to the `KEEP_LIMIT_CONCURRENCY` setti
 | **MAINTENANCE_WINDOW_STRATEGY**  |          Choose the strategy                |           No            |    "default"  |      "default" or "recover_previous_status"       |
 | **WATCHER_LAPSED_TIME**          | Time in seconds to execute the alert review |           No            |       60      |             Valid positive integer                |
 
+### Alert Retention
+
+<Info>
+  Alert retention deletes alerts older than the configured number of days from
+  the database, together with their incident links. It is disabled unless
+  KEEP_ALERT_RETENTION_DAYS is set to a positive value.
+</Info>
+
+|                Env var                |                           Purpose                            |        Required         | Default Value |               Valid options                       |
+| :-----------------------------------: | :----------------------------------------------------------: | :---------------------: | :-----------: | :-----------------------------------------------: |
+| **KEEP_ALERT_RETENTION_DAYS**         | Delete alerts older than this number of days (0 disables it) |           No            |       0       |             Valid positive integer                |
+| **KEEP_ALERT_RETENTION_INTERVAL**     |        Time in seconds between retention runs                |           No            |     86400     |             Valid positive integer                |
+| **KEEP_ALERT_RETENTION_BATCH_SIZE**   |        Number of alerts deleted per batch                    |           No            |     1000      |             Valid positive integer                |
+
 ## Frontend Environment Variables
 
 <Info>

keep/api/api.py

@@ -25,7 +25,7 @@
 from keep.api.arq_pool import get_pool
 import keep.api.logging
 import keep.api.observability
-from keep.api.tasks import process_watcher_task
+from keep.api.tasks import process_retention_task, process_watcher_task
 import keep.api.utils.import_ee
 from keep.api.core.config import config
 from keep.api.core.db import dispose_session
@@ -67,7 +67,7 @@
     IdentityManagerTypes,
 )
 from keep.topologies.topology_processor import TopologyProcessor
-from keep.api.consts import KEEP_ARQ_QUEUE_MAINTENANCE, MAINTENANCE_WINDOW_ALERT_STRATEGY, REDIS
+from keep.api.consts import KEEP_ALERT_RETENTION_DAYS, KEEP_ARQ_QUEUE_MAINTENANCE, MAINTENANCE_WINDOW_ALERT_STRATEGY, REDIS
 
 # load all providers into cache
 from keep.workflowmanager.workflowmanager import WorkflowManager
@@ -187,6 +187,27 @@ async def startup():
                     "task": "task",
                 },
             )
+
+    if KEEP_ALERT_RETENTION_DAYS > 0:
+        if REDIS:
+            try:
+                logger.info("Starting the alert retention process")
+                redis: ArqRedis = await get_pool()
+                job = await redis.enqueue_job(
+                    "async_process_retention",
+                    _queue_name=KEEP_ARQ_QUEUE_MAINTENANCE,
+                )
+                logger.info(
+                    "Enqueued job",
+                    extra={
+                        "job_id": job.job_id,
+                        "queue": KEEP_ARQ_QUEUE_MAINTENANCE,
+                    },
+                )
+            except Exception:
+                logger.exception("Failed to start the alert retention process")
+        else:
+            asyncio.create_task(process_retention_task.async_process_retention())
     logger.info("Services started successfully")
 
 

keep/api/arq_worker.py

@@ -14,6 +14,7 @@
 
 import keep.api.logging
 from keep.api.consts import (
+    KEEP_ALERT_RETENTION_DAYS,
     KEEP_ARQ_QUEUE_BASIC,
     KEEP_ARQ_TASK_POOL,
     KEEP_ARQ_TASK_POOL_ALL,
@@ -158,6 +159,8 @@ class WorkerSettings:
     timeout = 30
     functions: list = FUNCTIONS
     cron_jobs: list = [cron("keep.api.tasks.process_watcher_task.async_process_watcher", second=max(0, WATCHER_LAPSED_TIME-1))]
+    if KEEP_ALERT_RETENTION_DAYS > 0:
+        cron_jobs.append(cron("keep.api.tasks.process_retention_task.async_process_retention", hour=2, minute=0))
     queue_name: str
     health_check_interval: int = 10
     health_check_key: str

keep/api/consts.py

@@ -32,6 +32,13 @@
     "MAINTENANCE_WINDOW_STRATEGY", "default"
 )  # recover_previous_status or default
 WATCHER_LAPSED_TIME = int(os.environ.get("KEEP_WATCHER_LAPSED_TIME", 60))  # in seconds
+KEEP_ALERT_RETENTION_DAYS = int(os.environ.get("KEEP_ALERT_RETENTION_DAYS", 0))
+KEEP_ALERT_RETENTION_INTERVAL = int(
+    os.environ.get("KEEP_ALERT_RETENTION_INTERVAL", 86400)
+)  # in seconds
+KEEP_ALERT_RETENTION_BATCH_SIZE = int(
+    os.environ.get("KEEP_ALERT_RETENTION_BATCH_SIZE", 1000)
+)
 ###
 # Set ARQ_TASK_POOL_TO_EXECUTE to "none", "all", "basic_processing" or "ai"
 # to split the tasks between the workers.

keep/api/core/db.py

@@ -5989,4 +5989,66 @@ def recover_prev_alert_status(alert: Alert, session: Optional[Session] = None):
             )
         )
         session.exec(query)
-        session.commit()
+        session.commit()
+
+
+@retry_on_db_error
+def delete_alerts_by_retention(
+    tenant_id: str,
+    purge_before: datetime,
+    batch_size: int = 1000,
+    session: Optional[Session] = None,
+) -> int:
+    total_deleted = 0
+    with existed_or_new_session(session) as session:
+        while True:
+            alert_ids = session.exec(
+                select(Alert.id)
+                .where(
+                    Alert.tenant_id == tenant_id,
+                    Alert.timestamp < purge_before,
+                )
+                .limit(batch_size)
+            ).all()
+            if not alert_ids:
+                break
+
+            fingerprints = session.exec(
+                select(LastAlert.fingerprint).where(
+                    LastAlert.tenant_id == tenant_id,
+                    LastAlert.alert_id.in_(alert_ids),
+                )
+            ).all()
+
+            if fingerprints:
+                session.query(LastAlertToIncident).filter(
+                    LastAlertToIncident.tenant_id == tenant_id,
+                    LastAlertToIncident.fingerprint.in_(fingerprints),
+                ).delete(synchronize_session=False)
+
+            session.query(LastAlert).filter(
+                LastAlert.tenant_id == tenant_id,
+                LastAlert.alert_id.in_(alert_ids),
+            ).delete(synchronize_session=False)
+
+            session.query(AlertToIncident).filter(
+                AlertToIncident.tenant_id == tenant_id,
+                AlertToIncident.alert_id.in_(alert_ids),
+            ).delete(synchronize_session=False)
+
+            deleted = (
+                session.query(Alert)
+                .filter(Alert.id.in_(alert_ids))
+                .delete(synchronize_session=False)
+            )
+            session.commit()
+            total_deleted += deleted
+            logger.info(
+                "Deleted batch of alerts by retention",
+                extra={
+                    "tenant_id": tenant_id,
+                    "deleted": deleted,
+                    "total_deleted": total_deleted,
+                },
+            )
+    return total_deleted

keep/api/tasks/process_retention_task.py

@@ -0,0 +1,76 @@
+import asyncio
+import datetime
+import logging
+
+import redis
+from filelock import FileLock, Timeout
+
+from keep.api.consts import (
+    KEEP_ALERT_RETENTION_BATCH_SIZE,
+    KEEP_ALERT_RETENTION_DAYS,
+    KEEP_ALERT_RETENTION_INTERVAL,
+    REDIS,
+)
+from keep.api.core.db import delete_alerts_by_retention, get_tenants
+
+logger = logging.getLogger(__name__)
+
+
+def process_retention(logger):
+    if KEEP_ALERT_RETENTION_DAYS <= 0:
+        return
+    purge_before = datetime.datetime.utcnow() - datetime.timedelta(
+        days=KEEP_ALERT_RETENTION_DAYS
+    )
+    for tenant in get_tenants():
+        deleted = delete_alerts_by_retention(
+            tenant.id, purge_before, KEEP_ALERT_RETENTION_BATCH_SIZE
+        )
+        if deleted:
+            logger.info(
+                "Deleted alerts by retention policy",
+                extra={"tenant_id": tenant.id, "deleted": deleted},
+            )
+
+
+async def async_process_retention(*args):
+    if REDIS:
+        ctx = args[0]
+        redis_instance: redis.Redis = ctx.get("redis")
+        lock_key = "lock:retention:process"
+        lock_acquired = await redis_instance.set(lock_key, "1", ex=3600, nx=True)
+        if not lock_acquired:
+            logger.info("Retention process is already running, skipping this run.")
+            return
+        logger.info("Retention process started, acquiring lock.")
+        try:
+            loop = asyncio.get_running_loop()
+            await loop.run_in_executor(ctx.get("pool"), process_retention, logger)
+        except Exception as e:
+            logger.error("Error in retention process: %s", e, exc_info=True)
+            raise
+        finally:
+            await redis_instance.delete(lock_key)
+            logger.info("Retention process completed and lock released.")
+    else:
+        while True:
+            init_time = datetime.datetime.now()
+            try:
+                with FileLock(
+                    "/tmp/retention_process.lock",
+                    timeout=KEEP_ALERT_RETENTION_INTERVAL // 2,
+                ):
+                    logger.info("Retention process started, acquiring lock.")
+                    loop = asyncio.get_running_loop()
+                    await loop.run_in_executor(None, process_retention, logger)
+                    complete_time = datetime.datetime.now()
+                    await asyncio.sleep(
+                        max(
+                            0,
+                            KEEP_ALERT_RETENTION_INTERVAL
+                            - (complete_time - init_time).total_seconds(),
+                        )
+                    )
+                    logger.info("Retention process completed.")
+            except Timeout:
+                logger.info("Retention process is already running, skipping this run.")