feat: opt-in event log retention

[bhark]

Summary & Motivation

The event_logs table currently grows without bounds. This is a problem at anything beyond trivial scale. Operators have to work around this with custom K8s jobs, in-Dagster cleanup runs, manual SQL and similar hacky tricks. All of this is described in database tuning docs.

This PR adds an opt-in retention.event_logs.purge_after_days conf knob, sitting cleanly next to existing retention policies. Events related to assets are always preserved - this only affects operational rows.

The work follows the existing FreshnessDaemon precedent.

This has been a long-standing ask from the community: closes #33785, partially addresses #4100, and is related to discussion #12985 and discussion #12047. Tangential to #4497.

Test Plan

• Storage test which verifies type + timestamp filtering and chunked deletion
• Daemon iteration test verifying config-driven purge e2e

Both pass locally against my sqlite backend.

Changelog

• retention.event_logs.purge_after_days is now configurable in dagster.yaml
• New EventLogRetentionDaemon which runs hourly by default

[greptile-apps]

Greptile Summary

This PR adds an opt-in retention.event_logs.purge_after_days configuration knob that periodically deletes non-asset operational event log rows, addressing long-standing unbounded table growth in production deployments.

• A new EventLogRetentionDaemon (hourly IntervalDaemon) reads the config, computes a cutoff, and drives chunked 1 000-row deletes via a new purge_events method on SqlEventLogStorage; asset-related events (ASSET_MATERIALIZATION, ASSET_OBSERVATION, FRESHNESS_STATE_CHANGE, checks, health, wipe, and failed materializations) are never deleted.
• An Alembic migration (049) adds idx_event_timestamp with CONCURRENTLY on Postgres, matched by a new db.Index definition in schema.py for fresh installs; the revision chain correctly targets the current tip (29b539ebc72a).
• Per-run SQLite sharding is explicitly documented and emits a one-time warning; the purge_events cutoff is computed as naive UTC to match stored timestamps.

Confidence Score: 5/5

Safe to merge; the feature is opt-in with no effect on deployments that do not configure purge_after_days.

All changed code is additive and gated behind an opt-in config key. The major correctness concerns from earlier review rounds (timezone mismatch, per-run shard limitation, missing ORDER BY, noisy logging) have been addressed. The only remaining items are a Python bool-subclass edge case in the validation guard and a minor documentation omission, neither of which affects deployments using correctly-typed integer values.

python_modules/dagster/dagster/_core/instance/config.py — the get_event_log_retention_days helper has a bool-as-int edge case worth a one-line guard.

Important Files Changed

Filename	Overview
python_modules/dagster/dagster/_daemon/event_log_retention.py	New IntervalDaemon that runs hourly, reads retention config, and calls purge_events; correctly handles per-run-sharded SQLite with a one-time warning.
python_modules/dagster/dagster/_core/storage/event_log/sql_event_log.py	Adds purge_events with chunked DELETE (1 000 rows), ORDER BY id for stable cursor, and naive-UTC cutoff matching stored timestamps.
python_modules/dagster/dagster/_core/storage/alembic/versions/049_070de2d90a83_add_event_timestamp_idx.py	Adds idx_event_timestamp on event_logs.timestamp using CONCURRENTLY for Postgres, consistent with existing migration conventions; revision chain correctly follows 29b539ebc72a.
python_modules/dagster/dagster/_core/events/init.py	Introduces RETAINED_EVENT_TYPES and PURGEABLE_EVENT_TYPES set-arithmetic constants; ASSET_FAILED_TO_MATERIALIZE is preserved but not mentioned in config description.
python_modules/dagster/dagster/_core/instance/config.py	Adds retention config schema and get_event_log_retention_days helper; bool-subclass-of-int edge case means purge_after_days: true silently enables 1-day retention.

Sequence Diagram

sequenceDiagram
    participant Cfg as dagster.yaml
    participant Inst as DagsterInstance
    participant Ctrl as DaemonController
    participant ELD as EventLogRetentionDaemon
    participant Store as SqlEventLogStorage
    participant DB as Database

    Cfg->>Inst: "retention.event_logs.purge_after_days = N"
    Inst->>Ctrl: event_log_retention_enabled → add to required daemons
    loop Every hour (default)
        Ctrl->>ELD: run_iteration(workspace_process_context)
        ELD->>Inst: event_log_retention_days → N
        ELD->>ELD: "cutoff = now() - timedelta(days=N)"
        ELD->>Store: purge_events(cutoff.timestamp())
        loop Chunks of 1000
            Store->>DB: "SELECT id WHERE timestamp < cutoff AND event_type IN (purgeable) ORDER BY id LIMIT 1000"
            DB-->>Store: ids[]
            Store->>DB: DELETE WHERE id IN (ids)
        end
        Store-->>ELD: total_deleted
        ELD->>ELD: log info (if deleted) / debug (if nothing to purge)
    end

Loading

_{Reviews (3): Last reviewed commit: "fix(storage): strip tzinfo on event log ..." | Re-trigger Greptile}