diff --git a/docs/release-control/v6/internal/subsystems/alerts.md b/docs/release-control/v6/internal/subsystems/alerts.md index a23f7815de..4f53856f97 100644 --- a/docs/release-control/v6/internal/subsystems/alerts.md +++ b/docs/release-control/v6/internal/subsystems/alerts.md @@ -1420,7 +1420,12 @@ evidence. Failed asynchronous batches do not advance the successful-write counter, and `Flush` must report the failure or timeout rather than falsely claiming the batch landed. An unavailable or failed lifecycle store degrades history reads to the recovery model without hiding live active-alert truth. -Events retain for 90 days and prune hourly. Fired and +Events retain for 90 days and prune hourly. Retention alone does not bound disk +use: one flapping alert can write thousands of full alert-state snapshots inside +the window, so the hourly prune also enforces a total stored snapshot-payload +cap and removes the oldest events first when it is exceeded. A volume prune +raises the same retention revision as age pruning, so the history projection +rebuilds from the remaining log rather than reusing a stale fold. Fired and refired lifecycle events come only from the reducer core's explicit activation events: canonical lifecycle reactivation maps `EventRefired` separately, while shared metric activation records one fired event when its pending incident diff --git a/internal/alerts/eventlog/eventlog.go b/internal/alerts/eventlog/eventlog.go index 5caa8f5758..ce1d6a3934 100644 --- a/internal/alerts/eventlog/eventlog.go +++ b/internal/alerts/eventlog/eventlog.go @@ -103,6 +103,14 @@ const ( defaultRetention = 90 * 24 * time.Hour pruneInterval = time.Hour eventLogFileName = "events.db" + // maxRetainedSnapshotBytes bounds the total lifecycle-snapshot payload the + // log keeps. Retention bounds how long an event lives; a single flapping + // alert can still write hundreds of megabytes of full state snapshots + // inside that window and fill the host disk. When the stored payload + // exceeds this cap the oldest events are removed first, exactly as age + // pruning removes them, so the history projection rebuilds from what + // remains. + maxRetainedSnapshotBytes = 256 << 20 ) // Store is the SQLite-backed event log. A nil *Store is valid: Append and @@ -612,10 +620,91 @@ func (s *Store) forgetFailedDeliveryEpisodes(events []Event) { } func (s *Store) pruneOld() { - if s.retention <= 0 { + if s.retention > 0 { + s.pruneEventsBefore(time.Now().Add(-s.retention)) + } + s.pruneSnapshotVolume(maxRetainedSnapshotBytes) +} + +// pruneSnapshotVolume removes the oldest events until the total stored snapshot +// payload is within maxBytes. Age retention alone cannot bound disk use: one +// flapping alert can write thousands of full state snapshots inside the +// retention window. Deleting the oldest rows raises the retention revision so +// the history projection rebuilds from the remaining log. +func (s *Store) pruneSnapshotVolume(maxBytes int64) { + if s == nil || maxBytes <= 0 { return } - s.pruneEventsBefore(time.Now().Add(-s.retention)) + var total int64 + if err := s.db.QueryRow(`SELECT COALESCE(SUM(length(snapshot)), 0) FROM alert_events`).Scan(&total); err != nil { + log.Error().Err(err).Msg("alert event log volume check failed") + return + } + if total <= maxBytes { + return + } + // Read only ids and payload lengths, oldest first, until enough bytes are + // scheduled for removal. The cursor must be closed before deleting because + // the store serializes writes on a single connection. + rows, err := s.db.Query(`SELECT id, length(snapshot) FROM alert_events ORDER BY occurred_at ASC, id ASC`) + if err != nil { + log.Error().Err(err).Msg("alert event log volume scan failed") + return + } + ids := make([]int64, 0) + for total > maxBytes && rows.Next() { + var id, size int64 + if scanErr := rows.Scan(&id, &size); scanErr != nil { + rows.Close() + log.Error().Err(scanErr).Msg("alert event log volume scan failed") + return + } + ids = append(ids, id) + total -= size + } + scanErr := rows.Err() + rows.Close() + if scanErr != nil { + log.Error().Err(scanErr).Msg("alert event log volume scan failed") + return + } + if len(ids) == 0 { + return + } + s.deleteEventsByID(ids) +} + +func (s *Store) deleteEventsByID(ids []int64) { + tx, err := s.db.Begin() + if err != nil { + log.Error().Err(err).Msg("alert event log volume prune failed") + return + } + defer tx.Rollback() + for start := 0; start < len(ids); start += 500 { + end := start + 500 + if end > len(ids) { + end = len(ids) + } + placeholders := make([]string, 0, end-start) + args := make([]any, 0, end-start) + for _, id := range ids[start:end] { + placeholders = append(placeholders, "?") + args = append(args, id) + } + if _, err = tx.Exec("DELETE FROM alert_events WHERE id IN ("+strings.Join(placeholders, ",")+")", args...); err != nil { + log.Error().Err(err).Msg("alert event log volume prune failed") + return + } + } + if _, err = tx.Exec(`INSERT INTO alert_store_meta (key, value) VALUES ('retention_revision', '1') + ON CONFLICT(key) DO UPDATE SET value = CAST(value AS INTEGER) + 1`); err != nil { + log.Error().Err(err).Msg("alert event log volume prune failed") + return + } + if err = tx.Commit(); err != nil { + log.Error().Err(err).Msg("alert event log volume prune failed") + } } func (s *Store) pruneEventsBefore(cutoff time.Time) { diff --git a/internal/alerts/eventlog/eventlog_snapshot_test.go b/internal/alerts/eventlog/eventlog_snapshot_test.go index 69d93d21f9..0de9673542 100644 --- a/internal/alerts/eventlog/eventlog_snapshot_test.go +++ b/internal/alerts/eventlog/eventlog_snapshot_test.go @@ -1,8 +1,10 @@ package eventlog import ( + "bytes" "database/sql" "encoding/json" + "fmt" "path/filepath" "sync" "testing" @@ -312,3 +314,61 @@ func TestConcurrentUnchangedDeliveryDecisionAdmitsOneEpisode(t *testing.T) { t.Fatalf("concurrent reevaluations wrote %d episodes, want one", len(events)) } } + +func TestSnapshotVolumeBoundPrunesOldestAndAdvancesRetention(t *testing.T) { + store := newTestStore(t) + base := time.Now().Add(-time.Hour).UTC() + const eventCount = 8 + payload := bytes.Repeat([]byte("x"), 1024) + for i := 0; i < eventCount; i++ { + snapshot, err := json.Marshal(map[string]any{"id": fmt.Sprintf("a%d", i), "blob": string(payload)}) + if err != nil { + t.Fatal(err) + } + if err := store.AppendDurable(Event{ + OccurredAt: base.Add(time.Duration(i) * time.Minute), + Type: TypeFired, + AlertID: fmt.Sprintf("a%d", i), + Snapshot: snapshot, + }); err != nil { + t.Fatal(err) + } + } + before, err := store.ReplayBoundary() + if err != nil { + t.Fatal(err) + } + // Keep roughly three events' payload; the older rows must be removed. + store.pruneSnapshotVolume(3 * 1200) + after, err := store.ReplayBoundary() + if err != nil { + t.Fatal(err) + } + if after.RetentionRevision <= before.RetentionRevision { + t.Fatalf("volume prune did not advance retention revision: before=%+v after=%+v", before, after) + } + if after.LastID != before.LastID { + t.Fatalf("volume prune changed the newest id: before=%+v after=%+v", before, after) + } + remaining, err := store.Query(Filter{}) + if err != nil { + t.Fatal(err) + } + if len(remaining) == 0 || len(remaining) >= eventCount { + t.Fatalf("volume prune left %d of %d events", len(remaining), eventCount) + } + for _, event := range remaining { + if event.AlertID == "a0" { + t.Fatalf("oldest event survived volume prune: %+v", remaining) + } + } + // A second prune already within the cap must not change the boundary. + store.pruneSnapshotVolume(3 * 1200) + again, err := store.ReplayBoundary() + if err != nil { + t.Fatal(err) + } + if again != after { + t.Fatalf("prune within the cap changed the boundary: %+v -> %+v", after, again) + } +}