diff --git a/pkg/storage/unified/resource/eventstore.go b/pkg/storage/unified/resource/eventstore.go index f2a3bc4e028..651fcb52092 100644 --- a/pkg/storage/unified/resource/eventstore.go +++ b/pkg/storage/unified/resource/eventstore.go @@ -7,6 +7,9 @@ import ( "iter" "strconv" "strings" + "time" + + "github.com/bwmarrin/snowflake" ) const ( @@ -224,3 +227,30 @@ func (n *eventStore) ListSince(ctx context.Context, sinceRV int64) iter.Seq2[Eve } } } + +// CleanupOldEvents deletes events older than the specified retention period. +func (n *eventStore) CleanupOldEvents(ctx context.Context, cutoff time.Time) (int, error) { + deletedCount := 0 + + // Keys are stored in the format of "resource_version~namespace~group~resource~name" + // With a start key of "1" and an end key of the cutoff time we can get all expired events. + endKey := fmt.Sprintf("%d", snowflakeFromTime(cutoff)) + for key, err := range n.kv.Keys(ctx, eventsSection, ListOptions{StartKey: "1", EndKey: endKey}) { + if err != nil { + return deletedCount, fmt.Errorf("failed to list event keys: %w", err) + } + + // TODO should use batch deletes here when available + if err := n.kv.Delete(ctx, eventsSection, key); err != nil { + return deletedCount, fmt.Errorf("failed to delete event key %s: %w", key, err) + } + deletedCount++ + } + + return deletedCount, nil +} + +// snowflake id with last two sections set to 0 (machine id and sequence) +func snowflakeFromTime(t time.Time) int64 { + return (t.UnixMilli() - snowflake.Epoch) << (snowflake.NodeBits + snowflake.StepBits) +} diff --git a/pkg/storage/unified/resource/eventstore_test.go b/pkg/storage/unified/resource/eventstore_test.go index 7782e879094..63ddc999567 100644 --- a/pkg/storage/unified/resource/eventstore_test.go +++ b/pkg/storage/unified/resource/eventstore_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -468,3 +469,135 @@ func TestEventStore_Save_InvalidJSON(t *testing.T) { err := store.Save(ctx, event) assert.NoError(t, err) } + +func TestEventStore_CleanupOldEvents(t *testing.T) { + ctx := context.Background() + store := setupTestEventStore(t) + + now := time.Now() + oldRV := snowflakeFromTime(now.Add(-48 * time.Hour)) // 48 hours ago + recentRV := snowflakeFromTime(now.Add(-1 * time.Hour)) // 1 hour ago + + oldEvent := Event{ + Namespace: "default", + Group: "apps", + Resource: "resource", + Name: "old-resource", + ResourceVersion: oldRV, + Action: DataActionCreated, + Folder: "test-folder", + PreviousRV: 999, + } + + recentEvent := Event{ + Namespace: "default", + Group: "apps", + Resource: "resource", + Name: "recent-resource", + ResourceVersion: recentRV, + Action: DataActionCreated, + Folder: "test-folder", + PreviousRV: 999, + } + + // Save both events + err := store.Save(ctx, oldEvent) + require.NoError(t, err) + err = store.Save(ctx, recentEvent) + require.NoError(t, err) + + // Verify both events exist + _, err = store.Get(ctx, EventKey{ + Namespace: oldEvent.Namespace, + Group: oldEvent.Group, + Resource: oldEvent.Resource, + Name: oldEvent.Name, + ResourceVersion: oldEvent.ResourceVersion, + Action: oldEvent.Action, + }) + require.NoError(t, err) + + _, err = store.Get(ctx, EventKey{ + Namespace: recentEvent.Namespace, + Group: recentEvent.Group, + Resource: recentEvent.Resource, + Name: recentEvent.Name, + ResourceVersion: recentEvent.ResourceVersion, + Action: recentEvent.Action, + }) + require.NoError(t, err) + + // Clean up events older than 24 hours + deletedCount, err := store.CleanupOldEvents(ctx, time.Now().Add(-24*time.Hour)) + require.NoError(t, err) + assert.Equal(t, 1, deletedCount, "Should have deleted 1 old event") + + // Verify old event was deleted + _, err = store.Get(ctx, EventKey{ + Namespace: oldEvent.Namespace, + Group: oldEvent.Group, + Resource: oldEvent.Resource, + Name: oldEvent.Name, + ResourceVersion: oldEvent.ResourceVersion, + Action: oldEvent.Action, + }) + assert.Error(t, err, "Old event should have been deleted") + + // Verify recent event still exists + _, err = store.Get(ctx, EventKey{ + Namespace: recentEvent.Namespace, + Group: recentEvent.Group, + Resource: recentEvent.Resource, + Name: recentEvent.Name, + ResourceVersion: recentEvent.ResourceVersion, + Action: recentEvent.Action, + }) + require.NoError(t, err, "Recent event should still exist") +} + +func TestEventStore_CleanupOldEvents_NoOldEvents(t *testing.T) { + ctx := context.Background() + store := setupTestEventStore(t) + + // Create an event 1 hour old + rv := snowflakeFromTime(time.Now().Add(-1 * time.Hour)) + event := Event{ + Namespace: "default", + Group: "apps", + Resource: "resource", + Name: "recent-resource", + ResourceVersion: rv, + Action: DataActionCreated, + Folder: "test-folder", + PreviousRV: 999, + } + + err := store.Save(ctx, event) + require.NoError(t, err) + + // Clean up events older than 24 hours + deletedCount, err := store.CleanupOldEvents(ctx, time.Now().Add(-24*time.Hour)) + require.NoError(t, err) + assert.Equal(t, 0, deletedCount, "Should not have deleted any events") + + // Verify event still exists + _, err = store.Get(ctx, EventKey{ + Namespace: event.Namespace, + Group: event.Group, + Resource: event.Resource, + Name: event.Name, + ResourceVersion: event.ResourceVersion, + Action: event.Action, + }) + require.NoError(t, err, "Recent event should still exist") +} + +func TestEventStore_CleanupOldEvents_EmptyStore(t *testing.T) { + ctx := context.Background() + store := setupTestEventStore(t) + + // Clean up events from empty store + deletedCount, err := store.CleanupOldEvents(ctx, time.Now().Add(-24*time.Hour)) + require.NoError(t, err) + assert.Equal(t, 0, deletedCount, "Should not have deleted any events from empty store") +} diff --git a/pkg/storage/unified/resource/storage_backend.go b/pkg/storage/unified/resource/storage_backend.go index 4f5490f7c7d..54bd33ba67f 100644 --- a/pkg/storage/unified/resource/storage_backend.go +++ b/pkg/storage/unified/resource/storage_backend.go @@ -25,22 +25,26 @@ import ( ) const ( - defaultListBufferSize = 100 - prunerMaxEvents = 20 + defaultListBufferSize = 100 + prunerMaxEvents = 20 + defaultEventRetentionPeriod = 1 * time.Hour + defaultEventPruningInterval = 5 * time.Minute ) // kvStorageBackend Unified storage backend based on KV storage. type kvStorageBackend struct { - snowflake *snowflake.Node - kv KV - dataStore *dataStore - metaStore *metadataStore - eventStore *eventStore - notifier *notifier - builder DocumentBuilder - log logging.Logger - withPruner bool - historyPruner Pruner + snowflake *snowflake.Node + kv KV + dataStore *dataStore + metaStore *metadataStore + eventStore *eventStore + notifier *notifier + builder DocumentBuilder + log logging.Logger + withPruner bool + eventRetentionPeriod time.Duration + eventPruningInterval time.Duration + historyPruner Pruner //tracer trace.Tracer //reg prometheus.Registerer } @@ -48,10 +52,12 @@ type kvStorageBackend struct { var _ StorageBackend = &kvStorageBackend{} type KvBackendOptions struct { - KvStore KV - WithPruner bool - Tracer trace.Tracer // TODO add tracing - Reg prometheus.Registerer // TODO add metrics + KvStore KV + WithPruner bool + EventRetentionPeriod time.Duration // How long to keep events (default: 1 hour) + EventPruningInterval time.Duration // How often to run the event pruning (default: 5 minutes) + Tracer trace.Tracer // TODO add tracing + Reg prometheus.Registerer // TODO add metrics } func NewKvStorageBackend(opts KvBackendOptions) (StorageBackend, error) { @@ -63,23 +69,71 @@ func NewKvStorageBackend(opts KvBackendOptions) (StorageBackend, error) { return nil, fmt.Errorf("failed to create snowflake node: %w", err) } eventStore := newEventStore(kv) + + eventRetentionPeriod := opts.EventRetentionPeriod + if eventRetentionPeriod <= 0 { + eventRetentionPeriod = defaultEventRetentionPeriod + } + + eventPruningInterval := opts.EventPruningInterval + if eventPruningInterval <= 0 { + eventPruningInterval = defaultEventPruningInterval + } + backend := &kvStorageBackend{ - kv: kv, - dataStore: newDataStore(kv), - metaStore: newMetadataStore(kv), - eventStore: eventStore, - notifier: newNotifier(eventStore, notifierOptions{}), - snowflake: s, - builder: StandardDocumentBuilder(), // For now we use the standard document builder. - log: &logging.NoOpLogger{}, // Make this configurable + kv: kv, + dataStore: newDataStore(kv), + metaStore: newMetadataStore(kv), + eventStore: eventStore, + notifier: newNotifier(eventStore, notifierOptions{}), + snowflake: s, + builder: StandardDocumentBuilder(), // For now we use the standard document builder. + log: &logging.NoOpLogger{}, // Make this configurable + eventRetentionPeriod: eventRetentionPeriod, + eventPruningInterval: eventPruningInterval, } err = backend.initPruner(ctx) if err != nil { return nil, fmt.Errorf("failed to initialize pruner: %w", err) } + + // Start the event cleanup background job + go backend.runCleanupOldEvents(ctx) + return backend, nil } +// runCleanupOldEvents starts a background goroutine that periodically cleans up old events +func (k *kvStorageBackend) runCleanupOldEvents(ctx context.Context) { + // Run cleanup every hour + ticker := time.NewTicker(k.eventPruningInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + k.log.Debug("Event cleanup stopped due to context cancellation") + return + case <-ticker.C: + k.cleanupOldEvents(ctx) + } + } +} + +// cleanupOldEvents performs the actual cleanup of old events +func (k *kvStorageBackend) cleanupOldEvents(ctx context.Context) { + cutoff := time.Now().Add(-k.eventRetentionPeriod) + deletedCount, err := k.eventStore.CleanupOldEvents(ctx, cutoff) + if err != nil { + k.log.Error("Failed to cleanup old events", "error", err) + return + } + + if deletedCount == 0 { + k.log.Info("Cleaned up old events", "deleted_count", deletedCount, "retention_period", k.eventRetentionPeriod) + } +} + func (k *kvStorageBackend) pruneEvents(ctx context.Context, key PruningKey) error { if !key.Validate() { return fmt.Errorf("invalid pruning key, all fields must be set: %+v", key)