Refactor watchcache to support bookmark

staging/src/k8s.io/apiserver/pkg/registry/generic/registry/storage_factory.go

@@ -34,19 +34,21 @@ import (
 func StorageWithCacher(capacity int) generic.StorageDecorator {
 	return func(
 		storageConfig *storagebackend.Config,
-		objectType runtime.Object,
 		resourcePrefix string,
 		keyFunc func(obj runtime.Object) (string, error),
+		newFunc func() runtime.Object,
 		newListFunc func() runtime.Object,
 		getAttrsFunc storage.AttrFunc,
 		triggerFunc storage.TriggerPublisherFunc) (storage.Interface, factory.DestroyFunc) {
 
 		s, d := generic.NewRawStorage(storageConfig)
 		if capacity <= 0 {
-			klog.V(5).Infof("Storage caching is disabled for %T", objectType)
+			klog.V(5).Infof("Storage caching is disabled for %T", newFunc())
 			return s, d
 		}
-		klog.V(5).Infof("Storage caching is enabled for %T with capacity %v", objectType, capacity)
+		if klog.V(5) {
+			klog.Infof("Storage caching is enabled for %T with capacity %v", newFunc(), capacity)
+		}
 
 		// TODO: we would change this later to make storage always have cacher and hide low level KV layer inside.
 		// Currently it has two layers of same storage interface -- cacher and low level kv.
@@ -54,9 +56,9 @@ func StorageWithCacher(capacity int) generic.StorageDecorator {
 			CacheCapacity:        capacity,
 			Storage:              s,
 			Versioner:            etcdstorage.APIObjectVersioner{},
-			Type:                 objectType,
 			ResourcePrefix:       resourcePrefix,
 			KeyFunc:              keyFunc,
+			NewFunc:              newFunc,
 			NewListFunc:          newListFunc,
 			GetAttrsFunc:         getAttrsFunc,
 			TriggerPublisherFunc: triggerFunc,

staging/src/k8s.io/apiserver/pkg/registry/generic/registry/store.go

@@ -1288,9 +1288,9 @@ func (e *Store) CompleteWithOptions(options *generic.StoreOptions) error {
 		e.Storage.Codec = opts.StorageConfig.Codec
 		e.Storage.Storage, e.DestroyFunc = opts.Decorator(
 			opts.StorageConfig,
-			e.NewFunc(),
 			prefix,
 			keyFunc,
+			e.NewFunc,
 			e.NewListFunc,
 			attrFunc,
 			triggerFunc,

staging/src/k8s.io/apiserver/pkg/registry/generic/registry/store_test.go

@@ -1554,10 +1554,10 @@ func newTestGenericStoreRegistry(t *testing.T, scheme *runtime.Scheme, hasCacheE
 			CacheCapacity:  10,
 			Storage:        s,
 			Versioner:      etcdstorage.APIObjectVersioner{},
-			Type:           &example.Pod{},
 			ResourcePrefix: podPrefix,
 			KeyFunc:        func(obj runtime.Object) (string, error) { return storage.NoNamespaceKeyFunc(podPrefix, obj) },
 			GetAttrsFunc:   getPodAttrs,
+			NewFunc:        func() runtime.Object { return &example.Pod{} },
 			NewListFunc:    func() runtime.Object { return &example.PodList{} },
 			Codec:          sc.Codec,
 		}

staging/src/k8s.io/apiserver/pkg/registry/generic/storage_decorator.go

@@ -28,9 +28,9 @@ import (
 // and an associated DestroyFunc from given parameters.
 type StorageDecorator func(
 	config *storagebackend.Config,
-	objectType runtime.Object,
 	resourcePrefix string,
 	keyFunc func(obj runtime.Object) (string, error),
+	newFunc func() runtime.Object,
 	newListFunc func() runtime.Object,
 	getAttrsFunc storage.AttrFunc,
 	trigger storage.TriggerPublisherFunc) (storage.Interface, factory.DestroyFunc)
@@ -39,9 +39,9 @@ type StorageDecorator func(
 // without any decoration.
 func UndecoratedStorage(
 	config *storagebackend.Config,
-	objectType runtime.Object,
 	resourcePrefix string,
 	keyFunc func(obj runtime.Object) (string, error),
+	newFunc func() runtime.Object,
 	newListFunc func() runtime.Object,
 	getAttrsFunc storage.AttrFunc,
 	trigger storage.TriggerPublisherFunc) (storage.Interface, factory.DestroyFunc) {

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

@@ -73,7 +73,6 @@ type Config struct {
 
 	// The Cache will be caching objects of a given Type and assumes that they
 	// are all stored under ResourcePrefix directory in the underlying database.
-	Type           interface{}
 	ResourcePrefix string
 
 	// KeyFunc is used to get a key in the underlying storage for a given object.
@@ -86,6 +85,9 @@ type Config struct {
 	// needs to process an incoming event.
 	TriggerPublisherFunc storage.TriggerPublisherFunc
 
+	// NewFunc is a function that creates new empty object storing a object of type Type.
+	NewFunc func() runtime.Object
+
 	// NewList is a function that creates new empty object storing a list of
 	// objects of type Type.
 	NewListFunc func() runtime.Object
@@ -230,21 +232,20 @@ func NewCacherFromConfig(config Config) *Cacher {
 	listerWatcher := newCacherListerWatcher(config.Storage, config.ResourcePrefix, config.NewListFunc)
 	reflectorName := "storage/cacher.go:" + config.ResourcePrefix
 
+	obj := config.NewFunc()
 	// Give this error when it is constructed rather than when you get the
 	// first watch item, because it's much easier to track down that way.
-	if obj, ok := config.Type.(runtime.Object); ok {
-		if err := runtime.CheckCodec(config.Codec, obj); err != nil {
-			panic("storage codec doesn't seem to match given type: " + err.Error())
-		}
+	if err := runtime.CheckCodec(config.Codec, obj); err != nil {
+		panic("storage codec doesn't seem to match given type: " + err.Error())
 	}
 
 	stopCh := make(chan struct{})
 	cacher := &Cacher{
 		ready:       newReady(),
 		storage:     config.Storage,
-		objectType:  reflect.TypeOf(config.Type),
+		objectType:  reflect.TypeOf(obj),
 		watchCache:  watchCache,
-		reflector:   cache.NewNamedReflector(reflectorName, listerWatcher, config.Type, watchCache, 0),
+		reflector:   cache.NewNamedReflector(reflectorName, listerWatcher, obj, watchCache, 0),
 		versioner:   config.Versioner,
 		triggerFunc: config.TriggerPublisherFunc,
 		watcherIdx:  0,

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

@@ -252,10 +252,10 @@ func newTestCacher(s storage.Interface, cap int) (*Cacher, storage.Versioner) {
 		CacheCapacity:  cap,
 		Storage:        s,
 		Versioner:      testVersioner{},
-		Type:           &example.Pod{},
 		ResourcePrefix: prefix,
 		KeyFunc:        func(obj runtime.Object) (string, error) { return storage.NamespaceKeyFunc(prefix, obj) },
 		GetAttrsFunc:   func(obj runtime.Object) (labels.Set, fields.Set, error) { return nil, nil, nil },
+		NewFunc:        func() runtime.Object { return &example.Pod{} },
 		NewListFunc:    func() runtime.Object { return &example.PodList{} },
 		Codec:          codecs.LegacyCodec(examplev1.SchemeGroupVersion),
 	}

staging/src/k8s.io/apiserver/pkg/storage/tests/cacher_test.go

@@ -104,10 +104,10 @@ func newTestCacher(s storage.Interface, cap int) (*cacherstorage.Cacher, storage
 		CacheCapacity:  cap,
 		Storage:        s,
 		Versioner:      v,
-		Type:           &example.Pod{},
 		ResourcePrefix: prefix,
 		KeyFunc:        func(obj runtime.Object) (string, error) { return storage.NamespaceKeyFunc(prefix, obj) },
 		GetAttrsFunc:   GetAttrs,
+		NewFunc:        func() runtime.Object { return &example.Pod{} },
 		NewListFunc:    func() runtime.Object { return &example.PodList{} },
 		Codec:          codecs.LegacyCodec(examplev1.SchemeGroupVersion),
 	}