Merge pull request kubernetes#80160 from wojtek-t/propagate_error_from_cacher_creation

k8s-ci-robot · web-flow · commit 13af8f286de9 · 2019-07-15T16:33:49.000-07:00
Propagate error from creating cacher and storage decorators up
diff --git a/pkg/registry/core/pod/rest/log_test.go b/pkg/registry/core/pod/rest/log_test.go
@@ -30,7 +30,10 @@ import (
 func TestPodLogValidates(t *testing.T) {
 	config, server := registrytest.NewEtcdStorage(t, "")
 	defer server.Terminate(t)
-	s, destroyFunc := generic.NewRawStorage(config)
+	s, destroyFunc, err := generic.NewRawStorage(config)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
 	defer destroyFunc()
 	store := &genericregistry.Store{
 		Storage: genericregistry.DryRunnableStorage{Storage: s},
@@ -46,7 +49,7 @@ func TestPodLogValidates(t *testing.T) {
 	for _, tc := range testCases {
 		_, err := logRest.Get(genericapirequest.NewDefaultContext(), "test", tc)
 		if !errors.IsInvalid(err) {
-			t.Fatalf("unexpected error: %v", err)
+			t.Fatalf("Unexpected error: %v", err)
 		}
 	}
 }
diff --git a/pkg/registry/core/service/allocator/storage/storage.go b/pkg/registry/core/service/allocator/storage/storage.go
@@ -61,7 +61,10 @@ var _ rangeallocation.RangeRegistry = &Etcd{}
 // NewEtcd returns an allocator that is backed by Etcd and can manage
 // persisting the snapshot state of allocation after each allocation is made.
 func NewEtcd(alloc allocator.Snapshottable, baseKey string, resource schema.GroupResource, config *storagebackend.Config) *Etcd {
-	storage, d := generic.NewRawStorage(config)
+	storage, d, err := generic.NewRawStorage(config)
+	if err != nil {
+		panic(err) // TODO: Propagate error up
+	}
 
 	// TODO : Remove RegisterStorageCleanup below when PR
 	// https://github.com/kubernetes/kubernetes/pull/50690
diff --git a/pkg/registry/core/service/ipallocator/storage/storage_test.go b/pkg/registry/core/service/ipallocator/storage/storage_test.go
@@ -47,7 +47,10 @@ func newStorage(t *testing.T) (*etcd3testing.EtcdTestServer, ipallocator.Interfa
 		etcd := allocatorstore.NewEtcd(mem, "/ranges/serviceips", api.Resource("serviceipallocations"), etcdStorage)
 		return etcd
 	})
-	s, d := generic.NewRawStorage(etcdStorage)
+	s, d, err := generic.NewRawStorage(etcdStorage)
+	if err != nil {
+		t.Fatalf("Couldn't create storage: %v", err)
+	}
 	destroyFunc := func() {
 		d()
 		server.Terminate(t)
diff --git a/pkg/registry/extensions/controller/storage/storage_test.go b/pkg/registry/extensions/controller/storage/storage_test.go
@@ -34,7 +34,10 @@ import (
 func newStorage(t *testing.T) (*ScaleREST, *etcd3testing.EtcdTestServer, storage.Interface, factory.DestroyFunc) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, "")
 	restOptions := generic.RESTOptions{StorageConfig: etcdStorage, Decorator: generic.UndecoratedStorage, DeleteCollectionWorkers: 1, ResourcePrefix: "controllers"}
-	s, d := generic.NewRawStorage(etcdStorage)
+	s, d, err := generic.NewRawStorage(etcdStorage)
+	if err != nil {
+		t.Fatalf("Couldn't create storage: %v", err)
+	}
 	destroyFunc := func() {
 		d()
 		server.Terminate(t)
diff --git a/staging/src/k8s.io/apiserver/pkg/registry/generic/BUILD b/staging/src/k8s.io/apiserver/pkg/registry/generic/BUILD
@@ -23,7 +23,6 @@ go_library(
         "//staging/src/k8s.io/apiserver/pkg/storage:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/storage/storagebackend:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/storage/storagebackend/factory:go_default_library",
-        "//vendor/k8s.io/klog:go_default_library",
     ],
 )
 
diff --git a/staging/src/k8s.io/apiserver/pkg/registry/generic/registry/storage_factory.go b/staging/src/k8s.io/apiserver/pkg/registry/generic/registry/storage_factory.go
@@ -39,12 +39,15 @@ func StorageWithCacher(capacity int) generic.StorageDecorator {
 		newFunc func() runtime.Object,
 		newListFunc func() runtime.Object,
 		getAttrsFunc storage.AttrFunc,
-		triggerFunc storage.TriggerPublisherFunc) (storage.Interface, factory.DestroyFunc) {
+		triggerFunc storage.TriggerPublisherFunc) (storage.Interface, factory.DestroyFunc, error) {
 
-		s, d := generic.NewRawStorage(storageConfig)
+		s, d, err := generic.NewRawStorage(storageConfig)
+		if err != nil {
+			return s, d, err
+		}
 		if capacity <= 0 {
 			klog.V(5).Infof("Storage caching is disabled for %T", newFunc())
-			return s, d
+			return s, d, nil
 		}
 		if klog.V(5) {
 			klog.Infof("Storage caching is enabled for %T with capacity %v", newFunc(), capacity)
@@ -64,7 +67,10 @@ func StorageWithCacher(capacity int) generic.StorageDecorator {
 			TriggerPublisherFunc: triggerFunc,
 			Codec:                storageConfig.Codec,
 		}
-		cacher := cacherstorage.NewCacherFromConfig(cacherConfig)
+		cacher, err := cacherstorage.NewCacherFromConfig(cacherConfig)
+		if err != nil {
+			return nil, func() {}, err
+		}
 		destroyFunc := func() {
 			cacher.Stop()
 			d()
@@ -75,7 +81,7 @@ func StorageWithCacher(capacity int) generic.StorageDecorator {
 		// merges as that shuts down storage properly
 		RegisterStorageCleanup(destroyFunc)
 
-		return cacher, destroyFunc
+		return cacher, destroyFunc, nil
 	}
 }
 
diff --git a/staging/src/k8s.io/apiserver/pkg/registry/generic/registry/store.go b/staging/src/k8s.io/apiserver/pkg/registry/generic/registry/store.go
@@ -1310,7 +1310,8 @@ func (e *Store) CompleteWithOptions(options *generic.StoreOptions) error {
 
 	if e.Storage.Storage == nil {
 		e.Storage.Codec = opts.StorageConfig.Codec
-		e.Storage.Storage, e.DestroyFunc = opts.Decorator(
+		var err error
+		e.Storage.Storage, e.DestroyFunc, err = opts.Decorator(
 			opts.StorageConfig,
 			prefix,
 			keyFunc,
@@ -1319,6 +1320,9 @@ func (e *Store) CompleteWithOptions(options *generic.StoreOptions) error {
 			attrFunc,
 			triggerFunc,
 		)
+		if err != nil {
+			return err
+		}
 		e.StorageVersioner = opts.StorageConfig.EncodeVersioner
 
 		if opts.CountMetricPollPeriod > 0 {
diff --git a/staging/src/k8s.io/apiserver/pkg/registry/generic/registry/store_test.go b/staging/src/k8s.io/apiserver/pkg/registry/generic/registry/store_test.go
@@ -1561,7 +1561,10 @@ func newTestGenericStoreRegistry(t *testing.T, scheme *runtime.Scheme, hasCacheE
 			NewListFunc:    func() runtime.Object { return &example.PodList{} },
 			Codec:          sc.Codec,
 		}
-		cacher := cacherstorage.NewCacherFromConfig(config)
+		cacher, err := cacherstorage.NewCacherFromConfig(config)
+		if err != nil {
+			t.Fatalf("Couldn't create cacher: %v", err)
+		}
 		d := destroyFunc
 		s = cacher
 		destroyFunc = func() {
diff --git a/staging/src/k8s.io/apiserver/pkg/registry/generic/storage_decorator.go b/staging/src/k8s.io/apiserver/pkg/registry/generic/storage_decorator.go
@@ -21,7 +21,6 @@ import (
 	"k8s.io/apiserver/pkg/storage"
 	"k8s.io/apiserver/pkg/storage/storagebackend"
 	"k8s.io/apiserver/pkg/storage/storagebackend/factory"
-	"k8s.io/klog"
 )
 
 // StorageDecorator is a function signature for producing a storage.Interface
@@ -33,7 +32,7 @@ type StorageDecorator func(
 	newFunc func() runtime.Object,
 	newListFunc func() runtime.Object,
 	getAttrsFunc storage.AttrFunc,
-	trigger storage.TriggerPublisherFunc) (storage.Interface, factory.DestroyFunc)
+	trigger storage.TriggerPublisherFunc) (storage.Interface, factory.DestroyFunc, error)
 
 // UndecoratedStorage returns the given a new storage from the given config
 // without any decoration.
@@ -44,17 +43,13 @@ func UndecoratedStorage(
 	newFunc func() runtime.Object,
 	newListFunc func() runtime.Object,
 	getAttrsFunc storage.AttrFunc,
-	trigger storage.TriggerPublisherFunc) (storage.Interface, factory.DestroyFunc) {
+	trigger storage.TriggerPublisherFunc) (storage.Interface, factory.DestroyFunc, error) {
 	return NewRawStorage(config)
 }
 
 // NewRawStorage creates the low level kv storage. This is a work-around for current
 // two layer of same storage interface.
 // TODO: Once cacher is enabled on all registries (event registry is special), we will remove this method.
-func NewRawStorage(config *storagebackend.Config) (storage.Interface, factory.DestroyFunc) {
-	s, d, err := factory.Create(*config)
-	if err != nil {
-		klog.Fatalf("Unable to create storage backend: config (%v), err (%v)", config, err)
-	}
-	return s, d
+func NewRawStorage(config *storagebackend.Config) (storage.Interface, factory.DestroyFunc, error) {
+	return factory.Create(*config)
 }
diff --git a/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go b/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go
@@ -291,13 +291,13 @@ type Cacher struct {
 // NewCacherFromConfig creates a new Cacher responsible for servicing WATCH and LIST requests from
 // its internal cache and updating its cache in the background based on the
 // given configuration.
-func NewCacherFromConfig(config Config) *Cacher {
+func NewCacherFromConfig(config Config) (*Cacher, error) {
 	stopCh := make(chan struct{})
 	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 err := runtime.CheckCodec(config.Codec, obj); err != nil {
-		panic("storage codec doesn't seem to match given type: " + err.Error())
+		return nil, fmt.Errorf("storage codec doesn't seem to match given type: %v", err)
 	}
 
 	clock := clock.RealClock{}
@@ -363,7 +363,7 @@ func NewCacherFromConfig(config Config) *Cacher {
 		)
 	}()
 
-	return cacher
+	return cacher, nil
 }
 
 func (c *Cacher) startCaching(stopChannel <-chan struct{}) {
diff --git a/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_whitebox_test.go b/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_whitebox_test.go
@@ -257,7 +257,7 @@ func init() {
 	utilruntime.Must(examplev1.AddToScheme(scheme))
 }
 
-func newTestCacher(s storage.Interface, cap int) (*Cacher, storage.Versioner) {
+func newTestCacher(s storage.Interface, cap int) (*Cacher, storage.Versioner, error) {
 	prefix := "pods"
 	config := Config{
 		CacheCapacity:  cap,
@@ -270,7 +270,8 @@ func newTestCacher(s storage.Interface, cap int) (*Cacher, storage.Versioner) {
 		NewListFunc:    func() runtime.Object { return &example.PodList{} },
 		Codec:          codecs.LegacyCodec(examplev1.SchemeGroupVersion),
 	}
-	return NewCacherFromConfig(config), testVersioner{}
+	cacher, err := NewCacherFromConfig(config)
+	return cacher, testVersioner{}, err
 }
 
 type dummyStorage struct {
@@ -328,7 +329,10 @@ func (d *dummyStorage) Count(_ string) (int64, error) {
 
 func TestListWithLimitAndRV0(t *testing.T) {
 	backingStorage := &dummyStorage{}
-	cacher, _ := newTestCacher(backingStorage, 0)
+	cacher, _, err := newTestCacher(backingStorage, 0)
+	if err != nil {
+		t.Fatalf("Couldn't create cacher: %v", err)
+	}
 	defer cacher.Stop()
 
 	pred := storage.SelectionPredicate{
@@ -341,7 +345,7 @@ func TestListWithLimitAndRV0(t *testing.T) {
 
 	// Inject error to underlying layer and check if cacher is not bypassed.
 	backingStorage.err = errDummy
-	err := cacher.List(context.TODO(), "pods/ns", "0", pred, result)
+	err = cacher.List(context.TODO(), "pods/ns", "0", pred, result)
 	if err != nil {
 		t.Errorf("List with Limit and RV=0 should be served from cache: %v", err)
 	}
@@ -354,7 +358,10 @@ func TestListWithLimitAndRV0(t *testing.T) {
 
 func TestGetToListWithLimitAndRV0(t *testing.T) {
 	backingStorage := &dummyStorage{}
-	cacher, _ := newTestCacher(backingStorage, 0)
+	cacher, _, err := newTestCacher(backingStorage, 0)
+	if err != nil {
+		t.Fatalf("Couldn't create cacher: %v", err)
+	}
 	defer cacher.Stop()
 
 	pred := storage.SelectionPredicate{
@@ -367,7 +374,7 @@ func TestGetToListWithLimitAndRV0(t *testing.T) {
 
 	// Inject error to underlying layer and check if cacher is not bypassed.
 	backingStorage.err = errDummy
-	err := cacher.GetToList(context.TODO(), "pods/ns", "0", pred, result)
+	err = cacher.GetToList(context.TODO(), "pods/ns", "0", pred, result)
 	if err != nil {
 		t.Errorf("GetToList with Limit and RV=0 should be served from cache: %v", err)
 	}
@@ -380,7 +387,10 @@ func TestGetToListWithLimitAndRV0(t *testing.T) {
 
 func TestWatcherNotGoingBackInTime(t *testing.T) {
 	backingStorage := &dummyStorage{}
-	cacher, _ := newTestCacher(backingStorage, 1000)
+	cacher, _, err := newTestCacher(backingStorage, 1000)
+	if err != nil {
+		t.Fatalf("Couldn't create cacher: %v", err)
+	}
 	defer cacher.Stop()
 
 	// Wait until cacher is initialized.
@@ -498,7 +508,10 @@ func TestCacheWatcherStoppedInAnotherGoroutine(t *testing.T) {
 
 func TestCacheWatcherStoppedOnDestroy(t *testing.T) {
 	backingStorage := &dummyStorage{}
-	cacher, _ := newTestCacher(backingStorage, 1000)
+	cacher, _, err := newTestCacher(backingStorage, 1000)
+	if err != nil {
+		t.Fatalf("Couldn't create cacher: %v", err)
+	}
 	defer cacher.Stop()
 
 	// Wait until cacher is initialized.
@@ -577,7 +590,10 @@ func TestTimeBucketWatchersBasic(t *testing.T) {
 func testCacherSendBookmarkEvents(t *testing.T, watchCacheEnabled, allowWatchBookmarks, expectedBookmarks bool) {
 	defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WatchBookmark, watchCacheEnabled)()
 	backingStorage := &dummyStorage{}
-	cacher, _ := newTestCacher(backingStorage, 1000)
+	cacher, _, err := newTestCacher(backingStorage, 1000)
+	if err != nil {
+		t.Fatalf("Couldn't create cacher: %v", err)
+	}
 	defer cacher.Stop()
 
 	// Wait until cacher is initialized.
@@ -676,7 +692,10 @@ func TestCacherSendBookmarkEvents(t *testing.T) {
 func TestDispatchingBookmarkEventsWithConcurrentStop(t *testing.T) {
 	defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WatchBookmark, true)()
 	backingStorage := &dummyStorage{}
-	cacher, _ := newTestCacher(backingStorage, 1000)
+	cacher, _, err := newTestCacher(backingStorage, 1000)
+	if err != nil {
+		t.Fatalf("Couldn't create cacher: %v", err)
+	}
 	defer cacher.Stop()
 
 	// Wait until cacher is initialized.
@@ -686,7 +705,7 @@ func TestDispatchingBookmarkEventsWithConcurrentStop(t *testing.T) {
 	cacher.dispatchTimeoutBudget.returnUnused(100 * time.Millisecond)
 
 	resourceVersion := uint64(1000)
-	err := cacher.watchCache.Add(&examplev1.Pod{
+	err = cacher.watchCache.Add(&examplev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:            fmt.Sprintf("pod-0"),
 			Namespace:       "ns",
@@ -746,7 +765,10 @@ func TestDispatchingBookmarkEventsWithConcurrentStop(t *testing.T) {
 
 func TestDispatchEventWillNotBeBlockedByTimedOutWatcher(t *testing.T) {
 	backingStorage := &dummyStorage{}
-	cacher, _ := newTestCacher(backingStorage, 1000)
+	cacher, _, err := newTestCacher(backingStorage, 1000)
+	if err != nil {
+		t.Fatalf("Couldn't create cacher: %v", err)
+	}
 	defer cacher.Stop()
 
 	// Wait until cacher is initialized.
diff --git a/staging/src/k8s.io/apiserver/pkg/storage/tests/cacher_test.go b/staging/src/k8s.io/apiserver/pkg/storage/tests/cacher_test.go

Original file line number	Diff line number	Diff line change
`@@ -23,7 +23,6 @@ go_library(`
`23`	`23`	`"//staging/src/k8s.io/apiserver/pkg/storage:go_default_library",`
`24`	`24`	`"//staging/src/k8s.io/apiserver/pkg/storage/storagebackend:go_default_library",`
`25`	`25`	`"//staging/src/k8s.io/apiserver/pkg/storage/storagebackend/factory:go_default_library",`
`26`		`- "//vendor/k8s.io/klog:go_default_library",`
`27`	`26`	`],`
`28`	`27`	`)`
`29`	`28`