diff --git a/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_test.go b/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_test.go index 039aff032b4..90a173d0d34 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_test.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_test.go @@ -187,7 +187,7 @@ func TestLists(t *testing.T) { t.Parallel() ctx, cacher, server, terminate := testSetupWithEtcdServer(t) t.Cleanup(terminate) - storagetesting.RunTestList(ctx, t, cacher, increaseRV(server.V3Client.Client), true, server.V3Client.Kubernetes.(*storagetesting.KubernetesRecorder)) + storagetesting.RunTestList(ctx, t, cacher, increaseRV(server.V3Client.Client), compactStore(cacher, server.V3Client.Client), true, server.V3Client.Kubernetes.(*storagetesting.KubernetesRecorder)) }) t.Run("ConsistentList", func(t *testing.T) { @@ -300,7 +300,7 @@ func TestWatch(t *testing.T) { func TestWatchFromZero(t *testing.T) { ctx, cacher, server, terminate := testSetupWithEtcdServer(t) t.Cleanup(terminate) - storagetesting.RunTestWatchFromZero(ctx, t, cacher, compactWatchCache(cacher, server.V3Client.Client)) + storagetesting.RunTestWatchFromZero(ctx, t, cacher, compactWatch(cacher, server.V3Client.Client)) } func TestDeleteTriggerWatch(t *testing.T) { diff --git a/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_testing_utils_test.go b/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_testing_utils_test.go index 9090e4c6923..f14564826fb 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_testing_utils_test.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_testing_utils_test.go @@ -79,7 +79,7 @@ func computePodKey(obj *example.Pod) string { return fmt.Sprintf("/pods/%s/%s", obj.Namespace, obj.Name) } -func compactWatchCache(c *CacheDelegator, client *clientv3.Client) storagetesting.Compaction { +func compactWatch(c *CacheDelegator, client *clientv3.Client) storagetesting.Compaction { return func(ctx context.Context, t *testing.T, resourceVersion string) { versioner := storage.APIObjectVersioner{} rv, err := versioner.ParseResourceVersion(resourceVersion) @@ -124,6 +124,20 @@ func compactWatchCache(c *CacheDelegator, client *clientv3.Client) storagetestin } } +func compactStore(c *CacheDelegator, client *clientv3.Client) storagetesting.Compaction { + return func(ctx context.Context, t *testing.T, resourceVersion string) { + versioner := storage.APIObjectVersioner{} + rv, err := versioner.ParseResourceVersion(resourceVersion) + if err != nil { + t.Fatal(err) + } + if _, err := client.Compact(ctx, int64(rv)); err != nil { + t.Fatalf("Unable to compact, %v", err) + } + c.cacher.watchCache.Compact(int64(rv)) + } +} + func increaseRV(client *clientv3.Client) storagetesting.IncreaseRVFunc { return func(ctx context.Context, t *testing.T) { if _, err := client.KV.Put(ctx, "increaseRV", "ok"); err != nil { diff --git a/staging/src/k8s.io/apiserver/pkg/storage/cacher/watch_cache.go b/staging/src/k8s.io/apiserver/pkg/storage/cacher/watch_cache.go index a5093435643..f992e9b01b0 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/cacher/watch_cache.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/cacher/watch_cache.go @@ -935,3 +935,10 @@ func (w *watchCache) getIntervalFromStoreLocked(key string, matchesSingle bool) } return ci, nil } + +func (w *watchCache) Compact(rev int64) { + if w.snapshots == nil { + return + } + w.snapshots.RemoveLess(uint64(rev)) +} diff --git a/staging/src/k8s.io/apiserver/pkg/storage/etcd3/store_test.go b/staging/src/k8s.io/apiserver/pkg/storage/etcd3/store_test.go index 2ee950e2d8f..e99b54e1ea5 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/etcd3/store_test.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/etcd3/store_test.go @@ -251,7 +251,7 @@ func TestTransformationFailure(t *testing.T) { func TestList(t *testing.T) { ctx, store, client := testSetup(t) - storagetesting.RunTestList(ctx, t, store, increaseRV(client.Client), false, client.Kubernetes.(*storagetesting.KubernetesRecorder)) + storagetesting.RunTestList(ctx, t, store, increaseRV(client.Client), compactStorage(client.Client), false, client.Kubernetes.(*storagetesting.KubernetesRecorder)) } func TestConsistentList(t *testing.T) { diff --git a/staging/src/k8s.io/apiserver/pkg/storage/testing/store_tests.go b/staging/src/k8s.io/apiserver/pkg/storage/testing/store_tests.go index 22f76ff127a..4ba01e4b616 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/testing/store_tests.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/testing/store_tests.go @@ -627,7 +627,7 @@ func RunTestPreconditionalDeleteWithOnlySuggestionPass(ctx context.Context, t *t expectNoDiff(t, "incorrect pod:", updatedPod, out) } -func RunTestList(ctx context.Context, t *testing.T, store storage.Interface, increaseRV IncreaseRVFunc, watchCacheEnabled bool, recorder *KubernetesRecorder) { +func RunTestList(ctx context.Context, t *testing.T, store storage.Interface, increaseRV IncreaseRVFunc, compact Compaction, watchCacheEnabled bool, recorder *KubernetesRecorder) { initialRV, createdPods, updatedPod, err := seedMultiLevelData(ctx, store) if err != nil { t.Fatal(err) @@ -649,6 +649,8 @@ func RunTestList(ctx context.Context, t *testing.T, store storage.Interface, inc t.Fatal(err) } + initialRV2, _ := strconv.Atoi(initialRV) + getAttrs := func(obj runtime.Object) (labels.Set, fields.Set, error) { pod := obj.(*example.Pod) return nil, fields.Set{"metadata.name": pod.Name, "spec.nodeName": pod.Spec.NodeName}, nil @@ -656,6 +658,7 @@ func RunTestList(ctx context.Context, t *testing.T, store storage.Interface, inc // Increase RV to test consistent List. increaseRV(ctx, t) currentRV := fmt.Sprintf("%d", continueRV+1) + compact(ctx, t, createdPods[0].ResourceVersion) tests := []struct { name string @@ -671,6 +674,8 @@ func RunTestList(ctx context.Context, t *testing.T, store storage.Interface, inc expectedRemainingItemCount *int64 expectError bool expectRVTooLarge bool + expectRVTooOld bool + expectContinueTooOld bool expectRV string expectRVFunc func(string) error expectEtcdRequest func() []RecordedList @@ -715,13 +720,39 @@ func RunTestList(ctx context.Context, t *testing.T, store storage.Interface, inc rv: "0", }, { - name: "test List on existing key with resource version set before first write, match=Exact", + name: "test List on existing key with resource version set before update, match=Exact", prefix: "/pods/first/", pred: storage.Everything, - expectedOut: []example.Pod{}, - rv: initialRV, + expectedOut: []example.Pod{*createdPods[0]}, + rv: createdPods[0].ResourceVersion, rvMatch: metav1.ResourceVersionMatchExact, - expectRV: initialRV, + expectRV: createdPods[0].ResourceVersion, + }, + { + name: "test List on existing key with resource version set before creation, match=Exact", + prefix: "/pods/second/", + pred: storage.Everything, + expectedOut: []example.Pod{}, + rv: createdPods[0].ResourceVersion, + rvMatch: metav1.ResourceVersionMatchExact, + expectRV: createdPods[0].ResourceVersion, + }, + { + name: "test List on existing key with resource version set after creation, match=Exact", + prefix: "/pods/second/", + pred: storage.Everything, + expectedOut: []example.Pod{*createdPods[1]}, + rv: createdPods[1].ResourceVersion, + rvMatch: metav1.ResourceVersionMatchExact, + expectRV: createdPods[1].ResourceVersion, + }, + { + name: "test List on existing key with resource version set before first write, match=Exact", + prefix: "/pods/first/", + pred: storage.Everything, + rv: initialRV, + rvMatch: metav1.ResourceVersionMatchExact, + expectRVTooOld: true, }, { name: "test List on existing key with resource version set to 0, match=NotOlderThan", @@ -916,7 +947,7 @@ func RunTestList(ctx context.Context, t *testing.T, store storage.Interface, inc expectRVFunc: resourceVersionNotOlderThan(list.ResourceVersion), }, { - name: "test List with limit at resource version before first write and match=Exact", + name: "test List with limit at resource version before created and match=Exact", prefix: "/pods/second/", pred: storage.SelectionPredicate{ Label: labels.Everything(), @@ -925,9 +956,23 @@ func RunTestList(ctx context.Context, t *testing.T, store storage.Interface, inc }, expectedOut: []example.Pod{}, expectContinue: false, - rv: initialRV, + rv: createdPods[0].ResourceVersion, rvMatch: metav1.ResourceVersionMatchExact, - expectRV: initialRV, + expectRV: createdPods[0].ResourceVersion, + }, + { + name: "test List with limit at resource version after created and match=Exact", + prefix: "/pods/second/", + pred: storage.SelectionPredicate{ + Label: labels.Everything(), + Field: fields.Everything(), + Limit: 1, + }, + expectedOut: []example.Pod{*createdPods[1]}, + expectContinue: false, + rv: createdPods[1].ResourceVersion, + rvMatch: metav1.ResourceVersionMatchExact, + expectRV: createdPods[1].ResourceVersion, }, { name: "test List with pregenerated continue token", @@ -1207,13 +1252,12 @@ func RunTestList(ctx context.Context, t *testing.T, store storage.Interface, inc }, // match=Exact { - name: "test List with resource version set before first write, match=Exact", - prefix: "/pods/", - pred: storage.Everything, - expectedOut: []example.Pod{}, - rv: initialRV, - rvMatch: metav1.ResourceVersionMatchExact, - expectRV: initialRV, + name: "test List with resource version set before first write, match=Exact", + prefix: "/pods/", + pred: storage.Everything, + rv: initialRV, + rvMatch: metav1.ResourceVersionMatchExact, + expectRVTooOld: true, }, { name: "test List with resource version of first write, match=Exact", @@ -1428,6 +1472,29 @@ func RunTestList(ctx context.Context, t *testing.T, store storage.Interface, inc expectedRemainingItemCount: utilpointer.Int64(4), }, // Continue + { + name: "test List with continue, resource version before first write", + prefix: "/pods/", + pred: storage.SelectionPredicate{ + Label: labels.Everything(), + Field: fields.Everything(), + Limit: 1, + Continue: encodeContinueOrDie(createdPods[0].Namespace+"/"+createdPods[0].Name+"\x00", int64(initialRV2)), + }, + expectContinueTooOld: true, + }, + { + name: "test List with continue, resource version of first write", + prefix: "/pods/", + pred: storage.SelectionPredicate{ + Label: labels.Everything(), + Field: fields.Everything(), + Limit: 1, + Continue: encodeContinueOrDie(createdPods[0].Namespace+"/"+createdPods[0].Name+"\x00", int64(mustAtoi(createdPods[0].ResourceVersion))), + }, + expectedOut: []example.Pod{}, + expectRV: createdPods[0].ResourceVersion, + }, { name: "test List with continue, resource version of second write", prefix: "/pods/", @@ -1614,6 +1681,18 @@ func RunTestList(ctx context.Context, t *testing.T, store storage.Interface, inc } return } + if tt.expectRVTooOld { + if err == nil || !strings.Contains(err.Error(), "The resourceVersion for the provided list is too old") { + t.Fatalf("expecting resource version too old error, but get: %v", err) + } + return + } + if tt.expectContinueTooOld { + if err == nil || !strings.Contains(err.Error(), "The provided continue parameter is too old to display a consistent list result") { + t.Fatalf("expecting continue too old error, but get: %v", err) + } + return + } if err != nil { if !tt.expectError {