Refactor etcd3 list consistency test

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

@@ -22,8 +22,6 @@ import (
 	"io/ioutil"
 	"os"
 	"reflect"
-	"strconv"
-	"sync"
 	"sync/atomic"
 	"testing"
 
@@ -33,8 +31,6 @@ import (
 
 	"k8s.io/apimachinery/pkg/api/apitesting"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/fields"
-	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/runtime/serializer"
@@ -269,6 +265,11 @@ func TestListInconsistentContinuation(t *testing.T) {
 	storagetesting.RunTestListInconsistentContinuation(ctx, t, store, compactStorage(client))
 }
 
+func TestConsistentList(t *testing.T) {
+	ctx, store, _ := testSetup(t)
+	storagetesting.RunTestConsistentList(ctx, t, &storeWithPrefixTransformer{store})
+}
+
 func TestCount(t *testing.T) {
 	ctx, store, _ := testSetup(t)
 	storagetesting.RunTestCount(ctx, t, store)
@@ -509,12 +510,6 @@ func withoutPaging() setupOption {
 	}
 }
 
-func withTransformer(transformer value.Transformer) setupOption {
-	return func(options *setupOptions) {
-		options.transformer = transformer
-	}
-}
-
 func withLeaseConfig(leaseConfig LeaseManagerConfig) setupOption {
 	return func(options *setupOptions) {
 		options.leaseConfig = leaseConfig
@@ -565,111 +560,3 @@ func testSetup(t *testing.T, opts ...setupOption) (context.Context, *store, *cli
 	ctx := context.Background()
 	return ctx, store, client
 }
-
-// fancyTransformer creates next object on each call to
-// TransformFromStorage call.
-type fancyTransformer struct {
-	transformer value.Transformer
-	store       *store
-
-	lock  sync.Mutex
-	index int
-}
-
-func (t *fancyTransformer) TransformFromStorage(ctx context.Context, data []byte, dataCtx value.Context) ([]byte, bool, error) {
-	if err := t.createObject(ctx); err != nil {
-		return nil, false, err
-	}
-	return t.transformer.TransformFromStorage(ctx, data, dataCtx)
-}
-
-func (t *fancyTransformer) TransformToStorage(ctx context.Context, data []byte, dataCtx value.Context) ([]byte, error) {
-	return t.transformer.TransformToStorage(ctx, data, dataCtx)
-}
-
-func (t *fancyTransformer) createObject(ctx context.Context) error {
-	t.lock.Lock()
-	defer t.lock.Unlock()
-
-	t.index++
-	key := fmt.Sprintf("pod-%d", t.index)
-	obj := &example.Pod{
-		ObjectMeta: metav1.ObjectMeta{
-			Name: key,
-			Labels: map[string]string{
-				"even": strconv.FormatBool(t.index%2 == 0),
-			},
-		},
-	}
-	out := &example.Pod{}
-	return t.store.Create(ctx, key, obj, out, 0)
-}
-
-func TestConsistentList(t *testing.T) {
-	transformer := &fancyTransformer{
-		transformer: newTestTransformer(),
-	}
-	ctx, store, _ := testSetup(t, withTransformer(transformer))
-	transformer.store = store
-
-	for i := 0; i < 5; i++ {
-		if err := transformer.createObject(ctx); err != nil {
-			t.Fatalf("failed to create object: %v", err)
-		}
-	}
-
-	getAttrs := func(obj runtime.Object) (labels.Set, fields.Set, error) {
-		pod, ok := obj.(*example.Pod)
-		if !ok {
-			return nil, nil, fmt.Errorf("invalid object")
-		}
-		return labels.Set(pod.Labels), nil, nil
-	}
-	predicate := storage.SelectionPredicate{
-		Label:    labels.Set{"even": "true"}.AsSelector(),
-		GetAttrs: getAttrs,
-		Limit:    4,
-	}
-
-	result1 := example.PodList{}
-	options := storage.ListOptions{
-		Predicate: predicate,
-		Recursive: true,
-	}
-	if err := store.GetList(ctx, "/", options, &result1); err != nil {
-		t.Fatalf("failed to list objects: %v", err)
-	}
-
-	// List objects from the returned resource version.
-	options = storage.ListOptions{
-		Predicate:            predicate,
-		ResourceVersion:      result1.ResourceVersion,
-		ResourceVersionMatch: metav1.ResourceVersionMatchExact,
-		Recursive:            true,
-	}
-
-	result2 := example.PodList{}
-	if err := store.GetList(ctx, "/", options, &result2); err != nil {
-		t.Fatalf("failed to list objects: %v", err)
-	}
-
-	storagetesting.ExpectNoDiff(t, "incorrect lists", result1, result2)
-
-	// Now also verify the  ResourceVersionMatchNotOlderThan.
-	options.ResourceVersionMatch = metav1.ResourceVersionMatchNotOlderThan
-
-	result3 := example.PodList{}
-	if err := store.GetList(ctx, "/", options, &result3); err != nil {
-		t.Fatalf("failed to list objects: %v", err)
-	}
-
-	options.ResourceVersion = result3.ResourceVersion
-	options.ResourceVersionMatch = metav1.ResourceVersionMatchExact
-
-	result4 := example.PodList{}
-	if err := store.GetList(ctx, "/", options, &result4); err != nil {
-		t.Fatalf("failed to list objects: %v", err)
-	}
-
-	storagetesting.ExpectNoDiff(t, "incorrect lists", result3, result4)
-}

staging/src/k8s.io/apiserver/pkg/storage/testing/store_tests.go

@@ -1574,6 +1574,94 @@ type InterfaceWithPrefixTransformer interface {
 	UpdatePrefixTransformer(PrefixTransformerModifier) func()
 }
 
+func RunTestConsistentList(ctx context.Context, t *testing.T, store InterfaceWithPrefixTransformer) {
+	nextPod := func(index uint32) (string, *example.Pod) {
+		key := fmt.Sprintf("pod-%d", index)
+		obj := &example.Pod{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: key,
+				Labels: map[string]string{
+					"even": strconv.FormatBool(index%2 == 0),
+				},
+			},
+		}
+		return key, obj
+	}
+
+	transformer := &reproducingTransformer{
+		store:      store,
+		nextObject: nextPod,
+	}
+
+	revertTransformer := store.UpdatePrefixTransformer(
+		func(previousTransformer *PrefixTransformer) value.Transformer {
+			transformer.wrapped = previousTransformer
+			return transformer
+		})
+	defer revertTransformer()
+
+	for i := 0; i < 5; i++ {
+		if err := transformer.createObject(ctx); err != nil {
+			t.Fatalf("failed to create object: %v", err)
+		}
+	}
+
+	getAttrs := func(obj runtime.Object) (labels.Set, fields.Set, error) {
+		pod, ok := obj.(*example.Pod)
+		if !ok {
+			return nil, nil, fmt.Errorf("invalid object")
+		}
+		return labels.Set(pod.Labels), nil, nil
+	}
+	predicate := storage.SelectionPredicate{
+		Label:    labels.Set{"even": "true"}.AsSelector(),
+		GetAttrs: getAttrs,
+		Limit:    4,
+	}
+
+	result1 := example.PodList{}
+	options := storage.ListOptions{
+		Predicate: predicate,
+		Recursive: true,
+	}
+	if err := store.GetList(ctx, "/", options, &result1); err != nil {
+		t.Fatalf("failed to list objects: %v", err)
+	}
+
+	// List objects from the returned resource version.
+	options = storage.ListOptions{
+		Predicate:            predicate,
+		ResourceVersion:      result1.ResourceVersion,
+		ResourceVersionMatch: metav1.ResourceVersionMatchExact,
+		Recursive:            true,
+	}
+
+	result2 := example.PodList{}
+	if err := store.GetList(ctx, "/", options, &result2); err != nil {
+		t.Fatalf("failed to list objects: %v", err)
+	}
+
+	ExpectNoDiff(t, "incorrect lists", result1, result2)
+
+	// Now also verify the  ResourceVersionMatchNotOlderThan.
+	options.ResourceVersionMatch = metav1.ResourceVersionMatchNotOlderThan
+
+	result3 := example.PodList{}
+	if err := store.GetList(ctx, "/", options, &result3); err != nil {
+		t.Fatalf("failed to list objects: %v", err)
+	}
+
+	options.ResourceVersion = result3.ResourceVersion
+	options.ResourceVersionMatch = metav1.ResourceVersionMatchExact
+
+	result4 := example.PodList{}
+	if err := store.GetList(ctx, "/", options, &result4); err != nil {
+		t.Fatalf("failed to list objects: %v", err)
+	}
+
+	ExpectNoDiff(t, "incorrect lists", result3, result4)
+}
+
 func RunTestGuaranteedUpdate(ctx context.Context, t *testing.T, store InterfaceWithPrefixTransformer, validation KeyValidation) {
 	key := "/testkey"
 

staging/src/k8s.io/apiserver/pkg/storage/testing/utils.go

@@ -235,3 +235,33 @@ func (p *PrefixTransformer) TransformToStorage(ctx context.Context, data []byte,
 func (p *PrefixTransformer) GetReadsAndReset() uint64 {
 	return atomic.SwapUint64(&p.reads, 0)
 }
+
+// reproducingTransformer is a custom test-only transformer used purely
+// for testing consistency.
+// It allows for creating predefined objects on TransformFromStorage operations,
+// which allows for precise in time injection of new objects in the middle of
+// read operations.
+type reproducingTransformer struct {
+	wrapped value.Transformer
+	store   storage.Interface
+
+	index      uint32
+	nextObject func(uint32) (string, *example.Pod)
+}
+
+func (rt *reproducingTransformer) TransformFromStorage(ctx context.Context, data []byte, dataCtx value.Context) ([]byte, bool, error) {
+	if err := rt.createObject(ctx); err != nil {
+		return nil, false, err
+	}
+	return rt.wrapped.TransformFromStorage(ctx, data, dataCtx)
+}
+
+func (rt *reproducingTransformer) TransformToStorage(ctx context.Context, data []byte, dataCtx value.Context) ([]byte, error) {
+	return rt.wrapped.TransformToStorage(ctx, data, dataCtx)
+}
+
+func (rt *reproducingTransformer) createObject(ctx context.Context) error {
+	key, obj := rt.nextObject(atomic.AddUint32(&rt.index, 1))
+	out := &example.Pod{}
+	return rt.store.Create(ctx, key, obj, out, 0)
+}