Merge pull request kubernetes#130400 from serathius/storage-current-rv

Move GetCurrentResourceVersion to storage.Interface

Author: k8s-ci-robot

staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go

@@ -1248,7 +1248,7 @@ func (c *Cacher) getWatchCacheResourceVersion(ctx context.Context, parsedWatchRe
 	if !utilfeature.DefaultFeatureGate.Enabled(features.WatchFromStorageWithoutResourceVersion) && opts.SendInitialEvents == nil && opts.ResourceVersion == "" {
 		return 0, nil
 	}
-	rv, err := storage.GetCurrentResourceVersionFromStorage(ctx, c.storage, c.newListFunc, c.resourcePrefix, c.objectType.String())
+	rv, err := c.storage.GetCurrentResourceVersion(ctx)
 	return rv, err
 }
 

staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_whitebox_test.go

@@ -113,6 +113,7 @@ type dummyStorage struct {
 	sync.RWMutex
 	err       error
 	getListFn func(_ context.Context, _ string, _ storage.ListOptions, listObj runtime.Object) error
+	getRVFn   func(_ context.Context) (uint64, error)
 	watchFn   func(_ context.Context, _ string, _ storage.ListOptions) (watch.Interface, error)
 
 	// use getRequestWatchProgressCounter when reading
@@ -199,6 +200,13 @@ func (d *dummyStorage) injectError(err error) {
 	d.err = err
 }
 
+func (d *dummyStorage) GetCurrentResourceVersion(ctx context.Context) (uint64, error) {
+	if d.getRVFn != nil {
+		return d.getRVFn(ctx)
+	}
+	return 100, nil
+}
+
 func TestGetListCacheBypass(t *testing.T) {
 	type opts struct {
 		ResourceVersion      string
@@ -463,6 +471,14 @@ apiserver_watch_cache_consistent_read_total{fallback="true", resource="pods", su
 				podList.ResourceVersion = tc.storageRV
 				return nil
 			}
+			backingStorage.getRVFn = func(_ context.Context) (uint64, error) {
+				requestToStorageCount += 1
+				rv, err := strconv.Atoi(tc.storageRV)
+				if err != nil {
+					t.Fatalf("failed to parse RV: %s", err)
+				}
+				return uint64(rv), nil
+			}
 			result := &example.PodList{}
 
 			ctx, clockStepCancelFn := context.WithTimeout(context.TODO(), time.Minute)
@@ -2111,6 +2127,15 @@ func TestWaitUntilWatchCacheFreshAndForceAllEvents(t *testing.T) {
 					listAccessor.SetResourceVersion("105")
 					return nil
 				}
+				s.getRVFn = func(_ context.Context) (uint64, error) {
+					// the first call to this function
+					// primes the cacher
+					if !hasBeenPrimed {
+						hasBeenPrimed = true
+						return 100, nil
+					}
+					return 105, nil
+				}
 				return s
 			}(),
 			verifyBackingStore: func(t *testing.T, s *dummyStorage) {
@@ -2146,6 +2171,15 @@ func TestWaitUntilWatchCacheFreshAndForceAllEvents(t *testing.T) {
 					listAccessor.SetResourceVersion("105")
 					return nil
 				}
+				s.getRVFn = func(_ context.Context) (uint64, error) {
+					// the first call to this function
+					// primes the cacher
+					if !hasBeenPrimed {
+						hasBeenPrimed = true
+						return 100, nil
+					}
+					return 105, nil
+				}
 				return s
 			}(),
 			verifyBackingStore: func(t *testing.T, s *dummyStorage) {

staging/src/k8s.io/apiserver/pkg/storage/cacher/delegator.go

@@ -57,6 +57,10 @@ func (c *CacheDelegator) Create(ctx context.Context, key string, obj, out runtim
 	return c.storage.Create(ctx, key, obj, out, ttl)
 }
 
+func (c *CacheDelegator) GetCurrentResourceVersion(ctx context.Context) (uint64, error) {
+	return c.storage.GetCurrentResourceVersion(ctx)
+}
+
 func (c *CacheDelegator) Delete(ctx context.Context, key string, out runtime.Object, preconditions *storage.Preconditions, validateDeletion storage.ValidateObjectFunc, cachedExistingObject runtime.Object, opts storage.DeleteOptions) error {
 	// Ignore the suggestion and try to pass down the current version of the object
 	// read from cache.
@@ -160,7 +164,7 @@ func (c *CacheDelegator) GetList(ctx context.Context, key string, opts storage.L
 	requestWatchProgressSupported := etcdfeature.DefaultFeatureSupportChecker.Supports(storage.RequestWatchProgress)
 	consistentRead := opts.ResourceVersion == "" && utilfeature.DefaultFeatureGate.Enabled(features.ConsistentListFromCache) && requestWatchProgressSupported
 	if consistentRead {
-		listRV, err = storage.GetCurrentResourceVersionFromStorage(ctx, c.storage, c.cacher.newListFunc, c.cacher.resourcePrefix, c.cacher.objectType.String())
+		listRV, err = c.storage.GetCurrentResourceVersion(ctx)
 		if err != nil {
 			return err
 		}

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

@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"path"
 	"reflect"
+	"strconv"
 	"strings"
 	"time"
 
@@ -35,6 +36,8 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/conversion"
+	"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/watch"
@@ -84,6 +87,9 @@ type store struct {
 	leaseManager        *leaseManager
 	decoder             Decoder
 	listErrAggrFactory  func() ListErrorAggregator
+
+	resourcePrefix string
+	newListFunc    func() runtime.Object
 }
 
 func (s *store) RequestWatchProgress(ctx context.Context) error {
@@ -185,10 +191,13 @@ func newStore(c *kubernetes.Client, codec runtime.Codec, newFunc, newListFunc fu
 		leaseManager:        newDefaultLeaseManager(c.Client, leaseManagerConfig),
 		decoder:             decoder,
 		listErrAggrFactory:  listErrAggrFactory,
+
+		resourcePrefix: resourcePrefix,
+		newListFunc:    newListFunc,
 	}
 
 	w.getCurrentStorageRV = func(ctx context.Context) (uint64, error) {
-		return storage.GetCurrentResourceVersionFromStorage(ctx, s, newListFunc, resourcePrefix, w.objectType)
+		return s.GetCurrentResourceVersion(ctx)
 	}
 	if utilfeature.DefaultFeatureGate.Enabled(features.ConsistentListFromCache) || utilfeature.DefaultFeatureGate.Enabled(features.WatchList) {
 		etcdfeature.DefaultFeatureSupportChecker.CheckClient(c.Ctx(), c, storage.RequestWatchProgress)
@@ -677,6 +686,37 @@ func (s *store) resolveGetListRev(continueKey string, continueRV int64, opts sto
 	return withRev, nil
 }
 
+func (s *store) GetCurrentResourceVersion(ctx context.Context) (uint64, error) {
+	emptyList := s.newListFunc()
+	pred := storage.SelectionPredicate{
+		Label: labels.Everything(),
+		Field: fields.Everything(),
+		Limit: 1, // just in case we actually hit something
+	}
+
+	err := s.GetList(ctx, s.resourcePrefix, storage.ListOptions{Predicate: pred}, emptyList)
+	if err != nil {
+		return 0, err
+	}
+	emptyListAccessor, err := meta.ListAccessor(emptyList)
+	if err != nil {
+		return 0, err
+	}
+	if emptyListAccessor == nil {
+		return 0, fmt.Errorf("unable to extract a list accessor from %T", emptyList)
+	}
+
+	currentResourceVersion, err := strconv.Atoi(emptyListAccessor.GetResourceVersion())
+	if err != nil {
+		return 0, err
+	}
+
+	if currentResourceVersion == 0 {
+		return 0, fmt.Errorf("the current resource version must be greater than 0")
+	}
+	return uint64(currentResourceVersion), nil
+}
+
 // GetList implements storage.Interface.
 func (s *store) GetList(ctx context.Context, key string, opts storage.ListOptions, listObj runtime.Object) error {
 	keyPrefix, err := s.prepareKey(key)

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

@@ -28,6 +28,7 @@ import (
 	"testing"
 
 	"github.com/go-logr/logr"
+	"github.com/stretchr/testify/require"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.etcd.io/etcd/client/v3/kubernetes"
 	"go.etcd.io/etcd/server/v3/embed"
@@ -994,3 +995,49 @@ func BenchmarkStoreList(b *testing.B) {
 func computePodKey(obj *example.Pod) string {
 	return fmt.Sprintf("/pods/%s/%s", obj.Namespace, obj.Name)
 }
+
+func TestGetCurrentResourceVersion(t *testing.T) {
+	ctx, store, _ := testSetup(t)
+
+	makePod := func(name string) *example.Pod {
+		return &example.Pod{
+			ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: name},
+		}
+	}
+	createPod := func(obj *example.Pod) *example.Pod {
+		key := "pods/" + obj.Namespace + "/" + obj.Name
+		out := &example.Pod{}
+		err := store.Create(context.TODO(), key, obj, out, 0)
+		require.NoError(t, err)
+		return out
+	}
+	getPod := func(name, ns string) *example.Pod {
+		key := "pods/" + ns + "/" + name
+		out := &example.Pod{}
+		err := store.Get(context.TODO(), key, storage.GetOptions{}, out)
+		require.NoError(t, err)
+		return out
+	}
+
+	// create a pod and make sure its RV is equal to the one maintained by etcd
+	pod := createPod(makePod("pod-1"))
+	currentStorageRV, err := store.GetCurrentResourceVersion(context.TODO())
+	require.NoError(t, err)
+	podRV, err := store.versioner.ParseResourceVersion(pod.ResourceVersion)
+	require.NoError(t, err)
+	require.Equal(t, currentStorageRV, podRV, "expected the global etcd RV to be equal to pod's RV")
+
+	// now make unrelated write and make sure the target function returns global etcd RV
+	resp, err := store.client.KV.Put(ctx, "compact_rev_key", pod.ResourceVersion)
+	require.NoError(t, err)
+	currentStorageRV, err = store.GetCurrentResourceVersion(context.TODO())
+	require.NoError(t, err)
+	require.NoError(t, err)
+	require.Equal(t, currentStorageRV, uint64(resp.Header.Revision), "expected the global etcd RV to be equal to replicaset's RV")
+
+	// ensure that the pod's RV hasn't been changed
+	currentPod := getPod(pod.Name, pod.Namespace)
+	currentPodRV, err := store.versioner.ParseResourceVersion(currentPod.ResourceVersion)
+	require.NoError(t, err)
+	require.Equal(t, currentPodRV, podRV, "didn't expect to see the pod's RV changed")
+}

staging/src/k8s.io/apiserver/pkg/storage/interfaces.go

@@ -262,6 +262,10 @@ type Interface interface {
 	// TODO: Remove when storage.Interface will be separate from etc3.store.
 	// Deprecated: Added temporarily to simplify exposing RequestProgress for watch cache.
 	RequestWatchProgress(ctx context.Context) error
+
+	// GetCurrentResourceVersion gets the current resource version from etcd.
+	// This method issues an empty list request and reads only the ResourceVersion from the object metadata
+	GetCurrentResourceVersion(ctx context.Context) (uint64, error)
 }
 
 // GetOptions provides the options that may be provided for storage get operations.

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

@@ -1532,7 +1532,7 @@ func RunWatchSemantics(ctx context.Context, t *testing.T, store storage.Interfac
 			}
 
 			if scenario.useCurrentRV {
-				currentStorageRV, err := storage.GetCurrentResourceVersionFromStorage(ctx, store, func() runtime.Object { return &example.PodList{} }, "/pods", "")
+				currentStorageRV, err := store.GetCurrentResourceVersion(ctx)
 				require.NoError(t, err)
 				scenario.resourceVersion = fmt.Sprintf("%d", currentStorageRV)
 			}

staging/src/k8s.io/apiserver/pkg/storage/util.go

@@ -17,16 +17,12 @@ limitations under the License.
 package storage
 
 import (
-	"context"
 	"fmt"
-	"strconv"
 	"sync/atomic"
 
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/api/validation/path"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/fields"
-	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 )
 
@@ -81,45 +77,6 @@ func (hwm *HighWaterMark) Update(current int64) bool {
 	}
 }
 
-// GetCurrentResourceVersionFromStorage gets the current resource version from the underlying storage engine.
-// This method issues an empty list request and reads only the ResourceVersion from the object metadata
-func GetCurrentResourceVersionFromStorage(ctx context.Context, storage Interface, newListFunc func() runtime.Object, resourcePrefix, objectType string) (uint64, error) {
-	if storage == nil {
-		return 0, fmt.Errorf("storage wasn't provided for %s", objectType)
-	}
-	if newListFunc == nil {
-		return 0, fmt.Errorf("newListFunction wasn't provided for %s", objectType)
-	}
-	emptyList := newListFunc()
-	pred := SelectionPredicate{
-		Label: labels.Everything(),
-		Field: fields.Everything(),
-		Limit: 1, // just in case we actually hit something
-	}
-
-	err := storage.GetList(ctx, resourcePrefix, ListOptions{Predicate: pred}, emptyList)
-	if err != nil {
-		return 0, err
-	}
-	emptyListAccessor, err := meta.ListAccessor(emptyList)
-	if err != nil {
-		return 0, err
-	}
-	if emptyListAccessor == nil {
-		return 0, fmt.Errorf("unable to extract a list accessor from %T", emptyList)
-	}
-
-	currentResourceVersion, err := strconv.Atoi(emptyListAccessor.GetResourceVersion())
-	if err != nil {
-		return 0, err
-	}
-
-	if currentResourceVersion == 0 {
-		return 0, fmt.Errorf("the current resource version must be greater than 0")
-	}
-	return uint64(currentResourceVersion), nil
-}
-
 // AnnotateInitialEventsEndBookmark adds a special annotation to the given object
 // which indicates that the initial events have been sent.
 //