Add specific callback for the tracker to delete items eagerly. (#1235)

* Add specific callback for the tracker to delete items eagerly.

* Rename to OnDeletedObserver.

* Enhance FakeTracker and add tests.

reconciler/testing/tracker.go

@@ -20,6 +20,8 @@ import (
 	"sync"
 
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"knative.dev/pkg/kmeta"
 	"knative.dev/pkg/tracker"
 )
 
@@ -31,7 +33,7 @@ type NullTracker = FakeTracker
 // FakeTracker implements Tracker.
 type FakeTracker struct {
 	sync.Mutex
-	references []tracker.Reference
+	references map[tracker.Reference]map[types.NamespacedName]struct{}
 }
 
 var _ tracker.Interface = (*FakeTracker)(nil)
@@ -39,6 +41,25 @@ var _ tracker.Interface = (*FakeTracker)(nil)
 // OnChanged implements OnChanged.
 func (*FakeTracker) OnChanged(interface{}) {}
 
+// OnDeletedObserver implements OnDeletedObserver.
+func (n *FakeTracker) OnDeletedObserver(obj interface{}) {
+	item, err := kmeta.DeletionHandlingAccessor(obj)
+	if err != nil {
+		return
+	}
+	key := types.NamespacedName{Namespace: item.GetNamespace(), Name: item.GetName()}
+
+	n.Lock()
+	defer n.Unlock()
+
+	for ref, objs := range n.references {
+		delete(objs, key)
+		if len(objs) == 0 {
+			delete(n.references, ref)
+		}
+	}
+}
+
 // Track implements tracker.Interface.
 func (n *FakeTracker) Track(ref corev1.ObjectReference, obj interface{}) error {
 	return n.TrackReference(tracker.Reference{
@@ -51,10 +72,26 @@ func (n *FakeTracker) Track(ref corev1.ObjectReference, obj interface{}) error {
 
 // TrackReference implements tracker.Interface.
 func (n *FakeTracker) TrackReference(ref tracker.Reference, obj interface{}) error {
+	item, err := kmeta.DeletionHandlingAccessor(obj)
+	if err != nil {
+		return err
+	}
+	key := types.NamespacedName{Namespace: item.GetNamespace(), Name: item.GetName()}
+
 	n.Lock()
 	defer n.Unlock()
 
-	n.references = append(n.references, ref)
+	if n.references == nil {
+		n.references = make(map[tracker.Reference]map[types.NamespacedName]struct{}, 1)
+	}
+
+	objs := n.references[ref]
+	if objs == nil {
+		objs = make(map[types.NamespacedName]struct{}, 1)
+	}
+	objs[key] = struct{}{}
+	n.references[ref] = objs
+
 	return nil
 }
 
@@ -63,5 +100,10 @@ func (n *FakeTracker) References() []tracker.Reference {
 	n.Lock()
 	defer n.Unlock()
 
-	return append(n.references[:0:0], n.references...)
+	refs := make([]tracker.Reference, 0, len(n.references))
+	for ref := range n.references {
+		refs = append(refs, ref)
+	}
+
+	return refs
 }

reconciler/testing/tracker_test.go

@@ -0,0 +1,96 @@
+/*
+Copyright 2020 The Knative 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 testing
+
+import (
+	"testing"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	. "knative.dev/pkg/testing"
+	"knative.dev/pkg/tracker"
+)
+
+func TestFakeTracker(t *testing.T) {
+	t1 := &Resource{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "ns",
+			Name:      "foo",
+		},
+	}
+	t2 := &Resource{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "default",
+			Name:      "bar.baz.this-is-fine",
+		},
+	}
+
+	ref1 := tracker.Reference{
+		APIVersion: "fakeapi",
+		Kind:       "Fake",
+		Namespace:  "foo",
+		Name:       "bar",
+	}
+	ref2 := tracker.Reference{
+		APIVersion: "fakeapi2",
+		Kind:       "Fake2",
+		Namespace:  "foo2",
+		Name:       "bar2",
+	}
+
+	trk := &FakeTracker{}
+
+	// Adding t1 to ref1 and then removing it results in ref1 stopping tracking.
+	trk.TrackReference(ref1, t1)
+	if !isTracking(trk, ref1) {
+		t.Fatalf("Tracker is not tracking %v", ref1)
+	}
+	trk.OnDeletedObserver(t1)
+	if isTracking(trk, ref1) {
+		t.Fatalf("Tracker is still tracking %v", ref1)
+	}
+
+	// Adding t1, t2 to ref1 and t2 to ref2, then removing t2 results in ref2 stopping
+	// tracking.
+	trk.TrackReference(ref1, t1)
+	trk.TrackReference(ref1, t2)
+	trk.TrackReference(ref2, t2)
+	if !isTracking(trk, ref1) {
+		t.Fatalf("Tracker is not tracking %v", ref1)
+	}
+	if !isTracking(trk, ref2) {
+		t.Fatalf("Tracker is not tracking %v", ref2)
+	}
+	trk.OnDeletedObserver(t2)
+	if !isTracking(trk, ref1) {
+		t.Fatalf("Tracker is not tracking %v", ref1)
+	}
+	if isTracking(trk, ref2) {
+		t.Fatalf("Tracker is still tracking %v", ref2)
+	}
+	trk.OnDeletedObserver(t1)
+	if isTracking(trk, ref1) {
+		t.Fatalf("Tracker is still tracking %v", ref1)
+	}
+}
+
+func isTracking(tracker *FakeTracker, ref1 tracker.Reference) bool {
+	for _, tracking := range tracker.References() {
+		if tracking == ref1 {
+			return true
+		}
+	}
+	return false
+}

tracker/enqueue.go

@@ -275,3 +275,32 @@ func (i *impl) OnChanged(obj interface{}) {
 		}
 	}
 }
+
+// OnChanged implements Interface.
+func (i *impl) OnDeletedObserver(obj interface{}) {
+	item, err := kmeta.DeletionHandlingAccessor(obj)
+	if err != nil {
+		return
+	}
+
+	key := types.NamespacedName{Namespace: item.GetNamespace(), Name: item.GetName()}
+
+	i.m.Lock()
+	defer i.m.Unlock()
+
+	// Remove exact matches.
+	for ref, matchers := range i.exact {
+		delete(matchers, key)
+		if len(matchers) == 0 {
+			delete(i.exact, ref)
+		}
+	}
+
+	// Remove inexact matches.
+	for ref, matchers := range i.inexact {
+		delete(matchers, key)
+		if len(matchers) == 0 {
+			delete(i.exact, ref)
+		}
+	}
+}

tracker/enqueue_test.go

@@ -164,6 +164,15 @@ func TestHappyPathsExact(t *testing.T) {
 		}
 	}
 
+	// Stops tracking explicitly
+	{
+		trk.OnDeletedObserver(thing2)
+		trk.OnChanged(thing1)
+		if got, want := calls, 6; got != want {
+			t.Fatalf("OnChanged() = %v, wanted %v", got, want)
+		}
+	}
+
 	// Track bad object
 	{
 		if err := trk.Track(ref.ObjectReference(), struct{}{}); err == nil {
@@ -631,6 +640,15 @@ func TestHappyPathsInexact(t *testing.T) {
 		}
 	}
 
+	// Stops tracking explicitly
+	{
+		trk.OnDeletedObserver(thing2)
+		trk.OnChanged(thing1)
+		if got, want := calls, 7; got != want {
+			t.Fatalf("OnChanged() = %v, wanted %v", got, want)
+		}
+	}
+
 	// Track bad object
 	{
 		if err := trk.TrackReference(ref, struct{}{}); err == nil {
@@ -723,4 +741,13 @@ func TestHappyPathsByBoth(t *testing.T) {
 			t.Fatalf("OnChanged() = %v, wanted %v", got, want)
 		}
 	}
+
+	// Stops tracking explicitly
+	{
+		trk.OnDeletedObserver(thing2)
+		trk.OnChanged(thing1)
+		if got, want := calls, 4; got != want {
+			t.Fatalf("OnChanged() = %v, wanted %v", got, want)
+		}
+	}
 }

tracker/interface.go

@@ -70,6 +70,11 @@ type Interface interface {
 	// OnChanged is a callback to register with the InformerFactory
 	// so that we are notified for appropriate object changes.
 	OnChanged(obj interface{})
+
+	// OnDeletedObserver is a callback to register with the InformerFactory
+	// so that we are notified for deletions of a watching parent to
+	// remove the respective tracking.
+	OnDeletedObserver(obj interface{})
 }
 
 // GroupVersionKind returns the GroupVersion of the object referenced.