feat: opt-in event log retention

docs/docs/deployment/oss/dagster-yaml.md

@@ -147,6 +147,8 @@ retention:
       skipped: 7
       failure: 30
       success: -1
+  event_logs:
+    purge_after_days: 30
 
 sensors:
   use_threads: true
@@ -295,6 +297,8 @@ retention:
       skipped: 7
       failure: 30
       success: -1
+  event_logs:
+    purge_after_days: 30
 ```
 
 ### `sensors`

python_modules/dagster/dagster/_core/events/__init__.py

@@ -293,6 +293,18 @@ class DagsterEventType(str, Enum):
     DagsterEventType.ASSET_CHECK_EVALUATION_PLANNED,
 }
 
+# event types that retention policies will never delete; everything else is fair game
+RETAINED_EVENT_TYPES = (
+    ASSET_EVENTS
+    | ASSET_CHECK_EVENTS
+    | {
+        DagsterEventType.ASSET_HEALTH_CHANGED,
+        DagsterEventType.ASSET_WIPED,
+    }
+)
+
+PURGEABLE_EVENT_TYPES = {e for e in DagsterEventType} - RETAINED_EVENT_TYPES
+
 
 class RunFailureReasonSerializer(EnumSerializer):
     def unpack(self, value: str):

python_modules/dagster/dagster/_core/instance/config.py

@@ -323,17 +323,45 @@ def _tick_retention_config_schema() -> Field:
     )
 
 
+def _event_log_retention_config_schema() -> Field:
+    return Field(
+        {
+            "purge_after_days": Field(
+                int,
+                is_required=False,
+                description=(
+                    "Delete non-asset event log rows older than this many days. "
+                    "Asset materializations, observations, checks, freshness, health, "
+                    "and wipe events are always preserved."
+                ),
+            ),
+        },
+        is_required=False,
+    )
+
+
 def retention_config_schema() -> Field:
     return Field(
         {
             "schedule": _tick_retention_config_schema(),
             "sensor": _tick_retention_config_schema(),
             "auto_materialize": _tick_retention_config_schema(),
+            "event_logs": _event_log_retention_config_schema(),
         },
         is_required=False,
     )
 
 
+def get_event_log_retention_days(settings: Mapping[str, Any] | None) -> int | None:
+    """Return the configured retention in days, or None if not configured / disabled."""
+    if not settings:
+        return None
+    days = settings.get("purge_after_days")
+    if not isinstance(days, int) or days <= 0:
+        return None
+    return days
+
+
 def get_tick_retention_settings(
     settings: Mapping[str, Any] | None,
     default_retention_settings: Mapping["TickStatus", int],

python_modules/dagster/dagster/_core/instance/methods/daemon_methods.py

@@ -54,6 +54,14 @@ def auto_materialize_use_sensors(self) -> bool: ...
     @abstractmethod
     def freshness_enabled(self) -> bool: ...
 
+    @property
+    @abstractmethod
+    def event_log_retention_enabled(self) -> bool: ...
+
+    @property
+    @abstractmethod
+    def event_log_retention_days(self) -> int | None: ...
+
     def add_daemon_heartbeat(self, daemon_heartbeat: "DaemonHeartbeat") -> None:
         """Called on a regular interval by the daemon."""
         self.run_storage.add_daemon_heartbeat(daemon_heartbeat)
@@ -76,6 +84,7 @@ def get_required_daemon_types(self) -> Sequence[str]:
             SchedulerDaemon,
             SensorDaemon,
         )
+        from dagster._daemon.event_log_retention import EventLogRetentionDaemon
         from dagster._daemon.freshness import FreshnessDaemon
         from dagster._daemon.run_coordinator.queued_run_coordinator_daemon import (
             QueuedRunCoordinatorDaemon,
@@ -97,6 +106,8 @@ def get_required_daemon_types(self) -> Sequence[str]:
             daemons.append(AssetDaemon.daemon_type())
         if self.freshness_enabled:
             daemons.append(FreshnessDaemon.daemon_type())
+        if self.event_log_retention_enabled:
+            daemons.append(EventLogRetentionDaemon.daemon_type())
         return daemons
 
     def get_daemon_statuses(

python_modules/dagster/dagster/_core/instance/methods/settings_methods.py

@@ -165,6 +165,17 @@ def auto_materialize_enabled(self) -> bool:
     def freshness_enabled(self) -> bool:
         return self.get_settings("freshness").get("enabled", True)
 
+    @property
+    def event_log_retention_days(self) -> int | None:
+        """Configured retention horizon for non-asset event log rows. None if disabled."""
+        from dagster._core.instance.config import get_event_log_retention_days
+
+        return get_event_log_retention_days(self.get_settings("retention").get("event_logs"))
+
+    @property
+    def event_log_retention_enabled(self) -> bool:
+        return self.event_log_retention_days is not None
+
     @property
     def auto_materialize_minimum_interval_seconds(self) -> int:
         return self.get_settings("auto_materialize").get("minimum_interval_seconds")

python_modules/dagster/dagster/_core/storage/event_log/base.py

@@ -295,6 +295,15 @@ def store_event_batch(self, events: Sequence["EventLogEntry"]) -> None:
     def delete_events(self, run_id: str) -> None:
         """Remove events for a given run id."""
 
+    def purge_events(self, before_timestamp: float) -> int:
+        """Delete non-asset event rows older than the given UTC timestamp.
+
+        Asset materializations, observations, planned events, checks, freshness, health and wipe
+        events are always preserved. Default is a no-op; SQL backends override. Returns the number
+        of rows deleted (0 when unimplemented).
+        """
+        return 0
+
     @abstractmethod
     def upgrade(self) -> None:
         """This method should perform any schema migrations necessary to bring an

python_modules/dagster/dagster/_core/storage/event_log/sql_event_log.py

@@ -44,6 +44,7 @@
     ASSET_CHECK_EVENTS,
     ASSET_EVENTS,
     EVENT_TYPE_TO_PIPELINE_RUN_STATUS,
+    PURGEABLE_EVENT_TYPES,
     DagsterEventType,
 )
 from dagster._core.events.log import EventLogEntry
@@ -708,6 +709,35 @@ def delete_events_for_run(self, conn: Connection, run_id: str) -> None:
                 )
             )
 
+    def purge_events(self, before_timestamp: float) -> int:
+        # operates on the index shard; per-run sqlite backends only mirror run status events here,
+        # shared backends (postgres/mysql/consolidated sqlite) carry the full table
+        cutoff = datetime.fromtimestamp(before_timestamp, tz=timezone.utc)
+        purgeable_values = {e.value for e in PURGEABLE_EVENT_TYPES}
+        chunk_size = 1000
+        total_deleted = 0
+        while True:
+            with self.index_connection() as conn:
+                rows = conn.execute(
+                    db_select([SqlEventLogStorageTable.c.id])
+                    .where(
+                        db.and_(
+                            SqlEventLogStorageTable.c.timestamp < cutoff,
+                            SqlEventLogStorageTable.c.dagster_event_type.in_(purgeable_values),
+                        )
+                    )
+                    .limit(chunk_size)
+                ).fetchall()
+                if not rows:
+                    return total_deleted
+                ids = [row[0] for row in rows]
+                conn.execute(
+                    SqlEventLogStorageTable.delete().where(SqlEventLogStorageTable.c.id.in_(ids))
+                )
+                total_deleted += len(ids)
+                if len(ids) < chunk_size:
+                    return total_deleted
+
     @property
     def is_persistent(self) -> bool:
         return True

python_modules/dagster/dagster/_daemon/controller.py

@@ -24,6 +24,7 @@
     SchedulerDaemon,
     SensorDaemon,
 )
+from dagster._daemon.event_log_retention import EventLogRetentionDaemon
 from dagster._daemon.freshness import FreshnessDaemon
 from dagster._daemon.run_coordinator.queued_run_coordinator_daemon import QueuedRunCoordinatorDaemon
 from dagster._daemon.types import DaemonHeartbeat, DaemonStatus
@@ -380,6 +381,8 @@ def create_daemon_of_type(daemon_type: str, instance: DagsterInstance) -> Dagste
         )
     elif daemon_type == FreshnessDaemon.daemon_type():
         return FreshnessDaemon()
+    elif daemon_type == EventLogRetentionDaemon.daemon_type():
+        return EventLogRetentionDaemon()
     else:
         raise Exception(f"Unexpected daemon type {daemon_type}")
 

python_modules/dagster/dagster/_daemon/event_log_retention.py

@@ -0,0 +1,29 @@
+import datetime
+import os
+
+from dagster._core.workspace.context import IWorkspaceProcessContext
+from dagster._daemon.daemon import IntervalDaemon
+from dagster._time import get_current_datetime
+
+_INTERVAL_SECONDS = int(os.environ.get("EVENT_LOG_RETENTION_DAEMON_INTERVAL_SECONDS", "3600"))
+
+
+class EventLogRetentionDaemon(IntervalDaemon):
+    def __init__(self, interval_seconds: int = _INTERVAL_SECONDS):
+        super().__init__(interval_seconds=interval_seconds)
+
+    @classmethod
+    def daemon_type(cls):
+        return "EVENT_LOG_RETENTION"
+
+    def run_iteration(self, workspace_process_context: IWorkspaceProcessContext):
+        yield
+        instance = workspace_process_context.instance
+        days = instance.event_log_retention_days
+        if days is None:
+            return
+        cutoff = (get_current_datetime() - datetime.timedelta(days=days)).timestamp()
+        deleted = instance.event_log_storage.purge_events(cutoff)
+        self._logger.info(
+            f"Purged {deleted} non-asset event log row(s) older than {days} day(s).",
+        )

python_modules/dagster/dagster_tests/daemon_tests/test_event_log_retention_daemon.py

@@ -0,0 +1,78 @@
+import os
+import tempfile
+import time
+from types import SimpleNamespace
+
+import dagster as dg
+from dagster._core.events import DagsterEventType, EngineEventData, StepMaterializationData
+from dagster._core.execution.plan.objects import StepSuccessData
+from dagster._core.utils import make_new_run_id
+from dagster._daemon.event_log_retention import EventLogRetentionDaemon
+
+
+def _entry(run_id, timestamp, event_type, event_specific_data=None):
+    return dg.EventLogEntry(
+        error_info=None,
+        user_message="",
+        level="debug",
+        run_id=run_id,
+        timestamp=timestamp,
+        dagster_event=dg.DagsterEvent(
+            event_type_value=event_type.value,
+            job_name="nonce",
+            event_specific_data=event_specific_data,
+        ),
+    )
+
+
+def test_event_log_retention_daemon_iteration_purges_per_config():
+    """One full daemon iteration with retention configured deletes old non-asset rows only."""
+    run_id = make_new_run_id()
+    old_ts = time.time() - 60 * 60 * 24 * 30  # 30 days ago
+    recent_ts = time.time()
+
+    with (
+        tempfile.TemporaryDirectory(dir=os.getcwd()) as tmpdir,
+        dg.instance_for_test(
+            temp_dir=tmpdir,
+            overrides={
+                "retention": {"event_logs": {"purge_after_days": 7}},
+                "event_log_storage": {
+                    "module": "dagster._core.storage.event_log",
+                    "class": "ConsolidatedSqliteEventLogStorage",
+                    "config": {"base_dir": tmpdir},
+                },
+            },
+        ) as instance,
+    ):
+        instance.store_event(
+            _entry(
+                run_id,
+                old_ts,
+                DagsterEventType.ASSET_MATERIALIZATION,
+                StepMaterializationData(dg.AssetMaterialization(asset_key=dg.AssetKey("a"))),
+            )
+        )
+        instance.store_event(
+            _entry(
+                run_id,
+                old_ts,
+                DagsterEventType.ENGINE_EVENT,
+                EngineEventData.in_process(1),
+            )
+        )
+        instance.store_event(
+            _entry(
+                run_id,
+                recent_ts,
+                DagsterEventType.STEP_SUCCESS,
+                StepSuccessData(duration_ms=1.0),
+            )
+        )
+
+        list(EventLogRetentionDaemon().run_iteration(SimpleNamespace(instance=instance)))
+
+        remaining = {entry.dagster_event.event_type_value for entry in instance.all_logs(run_id)}
+        assert DagsterEventType.ASSET_MATERIALIZATION.value in remaining
+        assert DagsterEventType.STEP_SUCCESS.value in remaining
+        assert DagsterEventType.ENGINE_EVENT.value not in remaining