From 04d39cbbc69ab27df565d53ac612cb72097e1cad Mon Sep 17 00:00:00 2001 From: Georges Chaudy Date: Fri, 23 May 2025 15:00:18 +0200 Subject: [PATCH] unistore: split ListIterator and ListHistory in StorageBackend (#105654) split listIterator from ListHistory --- .../apis/dashboard/legacy/sql_dashboards.go | 5 ---- pkg/registry/apis/dashboard/legacy/storage.go | 7 ++++- pkg/storage/unified/resource/cdk_backend.go | 13 +++------ pkg/storage/unified/resource/server.go | 28 +++++++++++-------- pkg/storage/unified/sql/backend.go | 15 ++++++---- pkg/storage/unified/sql/list_iterator.go | 16 +++++------ pkg/storage/unified/sql/list_iterator_test.go | 11 ++++---- 7 files changed, 50 insertions(+), 45 deletions(-) diff --git a/pkg/registry/apis/dashboard/legacy/sql_dashboards.go b/pkg/registry/apis/dashboard/legacy/sql_dashboards.go index 5238b7ffd83..555395fbee4 100644 --- a/pkg/registry/apis/dashboard/legacy/sql_dashboards.go +++ b/pkg/registry/apis/dashboard/legacy/sql_dashboards.go @@ -186,11 +186,6 @@ func (r *rowsWrapper) ContinueToken() string { return r.row.token.String() } -// ContinueTokenWithCurrentRV implements resource.ListIterator. -func (r *rowsWrapper) ContinueTokenWithCurrentRV() string { - return r.row.token.String() -} - // Error implements resource.ListIterator. func (r *rowsWrapper) Error() error { return r.err diff --git a/pkg/registry/apis/dashboard/legacy/storage.go b/pkg/registry/apis/dashboard/legacy/storage.go index 5c9e392a862..5c47a5f6cfc 100644 --- a/pkg/registry/apis/dashboard/legacy/storage.go +++ b/pkg/registry/apis/dashboard/legacy/storage.go @@ -229,7 +229,12 @@ func (a *dashboardSqlAccess) ReadResource(ctx context.Context, req *resourcepb.R return rsp } -// List implements AppendingStore. +// ListHistory implements StorageBackend. +func (a *dashboardSqlAccess) ListHistory(ctx context.Context, req *resourcepb.ListRequest, cb func(resource.ListIterator) error) (int64, error) { + return a.ListIterator(ctx, req, cb) +} + +// List implements StorageBackend. func (a *dashboardSqlAccess) ListIterator(ctx context.Context, req *resourcepb.ListRequest, cb func(resource.ListIterator) error) (int64, error) { if req.ResourceVersion != 0 { return 0, apierrors.NewBadRequest("List with explicit resourceVersion is not supported with this storage backend") diff --git a/pkg/storage/unified/resource/cdk_backend.go b/pkg/storage/unified/resource/cdk_backend.go index 725feb11585..c47ec01d378 100644 --- a/pkg/storage/unified/resource/cdk_backend.go +++ b/pkg/storage/unified/resource/cdk_backend.go @@ -235,10 +235,6 @@ func isDeletedValue(raw []byte) bool { } func (s *cdkBackend) ListIterator(ctx context.Context, req *resourcepb.ListRequest, cb func(ListIterator) error) (int64, error) { - if req.Source != resourcepb.ListRequest_STORE { - return 0, fmt.Errorf("listing from history not supported in CDK backend") - } - resources, err := buildTree(ctx, s, req.Options.Key) if err != nil { return 0, err @@ -247,6 +243,10 @@ func (s *cdkBackend) ListIterator(ctx context.Context, req *resourcepb.ListReque return resources.listRV, err } +func (s *cdkBackend) ListHistory(ctx context.Context, req *resourcepb.ListRequest, cb func(ListIterator) error) (int64, error) { + return 0, fmt.Errorf("listing from history not supported in CDK backend") +} + func (s *cdkBackend) WatchWriteEvents(ctx context.Context) (<-chan *WrittenEvent, error) { s.mutex.Lock() defer s.mutex.Unlock() @@ -336,11 +336,6 @@ func (c *cdkListIterator) ContinueToken() string { return fmt.Sprintf("index:%d/key:%s", c.index, c.currentKey) } -// ContinueTokenWithCurrentRV implements ListIterator. -func (c *cdkListIterator) ContinueTokenWithCurrentRV() string { - return fmt.Sprintf("index:%d/key:%s", c.index, c.currentKey) -} - // Name implements ListIterator. func (c *cdkListIterator) Name() string { return c.currentKey // TODO (parse name from key) diff --git a/pkg/storage/unified/resource/server.go b/pkg/storage/unified/resource/server.go index 42474e706a3..2aa15a547ca 100644 --- a/pkg/storage/unified/resource/server.go +++ b/pkg/storage/unified/resource/server.go @@ -46,9 +46,6 @@ type ListIterator interface { // ContinueToken returns the token that can be used to start iterating *after* this item ContinueToken() string - // ContinueTokenWithCurrentRV returns the token that can be used to start iterating *before* this item - ContinueTokenWithCurrentRV() string - // ResourceVersion of the current item ResourceVersion() int64 @@ -102,6 +99,9 @@ type StorageBackend interface { // but are the the final answer. ListIterator(context.Context, *resourcepb.ListRequest, func(ListIterator) error) (int64, error) + // ListHistory is like ListIterator, but it returns the history of a resource + ListHistory(context.Context, *resourcepb.ListRequest, func(ListIterator) error) (int64, error) + // Get all events from the store // For HA setups, this will be more events than the local WriteEvent above! WatchWriteEvents(ctx context.Context) (<-chan *WrittenEvent, error) @@ -808,7 +808,7 @@ func (s *server) List(ctx context.Context, req *resourcepb.ListRequest) (*resour }}, nil } - rv, err := s.backend.ListIterator(ctx, req, func(iter ListIterator) error { + iterFunc := func(iter ListIterator) error { for iter.Next() { if err := iter.Error(); err != nil { return err @@ -827,13 +827,6 @@ func (s *server) List(ctx context.Context, req *resourcepb.ListRequest) (*resour rsp.Items = append(rsp.Items, item) if len(rsp.Items) >= int(req.Limit) || pageBytes >= maxPageBytes { t := iter.ContinueToken() - if req.Source == resourcepb.ListRequest_HISTORY || req.Source == resourcepb.ListRequest_TRASH { - // For history lists, we need to use the current RV in the continue token - // to ensure consistent pagination. The order depends on VersionMatch: - // - NotOlderThan: ascending order (oldest to newest) - // - Unset: descending order (newest to oldest) - t = iter.ContinueTokenWithCurrentRV() - } if iter.Next() { rsp.NextPageToken = t } @@ -841,7 +834,18 @@ func (s *server) List(ctx context.Context, req *resourcepb.ListRequest) (*resour } } return iter.Error() - }) + } + + var rv int64 + switch req.Source { + case resourcepb.ListRequest_STORE: + rv, err = s.backend.ListIterator(ctx, req, iterFunc) + case resourcepb.ListRequest_HISTORY, resourcepb.ListRequest_TRASH: + rv, err = s.backend.ListHistory(ctx, req, iterFunc) + default: + return nil, apierrors.NewBadRequest(fmt.Sprintf("invalid list source: %v", req.Source)) + } + if err != nil { rsp.Error = AsErrorResult(err) return rsp, nil diff --git a/pkg/storage/unified/sql/backend.go b/pkg/storage/unified/sql/backend.go index 51d77335c4a..0c5388c71f7 100644 --- a/pkg/storage/unified/sql/backend.go +++ b/pkg/storage/unified/sql/backend.go @@ -569,10 +569,6 @@ func (b *backend) ListIterator(ctx context.Context, req *resourcepb.ListRequest, return 0, fmt.Errorf("missing group or resource") } - if req.Source != resourcepb.ListRequest_STORE { - return b.getHistory(ctx, req, cb) - } - // TODO: think about how to handler VersionMatch. We should be able to use latest for the first page (only). // TODO: add support for RemainingItemCount @@ -583,6 +579,13 @@ func (b *backend) ListIterator(ctx context.Context, req *resourcepb.ListRequest, return b.listLatest(ctx, req, cb) } +func (b *backend) ListHistory(ctx context.Context, req *resourcepb.ListRequest, cb func(resource.ListIterator) error) (int64, error) { + ctx, span := b.tracer.Start(ctx, tracePrefix+"ListHistory") + defer span.End() + + return b.getHistory(ctx, req, cb) +} + // listLatest fetches the resources from the resource table. func (b *backend) listLatest(ctx context.Context, req *resourcepb.ListRequest, cb func(resource.ListIterator) error) (int64, error) { ctx, span := b.tracer.Start(ctx, tracePrefix+"listLatest") @@ -733,7 +736,9 @@ func (b *backend) getHistory(ctx context.Context, req *resourcepb.ListRequest, c // for Unset (default) and Exact matching. listReq.SortAscending = req.GetVersionMatchV2() == resourcepb.ResourceVersionMatchV2_NotOlderThan - iter := &listIter{} + iter := &listIter{ + useCurrentRV: true, // use the current RV for the continue token instead of the listRV + } if req.NextPageToken != "" { continueToken, err := resource.GetContinueToken(req.NextPageToken) if err != nil { diff --git a/pkg/storage/unified/sql/list_iterator.go b/pkg/storage/unified/sql/list_iterator.go index 96642bf58f3..8689dc20f45 100644 --- a/pkg/storage/unified/sql/list_iterator.go +++ b/pkg/storage/unified/sql/list_iterator.go @@ -8,10 +8,11 @@ import ( var _ resource.ListIterator = (*listIter)(nil) type listIter struct { - rows db.Rows - offset int64 - listRV int64 - sortAsc bool + rows db.Rows + offset int64 + listRV int64 + sortAsc bool + useCurrentRV bool // any error err error @@ -29,13 +30,12 @@ type listIter struct { // ContinueToken implements resource.ListIterator. func (l *listIter) ContinueToken() string { + if l.useCurrentRV { + return resource.ContinueToken{ResourceVersion: l.rv, StartOffset: l.offset, SortAscending: l.sortAsc}.String() + } return resource.ContinueToken{ResourceVersion: l.listRV, StartOffset: l.offset, SortAscending: l.sortAsc}.String() } -func (l *listIter) ContinueTokenWithCurrentRV() string { - return resource.ContinueToken{ResourceVersion: l.rv, StartOffset: l.offset, SortAscending: l.sortAsc}.String() -} - func (l *listIter) Error() error { return l.err } diff --git a/pkg/storage/unified/sql/list_iterator_test.go b/pkg/storage/unified/sql/list_iterator_test.go index ae827237787..e67f5c388db 100644 --- a/pkg/storage/unified/sql/list_iterator_test.go +++ b/pkg/storage/unified/sql/list_iterator_test.go @@ -219,7 +219,7 @@ func TestListIter(t *testing.T) { require.Equal(t, expected, actual) }) - t.Run("ContinueTokenWithCurrentRV uses current row's RV", func(t *testing.T) { + t.Run("ContinueToken uses the current row's RV", func(t *testing.T) { listReq := sqlResourceListRequest{ SQLTemplate: sqltemplate.New(dialect), Request: new(resourcepb.ListRequest), @@ -229,14 +229,15 @@ func TestListIter(t *testing.T) { require.NoError(t, err) iter := &listIter{ - rows: rows, - listRV: 300, - sortAsc: true, + rows: rows, + listRV: 300, + sortAsc: true, + useCurrentRV: true, // use the current RV for the continue token instead of the listRV } require.True(t, iter.Next()) - token := iter.ContinueTokenWithCurrentRV() + token := iter.ContinueToken() var actual resource.ContinueToken b, err := base64.StdEncoding.DecodeString(token)