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

kvstore: fix events lookback + startkey (#113092)

Browse Source 
* fix snowflakes events

* add tests
This commit is contained in:
Georges Chaudy
2025-10-28 17:25:55 +01:00
committed by GitHub
parent 0973a44e6a
commit 147ff4279b
5 changed files with 153 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/storage/unified/resource/eventstore.go
+15 -4
View File
@@ -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)
}
pkg/storage/unified/resource/eventstore_test.go
+131
View File
@@ -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)
}
pkg/storage/unified/resource/notifier.go
+2 -2
View File
@@ -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
pkg/storage/unified/resource/storage_backend.go
+1 -1
View File
@@ -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
pkg/storage/unified/resource/storage_backend_test.go
+4 -19
View File
@@ -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