public_git
/
orchard
mirror of https://github.com/cirruslabs/orchard.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Address etcd event pagination review

This commit is contained in:
Jonathan Haas 2026-06-19 22:28:44 -07:00
parent b8dfa03ba9
commit 4945127f76
4 changed files with 118 additions and 72 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

119
internal/controller/store/etcd/etcd_events.go
View File

@ -5,7 +5,6 @@ import (
"encoding/json"
"fmt"
"path"
"sort"
"time"
storepkg "github.com/cirruslabs/orchard/internal/controller/store"
@ -63,79 +62,33 @@ func (txn *Transaction) ListEventsPage(options storepkg.ListOptions, scope ...st
logicalPrefix := scopePrefix(scope)
physicalPrefix := txn.store.keyPrefix(logicalPrefix)
response, err := txn.store.client.Get(txn.ctx, physicalPrefix, clientv3.WithPrefix(), clientv3.WithSort(
clientv3.SortByKey, clientv3.SortAscend,
))
getKey, getOptions := txn.listEventsPageQueryOptions(physicalPrefix, logicalPrefix, options)
response, err := txn.store.client.Get(txn.ctx, getKey, getOptions...)
if err != nil {
return result, mapErr(err)
}
txn.prefixReadRevisions[physicalPrefix] = response.Header.Revision
type keyedEvent struct {
key string
event v1.Event
}
keyedEventsByKey := map[string]v1.Event{}
limit := options.Limit
for _, kv := range response.Kvs {
key := string(kv.Key)
if txn.isDeleted(key) {
continue
}
var event v1.Event
if err := json.Unmarshal(kv.Value, &event); err != nil {
return result, err
}
keyedEventsByKey[string(kv.Key)] = event
}
for key, value := range txn.puts {
if !hasPrefix(key, physicalPrefix) {
continue
}
var event v1.Event
if err := json.Unmarshal([]byte(value), &event); err != nil {
return result, err
}
keyedEventsByKey[key] = event
}
keyedEvents := make([]keyedEvent, 0, len(keyedEventsByKey))
for key, event := range keyedEventsByKey {
if _, deleted := txn.deletes[key]; deleted {
continue
}
keyedEvents = append(keyedEvents, keyedEvent{key: key, event: event})
}
sort.Slice(keyedEvents, func(i, j int) bool {
if options.Order == storepkg.ListOrderDesc {
return keyedEvents[i].key > keyedEvents[j].key
}
return keyedEvents[i].key < keyedEvents[j].key
})
startIndex := 0
if len(options.Cursor) > 0 {
cursor := eventCursor(physicalPrefix, logicalPrefix, options.Cursor)
for index, keyedEvent := range keyedEvents {
if keyedEvent.key == cursor {
startIndex = index + 1
break
}
}
}
for index := startIndex; index < len(keyedEvents); index++ {
result.Items = append(result.Items, keyedEvents[index].event)
if options.Limit > 0 && len(result.Items) >= options.Limit {
if index+1 < len(keyedEvents) {
result.NextCursor = bytes.TrimPrefix([]byte(keyedEvents[index].key), []byte(physicalPrefix))
}
if limit > 0 && len(result.Items) >= limit {
break
}
result.Items = append(result.Items, event)
if limit > 0 && len(result.Items) == limit && len(response.Kvs) > limit {
result.NextCursor = bytes.TrimPrefix([]byte(key), []byte(physicalPrefix))
}
}
return result, nil
@ -143,20 +96,54 @@ func (txn *Transaction) ListEventsPage(options storepkg.ListOptions, scope ...st
func (txn *Transaction) DeleteEvents(scope ...string) error {
physicalPrefix := txn.store.keyPrefix(scopePrefix(scope))
response, err := txn.store.client.Get(txn.ctx, physicalPrefix, clientv3.WithPrefix(), clientv3.WithKeysOnly())
response, err := txn.store.client.Get(txn.ctx, physicalPrefix, clientv3.WithPrefix(), clientv3.WithKeysOnly(),
clientv3.WithLimit(1))
if err != nil {
return mapErr(err)
}
txn.prefixReadRevisions[physicalPrefix] = response.Header.Revision
for _, kv := range response.Kvs {
txn.deletes[string(kv.Key)] = struct{}{}
delete(txn.puts, string(kv.Key))
txn.prefixDeletes[physicalPrefix] = struct{}{}
for key := range txn.puts {
if hasPrefix(key, physicalPrefix) {
delete(txn.puts, key)
}
}
return nil
}
func (txn *Transaction) listEventsPageQueryOptions(
physicalPrefix string,
logicalPrefix string,
options storepkg.ListOptions,
) (string, []clientv3.OpOption) {
rangeEnd := clientv3.GetPrefixRangeEnd(physicalPrefix)
getKey := physicalPrefix
getRangeEnd := rangeEnd
if len(options.Cursor) > 0 {
cursor := eventCursor(physicalPrefix, logicalPrefix, options.Cursor)
if options.Order == storepkg.ListOrderDesc {
getRangeEnd = cursor
} else {
getKey = cursor + "\x00"
}
}
getOptions := []clientv3.OpOption{
clientv3.WithRange(getRangeEnd),
clientv3.WithSort(clientv3.SortByKey, clientv3.SortAscend),
}
if options.Order == storepkg.ListOrderDesc {
getOptions[1] = clientv3.WithSort(clientv3.SortByKey, clientv3.SortDescend)
}
if options.Limit > 0 {
getOptions = append(getOptions, clientv3.WithLimit(int64(options.Limit+1)))
}
return getKey, getOptions
}
func eventCursor(physicalPrefix, logicalPrefix string, cursor []byte) string {
if bytes.HasPrefix(cursor, []byte(physicalPrefix)) {
return string(cursor)

8
internal/controller/store/etcd/etcd_generic.go
View File

@ -23,9 +23,6 @@ func genericGet[T any, PT interface {
SetVersion(uint64)
*T
}](txn *Transaction, key string) (*T, error) {
if _, deleted := txn.deletes[key]; deleted {
return nil, storepkg.ErrNotFound
}
if value, ok := txn.puts[key]; ok {
var obj T
if err := json.Unmarshal([]byte(value), &obj); err != nil {
@ -34,6 +31,9 @@ func genericGet[T any, PT interface {
return &obj, nil
}
if txn.isDeleted(key) {
return nil, storepkg.ErrNotFound
}
response, err := txn.store.client.Get(txn.ctx, key)
if err != nil {
@ -101,7 +101,7 @@ func genericList[T any, PT interface {
result := []T{}
for _, key := range keys {
if _, deleted := txn.deletes[key]; deleted {
if txn.isDeleted(key) {
continue
}

23
internal/controller/store/etcd/etcd_store.go
View File

@ -28,6 +28,7 @@ type Transaction struct {
prefixReadRevisions map[string]int64
puts map[string]string
deletes map[string]struct{}
prefixDeletes map[string]struct{}
}
func NewEtcdStore(endpoints []string, keyPrefix string, logger *zap.SugaredLogger) (storepkg.Store, error) {
@ -77,6 +78,7 @@ func (store *Store) newTransaction(ctx context.Context) *Transaction {
prefixReadRevisions: map[string]int64{},
puts: map[string]string{},
deletes: map[string]struct{}{},
prefixDeletes: map[string]struct{}{},
}
}
@ -92,7 +94,7 @@ func (store *Store) keyPrefix(logicalPrefix string) string {
}
func (txn *Transaction) commit() error {
if len(txn.puts) == 0 && len(txn.deletes) == 0 {
if len(txn.puts) == 0 && len(txn.deletes) == 0 && len(txn.prefixDeletes) == 0 {
return nil
}
@ -108,7 +110,10 @@ func (txn *Transaction) commit() error {
comparisons = append(comparisons, clientv3.Compare(clientv3.ModRevision(prefix).WithPrefix(), "<", revision+1))
}
operations := make([]clientv3.Op, 0, len(txn.puts)+len(txn.deletes))
operations := make([]clientv3.Op, 0, len(txn.puts)+len(txn.deletes)+len(txn.prefixDeletes))
for prefix := range txn.prefixDeletes {
operations = append(operations, clientv3.OpDelete(prefix, clientv3.WithPrefix()))
}
for key, value := range txn.puts {
if _, deleted := txn.deletes[key]; deleted {
continue
@ -131,6 +136,20 @@ func (txn *Transaction) commit() error {
return nil
}
func (txn *Transaction) isDeleted(key string) bool {
if _, deleted := txn.deletes[key]; deleted {
return true
}
for prefix := range txn.prefixDeletes {
if hasPrefix(key, prefix) {
return true
}
}
return false
}
func normalizePrefix(prefix string) string {
prefix = strings.TrimSpace(prefix)
if prefix == "" || prefix == "/" {

40
internal/controller/store/events_test.go
View File

@ -81,3 +81,43 @@ func testListEventsPage(t *testing.T, store storepkg.Store) {
require.Equal(t, []v1.Event{events[1], events[0]}, descPage2.Items)
require.Empty(t, descPage2.NextCursor)
}
func TestDeleteManyEvents(t *testing.T) {
logger := zap.NewNop().Sugar()
for _, storeImpl := range testStores(logger) {
t.Run(storeImpl.Name, func(t *testing.T) {
store := storeImpl.Init(t)
testDeleteManyEvents(t, store)
})
}
}
func testDeleteManyEvents(t *testing.T, store storepkg.Store) {
events := make([]v1.Event, 200)
for i := range events {
events[i] = v1.Event{Kind: v1.EventKindLogLine, Timestamp: int64(i), Payload: "log line"}
}
for _, eventBatch := range [][]v1.Event{events[:100], events[100:]} {
err := store.Update(func(txn storepkg.Transaction) error {
return txn.AppendEvents(eventBatch, "vms", "vm-with-many-events")
})
require.NoError(t, err)
}
err := store.Update(func(txn storepkg.Transaction) error {
return txn.DeleteEvents("vms", "vm-with-many-events")
})
require.NoError(t, err)
var remaining []v1.Event
err = store.View(func(txn storepkg.Transaction) error {
var err error
remaining, err = txn.ListEvents("vms", "vm-with-many-events")
return err
})
require.NoError(t, err)
require.Empty(t, remaining)
}