diff --git a/pkg/storage/unified/resource/search.go b/pkg/storage/unified/resource/search.go index a98d5d0cfc1..f6fa4e02475 100644 --- a/pkg/storage/unified/resource/search.go +++ b/pkg/storage/unified/resource/search.go @@ -563,7 +563,7 @@ func (s *searchSupport) build(ctx context.Context, nsr NamespacedResource, size return err } } - return err + return iter.Error() }) return rv, err }) diff --git a/pkg/storage/unified/resource/server.go b/pkg/storage/unified/resource/server.go index 81fc94ae146..ab54a68fb61 100644 --- a/pkg/storage/unified/resource/server.go +++ b/pkg/storage/unified/resource/server.go @@ -18,6 +18,7 @@ import ( "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" claims "github.com/grafana/authlib/types" + "github.com/grafana/grafana/pkg/apimachinery/utils" ) @@ -32,15 +33,18 @@ type ResourceServer interface { } type ListIterator interface { + // Next advances iterator and returns true if there is next value is available from the iterator. + // Error() should be checked after every call of Next(), even when Next() returns true. Next() bool // sql.Rows - // Iterator error (if exts) + // Error returns iterator error, if any. This should be checked after any Next() call. + // (Some iterator implementations return true from Next, but also set the error at the same time). Error() error - // The token that can be used to start iterating *after* this item + // ContinueToken returns the token that can be used to start iterating *after* this item ContinueToken() string - // The token that can be used to start iterating *before* this item + // ContinueTokenWithCurrentRV returns the token that can be used to start iterating *before* this item ContinueTokenWithCurrentRV() string // ResourceVersion of the current item @@ -763,11 +767,10 @@ func (s *server) List(ctx context.Context, req *ListRequest) (*ListResponse, err if iter.Next() { rsp.NextPageToken = t } - - break + return iter.Error() } } - return nil + return iter.Error() }) if err != nil { rsp.Error = AsErrorResult(err) @@ -871,7 +874,7 @@ func (s *server) Watch(req *WatchRequest, srv ResourceStore_WatchServer) error { return err } } - return nil + return iter.Error() }) if err != nil { return err diff --git a/pkg/storage/unified/testing/storage_backend.go b/pkg/storage/unified/testing/storage_backend.go index a81a69a95fa..e6e831dabc3 100644 --- a/pkg/storage/unified/testing/storage_backend.go +++ b/pkg/storage/unified/testing/storage_backend.go @@ -19,6 +19,7 @@ import ( "github.com/grafana/authlib/authn" "github.com/grafana/authlib/types" + "github.com/grafana/grafana/pkg/apimachinery/utils" "github.com/grafana/grafana/pkg/storage/unified/resource" "github.com/grafana/grafana/pkg/util/testutil" @@ -26,13 +27,14 @@ import ( // Test names for the storage backend test suite const ( - TestHappyPath = "happy path" - TestWatchWriteEvents = "watch write events from latest" - TestList = "list" - TestBlobSupport = "blob support" - TestGetResourceStats = "get resource stats" - TestListHistory = "list history" - TestCreateNewResource = "create new resource" + TestHappyPath = "happy path" + TestWatchWriteEvents = "watch write events from latest" + TestList = "list" + TestBlobSupport = "blob support" + TestGetResourceStats = "get resource stats" + TestListHistory = "list history" + TestListHistoryErrorReporting = "list history error reporting" + TestCreateNewResource = "create new resource" ) type NewBackendFunc func(ctx context.Context) resource.StorageBackend @@ -75,6 +77,7 @@ func RunStorageBackendTest(t *testing.T, newBackend NewBackendFunc, opts *TestOp {TestBlobSupport, runTestIntegrationBlobSupport}, {TestGetResourceStats, runTestIntegrationBackendGetResourceStats}, {TestListHistory, runTestIntegrationBackendListHistory}, + {TestListHistoryErrorReporting, runTestIntegrationBackendListHistoryErrorReporting}, {TestCreateNewResource, runTestIntegrationBackendCreateNewResource}, } @@ -476,7 +479,7 @@ func runTestIntegrationBackendList(t *testing.T, backend resource.StorageBackend } func runTestIntegrationBackendListHistory(t *testing.T, backend resource.StorageBackend, nsPrefix string) { - ctx := testutil.NewTestContext(t, time.Now().Add(5*time.Second)) + ctx := testutil.NewTestContext(t, time.Now().Add(30*time.Second)) server := newServer(t, backend) ns := nsPrefix + "-ns1" rv1, _ := writeEvent(ctx, backend, "item1", resource.WatchEvent_ADDED, WithNamespace(ns)) @@ -839,6 +842,58 @@ func runTestIntegrationBackendListHistory(t *testing.T, backend resource.Storage }) } +func runTestIntegrationBackendListHistoryErrorReporting(t *testing.T, backend resource.StorageBackend, nsPrefix string) { + ctx := testutil.NewTestContext(t, time.Now().Add(30*time.Second)) + server := newServer(t, backend) + + ns := nsPrefix + "-short" + const ( + name = "it1" + group = "group" + resourceName = "resource" + ) + + start := time.Now() + origRv, _ := writeEvent(ctx, backend, name, resource.WatchEvent_ADDED, WithNamespace(ns), WithGroup(group), WithResource(resourceName)) + require.Greater(t, origRv, int64(0)) + + const events = 500 + prevRv := origRv + for i := 0; i < events; i++ { + rv, err := writeEvent(ctx, backend, name, resource.WatchEvent_MODIFIED, WithNamespace(ns), WithGroup(group), WithResource(resourceName)) + require.NoError(t, err) + require.Greater(t, rv, prevRv) + prevRv = rv + } + t.Log("added events in ", time.Since(start)) + + req := &resource.ListRequest{ + Limit: 2 * events, + Source: resource.ListRequest_HISTORY, + ResourceVersion: origRv, + VersionMatchV2: resource.ResourceVersionMatchV2_NotOlderThan, + Options: &resource.ListOptions{ + Key: &resource.ResourceKey{ + Namespace: ns, + Group: group, + Resource: resourceName, + Name: name, + }, + }, + } + + shortContext, cancel := context.WithTimeout(ctx, 1*time.Millisecond) + defer cancel() + + res, err := server.List(shortContext, req) + // We expect context deadline error, but it may be reported as a res.Error object. + t.Log("list error:", err) + if res != nil { + t.Log("iterator error:", res.Error) + } + require.True(t, err != nil || (res != nil && res.Error != nil)) +} + func runTestIntegrationBlobSupport(t *testing.T, backend resource.StorageBackend, nsPrefix string) { ctx := testutil.NewTestContext(t, time.Now().Add(5*time.Second)) server := newServer(t, backend)