From 147ff4279b22e21ac8803ccb25f7f6f105dc2571 Mon Sep 17 00:00:00 2001 From: Georges Chaudy Date: Tue, 28 Oct 2025 17:25:55 +0100 Subject: [PATCH] kvstore: fix events lookback + startkey (#113092) * fix snowflakes events * add tests --- pkg/storage/unified/resource/eventstore.go | 19 ++- .../unified/resource/eventstore_test.go | 131 ++++++++++++++++++ pkg/storage/unified/resource/notifier.go | 4 +- .../unified/resource/storage_backend.go | 2 +- .../unified/resource/storage_backend_test.go | 23 +-- 5 files changed, 153 insertions(+), 26 deletions(-) diff --git a/pkg/storage/unified/resource/eventstore.go b/pkg/storage/unified/resource/eventstore.go index 828e3cececf..53e15253b15 100644 --- a/pkg/storage/unified/resource/eventstore.go +++ b/pkg/storage/unified/resource/eventstore.go @@ -183,10 +183,8 @@ func (n *eventStore) Get(ctx context.Context, key EventKey) (Event, error) { // ListSince returns a sequence of events since the given resource version. func (n *eventStore) ListKeysSince(ctx context.Context, sinceRV int64) iter.Seq2[string, error] { opts := ListOptions{ - Sort: SortOrderAsc, - StartKey: EventKey{ - ResourceVersion: sinceRV, - }.String(), + Sort: SortOrderAsc, + StartKey: fmt.Sprintf("%d", sinceRV), } return func(yield func(string, error) bool) { for evtKey, err := range n.kv.Keys(ctx, eventsSection, opts) { @@ -275,3 +273,16 @@ func (n *eventStore) batchDelete(ctx context.Context, keys []string) error { func snowflakeFromTime(t time.Time) int64 { return (t.UnixMilli() - snowflake.Epoch) << (snowflake.NodeBits + snowflake.StepBits) } + +// subtractDurationFromSnowflake subtracts a duration from a snowflake ID by +// converting it to time, subtracting the duration, and converting back to a snowflake ID +func subtractDurationFromSnowflake(snowflakeID int64, duration time.Duration) int64 { + // Extract timestamp from snowflake (returns milliseconds since epoch) + timestamp := snowflake.ID(snowflakeID).Time() + // Convert to time.Time + t := time.Unix(0, timestamp*int64(time.Millisecond)) + // Subtract duration + newTime := t.Add(-duration) + // Convert back to snowflake + return snowflakeFromTime(newTime) +} diff --git a/pkg/storage/unified/resource/eventstore_test.go b/pkg/storage/unified/resource/eventstore_test.go index 5744d81c985..a9d2ee93eb4 100644 --- a/pkg/storage/unified/resource/eventstore_test.go +++ b/pkg/storage/unified/resource/eventstore_test.go @@ -6,6 +6,7 @@ import ( "testing" "time" + "github.com/bwmarrin/snowflake" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -660,3 +661,133 @@ func TestEventStore_BatchDelete(t *testing.T) { require.Error(t, err, "Event should have been deleted") } } + +func TestSubtractDurationFromSnowflake(t *testing.T) { + baseTime := time.Date(2025, 1, 1, 12, 0, 0, 0, time.UTC) + + tests := []struct { + name string + addTime time.Duration + }{ + { + name: "subtract 1 hour", + addTime: -1 * time.Hour, + }, + { + name: "subtract 2 hours", + addTime: -2 * time.Hour, + }, + { + name: "subtract 24 hours", + addTime: -24 * time.Hour, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + // Generate a snowflake from the base time + baseSnowflake := snowflakeFromTime(baseTime) + + // Subtract the duration + resultSnowflake := subtractDurationFromSnowflake(baseSnowflake, tt.addTime) + + // Convert back to timestamp and verify + // Extract timestamp from the result snowflake + timestamp := snowflake.ID(resultSnowflake).Time() + resultTime := time.Unix(0, timestamp*int64(time.Millisecond)) + + // Compare with expected time (allowing for small differences due to snowflake precision) + expectedMillis := baseTime.Add(-tt.addTime).UnixMilli() + resultMillis := resultTime.UnixMilli() + assert.InDelta(t, expectedMillis, resultMillis, 1, + "Expected time %v, got %v (diff: %d ms)", + baseTime, resultTime, expectedMillis-resultMillis) + }) + } +} + +func TestSnowflakeFromTime(t *testing.T) { + testTime := time.Date(2024, 1, 1, 12, 0, 0, 0, time.UTC) + snowflakeID := snowflakeFromTime(testTime) + + // Extract timestamp and verify it matches + timestamp := snowflake.ID(snowflakeID).Time() + reconstructedTime := time.Unix(0, timestamp*int64(time.Millisecond)) + + // The times should match at millisecond precision + expectedMillis := testTime.UnixMilli() + resultMillis := reconstructedTime.UnixMilli() + + assert.Equal(t, expectedMillis, resultMillis, "Snowflake timestamp should match original time at millisecond precision") +} + +func TestListKeysSince_WithSnowflakeTime(t *testing.T) { + ctx := context.Background() + store := setupTestEventStore(t) + + // Create events with snowflake-based resource versions at different times + now := time.Now() + events := []Event{ + { + Namespace: "default", + Group: "apps", + Resource: "resource", + Name: "test-1", + ResourceVersion: snowflakeFromTime(now.Add(-2 * time.Hour)), + Action: DataActionCreated, + }, + { + Namespace: "default", + Group: "apps", + Resource: "resource", + Name: "test-2", + ResourceVersion: snowflakeFromTime(now.Add(-1 * time.Hour)), + Action: DataActionUpdated, + }, + { + Namespace: "default", + Group: "apps", + Resource: "resource", + Name: "test-3", + ResourceVersion: snowflakeFromTime(now.Add(-30 * time.Minute)), + Action: DataActionDeleted, + }, + } + + // Save all events + for _, event := range events { + err := store.Save(ctx, event) + require.NoError(t, err) + } + + // List events since 90 minutes ago using subtractDurationFromSnowflake + sinceRV := subtractDurationFromSnowflake(snowflakeFromTime(now), 90*time.Minute) + retrievedEvents := make([]string, 0) + for eventKey, err := range store.ListKeysSince(ctx, sinceRV) { + require.NoError(t, err) + retrievedEvents = append(retrievedEvents, eventKey) + } + + // Should return events from the last hour and 30 minutes + require.Len(t, retrievedEvents, 2) + evt1, err := ParseEventKey(retrievedEvents[0]) + require.NoError(t, err) + assert.Equal(t, "test-2", evt1.Name) + evt2, err := ParseEventKey(retrievedEvents[1]) + require.NoError(t, err) + assert.Equal(t, "test-3", evt2.Name) + + // List events since 30 minutes ago using subtractDurationFromSnowflake + sinceRV = subtractDurationFromSnowflake(snowflakeFromTime(now), 30*time.Minute) + retrievedEvents = make([]string, 0) + for eventKey, err := range store.ListKeysSince(ctx, sinceRV) { + require.NoError(t, err) + retrievedEvents = append(retrievedEvents, eventKey) + } + + // Should return events from the last hour and 30 minutes + require.Len(t, retrievedEvents, 1) + evt, err := ParseEventKey(retrievedEvents[0]) + require.NoError(t, err) + assert.Equal(t, "test-3", evt.Name) +} diff --git a/pkg/storage/unified/resource/notifier.go b/pkg/storage/unified/resource/notifier.go index d0b27e4ad11..f55db60623a 100644 --- a/pkg/storage/unified/resource/notifier.go +++ b/pkg/storage/unified/resource/notifier.go @@ -73,7 +73,7 @@ func (n *notifier) Watch(ctx context.Context, opts watchOptions) <-chan Event { initialRV, err := n.lastEventResourceVersion(ctx) if errors.Is(err, ErrNotFound) { - initialRV = 0 // No events yet, start from the beginning + initialRV = snowflakeFromTime(time.Now()) // No events yet, start from the beginning } else if err != nil { n.log.Error("Failed to get last event resource version", "error", err) } @@ -86,7 +86,7 @@ func (n *notifier) Watch(ctx context.Context, opts watchOptions) <-chan Event { case <-ctx.Done(): return case <-time.After(opts.PollInterval): - for evt, err := range n.eventStore.ListSince(ctx, lastRV-opts.LookbackPeriod.Nanoseconds()) { + for evt, err := range n.eventStore.ListSince(ctx, subtractDurationFromSnowflake(lastRV, opts.LookbackPeriod)) { if err != nil { n.log.Error("Failed to list events since", "error", err) continue diff --git a/pkg/storage/unified/resource/storage_backend.go b/pkg/storage/unified/resource/storage_backend.go index f7bca33766b..4de28e758d5 100644 --- a/pkg/storage/unified/resource/storage_backend.go +++ b/pkg/storage/unified/resource/storage_backend.go @@ -746,7 +746,7 @@ func (k *kvStorageBackend) listModifiedSinceEventStore(ctx context.Context, key return func(yield func(*ModifiedResource, error) bool) { // store all events ordered by RV for the given tenant here eventKeys := make([]EventKey, 0) - for evtKeyStr, err := range k.eventStore.ListKeysSince(ctx, sinceRv-defaultLookbackPeriod.Nanoseconds()) { + for evtKeyStr, err := range k.eventStore.ListKeysSince(ctx, subtractDurationFromSnowflake(sinceRv, defaultLookbackPeriod)) { if err != nil { yield(&ModifiedResource{}, err) return diff --git a/pkg/storage/unified/resource/storage_backend_test.go b/pkg/storage/unified/resource/storage_backend_test.go index 4ef60d1a4c4..05920a603c4 100644 --- a/pkg/storage/unified/resource/storage_backend_test.go +++ b/pkg/storage/unified/resource/storage_backend_test.go @@ -10,7 +10,6 @@ import ( "testing" "time" - "github.com/bwmarrin/snowflake" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" @@ -764,24 +763,10 @@ func randomStringGenerator() func() string { // creates 2 hour old snowflake for testing func generateOldSnowflake(t *testing.T) int64 { - // Generate a current snowflake first - node, err := snowflake.NewNode(1) - require.NoError(t, err) - currentSnowflake := node.Generate().Int64() - - // Extract its timestamp component by shifting right - currentTimestamp := currentSnowflake >> 22 - - // Subtract 2 hours (in milliseconds) from the timestamp - twoHoursMs := int64(2 * time.Hour / time.Millisecond) - oldTimestamp := currentTimestamp - twoHoursMs - - // Reconstruct snowflake: [timestamp:41][node:10][sequence:12] - // Keep the original node and sequence bits - nodeAndSequence := currentSnowflake & 0x3FFFFF // Bottom 22 bits (10 node + 12 sequence) - snowflakeID := (oldTimestamp << 22) | nodeAndSequence - - return snowflakeID + // Generate a snowflake for 2 hours ago using the snowflakeFromTime utility + // which properly handles the epoch + twoHoursAgo := time.Now().Add(-2 * time.Hour) + return snowflakeFromTime(twoHoursAgo) } // seedBackend seeds the kvstore with data and return the expected result for ListModifiedSince calls