diff --git a/docs/deployment/configuration.mdx b/docs/deployment/configuration.mdx
index e70738df53..4c825b2bfc 100644
--- a/docs/deployment/configuration.mdx
+++ b/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
+
+
+ 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.
+
+
+| 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
diff --git a/keep/api/api.py b/keep/api/api.py
index 4ae2072457..626585056e 100644
--- a/keep/api/api.py
+++ b/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")
diff --git a/keep/api/arq_worker.py b/keep/api/arq_worker.py
index f7550d0907..4567d5e697 100644
--- a/keep/api/arq_worker.py
+++ b/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
diff --git a/keep/api/consts.py b/keep/api/consts.py
index f2f8ce9e00..a8cb3fed53 100644
--- a/keep/api/consts.py
+++ b/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.
diff --git a/keep/api/core/db.py b/keep/api/core/db.py
index 878d0d877b..6e26eeb0c0 100644
--- a/keep/api/core/db.py
+++ b/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()
\ No newline at end of file
+ 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
\ No newline at end of file
diff --git a/keep/api/tasks/process_retention_task.py b/keep/api/tasks/process_retention_task.py
new file mode 100644
index 0000000000..23da54dd8d
--- /dev/null
+++ b/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.")
diff --git a/tests/test_alert_retention.py b/tests/test_alert_retention.py
new file mode 100644
index 0000000000..b11ae87297
--- /dev/null
+++ b/tests/test_alert_retention.py
@@ -0,0 +1,160 @@
+from datetime import datetime, timedelta
+
+from keep.api.core.db import delete_alerts_by_retention
+from keep.api.core.dependencies import SINGLE_TENANT_UUID
+from keep.api.models.db.alert import (
+ Alert,
+ AlertToIncident,
+ LastAlert,
+ LastAlertToIncident,
+)
+from keep.api.models.db.incident import Incident
+
+
+def _create_alert(db_session, fingerprint, timestamp):
+ alert = Alert(
+ tenant_id=SINGLE_TENANT_UUID,
+ provider_type="test",
+ provider_id="test",
+ event={"name": fingerprint, "fingerprint": fingerprint},
+ fingerprint=fingerprint,
+ timestamp=timestamp,
+ )
+ db_session.add(alert)
+ db_session.commit()
+ return alert
+
+
+def _create_last_alert(db_session, alert):
+ last_alert = LastAlert(
+ tenant_id=SINGLE_TENANT_UUID,
+ fingerprint=alert.fingerprint,
+ alert_id=alert.id,
+ timestamp=alert.timestamp,
+ first_timestamp=alert.timestamp,
+ )
+ db_session.add(last_alert)
+ db_session.commit()
+ return last_alert
+
+
+def _create_incident(db_session):
+ incident = Incident(
+ tenant_id=SINGLE_TENANT_UUID,
+ user_generated_name="test-incident",
+ user_summary="test",
+ generated_summary="test",
+ )
+ db_session.add(incident)
+ db_session.commit()
+ return incident
+
+
+def test_delete_alerts_by_retention_purges_old_alerts(db_session):
+ incident = _create_incident(db_session)
+ old_timestamp = datetime.utcnow() - timedelta(days=100)
+
+ for i in range(3):
+ alert = _create_alert(db_session, f"old-{i}", old_timestamp)
+ _create_last_alert(db_session, alert)
+ db_session.add(
+ AlertToIncident(
+ tenant_id=SINGLE_TENANT_UUID,
+ alert_id=alert.id,
+ incident_id=incident.id,
+ )
+ )
+ db_session.add(
+ LastAlertToIncident(
+ tenant_id=SINGLE_TENANT_UUID,
+ fingerprint=alert.fingerprint,
+ incident_id=incident.id,
+ )
+ )
+ db_session.commit()
+
+ for i in range(2):
+ alert = _create_alert(db_session, f"fresh-{i}", datetime.utcnow())
+ _create_last_alert(db_session, alert)
+
+ purge_before = datetime.utcnow() - timedelta(days=30)
+ deleted = delete_alerts_by_retention(
+ SINGLE_TENANT_UUID, purge_before, batch_size=2, session=db_session
+ )
+
+ assert deleted == 3
+ remaining_alerts = db_session.query(Alert).all()
+ assert sorted(alert.fingerprint for alert in remaining_alerts) == [
+ "fresh-0",
+ "fresh-1",
+ ]
+ remaining_last_alerts = db_session.query(LastAlert).all()
+ assert sorted(last_alert.fingerprint for last_alert in remaining_last_alerts) == [
+ "fresh-0",
+ "fresh-1",
+ ]
+ assert db_session.query(AlertToIncident).count() == 0
+ assert db_session.query(LastAlertToIncident).count() == 0
+
+
+def test_delete_alerts_by_retention_keeps_active_fingerprint(db_session):
+ old_alert = _create_alert(
+ db_session, "service-down", datetime.utcnow() - timedelta(days=100)
+ )
+ fresh_alert = _create_alert(db_session, "service-down", datetime.utcnow())
+ _create_last_alert(db_session, fresh_alert)
+ old_alert_id = old_alert.id
+ fresh_alert_id = fresh_alert.id
+
+ purge_before = datetime.utcnow() - timedelta(days=30)
+ deleted = delete_alerts_by_retention(
+ SINGLE_TENANT_UUID, purge_before, session=db_session
+ )
+
+ assert deleted == 1
+ assert old_alert_id != fresh_alert_id
+ remaining_alerts = db_session.query(Alert).all()
+ assert [alert.id for alert in remaining_alerts] == [fresh_alert_id]
+ remaining_last_alerts = db_session.query(LastAlert).all()
+ assert [last_alert.fingerprint for last_alert in remaining_last_alerts] == [
+ "service-down"
+ ]
+
+
+def test_delete_alerts_by_retention_removes_soft_deleted_incident_links(db_session):
+ incident = _create_incident(db_session)
+ alert = _create_alert(db_session, "stale", datetime.utcnow() - timedelta(days=100))
+ _create_last_alert(db_session, alert)
+ db_session.add(
+ LastAlertToIncident(
+ tenant_id=SINGLE_TENANT_UUID,
+ fingerprint=alert.fingerprint,
+ incident_id=incident.id,
+ deleted_at=datetime.utcnow() - timedelta(days=50),
+ )
+ )
+ db_session.commit()
+
+ purge_before = datetime.utcnow() - timedelta(days=30)
+ deleted = delete_alerts_by_retention(
+ SINGLE_TENANT_UUID, purge_before, session=db_session
+ )
+
+ assert deleted == 1
+ assert db_session.query(Alert).count() == 0
+ assert db_session.query(LastAlert).count() == 0
+ assert db_session.query(LastAlertToIncident).count() == 0
+
+
+def test_delete_alerts_by_retention_noop_when_nothing_expired(db_session):
+ alert = _create_alert(db_session, "fresh", datetime.utcnow())
+ _create_last_alert(db_session, alert)
+
+ purge_before = datetime.utcnow() - timedelta(days=30)
+ deleted = delete_alerts_by_retention(
+ SINGLE_TENANT_UUID, purge_before, session=db_session
+ )
+
+ assert deleted == 0
+ assert db_session.query(Alert).count() == 1
+ assert db_session.query(LastAlert).count() == 1
diff --git a/tests/test_process_retention_task.py b/tests/test_process_retention_task.py
new file mode 100644
index 0000000000..f60b687b6e
--- /dev/null
+++ b/tests/test_process_retention_task.py
@@ -0,0 +1,164 @@
+import asyncio
+from datetime import datetime, timedelta
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, call, patch
+
+import pytest
+
+from keep.api.tasks import process_retention_task
+
+
+def test_process_retention_skips_when_disabled():
+ with (
+ patch.object(process_retention_task, "KEEP_ALERT_RETENTION_DAYS", 0),
+ patch.object(process_retention_task, "get_tenants") as get_tenants,
+ patch.object(
+ process_retention_task, "delete_alerts_by_retention"
+ ) as delete_alerts,
+ ):
+ process_retention_task.process_retention(MagicMock())
+
+ get_tenants.assert_not_called()
+ delete_alerts.assert_not_called()
+
+
+def test_process_retention_applies_policy_to_each_tenant():
+ fixed_now = datetime(2026, 7, 26, 12, 0, 0)
+ datetime_class = MagicMock(wraps=datetime)
+ datetime_class.utcnow.return_value = fixed_now
+ logger = MagicMock()
+
+ with (
+ patch.object(process_retention_task, "KEEP_ALERT_RETENTION_DAYS", 30),
+ patch.object(process_retention_task, "KEEP_ALERT_RETENTION_BATCH_SIZE", 25),
+ patch.object(process_retention_task.datetime, "datetime", datetime_class),
+ patch.object(
+ process_retention_task,
+ "get_tenants",
+ return_value=[
+ SimpleNamespace(id="tenant-with-expired-alerts"),
+ SimpleNamespace(id="tenant-without-expired-alerts"),
+ ],
+ ),
+ patch.object(
+ process_retention_task,
+ "delete_alerts_by_retention",
+ side_effect=[3, 0],
+ ) as delete_alerts,
+ ):
+ process_retention_task.process_retention(logger)
+
+ purge_before = fixed_now - timedelta(days=30)
+ assert delete_alerts.call_args_list == [
+ call("tenant-with-expired-alerts", purge_before, 25),
+ call("tenant-without-expired-alerts", purge_before, 25),
+ ]
+ logger.info.assert_called_once_with(
+ "Deleted alerts by retention policy",
+ extra={"tenant_id": "tenant-with-expired-alerts", "deleted": 3},
+ )
+
+
+@pytest.mark.asyncio
+async def test_redis_retention_skips_when_lock_is_held():
+ redis_instance = AsyncMock()
+ redis_instance.set.return_value = False
+ process = MagicMock()
+
+ with (
+ patch.object(process_retention_task, "REDIS", True),
+ patch.object(process_retention_task, "process_retention", process),
+ ):
+ await process_retention_task.async_process_retention(
+ {"redis": redis_instance, "pool": None}
+ )
+
+ redis_instance.set.assert_awaited_once_with(
+ "lock:retention:process", "1", ex=3600, nx=True
+ )
+ process.assert_not_called()
+ redis_instance.delete.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_redis_retention_runs_in_pool_and_releases_lock():
+ redis_instance = AsyncMock()
+ redis_instance.set.return_value = True
+ pool = MagicMock()
+ loop = MagicMock()
+ loop.run_in_executor = AsyncMock()
+ process = MagicMock()
+
+ with (
+ patch.object(process_retention_task, "REDIS", True),
+ patch.object(process_retention_task, "process_retention", process),
+ patch.object(
+ process_retention_task.asyncio,
+ "get_running_loop",
+ return_value=loop,
+ ),
+ ):
+ await process_retention_task.async_process_retention(
+ {"redis": redis_instance, "pool": pool}
+ )
+
+ loop.run_in_executor.assert_awaited_once_with(
+ pool, process, process_retention_task.logger
+ )
+ redis_instance.delete.assert_awaited_once_with("lock:retention:process")
+
+
+@pytest.mark.asyncio
+async def test_redis_retention_releases_lock_after_failure():
+ redis_instance = AsyncMock()
+ redis_instance.set.return_value = True
+ loop = MagicMock()
+ loop.run_in_executor = AsyncMock(side_effect=RuntimeError("retention failed"))
+
+ with (
+ patch.object(process_retention_task, "REDIS", True),
+ patch.object(
+ process_retention_task.asyncio,
+ "get_running_loop",
+ return_value=loop,
+ ),
+ pytest.raises(RuntimeError, match="retention failed"),
+ ):
+ await process_retention_task.async_process_retention(
+ {"redis": redis_instance, "pool": None}
+ )
+
+ redis_instance.delete.assert_awaited_once_with("lock:retention:process")
+
+
+@pytest.mark.asyncio
+async def test_local_retention_runs_once_before_sleeping():
+ lock = MagicMock()
+ loop = MagicMock()
+ loop.run_in_executor = AsyncMock()
+ sleep = AsyncMock(side_effect=asyncio.CancelledError)
+ process = MagicMock()
+
+ with (
+ patch.object(process_retention_task, "REDIS", False),
+ patch.object(process_retention_task, "KEEP_ALERT_RETENTION_INTERVAL", 60),
+ patch.object(
+ process_retention_task, "FileLock", return_value=lock
+ ) as file_lock,
+ patch.object(process_retention_task, "process_retention", process),
+ patch.object(
+ process_retention_task.asyncio,
+ "get_running_loop",
+ return_value=loop,
+ ),
+ patch.object(process_retention_task.asyncio, "sleep", sleep),
+ pytest.raises(asyncio.CancelledError),
+ ):
+ await process_retention_task.async_process_retention()
+
+ file_lock.assert_called_once_with("/tmp/retention_process.lock", timeout=30)
+ loop.run_in_executor.assert_awaited_once_with(
+ None, process, process_retention_task.logger
+ )
+ sleep.assert_awaited_once()
+ assert 0 <= sleep.await_args.args[0] <= 60