Merge pull request #92743 from liggitt/gc

Fix GC uid races and handling of conflicting ownerReferences

cmd/kube-controller-manager/app/core.go

@@ -553,6 +553,7 @@ func startGarbageCollectorController(ctx ControllerContext) (http.Handler, bool,
 		ignoredResources[schema.GroupResource{Group: r.Group, Resource: r.Resource}] = struct{}{}
 	}
 	garbageCollector, err := garbagecollector.NewGarbageCollector(
+		gcClientset,
 		metadataClient,
 		ctx.RESTMapper,
 		ignoredResources,

pkg/controller/garbagecollector/BUILD

@@ -20,7 +20,9 @@ go_library(
     ],
     importpath = "k8s.io/kubernetes/pkg/controller/garbagecollector",
     deps = [
+        "//pkg/controller/apis/config/scheme:go_default_library",
         "//pkg/controller/garbagecollector/metaonly:go_default_library",
+        "//staging/src/k8s.io/api/core/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/api/errors:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/api/meta:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
@@ -33,8 +35,10 @@ go_library(
         "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
         "//staging/src/k8s.io/client-go/discovery:go_default_library",
         "//staging/src/k8s.io/client-go/kubernetes:go_default_library",
+        "//staging/src/k8s.io/client-go/kubernetes/typed/core/v1:go_default_library",
         "//staging/src/k8s.io/client-go/metadata:go_default_library",
         "//staging/src/k8s.io/client-go/tools/cache:go_default_library",
+        "//staging/src/k8s.io/client-go/tools/record:go_default_library",
         "//staging/src/k8s.io/client-go/util/retry:go_default_library",
         "//staging/src/k8s.io/client-go/util/workqueue:go_default_library",
         "//staging/src/k8s.io/controller-manager/pkg/informerfactory:go_default_library",
@@ -52,15 +56,18 @@ go_test(
     srcs = [
         "dump_test.go",
         "garbagecollector_test.go",
+        "graph_builder_test.go",
     ],
     embed = [":go_default_library"],
     deps = [
         "//pkg/api/legacyscheme:go_default_library",
         "//pkg/apis/core/install:go_default_library",
+        "//pkg/controller/garbagecollector/metaonly:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/api/meta:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/api/meta/testrestmapper:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/json:go_default_library",
@@ -71,14 +78,21 @@ go_test(
         "//staging/src/k8s.io/client-go/kubernetes:go_default_library",
         "//staging/src/k8s.io/client-go/kubernetes/fake:go_default_library",
         "//staging/src/k8s.io/client-go/metadata:go_default_library",
+        "//staging/src/k8s.io/client-go/metadata/fake:go_default_library",
         "//staging/src/k8s.io/client-go/metadata/metadatainformer:go_default_library",
         "//staging/src/k8s.io/client-go/rest:go_default_library",
+        "//staging/src/k8s.io/client-go/testing:go_default_library",
+        "//staging/src/k8s.io/client-go/tools/record:go_default_library",
         "//staging/src/k8s.io/client-go/util/workqueue:go_default_library",
         "//staging/src/k8s.io/controller-manager/pkg/informerfactory:go_default_library",
         "//vendor/github.com/davecgh/go-spew/spew:go_default_library",
+        "//vendor/github.com/golang/groupcache/lru:go_default_library",
+        "//vendor/github.com/google/go-cmp/cmp:go_default_library",
         "//vendor/github.com/stretchr/testify/assert:go_default_library",
+        "//vendor/golang.org/x/time/rate:go_default_library",
         "//vendor/gonum.org/v1/gonum/graph:go_default_library",
         "//vendor/gonum.org/v1/gonum/graph/simple:go_default_library",
+        "//vendor/k8s.io/utils/pointer:go_default_library",
     ],
 )
 

pkg/controller/garbagecollector/garbagecollector.go

@@ -18,6 +18,7 @@ package garbagecollector
 
 import (
 	"context"
+	goerrors "errors"
 	"fmt"
 	"reflect"
 	"sync"
@@ -25,6 +26,7 @@ import (
 
 	"k8s.io/klog/v2"
 
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -35,10 +37,14 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/discovery"
+	clientset "k8s.io/client-go/kubernetes"
+	v1core "k8s.io/client-go/kubernetes/typed/core/v1"
 	"k8s.io/client-go/metadata"
 	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/workqueue"
 	"k8s.io/controller-manager/pkg/informerfactory"
+	"k8s.io/kubernetes/pkg/controller/apis/config/scheme"
 
 	// import known versions
 	_ "k8s.io/client-go/kubernetes"
@@ -67,22 +73,29 @@ type GarbageCollector struct {
 	attemptToOrphan        workqueue.RateLimitingInterface
 	dependencyGraphBuilder *GraphBuilder
 	// GC caches the owners that do not exist according to the API server.
-	absentOwnerCache *UIDCache
+	absentOwnerCache *ReferenceCache
 
 	workerLock sync.RWMutex
 }
 
 // NewGarbageCollector creates a new GarbageCollector.
 func NewGarbageCollector(
+	kubeClient clientset.Interface,
 	metadataClient metadata.Interface,
 	mapper resettableRESTMapper,
 	ignoredResources map[schema.GroupResource]struct{},
 	sharedInformers informerfactory.InformerFactory,
 	informersStarted <-chan struct{},
 ) (*GarbageCollector, error) {
+
+	eventBroadcaster := record.NewBroadcaster()
+	eventBroadcaster.StartStructuredLogging(0)
+	eventBroadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: kubeClient.CoreV1().Events("")})
+	eventRecorder := eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "garbage-collector-controller"})
+
 	attemptToDelete := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "garbage_collector_attempt_to_delete")
 	attemptToOrphan := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "garbage_collector_attempt_to_orphan")
-	absentOwnerCache := NewUIDCache(500)
+	absentOwnerCache := NewReferenceCache(500)
 	gc := &GarbageCollector{
 		metadataClient:   metadataClient,
 		restMapper:       mapper,
@@ -91,6 +104,7 @@ func NewGarbageCollector(
 		absentOwnerCache: absentOwnerCache,
 	}
 	gc.dependencyGraphBuilder = &GraphBuilder{
+		eventRecorder:    eventRecorder,
 		metadataClient:   metadataClient,
 		informersStarted: informersStarted,
 		restMapper:       mapper,
@@ -281,6 +295,10 @@ func (gc *GarbageCollector) runAttemptToDeleteWorker() {
 	}
 }
 
+var enqueuedVirtualDeleteEventErr = goerrors.New("enqueued virtual delete event")
+
+var namespacedOwnerOfClusterScopedObjectErr = goerrors.New("cluster-scoped objects cannot refer to namespaced owners")
+
 func (gc *GarbageCollector) attemptToDeleteWorker() bool {
 	item, quit := gc.attemptToDelete.Get()
 	gc.workerLock.RLock()
@@ -294,8 +312,31 @@ func (gc *GarbageCollector) attemptToDeleteWorker() bool {
 		utilruntime.HandleError(fmt.Errorf("expect *node, got %#v", item))
 		return true
 	}
+
+	if !n.isObserved() {
+		nodeFromGraph, existsInGraph := gc.dependencyGraphBuilder.uidToNode.Read(n.identity.UID)
+		if !existsInGraph {
+			// this can happen if attemptToDelete loops on a requeued virtual node because attemptToDeleteItem returned an error,
+			// and in the meantime a deletion of the real object associated with that uid was observed
+			klog.V(5).Infof("item %s no longer in the graph, skipping attemptToDeleteItem", n)
+			return true
+		}
+		if nodeFromGraph.isObserved() {
+			// this can happen if attemptToDelete loops on a requeued virtual node because attemptToDeleteItem returned an error,
+			// and in the meantime the real object associated with that uid was observed
+			klog.V(5).Infof("item %s no longer virtual in the graph, skipping attemptToDeleteItem on virtual node", n)
+			return true
+		}
+	}
+
 	err := gc.attemptToDeleteItem(n)
-	if err != nil {
+	if err == enqueuedVirtualDeleteEventErr {
+		// a virtual event was produced and will be handled by processGraphChanges, no need to requeue this node
+		return true
+	} else if err == namespacedOwnerOfClusterScopedObjectErr {
+		// a cluster-scoped object referring to a namespaced owner is an error that will not resolve on retry, no need to requeue this node
+		return true
+	} else if err != nil {
 		if _, ok := err.(*restMappingError); ok {
 			// There are at least two ways this can happen:
 			// 1. The reference is to an object of a custom type that has not yet been
@@ -325,10 +366,20 @@ func (gc *GarbageCollector) attemptToDeleteWorker() bool {
 // returns its latest state.
 func (gc *GarbageCollector) isDangling(reference metav1.OwnerReference, item *node) (
 	dangling bool, owner *metav1.PartialObjectMetadata, err error) {
-	if gc.absentOwnerCache.Has(reference.UID) {
+
+	// check for recorded absent cluster-scoped parent
+	absentOwnerCacheKey := objectReference{OwnerReference: ownerReferenceCoordinates(reference)}
+	if gc.absentOwnerCache.Has(absentOwnerCacheKey) {
 		klog.V(5).Infof("according to the absentOwnerCache, object %s's owner %s/%s, %s does not exist", item.identity.UID, reference.APIVersion, reference.Kind, reference.Name)
 		return true, nil, nil
 	}
+	// check for recorded absent namespaced parent
+	absentOwnerCacheKey.Namespace = item.identity.Namespace
+	if gc.absentOwnerCache.Has(absentOwnerCacheKey) {
+		klog.V(5).Infof("according to the absentOwnerCache, object %s's owner %s/%s, %s does not exist in namespace %s", item.identity.UID, reference.APIVersion, reference.Kind, reference.Name, item.identity.Namespace)
+		return true, nil, nil
+	}
+
 	// TODO: we need to verify the reference resource is supported by the
 	// system. If it's not a valid resource, the garbage collector should i)
 	// ignore the reference when decide if the object should be deleted, and
@@ -339,6 +390,16 @@ func (gc *GarbageCollector) isDangling(reference metav1.OwnerReference, item *no
 	if err != nil {
 		return false, nil, err
 	}
+	if !namespaced {
+		absentOwnerCacheKey.Namespace = ""
+	}
+
+	if len(item.identity.Namespace) == 0 && namespaced {
+		// item is a cluster-scoped object referring to a namespace-scoped owner, which is not valid.
+		// return a marker error, rather than retrying on the lookup failure forever.
+		klog.V(2).Infof("object %s is cluster-scoped, but refers to a namespaced owner of type %s/%s", item.identity, reference.APIVersion, reference.Kind)
+		return false, nil, namespacedOwnerOfClusterScopedObjectErr
+	}
 
 	// TODO: It's only necessary to talk to the API server if the owner node
 	// is a "virtual" node. The local graph could lag behind the real
@@ -346,7 +407,7 @@ func (gc *GarbageCollector) isDangling(reference metav1.OwnerReference, item *no
 	owner, err = gc.metadataClient.Resource(resource).Namespace(resourceDefaultNamespace(namespaced, item.identity.Namespace)).Get(context.TODO(), reference.Name, metav1.GetOptions{})
 	switch {
 	case errors.IsNotFound(err):
-		gc.absentOwnerCache.Add(reference.UID)
+		gc.absentOwnerCache.Add(absentOwnerCacheKey)
 		klog.V(5).Infof("object %s's owner %s/%s, %s is not found", item.identity.UID, reference.APIVersion, reference.Kind, reference.Name)
 		return true, nil, nil
 	case err != nil:
@@ -355,7 +416,7 @@ func (gc *GarbageCollector) isDangling(reference metav1.OwnerReference, item *no
 
 	if owner.GetUID() != reference.UID {
 		klog.V(5).Infof("object %s's owner %s/%s, %s is not found, UID mismatch", item.identity.UID, reference.APIVersion, reference.Kind, reference.Name)
-		gc.absentOwnerCache.Add(reference.UID)
+		gc.absentOwnerCache.Add(absentOwnerCacheKey)
 		return true, nil, nil
 	}
 	return false, owner, nil
@@ -400,9 +461,15 @@ func ownerRefsToUIDs(refs []metav1.OwnerReference) []types.UID {
 	return ret
 }
 
+// attemptToDeleteItem looks up the live API object associated with the node,
+// and issues a delete IFF the uid matches, the item is not blocked on deleting dependents,
+// and all owner references are dangling.
+//
+// if the API get request returns a NotFound error, or the retrieved item's uid does not match,
+// a virtual delete event for the node is enqueued and enqueuedVirtualDeleteEventErr is returned.
 func (gc *GarbageCollector) attemptToDeleteItem(item *node) error {
 	klog.V(2).InfoS("Processing object", "object", klog.KRef(item.identity.Namespace, item.identity.Name),
-		"objectUID", item.identity.UID, "kind", item.identity.Kind)
+		"objectUID", item.identity.UID, "kind", item.identity.Kind, "virtual", !item.isObserved())
 
 	// "being deleted" is an one-way trip to the final deletion. We'll just wait for the final deletion, and then process the object's dependents.
 	if item.isBeingDeleted() && !item.isDeletingDependents() {
@@ -420,10 +487,7 @@ func (gc *GarbageCollector) attemptToDeleteItem(item *node) error {
 		// the virtual node from GraphBuilder.uidToNode.
 		klog.V(5).Infof("item %v not found, generating a virtual delete event", item.identity)
 		gc.dependencyGraphBuilder.enqueueVirtualDeleteEvent(item.identity)
-		// since we're manually inserting a delete event to remove this node,
-		// we don't need to keep tracking it as a virtual node and requeueing in attemptToDelete
-		item.markObserved()
-		return nil
+		return enqueuedVirtualDeleteEventErr
 	case err != nil:
 		return err
 	}
@@ -431,10 +495,7 @@ func (gc *GarbageCollector) attemptToDeleteItem(item *node) error {
 	if latest.GetUID() != item.identity.UID {
 		klog.V(5).Infof("UID doesn't match, item %v not found, generating a virtual delete event", item.identity)
 		gc.dependencyGraphBuilder.enqueueVirtualDeleteEvent(item.identity)
-		// since we're manually inserting a delete event to remove this node,
-		// we don't need to keep tracking it as a virtual node and requeueing in attemptToDelete
-		item.markObserved()
-		return nil
+		return enqueuedVirtualDeleteEventErr
 	}
 
 	// TODO: attemptToOrphanWorker() routine is similar. Consider merging

pkg/controller/garbagecollector/graph.go

@@ -61,6 +61,25 @@ type node struct {
 	owners []metav1.OwnerReference
 }
 
+// clone() must only be called from the single-threaded GraphBuilder.processGraphChanges()
+func (n *node) clone() *node {
+	c := &node{
+		identity:           n.identity,
+		dependents:         make(map[*node]struct{}, len(n.dependents)),
+		deletingDependents: n.deletingDependents,
+		beingDeleted:       n.beingDeleted,
+		virtual:            n.virtual,
+		owners:             make([]metav1.OwnerReference, 0, len(n.owners)),
+	}
+	for dep := range n.dependents {
+		c.dependents[dep] = struct{}{}
+	}
+	for _, owner := range n.owners {
+		c.owners = append(c.owners, owner)
+	}
+	return c
+}
+
 // An object is on a one way trip to its final deletion if it starts being
 // deleted, so we only provide a function to set beingDeleted to true.
 func (n *node) markBeingDeleted() {
@@ -148,6 +167,23 @@ func (n *node) blockingDependents() []*node {
 	return ret
 }
 
+// ownerReferenceCoordinates returns an owner reference containing only the coordinate fields
+// from the input reference (uid, name, kind, apiVersion)
+func ownerReferenceCoordinates(ref metav1.OwnerReference) metav1.OwnerReference {
+	return metav1.OwnerReference{
+		UID:        ref.UID,
+		Name:       ref.Name,
+		Kind:       ref.Kind,
+		APIVersion: ref.APIVersion,
+	}
+}
+
+// ownerReferenceMatchesCoordinates returns true if all of the coordinate fields match
+// between the two references (uid, name, kind, apiVersion)
+func ownerReferenceMatchesCoordinates(a, b metav1.OwnerReference) bool {
+	return a.UID == b.UID && a.Name == b.Name && a.Kind == b.Kind && a.APIVersion == b.APIVersion
+}
+
 // String renders node as a string using fmt. Acquires a read lock to ensure the
 // reflective dump of dependents doesn't race with any concurrent writes.
 func (n *node) String() string {

pkg/controller/garbagecollector/graph_builder.go

@@ -24,15 +24,18 @@ import (
 
 	"k8s.io/klog/v2"
 
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/metadata"
 	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/workqueue"
 	"k8s.io/controller-manager/pkg/informerfactory"
 	"k8s.io/kubernetes/pkg/controller/garbagecollector/metaonly"
@@ -60,6 +63,8 @@ const (
 )
 
 type event struct {
+	// virtual indicates this event did not come from an informer, but was constructed artificially
+	virtual   bool
 	eventType eventType
 	obj       interface{}
 	// the update event comes with an old object, but it's not used by the garbage collector.
@@ -89,6 +94,8 @@ type GraphBuilder struct {
 	// it is protected by monitorLock.
 	running bool
 
+	eventRecorder record.EventRecorder
+
 	metadataClient metadata.Interface
 	// monitors are the producer of the graphChanges queue, graphBuilder alters
 	// the in-memory graph according to the changes.
@@ -101,7 +108,7 @@ type GraphBuilder struct {
 	attemptToOrphan workqueue.RateLimitingInterface
 	// GraphBuilder and GC share the absentOwnerCache. Objects that are known to
 	// be non-existent are added to the cached.
-	absentOwnerCache *UIDCache
+	absentOwnerCache *ReferenceCache
 	sharedInformers  informerfactory.InformerFactory
 	ignoredResources map[schema.GroupResource]struct{}
 }
@@ -324,8 +331,11 @@ func DefaultIgnoredResources() map[schema.GroupResource]struct{} {
 // enqueueVirtualDeleteEvent is used to add a virtual delete event to be processed for virtual nodes
 // once it is determined they do not have backing objects in storage
 func (gb *GraphBuilder) enqueueVirtualDeleteEvent(ref objectReference) {
+	gv, _ := schema.ParseGroupVersion(ref.APIVersion)
 	gb.graphChanges.Add(&event{
+		virtual:   true,
 		eventType: deleteEvent,
+		gvk:       gv.WithKind(ref.Kind),
 		obj: &metaonly.MetadataOnlyObject{
 			TypeMeta:   metav1.TypeMeta{APIVersion: ref.APIVersion, Kind: ref.Kind},
 			ObjectMeta: metav1.ObjectMeta{Namespace: ref.Namespace, UID: ref.UID, Name: ref.Name},
@@ -338,6 +348,10 @@ func (gb *GraphBuilder) enqueueVirtualDeleteEvent(ref objectReference) {
 // the owner. The "virtual" node will be enqueued to the attemptToDelete, so that
 // attemptToDeleteItem() will verify if the owner exists according to the API server.
 func (gb *GraphBuilder) addDependentToOwners(n *node, owners []metav1.OwnerReference) {
+	// track if some of the referenced owners already exist in the graph and have been observed,
+	// and the dependent's ownerRef does not match their observed coordinates
+	hasPotentiallyInvalidOwnerReference := false
+
 	for _, owner := range owners {
 		ownerNode, ok := gb.uidToNode.Read(owner.UID)
 		if !ok {
@@ -345,7 +359,7 @@ func (gb *GraphBuilder) addDependentToOwners(n *node, owners []metav1.OwnerRefer
 			// exist in the graph yet.
 			ownerNode = &node{
 				identity: objectReference{
-					OwnerReference: owner,
+					OwnerReference: ownerReferenceCoordinates(owner),
 					Namespace:      n.identity.Namespace,
 				},
 				dependents: make(map[*node]struct{}),
@@ -361,8 +375,66 @@ func (gb *GraphBuilder) addDependentToOwners(n *node, owners []metav1.OwnerRefer
 			// event to delete it from the graph if API server confirms this
 			// owner doesn't exist.
 			gb.attemptToDelete.Add(ownerNode)
+		} else if !hasPotentiallyInvalidOwnerReference {
+			ownerIsNamespaced := len(ownerNode.identity.Namespace) > 0
+			if ownerIsNamespaced && ownerNode.identity.Namespace != n.identity.Namespace {
+				if ownerNode.isObserved() {
+					// The owner node has been observed via an informer
+					// the dependent's namespace doesn't match the observed owner's namespace, this is definitely wrong.
+					// cluster-scoped owners can be referenced as an owner from any namespace or cluster-scoped object.
+					klog.V(2).Infof("node %s references an owner %s but does not match namespaces", n.identity, ownerNode.identity)
+					gb.reportInvalidNamespaceOwnerRef(n, owner.UID)
+				}
+				hasPotentiallyInvalidOwnerReference = true
+			} else if !ownerReferenceMatchesCoordinates(owner, ownerNode.identity.OwnerReference) {
+				if ownerNode.isObserved() {
+					// The owner node has been observed via an informer
+					// n's owner reference doesn't match the observed identity, this might be wrong.
+					klog.V(2).Infof("node %s references an owner %s with coordinates that do not match the observed identity", n.identity, ownerNode.identity)
+				}
+				hasPotentiallyInvalidOwnerReference = true
+			}
 		}
 	}
+
+	if hasPotentiallyInvalidOwnerReference {
+		// Enqueue the potentially invalid dependent node into attemptToDelete.
+		// The garbage processor will verify whether the owner references are dangling
+		// and delete the dependent if all owner references are confirmed absent.
+		gb.attemptToDelete.Add(n)
+	}
+}
+
+func (gb *GraphBuilder) reportInvalidNamespaceOwnerRef(n *node, invalidOwnerUID types.UID) {
+	var invalidOwnerRef metav1.OwnerReference
+	var found = false
+	for _, ownerRef := range n.owners {
+		if ownerRef.UID == invalidOwnerUID {
+			invalidOwnerRef = ownerRef
+			found = true
+			break
+		}
+	}
+	if !found {
+		return
+	}
+	ref := &v1.ObjectReference{
+		Kind:       n.identity.Kind,
+		APIVersion: n.identity.APIVersion,
+		Namespace:  n.identity.Namespace,
+		Name:       n.identity.Name,
+		UID:        n.identity.UID,
+	}
+	invalidIdentity := objectReference{
+		OwnerReference: metav1.OwnerReference{
+			Kind:       invalidOwnerRef.Kind,
+			APIVersion: invalidOwnerRef.APIVersion,
+			Name:       invalidOwnerRef.Name,
+			UID:        invalidOwnerRef.UID,
+		},
+		Namespace: n.identity.Namespace,
+	}
+	gb.eventRecorder.Eventf(ref, v1.EventTypeWarning, "OwnerRefInvalidNamespace", "ownerRef %s does not exist in namespace %q", invalidIdentity, n.identity.Namespace)
 }
 
 // insertNode insert the node to gb.uidToNode; then it finds all owners as listed
@@ -522,6 +594,18 @@ func (gb *GraphBuilder) runProcessGraphChanges() {
 	}
 }
 
+func identityFromEvent(event *event, accessor metav1.Object) objectReference {
+	return objectReference{
+		OwnerReference: metav1.OwnerReference{
+			APIVersion: event.gvk.GroupVersion().String(),
+			Kind:       event.gvk.Kind,
+			UID:        accessor.GetUID(),
+			Name:       accessor.GetName(),
+		},
+		Namespace: accessor.GetNamespace(),
+	}
+}
+
 // Dequeueing an event from graphChanges, updating graph, populating dirty_queue.
 func (gb *GraphBuilder) processGraphChanges() bool {
 	item, quit := gb.graphChanges.Get()
@@ -540,27 +624,42 @@ func (gb *GraphBuilder) processGraphChanges() bool {
 		utilruntime.HandleError(fmt.Errorf("cannot access obj: %v", err))
 		return true
 	}
-	klog.V(5).Infof("GraphBuilder process object: %s/%s, namespace %s, name %s, uid %s, event type %v", event.gvk.GroupVersion().String(), event.gvk.Kind, accessor.GetNamespace(), accessor.GetName(), string(accessor.GetUID()), event.eventType)
+	klog.V(5).Infof("GraphBuilder process object: %s/%s, namespace %s, name %s, uid %s, event type %v, virtual=%v", event.gvk.GroupVersion().String(), event.gvk.Kind, accessor.GetNamespace(), accessor.GetName(), string(accessor.GetUID()), event.eventType, event.virtual)
 	// Check if the node already exists
 	existingNode, found := gb.uidToNode.Read(accessor.GetUID())
-	if found {
+	if found && !event.virtual && !existingNode.isObserved() {
 		// this marks the node as having been observed via an informer event
 		// 1. this depends on graphChanges only containing add/update events from the actual informer
 		// 2. this allows things tracking virtual nodes' existence to stop polling and rely on informer events
+		observedIdentity := identityFromEvent(event, accessor)
+		if observedIdentity != existingNode.identity {
+			// find dependents that don't match the identity we observed
+			_, potentiallyInvalidDependents := partitionDependents(existingNode.getDependents(), observedIdentity)
+			// add those potentially invalid dependents to the attemptToDelete queue.
+			// if their owners are still solid the attemptToDelete will be a no-op.
+			// this covers the bad child -> good parent observation sequence.
+			// the good parent -> bad child observation sequence is handled in addDependentToOwners
+			for _, dep := range potentiallyInvalidDependents {
+				if len(observedIdentity.Namespace) > 0 && dep.identity.Namespace != observedIdentity.Namespace {
+					// Namespace mismatch, this is definitely wrong
+					klog.V(2).Infof("node %s references an owner %s but does not match namespaces", dep.identity, observedIdentity)
+					gb.reportInvalidNamespaceOwnerRef(dep, observedIdentity.UID)
+				}
+				gb.attemptToDelete.Add(dep)
+			}
+
+			// make a copy (so we don't modify the existing node in place), store the observed identity, and replace the virtual node
+			klog.V(2).Infof("replacing virtual node %s with observed node %s", existingNode.identity, observedIdentity)
+			existingNode = existingNode.clone()
+			existingNode.identity = observedIdentity
+			gb.uidToNode.Write(existingNode)
+		}
 		existingNode.markObserved()
 	}
 	switch {
 	case (event.eventType == addEvent || event.eventType == updateEvent) && !found:
 		newNode := &node{
-			identity: objectReference{
-				OwnerReference: metav1.OwnerReference{
-					APIVersion: event.gvk.GroupVersion().String(),
-					Kind:       event.gvk.Kind,
-					UID:        accessor.GetUID(),
-					Name:       accessor.GetName(),
-				},
-				Namespace: accessor.GetNamespace(),
-			},
+			identity:           identityFromEvent(event, accessor),
 			dependents:         make(map[*node]struct{}),
 			owners:             accessor.GetOwnerReferences(),
 			deletingDependents: beingDeleted(accessor) && hasDeleteDependentsFinalizer(accessor),
@@ -595,25 +694,208 @@ func (gb *GraphBuilder) processGraphChanges() bool {
 			klog.V(5).Infof("%v doesn't exist in the graph, this shouldn't happen", accessor.GetUID())
 			return true
 		}
-		// removeNode updates the graph
-		gb.removeNode(existingNode)
-		existingNode.dependentsLock.RLock()
-		defer existingNode.dependentsLock.RUnlock()
-		if len(existingNode.dependents) > 0 {
-			gb.absentOwnerCache.Add(accessor.GetUID())
-		}
-		for dep := range existingNode.dependents {
-			gb.attemptToDelete.Add(dep)
-		}
-		for _, owner := range existingNode.owners {
-			ownerNode, found := gb.uidToNode.Read(owner.UID)
-			if !found || !ownerNode.isDeletingDependents() {
-				continue
+
+		removeExistingNode := true
+
+		if event.virtual {
+			// this is a virtual delete event, not one observed from an informer
+			deletedIdentity := identityFromEvent(event, accessor)
+			if existingNode.virtual {
+
+				// our existing node is also virtual, we're not sure of its coordinates.
+				// see if any dependents reference this owner with coordinates other than the one we got a virtual delete event for.
+				if matchingDependents, nonmatchingDependents := partitionDependents(existingNode.getDependents(), deletedIdentity); len(nonmatchingDependents) > 0 {
+
+					// some of our dependents disagree on our coordinates, so do not remove the existing virtual node from the graph
+					removeExistingNode = false
+
+					if len(matchingDependents) > 0 {
+						// mark the observed deleted identity as absent
+						gb.absentOwnerCache.Add(deletedIdentity)
+						// attempt to delete dependents that do match the verified deleted identity
+						for _, dep := range matchingDependents {
+							gb.attemptToDelete.Add(dep)
+						}
+					}
+
+					// if the delete event verified existingNode.identity doesn't exist...
+					if existingNode.identity == deletedIdentity {
+						// find an alternative identity our nonmatching dependents refer to us by
+						replacementIdentity := getAlternateOwnerIdentity(nonmatchingDependents, deletedIdentity)
+						if replacementIdentity != nil {
+							// replace the existing virtual node with a new one with one of our other potential identities
+							replacementNode := existingNode.clone()
+							replacementNode.identity = *replacementIdentity
+							gb.uidToNode.Write(replacementNode)
+							// and add the new virtual node back to the attemptToDelete queue
+							gb.attemptToDelete.AddRateLimited(replacementNode)
+						}
+					}
+				}
+
+			} else if existingNode.identity != deletedIdentity {
+				// do not remove the existing real node from the graph based on a virtual delete event
+				removeExistingNode = false
+
+				// our existing node which was observed via informer disagrees with the virtual delete event's coordinates
+				matchingDependents, _ := partitionDependents(existingNode.getDependents(), deletedIdentity)
+
+				if len(matchingDependents) > 0 {
+					// mark the observed deleted identity as absent
+					gb.absentOwnerCache.Add(deletedIdentity)
+					// attempt to delete dependents that do match the verified deleted identity
+					for _, dep := range matchingDependents {
+						gb.attemptToDelete.Add(dep)
+					}
+				}
+			}
+		}
+
+		if removeExistingNode {
+			// removeNode updates the graph
+			gb.removeNode(existingNode)
+			existingNode.dependentsLock.RLock()
+			defer existingNode.dependentsLock.RUnlock()
+			if len(existingNode.dependents) > 0 {
+				gb.absentOwnerCache.Add(identityFromEvent(event, accessor))
+			}
+			for dep := range existingNode.dependents {
+				gb.attemptToDelete.Add(dep)
+			}
+			for _, owner := range existingNode.owners {
+				ownerNode, found := gb.uidToNode.Read(owner.UID)
+				if !found || !ownerNode.isDeletingDependents() {
+					continue
+				}
+				// this is to let attempToDeleteItem check if all the owner's
+				// dependents are deleted, if so, the owner will be deleted.
+				gb.attemptToDelete.Add(ownerNode)
 			}
-			// this is to let attempToDeleteItem check if all the owner's
-			// dependents are deleted, if so, the owner will be deleted.
-			gb.attemptToDelete.Add(ownerNode)
 		}
 	}
 	return true
 }
+
+// partitionDependents divides the provided dependents into a list which have an ownerReference matching the provided identity,
+// and ones which have an ownerReference for the given uid that do not match the provided identity.
+// Note that a dependent with multiple ownerReferences for the target uid can end up in both lists.
+func partitionDependents(dependents []*node, matchOwnerIdentity objectReference) (matching, nonmatching []*node) {
+	ownerIsNamespaced := len(matchOwnerIdentity.Namespace) > 0
+	for i := range dependents {
+		dep := dependents[i]
+		foundMatch := false
+		foundMismatch := false
+		// if the dep namespace matches or the owner is cluster scoped ...
+		if ownerIsNamespaced && matchOwnerIdentity.Namespace != dep.identity.Namespace {
+			// all references to the parent do not match, since the dependent namespace does not match the owner
+			foundMismatch = true
+		} else {
+			for _, ownerRef := range dep.owners {
+				// ... find the ownerRef with a matching uid ...
+				if ownerRef.UID == matchOwnerIdentity.UID {
+					// ... and check if it matches all coordinates
+					if ownerReferenceMatchesCoordinates(ownerRef, matchOwnerIdentity.OwnerReference) {
+						foundMatch = true
+					} else {
+						foundMismatch = true
+					}
+				}
+			}
+		}
+
+		if foundMatch {
+			matching = append(matching, dep)
+		}
+		if foundMismatch {
+			nonmatching = append(nonmatching, dep)
+		}
+	}
+	return matching, nonmatching
+}
+
+func referenceLessThan(a, b objectReference) bool {
+	// kind/apiVersion are more significant than namespace,
+	// so that we get coherent ordering between kinds
+	// regardless of whether they are cluster-scoped or namespaced
+	if a.Kind != b.Kind {
+		return a.Kind < b.Kind
+	}
+	if a.APIVersion != b.APIVersion {
+		return a.APIVersion < b.APIVersion
+	}
+	// namespace is more significant than name
+	if a.Namespace != b.Namespace {
+		return a.Namespace < b.Namespace
+	}
+	// name is more significant than uid
+	if a.Name != b.Name {
+		return a.Name < b.Name
+	}
+	// uid is included for completeness, but is expected to be identical
+	// when getting alternate identities for an owner since they are keyed by uid
+	if a.UID != b.UID {
+		return a.UID < b.UID
+	}
+	return false
+}
+
+// getAlternateOwnerIdentity searches deps for owner references which match
+// verifiedAbsentIdentity.UID but differ in apiVersion/kind/name or namespace.
+// The first that follows verifiedAbsentIdentity (according to referenceLessThan) is returned.
+// If none follow verifiedAbsentIdentity, the first (according to referenceLessThan) is returned.
+// If no alternate identities are found, nil is returned.
+func getAlternateOwnerIdentity(deps []*node, verifiedAbsentIdentity objectReference) *objectReference {
+	absentIdentityIsClusterScoped := len(verifiedAbsentIdentity.Namespace) == 0
+
+	seenAlternates := map[objectReference]bool{verifiedAbsentIdentity: true}
+
+	// keep track of the first alternate reference (according to referenceLessThan)
+	var first *objectReference
+	// keep track of the first reference following verifiedAbsentIdentity (according to referenceLessThan)
+	var firstFollowing *objectReference
+
+	for _, dep := range deps {
+		for _, ownerRef := range dep.owners {
+			if ownerRef.UID != verifiedAbsentIdentity.UID {
+				// skip references that aren't the uid we care about
+				continue
+			}
+
+			if ownerReferenceMatchesCoordinates(ownerRef, verifiedAbsentIdentity.OwnerReference) {
+				if absentIdentityIsClusterScoped || verifiedAbsentIdentity.Namespace == dep.identity.Namespace {
+					// skip references that exactly match verifiedAbsentIdentity
+					continue
+				}
+			}
+
+			ref := objectReference{OwnerReference: ownerReferenceCoordinates(ownerRef), Namespace: dep.identity.Namespace}
+			if absentIdentityIsClusterScoped && ref.APIVersion == verifiedAbsentIdentity.APIVersion && ref.Kind == verifiedAbsentIdentity.Kind {
+				// we know this apiVersion/kind is cluster-scoped because of verifiedAbsentIdentity,
+				// so clear the namespace from the alternate identity
+				ref.Namespace = ""
+			}
+
+			if seenAlternates[ref] {
+				// skip references we've already seen
+				continue
+			}
+			seenAlternates[ref] = true
+
+			if first == nil || referenceLessThan(ref, *first) {
+				// this alternate comes first lexically
+				first = &ref
+			}
+			if referenceLessThan(verifiedAbsentIdentity, ref) && (firstFollowing == nil || referenceLessThan(ref, *firstFollowing)) {
+				// this alternate is the first following verifiedAbsentIdentity lexically
+				firstFollowing = &ref
+			}
+		}
+	}
+
+	// return the first alternate identity following the verified absent identity, if there is one
+	if firstFollowing != nil {
+		return firstFollowing
+	}
+	// otherwise return the first alternate identity
+	return first
+}

pkg/controller/garbagecollector/graph_builder_test.go

@@ -0,0 +1,212 @@
+/*
+Copyright 2020 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 garbagecollector
+
+import (
+	"reflect"
+	"testing"
+)
+
+func TestGetAlternateOwnerIdentity(t *testing.T) {
+	ns1child1 := makeID("v1", "Child", "ns1", "child1", "childuid11")
+	ns1child2 := makeID("v1", "Child", "ns1", "child2", "childuid12")
+
+	ns2child1 := makeID("v1", "Child", "ns2", "child1", "childuid21")
+
+	clusterchild1 := makeID("v1", "Child", "", "child1", "childuidc1")
+
+	var (
+		nsabsentparentns1 = makeID("v1", "NSParent", "ns1", "parentname", "parentuid")
+		nsabsentparentns2 = makeID("v1", "NSParent", "ns2", "parentname", "parentuid")
+
+		nsabsentparent_version = makeID("xx", "NSParent", "ns1", "parentname", "parentuid")
+		nsabsentparent_kind    = makeID("v1", "xxxxxxxx", "ns1", "parentname", "parentuid")
+		nsabsentparent_name    = makeID("v1", "NSParent", "ns1", "xxxxxxxxxx", "parentuid")
+
+		clusterabsentparent         = makeID("v1", "ClusterParent", "", "parentname", "parentuid")
+		clusterabsentparent_version = makeID("xx", "ClusterParent", "", "parentname", "parentuid")
+		clusterabsentparent_kind    = makeID("v1", "xxxxxxxxxxxxx", "", "parentname", "parentuid")
+		clusterabsentparent_name    = makeID("v1", "ClusterParent", "", "xxxxxxxxxx", "parentuid")
+
+		clusterabsentparent_ns1_version = makeID("xx", "ClusterParent", "ns1", "parentname", "parentuid")
+		clusterabsentparent_ns1_kind    = makeID("v1", "xxxxxxxxxxxxx", "ns1", "parentname", "parentuid")
+	)
+
+	orderedNamespacedReferences := []objectReference{
+		makeID("v1", "kind", "ns1", "name", "uid"),
+		makeID("v2", "kind", "ns1", "name", "uid"),
+		makeID("v3", "kind", "ns1", "name", "uid"),
+		makeID("v4", "kind", "ns1", "name", "uid"),
+		makeID("v5", "kind", "ns1", "name", "uid"),
+	}
+	orderedClusterReferences := []objectReference{
+		makeID("v1", "kind", "", "name", "uid"),
+		makeID("v2", "kind", "", "name", "uid"),
+		makeID("v3", "kind", "", "name", "uid"),
+		makeID("v4", "kind", "", "name", "uid"),
+		makeID("v5", "kind", "", "name", "uid"),
+	}
+
+	testcases := []struct {
+		name              string
+		deps              []*node
+		verifiedAbsent    objectReference
+		expectedAlternate *objectReference
+	}{
+		{
+			name: "namespaced alternate version",
+			deps: []*node{
+				makeNode(ns1child1, withOwners(nsabsentparentns1)),
+				makeNode(ns1child2, withOwners(nsabsentparent_version)),
+			},
+			verifiedAbsent:    nsabsentparentns1,
+			expectedAlternate: &nsabsentparent_version, // switch to alternate version
+		},
+		{
+			name: "namespaced alternate kind",
+			deps: []*node{
+				makeNode(ns1child1, withOwners(nsabsentparentns1)),
+				makeNode(ns1child2, withOwners(nsabsentparent_kind)),
+			},
+			verifiedAbsent:    nsabsentparentns1,
+			expectedAlternate: &nsabsentparent_kind, // switch to alternate kind
+		},
+		{
+			name: "namespaced alternate namespace",
+			deps: []*node{
+				makeNode(ns1child1, withOwners(nsabsentparentns1)),
+				makeNode(ns2child1, withOwners(nsabsentparentns2)),
+			},
+			verifiedAbsent:    nsabsentparentns1,
+			expectedAlternate: &nsabsentparentns2, // switch to alternate namespace
+		},
+		{
+			name: "namespaced alternate name",
+			deps: []*node{
+				makeNode(ns1child1, withOwners(nsabsentparentns1)),
+				makeNode(ns1child1, withOwners(nsabsentparent_name)),
+			},
+			verifiedAbsent:    nsabsentparentns1,
+			expectedAlternate: &nsabsentparent_name, // switch to alternate name
+		},
+
+		{
+			name: "cluster alternate version",
+			deps: []*node{
+				makeNode(ns1child1, withOwners(clusterabsentparent)),
+				makeNode(ns1child2, withOwners(clusterabsentparent_version)),
+			},
+			verifiedAbsent:    clusterabsentparent,
+			expectedAlternate: &clusterabsentparent_ns1_version, // switch to alternate version, namespaced to new dependent since we don't know the version is cluster-scoped
+		},
+		{
+			name: "cluster alternate kind",
+			deps: []*node{
+				makeNode(ns1child1, withOwners(clusterabsentparent)),
+				makeNode(ns1child2, withOwners(clusterabsentparent_kind)),
+			},
+			verifiedAbsent:    clusterabsentparent,
+			expectedAlternate: &clusterabsentparent_ns1_kind, // switch to alternate kind, namespaced to new dependent since we don't know the new kind is cluster-scoped
+		},
+		{
+			name: "cluster alternate namespace",
+			deps: []*node{
+				makeNode(ns1child1, withOwners(clusterabsentparent)),
+				makeNode(ns2child1, withOwners(clusterabsentparent)),
+			},
+			verifiedAbsent:    clusterabsentparent,
+			expectedAlternate: nil, // apiVersion/kind verified cluster-scoped, namespace delta ignored, no alternates found
+		},
+		{
+			name: "cluster alternate name",
+			deps: []*node{
+				makeNode(ns1child1, withOwners(clusterabsentparent)),
+				makeNode(ns1child1, withOwners(clusterabsentparent_name)),
+			},
+			verifiedAbsent:    clusterabsentparent,
+			expectedAlternate: &clusterabsentparent_name, // switch to alternate name, apiVersion/kind verified cluster-scoped, namespace dropped
+		},
+
+		{
+			name: "namespaced ref from namespaced child returns first if absent is sorted last",
+			deps: []*node{
+				makeNode(ns1child1, withOwners(orderedNamespacedReferences...)),
+			},
+			verifiedAbsent:    orderedNamespacedReferences[len(orderedNamespacedReferences)-1],
+			expectedAlternate: &orderedNamespacedReferences[0],
+		},
+		{
+			name: "namespaced ref from namespaced child returns next after absent",
+			deps: []*node{
+				makeNode(ns1child1, withOwners(orderedNamespacedReferences...)),
+			},
+			verifiedAbsent:    orderedNamespacedReferences[len(orderedNamespacedReferences)-2],
+			expectedAlternate: &orderedNamespacedReferences[len(orderedNamespacedReferences)-1],
+		},
+
+		{
+			name: "cluster ref from cluster child returns first if absent is sorted last",
+			deps: []*node{
+				makeNode(clusterchild1, withOwners(orderedClusterReferences...)),
+			},
+			verifiedAbsent:    orderedClusterReferences[len(orderedClusterReferences)-1],
+			expectedAlternate: &orderedClusterReferences[0],
+		},
+		{
+			name: "cluster ref from cluster child returns next after absent",
+			deps: []*node{
+				makeNode(clusterchild1, withOwners(orderedClusterReferences...)),
+			},
+			verifiedAbsent:    orderedClusterReferences[len(orderedClusterReferences)-2],
+			expectedAlternate: &orderedClusterReferences[len(orderedClusterReferences)-1],
+		},
+
+		{
+			name: "ignore unrelated",
+			deps: []*node{
+				makeNode(ns1child1, withOwners(clusterabsentparent, makeID("v1", "Parent", "ns1", "name", "anotheruid"))),
+			},
+			verifiedAbsent:    clusterabsentparent,
+			expectedAlternate: nil,
+		},
+		{
+			name: "ignore matches",
+			deps: []*node{
+				makeNode(ns1child1, withOwners(clusterabsentparent, clusterabsentparent)),
+			},
+			verifiedAbsent:    clusterabsentparent,
+			expectedAlternate: nil,
+		},
+		{
+			name: "collapse duplicates",
+			deps: []*node{
+				makeNode(clusterchild1, withOwners(clusterabsentparent, clusterabsentparent_kind, clusterabsentparent_kind)),
+			},
+			verifiedAbsent:    clusterabsentparent,
+			expectedAlternate: &clusterabsentparent_kind,
+		},
+	}
+
+	for _, tc := range testcases {
+		t.Run(tc.name, func(t *testing.T) {
+			alternate := getAlternateOwnerIdentity(tc.deps, tc.verifiedAbsent)
+			if !reflect.DeepEqual(alternate, tc.expectedAlternate) {
+				t.Errorf("expected\n%#v\ngot\n%#v", tc.expectedAlternate, alternate)
+			}
+		})
+	}
+}

pkg/controller/garbagecollector/operations.go

@@ -65,7 +65,13 @@ func (gc *GarbageCollector) getObject(item objectReference) (*metav1.PartialObje
 	if err != nil {
 		return nil, err
 	}
-	return gc.metadataClient.Resource(resource).Namespace(resourceDefaultNamespace(namespaced, item.Namespace)).Get(context.TODO(), item.Name, metav1.GetOptions{})
+	namespace := resourceDefaultNamespace(namespaced, item.Namespace)
+	if namespaced && len(namespace) == 0 {
+		// the type is namespaced, but we have no namespace coordinate.
+		// the only way this can happen is if a cluster-scoped object referenced this type as an owner.
+		return nil, namespacedOwnerOfClusterScopedObjectErr
+	}
+	return gc.metadataClient.Resource(resource).Namespace(namespace).Get(context.TODO(), item.Name, metav1.GetOptions{})
 }
 
 func (gc *GarbageCollector) patchObject(item objectReference, patch []byte, pt types.PatchType) (*metav1.PartialObjectMetadata, error) {

pkg/controller/garbagecollector/uid_cache.go

@@ -20,33 +20,32 @@ import (
 	"sync"
 
 	"github.com/golang/groupcache/lru"
-	"k8s.io/apimachinery/pkg/types"
 )
 
-// UIDCache is an LRU cache for uid.
-type UIDCache struct {
+// ReferenceCache is an LRU cache for uid.
+type ReferenceCache struct {
 	mutex sync.Mutex
 	cache *lru.Cache
 }
 
-// NewUIDCache returns a UIDCache.
-func NewUIDCache(maxCacheEntries int) *UIDCache {
-	return &UIDCache{
+// NewReferenceCache returns a ReferenceCache.
+func NewReferenceCache(maxCacheEntries int) *ReferenceCache {
+	return &ReferenceCache{
 		cache: lru.New(maxCacheEntries),
 	}
 }
 
 // Add adds a uid to the cache.
-func (c *UIDCache) Add(uid types.UID) {
+func (c *ReferenceCache) Add(reference objectReference) {
 	c.mutex.Lock()
 	defer c.mutex.Unlock()
-	c.cache.Add(uid, nil)
+	c.cache.Add(reference, nil)
 }
 
 // Has returns if a uid is in the cache.
-func (c *UIDCache) Has(uid types.UID) bool {
+func (c *ReferenceCache) Has(reference objectReference) bool {
 	c.mutex.Lock()
 	defer c.mutex.Unlock()
-	_, found := c.cache.Get(uid)
+	_, found := c.cache.Get(reference)
 	return found
 }

staging/src/k8s.io/client-go/tools/record/fake.go

@@ -27,17 +27,29 @@ import (
 // thrown away in this case.
 type FakeRecorder struct {
 	Events chan string
+
+	IncludeObject bool
+}
+
+func objectString(object runtime.Object, includeObject bool) string {
+	if !includeObject {
+		return ""
+	}
+	return fmt.Sprintf(" involvedObject{kind=%s,apiVersion=%s}",
+		object.GetObjectKind().GroupVersionKind().Kind,
+		object.GetObjectKind().GroupVersionKind().GroupVersion(),
+	)
 }
 
 func (f *FakeRecorder) Event(object runtime.Object, eventtype, reason, message string) {
 	if f.Events != nil {
-		f.Events <- fmt.Sprintf("%s %s %s", eventtype, reason, message)
+		f.Events <- fmt.Sprintf("%s %s %s%s", eventtype, reason, message, objectString(object, f.IncludeObject))
 	}
 }
 
 func (f *FakeRecorder) Eventf(object runtime.Object, eventtype, reason, messageFmt string, args ...interface{}) {
 	if f.Events != nil {
-		f.Events <- fmt.Sprintf(eventtype+" "+reason+" "+messageFmt, args...)
+		f.Events <- fmt.Sprintf(eventtype+" "+reason+" "+messageFmt, args...) + objectString(object, f.IncludeObject)
 	}
 }
 

test/integration/garbagecollector/BUILD

@@ -37,6 +37,7 @@ go_test(
         "//staging/src/k8s.io/controller-manager/pkg/informerfactory:go_default_library",
         "//test/integration:go_default_library",
         "//test/integration/framework:go_default_library",
+        "//vendor/k8s.io/utils/pointer:go_default_library",
     ],
 )
 

test/integration/garbagecollector/garbage_collector_test.go

@@ -50,6 +50,7 @@ import (
 	"k8s.io/kubernetes/pkg/controller/garbagecollector"
 	"k8s.io/kubernetes/test/integration"
 	"k8s.io/kubernetes/test/integration/framework"
+	"k8s.io/utils/pointer"
 )
 
 func getForegroundOptions() metav1.DeleteOptions {
@@ -246,6 +247,7 @@ func setupWithServer(t *testing.T, result *kubeapiservertesting.TestServer, work
 	alwaysStarted := make(chan struct{})
 	close(alwaysStarted)
 	gc, err := garbagecollector.NewGarbageCollector(
+		clientSet,
 		metadataClient,
 		restMapper,
 		garbagecollector.DefaultIgnoredResources(),
@@ -314,6 +316,143 @@ func deleteNamespaceOrDie(name string, c clientset.Interface, t *testing.T) {
 	}
 }
 
+func TestCrossNamespaceReferencesWithWatchCache(t *testing.T) {
+	testCrossNamespaceReferences(t, true)
+}
+func TestCrossNamespaceReferencesWithoutWatchCache(t *testing.T) {
+	testCrossNamespaceReferences(t, false)
+}
+
+func testCrossNamespaceReferences(t *testing.T, watchCache bool) {
+	var (
+		workers            = 5
+		validChildrenCount = 10
+		namespaceB         = "b"
+		namespaceA         = "a"
+	)
+
+	// Start the server
+	testServer := kubeapiservertesting.StartTestServerOrDie(t, nil, []string{fmt.Sprintf("--watch-cache=%v", watchCache)}, framework.SharedEtcd())
+	defer func() {
+		if testServer != nil {
+			testServer.TearDownFn()
+		}
+	}()
+	clientSet, err := clientset.NewForConfig(testServer.ClientConfig)
+	if err != nil {
+		t.Fatalf("error creating clientset: %v", err)
+	}
+
+	createNamespaceOrDie(namespaceB, clientSet, t)
+	parent, err := clientSet.CoreV1().ConfigMaps(namespaceB).Create(context.TODO(), &v1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: "parent"}}, metav1.CreateOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	for i := 0; i < validChildrenCount; i++ {
+		_, err := clientSet.CoreV1().Secrets(namespaceB).Create(context.TODO(), &v1.Secret{ObjectMeta: metav1.ObjectMeta{GenerateName: "child-", OwnerReferences: []metav1.OwnerReference{
+			{Name: "parent", Kind: "ConfigMap", APIVersion: "v1", UID: parent.UID, Controller: pointer.BoolPtr(false)},
+		}}}, metav1.CreateOptions{})
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	createNamespaceOrDie(namespaceA, clientSet, t)
+
+	// Construct invalid owner references:
+	invalidOwnerReferences := []metav1.OwnerReference{}
+	for i := 0; i < 25; i++ {
+		invalidOwnerReferences = append(invalidOwnerReferences, metav1.OwnerReference{Name: "invalid", UID: types.UID(fmt.Sprintf("invalid-%d", i)), APIVersion: "test/v1", Kind: fmt.Sprintf("invalid%d", i)})
+	}
+	invalidOwnerReferences = append(invalidOwnerReferences, metav1.OwnerReference{Name: "invalid", UID: parent.UID, APIVersion: "v1", Kind: "Pod", Controller: pointer.BoolPtr(false)})
+
+	invalidUIDs := []types.UID{}
+	for i := 0; i < workers; i++ {
+		invalidChildType1, err := clientSet.CoreV1().ConfigMaps(namespaceA).Create(context.TODO(), &v1.ConfigMap{ObjectMeta: metav1.ObjectMeta{GenerateName: "invalid-child-", OwnerReferences: invalidOwnerReferences}}, metav1.CreateOptions{})
+		if err != nil {
+			t.Fatal(err)
+		}
+		invalidChildType2, err := clientSet.CoreV1().Secrets(namespaceA).Create(context.TODO(), &v1.Secret{ObjectMeta: metav1.ObjectMeta{GenerateName: "invalid-child-a-", OwnerReferences: invalidOwnerReferences}}, metav1.CreateOptions{})
+		if err != nil {
+			t.Fatal(err)
+		}
+		invalidChildType3, err := clientSet.CoreV1().Secrets(namespaceA).Create(context.TODO(), &v1.Secret{
+			ObjectMeta: metav1.ObjectMeta{
+				Labels:          map[string]string{"single-bad-reference": "true"},
+				GenerateName:    "invalid-child-b-",
+				OwnerReferences: []metav1.OwnerReference{{Name: "invalid", UID: parent.UID, APIVersion: "v1", Kind: "Pod", Controller: pointer.BoolPtr(false)}},
+			},
+		}, metav1.CreateOptions{})
+		if err != nil {
+			t.Fatal(err)
+		}
+		invalidUIDs = append(invalidUIDs, invalidChildType1.UID, invalidChildType2.UID, invalidChildType3.UID)
+	}
+
+	// start GC with existing objects in place to simulate controller-manager restart
+	ctx := setupWithServer(t, testServer, workers)
+	defer ctx.tearDown()
+	testServer = nil
+
+	// Wait for the invalid children to be garbage collected
+	if err := wait.PollImmediate(time.Second, 10*time.Second, func() (bool, error) {
+		children, err := clientSet.CoreV1().Secrets(namespaceA).List(context.TODO(), metav1.ListOptions{LabelSelector: "single-bad-reference=true"})
+		if err != nil {
+			return false, err
+		}
+		if len(children.Items) > 0 {
+			t.Logf("expected 0 invalid children, got %d, will wait and relist", len(children.Items))
+			return false, nil
+		}
+		return true, nil
+	}); err != nil && err != wait.ErrWaitTimeout {
+		t.Error(err)
+	}
+
+	// Wait for a little while to make sure they didn't trigger deletion of the valid children
+	if err := wait.Poll(time.Second, 5*time.Second, func() (bool, error) {
+		children, err := clientSet.CoreV1().Secrets(namespaceB).List(context.TODO(), metav1.ListOptions{})
+		if err != nil {
+			return false, err
+		}
+		if len(children.Items) != validChildrenCount {
+			return false, fmt.Errorf("expected %d valid children, got %d", validChildrenCount, len(children.Items))
+		}
+		return false, nil
+	}); err != nil && err != wait.ErrWaitTimeout {
+		t.Error(err)
+	}
+
+	if !ctx.gc.GraphHasUID(parent.UID) {
+		t.Errorf("valid parent UID no longer exists in the graph")
+	}
+
+	// Now that our graph has correct data in it, add a new invalid child and see if it gets deleted
+	invalidChild, err := clientSet.CoreV1().Secrets(namespaceA).Create(context.TODO(), &v1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			GenerateName:    "invalid-child-c-",
+			OwnerReferences: []metav1.OwnerReference{{Name: "invalid", UID: parent.UID, APIVersion: "v1", Kind: "Pod", Controller: pointer.BoolPtr(false)}},
+		},
+	}, metav1.CreateOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	// Wait for the invalid child to be garbage collected
+	if err := wait.PollImmediate(time.Second, 10*time.Second, func() (bool, error) {
+		_, err := clientSet.CoreV1().Secrets(namespaceA).Get(context.TODO(), invalidChild.Name, metav1.GetOptions{})
+		if apierrors.IsNotFound(err) {
+			return true, nil
+		}
+		if err != nil {
+			return false, err
+		}
+		t.Logf("%s remains, waiting for deletion", invalidChild.Name)
+		return false, nil
+	}); err != nil {
+		t.Fatal(err)
+	}
+}
+
 // This test simulates the cascading deletion.
 func TestCascadingDeletion(t *testing.T) {
 	ctx := setup(t, 5)