Merge pull request #143320 from wojtek-t/remove_decorated_watcher_store

Remove decoratedWatcher and Store.Decorator

pkg/registry/core/service/storage/storage.go

@@ -331,9 +331,7 @@ func (r *REST) defaultOnReadIPFamilies(service *api.Service) {
 func (r *REST) afterDelete(obj runtime.Object, options *metav1.DeleteOptions) {
 	svc := obj.(*api.Service)
 
-	// Normally this defaulting is done automatically, but the hook (Decorator)
-	// is called at the end of this process, and we want the fully-formed
-	// object.
+	// Make sure the object has all fields we expect to be defaulted.
 	r.defaultOnReadService(svc)
 
 	// Only perform the cleanup if this is a non-dryrun deletion
@@ -376,8 +374,7 @@ func (r *REST) beginUpdate(ctx context.Context, obj, oldObj runtime.Object, opti
 	oldSvc := oldObj.(*api.Service)
 
 	// Make sure the existing object has all fields we expect to be defaulted.
-	// This might not be true if the saved object predates these fields (the
-	// Decorator hook is not called on 'old' in the update path.
+	// This might not be true if the saved object predates these fields.
 	r.defaultOnReadService(oldSvc)
 
 	// Fix up allocated values that the client may have not specified (for

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

@@ -1,91 +0,0 @@
-/*
-Copyright 2016 The Kubernetes Authors.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package registry
-
-import (
-	"context"
-	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/apimachinery/pkg/watch"
-)
-
-type decoratedWatcher struct {
-	w         watch.Interface
-	decorator func(runtime.Object)
-	cancel    context.CancelFunc
-	resultCh  chan watch.Event
-}
-
-func newDecoratedWatcher(ctx context.Context, w watch.Interface, decorator func(runtime.Object)) *decoratedWatcher {
-	ctx, cancel := context.WithCancel(ctx)
-	d := &decoratedWatcher{
-		w:         w,
-		decorator: decorator,
-		cancel:    cancel,
-		resultCh:  make(chan watch.Event),
-	}
-	go d.run(ctx)
-	return d
-}
-
-// run decorates watch events from the underlying watcher until its result channel
-// is closed or the passed in context is done.
-// When run() returns, decoratedWatcher#resultCh is closed.
-func (d *decoratedWatcher) run(ctx context.Context) {
-	var recv, send watch.Event
-	var ok bool
-	defer close(d.resultCh)
-	for {
-		select {
-		case recv, ok = <-d.w.ResultChan():
-			if !ok {
-				// The underlying channel was closed, cancel our context
-				d.cancel()
-				return
-			}
-			switch recv.Type {
-			case watch.Added, watch.Modified, watch.Deleted, watch.Bookmark:
-				d.decorator(recv.Object)
-				send = recv
-			case watch.Error:
-				send = recv
-			}
-			select {
-			case d.resultCh <- send:
-				// propagated event successfully
-			case <-ctx.Done():
-				// context timed out or was cancelled, stop the underlying watcher
-				d.w.Stop()
-				return
-			}
-		case <-ctx.Done():
-			// context timed out or was cancelled, stop the underlying watcher
-			d.w.Stop()
-			return
-		}
-	}
-}
-
-func (d *decoratedWatcher) Stop() {
-	// stop the underlying watcher
-	d.w.Stop()
-	// cancel our context
-	d.cancel()
-}
-
-func (d *decoratedWatcher) ResultChan() <-chan watch.Event {
-	return d.resultCh
-}

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

@@ -1,108 +0,0 @@
-/*
-Copyright 2016 The Kubernetes Authors.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package registry
-
-import (
-	"context"
-	"testing"
-	"time"
-
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/apimachinery/pkg/util/wait"
-	"k8s.io/apimachinery/pkg/watch"
-	"k8s.io/apiserver/pkg/apis/example"
-)
-
-func TestDecoratedWatcher(t *testing.T) {
-	w := watch.NewFake()
-	decorator := func(obj runtime.Object) {
-		if pod, ok := obj.(*example.Pod); ok {
-			pod.Annotations = map[string]string{"decorated": "true"}
-		}
-	}
-	ctx, cancel := context.WithCancel(context.Background())
-	dw := newDecoratedWatcher(ctx, w, decorator)
-	defer dw.Stop()
-
-	go func() {
-		w.Error(&metav1.Status{Status: "Failure"})
-		w.Add(&example.Pod{ObjectMeta: metav1.ObjectMeta{Name: "foo"}})
-		w.Error(&metav1.Status{Status: "Failure"})
-		w.Modify(&example.Pod{ObjectMeta: metav1.ObjectMeta{Name: "foo"}})
-		w.Error(&metav1.Status{Status: "Failure"})
-		w.Delete(&example.Pod{ObjectMeta: metav1.ObjectMeta{Name: "foo"}})
-	}()
-
-	expectErrorEvent(t, dw) // expect error is plumbed and doesn't force close the watcher
-	expectPodEvent(t, dw, watch.Added)
-	expectErrorEvent(t, dw) // expect error is plumbed and doesn't force close the watcher
-	expectPodEvent(t, dw, watch.Modified)
-	expectErrorEvent(t, dw) // expect error is plumbed and doesn't force close the watcher
-	expectPodEvent(t, dw, watch.Deleted)
-
-	// cancel the passed-in context to simulate request timeout
-	cancel()
-
-	// expect the decorated channel to be closed
-	select {
-	case e, ok := <-dw.ResultChan():
-		if ok {
-			t.Errorf("expected result chan closed, got %#v", e)
-		}
-	case <-time.After(wait.ForeverTestTimeout):
-		t.Errorf("timeout after %v", wait.ForeverTestTimeout)
-	}
-
-	// expect the underlying watcher to have been stopped as a result of the context cancellation
-	if !w.IsStopped() {
-		t.Errorf("expected underlying watcher to be stopped")
-	}
-}
-
-func expectPodEvent(t *testing.T, dw *decoratedWatcher, watchType watch.EventType) {
-	select {
-	case e := <-dw.ResultChan():
-		pod, ok := e.Object.(*example.Pod)
-		if !ok {
-			t.Fatalf("Should received object of type *api.Pod, get type (%T)", e.Object)
-		}
-		if pod.Annotations["decorated"] != "true" {
-			t.Fatalf("pod.Annotations[\"decorated\"], want=%s, get=%s", "true", pod.Labels["decorated"])
-		}
-		if e.Type != watchType {
-			t.Fatalf("expected type %s, got %s", watchType, e.Type)
-		}
-	case <-time.After(wait.ForeverTestTimeout):
-		t.Fatalf("timeout after %v", wait.ForeverTestTimeout)
-	}
-}
-
-func expectErrorEvent(t *testing.T, dw *decoratedWatcher) {
-	select {
-	case e := <-dw.ResultChan():
-		_, ok := e.Object.(*metav1.Status)
-		if !ok {
-			t.Fatalf("Should received object of type *metav1.Status, get type (%T)", e.Object)
-		}
-		if e.Type != watch.Error {
-			t.Fatalf("expected type %s, got %s", watch.Error, e.Type)
-		}
-	case <-time.After(wait.ForeverTestTimeout):
-		t.Fatalf("timeout after %v", wait.ForeverTestTimeout)
-	}
-}

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

@@ -176,42 +176,34 @@ type Store struct {
 	// are issued in parallel.
 	DeleteCollectionWorkers int
 
-	// Decorator is an optional exit hook on an object returned from the
-	// underlying storage. The returned object could be an individual object
-	// (e.g. Pod) or a list type (e.g. PodList). Decorator is intended for
-	// integrations that are above storage and should only be used for
-	// specific cases where storage of the value is not appropriate, since
-	// they cannot be watched.
-	Decorator func(runtime.Object)
-
 	// CreateStrategy implements resource-specific behavior during creation.
 	CreateStrategy rest.RESTCreateStrategy
 	// BeginCreate is an optional hook that returns a "transaction-like"
 	// commit/revert function which will be called at the end of the operation,
-	// but before AfterCreate and Decorator, indicating via the argument
+	// but before AfterCreate, indicating via the argument
 	// whether the operation succeeded.  If this returns an error, the function
 	// is not called.  Almost nobody should use this hook.
 	BeginCreate BeginCreateFunc
 	// AfterCreate implements a further operation to run after a resource is
-	// created and before it is decorated, optional.
+	// created, optional.
 	AfterCreate AfterCreateFunc
 
 	// UpdateStrategy implements resource-specific behavior during updates.
 	UpdateStrategy rest.RESTUpdateStrategy
 	// BeginUpdate is an optional hook that returns a "transaction-like"
 	// commit/revert function which will be called at the end of the operation,
-	// but before AfterUpdate and Decorator, indicating via the argument
+	// but before AfterUpdate, indicating via the argument
 	// whether the operation succeeded.  If this returns an error, the function
 	// is not called.  Almost nobody should use this hook.
 	BeginUpdate BeginUpdateFunc
 	// AfterUpdate implements a further operation to run after a resource is
-	// updated and before it is decorated, optional.
+	// updated, optional.
 	AfterUpdate AfterUpdateFunc
 
 	// DeleteStrategy implements resource-specific behavior during deletion.
 	DeleteStrategy rest.RESTDeleteStrategy
 	// AfterDelete implements a further operation to run after a resource is
-	// deleted and before it is decorated, optional.
+	// deleted, optional.
 	AfterDelete AfterDeleteFunc
 	// ReturnDeletedObject determines whether the Store returns the object
 	// that was deleted. Otherwise, return a generic success status response.
@@ -500,9 +492,6 @@ func (e *Store) List(ctx context.Context, options *metainternalversion.ListOptio
 	if err != nil {
 		return nil, err
 	}
-	if e.Decorator != nil {
-		e.Decorator(out)
-	}
 	return out, nil
 }
 
@@ -680,9 +669,6 @@ func (e *Store) create(ctx context.Context, obj runtime.Object, createValidation
 	if e.AfterCreate != nil {
 		e.AfterCreate(out, options)
 	}
-	if e.Decorator != nil {
-		e.Decorator(out)
-	}
 	return out, nil
 }
 
@@ -974,9 +960,6 @@ func (e *Store) Update(ctx context.Context, name string, objInfo rest.UpdatedObj
 			e.AfterUpdate(out, options)
 		}
 	}
-	if e.Decorator != nil {
-		e.Decorator(out)
-	}
 	return out, creating, nil
 }
 
@@ -1012,9 +995,6 @@ func (e *Store) Get(ctx context.Context, name string, options *metav1.GetOptions
 	if err := e.Storage.Get(ctx, key, storage.GetOptions{ResourceVersion: options.ResourceVersion}, obj); err != nil {
 		return nil, storeerr.InterpretGetError(err, e.qualifiedResourceFromContext(ctx), name)
 	}
-	if e.Decorator != nil {
-		e.Decorator(obj)
-	}
 	return obj, nil
 }
 
@@ -1547,15 +1527,12 @@ func (e *Store) DeleteCollection(ctx context.Context, deleteValidation rest.Vali
 }
 
 // finalizeDelete runs the Store's AfterDelete hook if runHooks is set and
-// returns the decorated deleted object if appropriate.
+// returns the deleted object if appropriate.
 func (e *Store) finalizeDelete(ctx context.Context, obj runtime.Object, runHooks bool, options *metav1.DeleteOptions) (runtime.Object, error) {
 	if runHooks && e.AfterDelete != nil {
 		e.AfterDelete(obj, options)
 	}
 	if e.ReturnDeletedObject {
-		if e.Decorator != nil {
-			e.Decorator(obj)
-		}
 		return obj, nil
 	}
 	// Return information about the deleted object, which enables clients to
@@ -1634,14 +1611,7 @@ func (e *Store) WatchPredicate(ctx context.Context, p storage.SelectionPredicate
 		// optimization is skipped
 	}
 
-	w, err := e.Storage.Watch(ctx, key, storageOpts)
-	if err != nil {
-		return nil, err
-	}
-	if e.Decorator != nil {
-		return newDecoratedWatcher(ctx, w, e.Decorator), nil
-	}
-	return w, nil
+	return e.Storage.Watch(ctx, key, storageOpts)
 }
 
 // calculateTTL is a helper for retrieving the updated TTL for an object or

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

@@ -947,7 +947,6 @@ func TestStoreCreateHooks(t *testing.T) {
 
 	testCases := []struct {
 		name        string
-		decorator   func(runtime.Object)
 		beginCreate BeginCreateFunc
 		afterCreate AfterCreateFunc
 		// the TTLFunc is an easy hook to force a failure
@@ -957,12 +956,6 @@ func TestStoreCreateHooks(t *testing.T) {
 		expectMilestones []string // to test sequence
 	}{{
 		name: "no hooks",
-	}, {
-		name: "Decorator mutation",
-		decorator: func(obj runtime.Object) {
-			setAnn(obj, "DecoratorWasCalled")
-		},
-		expectAnnotation: "DecoratorWasCalled",
 	}, {
 		name: "AfterCreate mutation",
 		afterCreate: func(obj runtime.Object, opts *metav1.CreateOptions) {
@@ -978,9 +971,6 @@ func TestStoreCreateHooks(t *testing.T) {
 		expectAnnotation: "BeginCreateWasCalled",
 	}, {
 		name: "success ordering",
-		decorator: func(obj runtime.Object) {
-			mile("Decorator")
-		},
 		afterCreate: func(obj runtime.Object, opts *metav1.CreateOptions) {
 			mile("AfterCreate")
 		},
@@ -990,12 +980,9 @@ func TestStoreCreateHooks(t *testing.T) {
 				mile(fmt.Sprintf("FinishCreate(%v)", success))
 			}, nil
 		},
-		expectMilestones: []string{"BeginCreate", "FinishCreate(true)", "AfterCreate", "Decorator"},
+		expectMilestones: []string{"BeginCreate", "FinishCreate(true)", "AfterCreate"},
 	}, {
 		name: "fail ordering",
-		decorator: func(obj runtime.Object) {
-			mile("Decorator")
-		},
 		afterCreate: func(obj runtime.Object, opts *metav1.CreateOptions) {
 			mile("AfterCreate")
 		},
@@ -1014,9 +1001,6 @@ func TestStoreCreateHooks(t *testing.T) {
 	}, {
 		name:        "fail BeginCreate ordering",
 		expectError: true,
-		decorator: func(obj runtime.Object) {
-			mile("Decorator")
-		},
 		afterCreate: func(obj runtime.Object, opts *metav1.CreateOptions) {
 			mile("AfterCreate")
 		},
@@ -1039,7 +1023,6 @@ func TestStoreCreateHooks(t *testing.T) {
 			testContext := genericapirequest.WithNamespace(genericapirequest.NewContext(), "test")
 			destroyFunc, registry := NewTestGenericStoreRegistry(t)
 			defer destroyFunc()
-			registry.Decorator = tc.decorator
 			registry.BeginCreate = tc.beginCreate
 			registry.AfterCreate = tc.afterCreate
 			registry.TTLFunc = tc.ttl
@@ -1254,8 +1237,7 @@ func TestStoreUpdateHooks(t *testing.T) {
 	}
 
 	testCases := []struct {
-		name      string
-		decorator func(runtime.Object)
+		name string
 		// create-on-update is tested elsewhere, but this proves non-use here
 		beginCreate      BeginCreateFunc
 		afterCreate      AfterCreateFunc
@@ -1266,12 +1248,6 @@ func TestStoreUpdateHooks(t *testing.T) {
 		expectMilestones []string // to test sequence
 	}{{
 		name: "no hooks",
-	}, {
-		name: "Decorator mutation",
-		decorator: func(obj runtime.Object) {
-			setAnn(obj, "DecoratorWasCalled")
-		},
-		expectAnnotation: "DecoratorWasCalled",
 	}, {
 		name: "AfterUpdate mutation",
 		afterUpdate: func(obj runtime.Object, opts *metav1.UpdateOptions) {
@@ -1287,9 +1263,6 @@ func TestStoreUpdateHooks(t *testing.T) {
 		expectAnnotation: "BeginUpdateWasCalled",
 	}, {
 		name: "success ordering",
-		decorator: func(obj runtime.Object) {
-			mile("Decorator")
-		},
 		afterCreate: func(obj runtime.Object, opts *metav1.CreateOptions) {
 			mile("AfterCreate")
 		},
@@ -1308,13 +1281,10 @@ func TestStoreUpdateHooks(t *testing.T) {
 				mile(fmt.Sprintf("FinishUpdate(%v)", success))
 			}, nil
 		},
-		expectMilestones: []string{"BeginUpdate", "FinishUpdate(true)", "AfterUpdate", "Decorator"},
+		expectMilestones: []string{"BeginUpdate", "FinishUpdate(true)", "AfterUpdate"},
 	}, /* fail ordering is covered in TestStoreUpdateHooksInnerRetry */ {
 		name:        "fail BeginUpdate ordering",
 		expectError: true,
-		decorator: func(obj runtime.Object) {
-			mile("Decorator")
-		},
 		afterUpdate: func(obj runtime.Object, opts *metav1.UpdateOptions) {
 			mile("AfterUpdate")
 		},
@@ -1347,7 +1317,6 @@ func TestStoreUpdateHooks(t *testing.T) {
 				t.Fatalf("Unexpected error: %v", err)
 			}
 			milestones = nil
-			registry.Decorator = tc.decorator
 			obj, _, err := registry.Update(testContext, pod.Name, rest.DefaultUpdatedObjectInfo(pod), rest.ValidateAllObjectFunc, rest.ValidateAllObjectUpdateFunc, false, &metav1.UpdateOptions{})
 			if err != nil && !tc.expectError {
 				t.Fatalf("Unexpected error: %v", err)
@@ -1385,7 +1354,6 @@ func TestStoreCreateOnUpdateHooks(t *testing.T) {
 
 	testCases := []struct {
 		name        string
-		decorator   func(runtime.Object)
 		beginCreate BeginCreateFunc
 		afterCreate AfterCreateFunc
 		beginUpdate BeginUpdateFunc
@@ -1398,9 +1366,6 @@ func TestStoreCreateOnUpdateHooks(t *testing.T) {
 		name: "no hooks",
 	}, {
 		name: "success ordering",
-		decorator: func(obj runtime.Object) {
-			mile("Decorator")
-		},
 		afterCreate: func(obj runtime.Object, opts *metav1.CreateOptions) {
 			mile("AfterCreate")
 		},
@@ -1419,12 +1384,9 @@ func TestStoreCreateOnUpdateHooks(t *testing.T) {
 				mile(fmt.Sprintf("FinishUpdate(%v)", success))
 			}, nil
 		},
-		expectMilestones: []string{"BeginCreate", "FinishCreate(true)", "AfterCreate", "Decorator"},
+		expectMilestones: []string{"BeginCreate", "FinishCreate(true)", "AfterCreate"},
 	}, {
 		name: "fail ordering",
-		decorator: func(obj runtime.Object) {
-			mile("Decorator")
-		},
 		afterCreate: func(obj runtime.Object, opts *metav1.CreateOptions) {
 			mile("AfterCreate")
 		},
@@ -1452,9 +1414,6 @@ func TestStoreCreateOnUpdateHooks(t *testing.T) {
 	}, {
 		name:        "fail BeginCreate ordering",
 		expectError: true,
-		decorator: func(obj runtime.Object) {
-			mile("Decorator")
-		},
 		afterCreate: func(obj runtime.Object, opts *metav1.CreateOptions) {
 			mile("AfterCreate")
 		},
@@ -1486,7 +1445,6 @@ func TestStoreCreateOnUpdateHooks(t *testing.T) {
 			testContext := genericapirequest.WithNamespace(genericapirequest.NewContext(), "test")
 			destroyFunc, registry := NewTestGenericStoreRegistry(t)
 			defer destroyFunc()
-			registry.Decorator = tc.decorator
 			registry.UpdateStrategy.(*testRESTStrategy).allowCreateOnUpdate = true
 			registry.BeginUpdate = tc.beginUpdate
 			registry.AfterUpdate = tc.afterUpdate
@@ -1539,7 +1497,6 @@ func TestStoreUpdateHooksInnerRetry(t *testing.T) {
 
 	testCases := []struct {
 		name        string
-		decorator   func(runtime.Object)
 		beginUpdate func(context.Context, runtime.Object, runtime.Object, *metav1.UpdateOptions) (FinishFunc, error)
 		afterUpdate AfterUpdateFunc
 		// the TTLFunc is an easy hook to force an inner-loop retry
@@ -1548,9 +1505,6 @@ func TestStoreUpdateHooksInnerRetry(t *testing.T) {
 		expectMilestones []string // to test sequence
 	}{{
 		name: "inner retry success",
-		decorator: func(obj runtime.Object) {
-			mile("Decorator")
-		},
 		afterUpdate: func(obj runtime.Object, opts *metav1.UpdateOptions) {
 			mile("AfterUpdate")
 		},
@@ -1561,12 +1515,9 @@ func TestStoreUpdateHooksInnerRetry(t *testing.T) {
 			}, nil
 		},
 		ttl:              ttlFailOnce,
-		expectMilestones: []string{"BeginUpdate", "TTLError", "FinishUpdate(false)", "BeginUpdate", "TTL", "FinishUpdate(true)", "AfterUpdate", "Decorator"},
+		expectMilestones: []string{"BeginUpdate", "TTLError", "FinishUpdate(false)", "BeginUpdate", "TTL", "FinishUpdate(true)", "AfterUpdate"},
 	}, {
 		name: "inner retry fail",
-		decorator: func(obj runtime.Object) {
-			mile("Decorator")
-		},
 		afterUpdate: func(obj runtime.Object, opts *metav1.UpdateOptions) {
 			mile("AfterUpdate")
 		},
@@ -1599,7 +1550,6 @@ func TestStoreUpdateHooksInnerRetry(t *testing.T) {
 				t.Fatalf("Unexpected error: %v", err)
 			}
 			milestones = nil
-			registry.Decorator = tc.decorator
 			ttlFailDone = false
 			registry.TTLFunc = tc.ttl
 			// force storage to use a cached object with a non-matching resourceVersion to guarantee a live lookup + retry