public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Unified Storage: Adds pruner for kv eventstore (#110785)

Browse Source 
* Adds pruner for eventstore - default 24 hours. Adds tests.

* update comment

* remove delay on startup. formatting

* updates log message type and removes useless comment

* caller handles goroutine for runCleanupOldEvents()

* simplify timestamp extraction

* adds config for event pruning interval

* uses start and end key to get all expired events

* remove sort when listing keys in event pruner - order doesnt matter

* use snowflake constants

* log when we delete 0 rows

* pass time.Time to cleanup old events func
This commit is contained in:
owensmallwood
2025-09-12 14:40:16 -06:00
committed by GitHub
parent c5ed2780ab
commit 7ce971cba1
3 changed files with 241 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/storage/unified/resource/eventstore.go
+30
View File
@@ -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)
}
pkg/storage/unified/resource/eventstore_test.go
+133
View File
@@ -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")
}
pkg/storage/unified/resource/storage_backend.go
+78 -24
View File
@@ -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)