Handle expired errors with RV>0 in pager, don't full list if 1st page is expired

Kubernetes-commit: 1f3dc14eea231c90819cbba06f25cb4221236b3f

tools/cache/reflector_test.go

@@ -26,6 +26,7 @@ import (
 	"time"
 
 	"k8s.io/api/core/v1"
+	apierrs "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -434,6 +435,194 @@ func TestReflectorWatchListPageSize(t *testing.T) {
 	}
 }
 
+// TestReflectorResyncWithResourceVersion ensures that a reflector keeps track of the ResourceVersion and sends
+// it in relist requests to prevent the reflector from traveling back in time if the relist is to a api-server or
+// etcd that is partitioned and serving older data than the reflector has already processed.
+func TestReflectorResyncWithResourceVersion(t *testing.T) {
+	stopCh := make(chan struct{})
+	s := NewStore(MetaNamespaceKeyFunc)
+	listCallRVs := []string{}
+
+	lw := &testLW{
+		WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
+			// Stop once the reflector begins watching since we're only interested in the list.
+			close(stopCh)
+			fw := watch.NewFake()
+			return fw, nil
+		},
+		ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
+			listCallRVs = append(listCallRVs, options.ResourceVersion)
+			pods := make([]v1.Pod, 8)
+			for i := 0; i < 8; i++ {
+				pods[i] = v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: fmt.Sprintf("pod-%d", i), ResourceVersion: fmt.Sprintf("%d", i)}}
+			}
+			switch options.ResourceVersion {
+			case "0":
+				return &v1.PodList{ListMeta: metav1.ListMeta{ResourceVersion: "10"}, Items: pods[0:4]}, nil
+			case "10":
+				return &v1.PodList{ListMeta: metav1.ListMeta{ResourceVersion: "11"}, Items: pods[0:8]}, nil
+			default:
+				t.Fatalf("Unrecognized ResourceVersion: %s", options.ResourceVersion)
+			}
+			return nil, nil
+		},
+	}
+	r := NewReflector(lw, &v1.Pod{}, s, 0)
+
+	// Initial list should use RV=0
+	r.ListAndWatch(stopCh)
+
+	results := s.List()
+	if len(results) != 4 {
+		t.Errorf("Expected 4 results, got %d", len(results))
+	}
+
+	// relist should use lastSyncResourceVersions (RV=10)
+	stopCh = make(chan struct{})
+	r.ListAndWatch(stopCh)
+
+	results = s.List()
+	if len(results) != 8 {
+		t.Errorf("Expected 8 results, got %d", len(results))
+	}
+
+	expectedRVs := []string{"0", "10"}
+	if !reflect.DeepEqual(listCallRVs, expectedRVs) {
+		t.Errorf("Expected series of list calls with resource versiosn of %v but got: %v", expectedRVs, listCallRVs)
+	}
+}
+
+// TestReflectorExpiredExactResourceVersion tests that a reflector handles the behavior of kubernetes 1.16 an earlier
+// where if the exact ResourceVersion requested is not available for a List request for a non-zero ResourceVersion,
+// an "Expired" error is returned if the ResourceVersion has expired (etcd has compacted it).
+// (In kubernetes 1.17, or when the watch cache is enabled, the List will instead return the list that is no older than
+// the requested ResourceVersion).
+func TestReflectorExpiredExactResourceVersion(t *testing.T) {
+	stopCh := make(chan struct{})
+	s := NewStore(MetaNamespaceKeyFunc)
+	listCallRVs := []string{}
+
+	lw := &testLW{
+		WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
+			// Stop once the reflector begins watching since we're only interested in the list.
+			close(stopCh)
+			fw := watch.NewFake()
+			return fw, nil
+		},
+		ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
+			listCallRVs = append(listCallRVs, options.ResourceVersion)
+			pods := make([]v1.Pod, 8)
+			for i := 0; i < 8; i++ {
+				pods[i] = v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: fmt.Sprintf("pod-%d", i), ResourceVersion: fmt.Sprintf("%d", i)}}
+			}
+			switch options.ResourceVersion {
+			case "0":
+				return &v1.PodList{ListMeta: metav1.ListMeta{ResourceVersion: "10"}, Items: pods[0:4]}, nil
+			case "10":
+				// When watch cache is disabled, if the exact ResourceVersion requested is not available, a "Expired" error is returned.
+				return nil, apierrs.NewResourceExpired("The resourceVersion for the provided watch is too old.")
+			case "":
+				return &v1.PodList{ListMeta: metav1.ListMeta{ResourceVersion: "11"}, Items: pods[0:8]}, nil
+			default:
+				t.Fatalf("Unrecognized ResourceVersion: %s", options.ResourceVersion)
+			}
+			return nil, nil
+		},
+	}
+	r := NewReflector(lw, &v1.Pod{}, s, 0)
+
+	// Initial list should use RV=0
+	r.ListAndWatch(stopCh)
+
+	results := s.List()
+	if len(results) != 4 {
+		t.Errorf("Expected 4 results, got %d", len(results))
+	}
+
+	// relist should use lastSyncResourceVersions (RV=10) and since RV=10 is expired, it should retry with RV="".
+	stopCh = make(chan struct{})
+	r.ListAndWatch(stopCh)
+
+	results = s.List()
+	if len(results) != 8 {
+		t.Errorf("Expected 8 results, got %d", len(results))
+	}
+
+	expectedRVs := []string{"0", "10", ""}
+	if !reflect.DeepEqual(listCallRVs, expectedRVs) {
+		t.Errorf("Expected series of list calls with resource versiosn of %v but got: %v", expectedRVs, listCallRVs)
+	}
+}
+
+func TestReflectorFullListIfExpired(t *testing.T) {
+	stopCh := make(chan struct{})
+	s := NewStore(MetaNamespaceKeyFunc)
+	listCallRVs := []string{}
+
+	lw := &testLW{
+		WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
+			// Stop once the reflector begins watching since we're only interested in the list.
+			close(stopCh)
+			fw := watch.NewFake()
+			return fw, nil
+		},
+		ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
+			listCallRVs = append(listCallRVs, options.ResourceVersion)
+			pods := make([]v1.Pod, 8)
+			for i := 0; i < 8; i++ {
+				pods[i] = v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: fmt.Sprintf("pod-%d", i), ResourceVersion: fmt.Sprintf("%d", i)}}
+			}
+			switch options.ResourceVersion {
+			case "0":
+				return &v1.PodList{ListMeta: metav1.ListMeta{ResourceVersion: "10"}, Items: pods[0:4]}, nil
+			case "10":
+				switch options.Limit {
+				case 4:
+					switch options.Continue {
+					case "":
+						return &v1.PodList{ListMeta: metav1.ListMeta{Continue: "C1", ResourceVersion: "11"}, Items: pods[0:4]}, nil
+					case "C1":
+						return nil, apierrs.NewResourceExpired("The resourceVersion for the provided watch is too old.")
+					default:
+						t.Fatalf("Unrecognized Continue: %s", options.Continue)
+					}
+				case 0:
+					return &v1.PodList{ListMeta: metav1.ListMeta{ResourceVersion: "11"}, Items: pods[0:8]}, nil
+				default:
+					t.Fatalf("Unrecognized Limit: %d", options.Limit)
+				}
+			default:
+				t.Fatalf("Unrecognized ResourceVersion: %s", options.ResourceVersion)
+			}
+			return nil, nil
+		},
+	}
+	r := NewReflector(lw, &v1.Pod{}, s, 0)
+	r.WatchListPageSize = 4
+
+	// Initial list should use RV=0
+	r.ListAndWatch(stopCh)
+
+	results := s.List()
+	if len(results) != 4 {
+		t.Errorf("Expected 4 results, got %d", len(results))
+	}
+
+	// relist should use lastSyncResourceVersions (RV=10) and since second page of RV=10 is expired, it should full list with RV=10
+	stopCh = make(chan struct{})
+	r.ListAndWatch(stopCh)
+
+	results = s.List()
+	if len(results) != 8 {
+		t.Errorf("Expected 8 results, got %d", len(results))
+	}
+
+	expectedRVs := []string{"0", "10", "10", "10"}
+	if !reflect.DeepEqual(listCallRVs, expectedRVs) {
+		t.Errorf("Expected series of list calls with resource versiosn of %v but got: %v", expectedRVs, listCallRVs)
+	}
+}
+
 func TestReflectorSetExpectedType(t *testing.T) {
 	obj := &unstructured.Unstructured{}
 	gvk := schema.GroupVersionKind{