Add testing compaction in storage List tests

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) {

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 {

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))
+}

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) {

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 {