pkg/storage/etcd3: correctly validate resourceVersions

In a number of tests, the underlying storage backend interaction will
return the revision (logical clock underpinning the MVCC implementation)
at the call-time of the RPC. Previously, the tests validated that this
returned revision was exactly equal to some previously seen revision.
This assertion is only true in systems where no other events are
advancing the logical clock. For instance, when using a single etcd
cluster as a shared fixture for these tests, the assertion is not valid
any longer. By checking that the returned revision is no older than the
previously seen revision, the validation logic is correct in all cases.

Signed-off-by: Steve Kuznetsov <skuznets@redhat.com>

staging/src/k8s.io/apiserver/pkg/storage/etcd3/store_test.go

@@ -1213,6 +1213,7 @@ func TestList(t *testing.T) {
 		expectError                bool
 		expectRVTooLarge           bool
 		expectRV                   string
+		expectRVFunc               func(string) error
 	}{
 		{
 			name:        "rejects invalid resource version",
@@ -1385,7 +1386,7 @@ func TestList(t *testing.T) {
 			expectContinue:             true,
 			expectedRemainingItemCount: utilpointer.Int64Ptr(1),
 			rv:                         "0",
-			expectRV:                   list.ResourceVersion,
+			expectRVFunc:               resourceVersionNotOlderThan(list.ResourceVersion),
 		},
 		{
 			name:   "test List with limit at resource version 0 match=NotOlderThan",
@@ -1400,7 +1401,7 @@ func TestList(t *testing.T) {
 			expectedRemainingItemCount: utilpointer.Int64Ptr(1),
 			rv:                         "0",
 			rvMatch:                    metav1.ResourceVersionMatchNotOlderThan,
-			expectRV:                   list.ResourceVersion,
+			expectRVFunc:               resourceVersionNotOlderThan(list.ResourceVersion),
 		},
 		{
 			name:   "test List with limit at resource version before first write and match=Exact",
@@ -1612,6 +1613,11 @@ func TestList(t *testing.T) {
 					t.Errorf("resourceVersion in list response want=%s, got=%s", tt.expectRV, out.ResourceVersion)
 				}
 			}
+			if tt.expectRVFunc != nil {
+				if err := tt.expectRVFunc(out.ResourceVersion); err != nil {
+					t.Errorf("resourceVersion in list response invalid: %v", err)
+				}
+			}
 			if len(tt.expectedOut) != len(out.Items) {
 				t.Fatalf("length of list want=%d, got=%d", len(tt.expectedOut), len(out.Items))
 			}
@@ -2106,11 +2112,13 @@ func TestListInconsistentContinuation(t *testing.T) {
 	if len(out.Continue) == 0 {
 		t.Fatalf("No continuation token set")
 	}
+	validateResourceVersion := resourceVersionNotOlderThan(lastRVString)
 	expectNoDiff(t, "incorrect second page", []example.Pod{*preset[1].storedObj}, out.Items)
-	if out.ResourceVersion != lastRVString {
-		t.Fatalf("Expected list resource version to be %s, got %s", lastRVString, out.ResourceVersion)
+	if err := validateResourceVersion(out.ResourceVersion); err != nil {
+		t.Fatal(err)
 	}
 	continueFromThirdItem := out.Continue
+	resolvedResourceVersionFromThirdItem := out.ResourceVersion
 	out = &example.PodList{}
 	options = storage.ListOptions{
 		ResourceVersion: "0",
@@ -2124,8 +2132,8 @@ func TestListInconsistentContinuation(t *testing.T) {
 		t.Fatalf("Unexpected continuation token set")
 	}
 	expectNoDiff(t, "incorrect third page", []example.Pod{*preset[2].storedObj}, out.Items)
-	if out.ResourceVersion != lastRVString {
-		t.Fatalf("Expected list resource version to be %s, got %s", lastRVString, out.ResourceVersion)
+	if out.ResourceVersion != resolvedResourceVersionFromThirdItem {
+		t.Fatalf("Expected list resource version to be %s, got %s", resolvedResourceVersionFromThirdItem, out.ResourceVersion)
 	}
 }
 

staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher_test.go

@@ -353,8 +353,13 @@ func TestProgressNotify(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Watch failed: %v", err)
 	}
-	result := &example.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: out.ResourceVersion}}
-	testCheckResult(t, watch.Bookmark, w, result)
+	testCheckResultFunc(t, watch.Bookmark, w, func(object runtime.Object) error {
+		pod, ok := object.(*example.Pod)
+		if !ok {
+			return fmt.Errorf("got %T, not *example.Pod", object)
+		}
+		return resourceVersionNotOlderThan(out.ResourceVersion)(pod.ResourceVersion)
+	})
 }
 
 type testWatchStruct struct {
@@ -382,14 +387,44 @@ func testCheckEventType(t *testing.T, expectEventType watch.EventType, w watch.I
 	}
 }
 
+// resourceVersionNotOlderThan returns a function to validate resource versions. Resource versions
+// referring to points in logical time before the sentinel generate an error. All logical times as
+// new as the sentinel or newer generate no error.
+func resourceVersionNotOlderThan(sentinel string) func(string) error {
+	return func(resourceVersion string) error {
+		objectVersioner := APIObjectVersioner{}
+		actualRV, err := objectVersioner.ParseResourceVersion(resourceVersion)
+		if err != nil {
+			return err
+		}
+		expectedRV, err := objectVersioner.ParseResourceVersion(sentinel)
+		if err != nil {
+			return err
+		}
+		if actualRV < expectedRV {
+			return fmt.Errorf("expected a resourceVersion no smaller than than %d, but got %d", expectedRV, actualRV)
+		}
+		return nil
+	}
+}
+
 func testCheckResult(t *testing.T, expectEventType watch.EventType, w watch.Interface, expectObj *example.Pod) {
+	testCheckResultFunc(t, expectEventType, w, func(object runtime.Object) error {
+		expectNoDiff(t, "incorrect object", expectObj, object)
+		return nil
+	})
+}
+
+func testCheckResultFunc(t *testing.T, expectEventType watch.EventType, w watch.Interface, check func(object runtime.Object) error) {
 	select {
 	case res := <-w.ResultChan():
 		if res.Type != expectEventType {
 			t.Errorf("event type want=%v, get=%v", expectEventType, res.Type)
 			return
 		}
-		expectNoDiff(t, "incorrect obj", expectObj, res.Object)
+		if err := check(res.Object); err != nil {
+			t.Error(err)
+		}
 	case <-time.After(wait.ForeverTestTimeout):
 		t.Errorf("time out after waiting %v on ResultChan", wait.ForeverTestTimeout)
 	}