UPSTREAM: <carry>: support logical clusters

Signed-off-by: Andy Goldstein <[email protected]>
Co-authored-by: Fabian von Feilitzsch <[email protected]>

Author: ncdc

pkg/cache/internal/cache_reader.go

@@ -21,6 +21,9 @@ import (
 	"fmt"
 	"reflect"
 
+	kcpcache "github.com/kcp-dev/apimachinery/pkg/cache"
+	"github.com/kcp-dev/logicalcluster"
+
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/fields"
@@ -54,11 +57,11 @@ type CacheReader struct {
 }
 
 // Get checks the indexer for the object and writes a copy of it if found.
-func (c *CacheReader) Get(_ context.Context, key client.ObjectKey, out client.Object) error {
+func (c *CacheReader) Get(ctx context.Context, key client.ObjectKey, out client.Object) error {
 	if c.scopeName == apimeta.RESTScopeNameRoot {
 		key.Namespace = ""
 	}
-	storeKey := objectKeyToStoreKey(key)
+	storeKey := objectKeyToStoreKey(ctx, key)
 
 	// Lookup the object from the indexer cache
 	obj, exists, err := c.indexer.GetByKey(storeKey)
@@ -105,13 +108,15 @@ func (c *CacheReader) Get(_ context.Context, key client.ObjectKey, out client.Ob
 }
 
 // List lists items out of the indexer and writes them to out.
-func (c *CacheReader) List(_ context.Context, out client.ObjectList, opts ...client.ListOption) error {
+func (c *CacheReader) List(ctx context.Context, out client.ObjectList, opts ...client.ListOption) error {
 	var objs []interface{}
 	var err error
 
 	listOpts := client.ListOptions{}
 	listOpts.ApplyOptions(opts)
 
+	clusterName, _ := logicalcluster.ClusterFromContext(ctx)
+
 	switch {
 	case listOpts.FieldSelector != nil:
 		// TODO(directxman12): support more complicated field selectors by
@@ -125,9 +130,17 @@ func (c *CacheReader) List(_ context.Context, out client.ObjectList, opts ...cli
 		// namespace.
 		objs, err = c.indexer.ByIndex(FieldIndexName(field), KeyToNamespacedKey(listOpts.Namespace, val))
 	case listOpts.Namespace != "":
-		objs, err = c.indexer.ByIndex(cache.NamespaceIndex, listOpts.Namespace)
+		if clusterName.Empty() {
+			objs, err = c.indexer.ByIndex(cache.NamespaceIndex, listOpts.Namespace)
+		} else {
+			objs, err = c.indexer.ByIndex(kcpcache.ClusterAndNamespaceIndexName, kcpcache.ToClusterAwareKey(clusterName.String(), listOpts.Namespace, ""))
+		}
 	default:
-		objs = c.indexer.List()
+		if clusterName.Empty() {
+			objs = c.indexer.List()
+		} else {
+			objs, err = c.indexer.ByIndex(kcpcache.ClusterIndexName, kcpcache.ToClusterAwareKey(clusterName.String(), "", ""))
+		}
 	}
 	if err != nil {
 		return err
@@ -179,7 +192,12 @@ func (c *CacheReader) List(_ context.Context, out client.ObjectList, opts ...cli
 // It's akin to MetaNamespaceKeyFunc.  It's separate from
 // String to allow keeping the key format easily in sync with
 // MetaNamespaceKeyFunc.
-func objectKeyToStoreKey(k client.ObjectKey) string {
+func objectKeyToStoreKey(ctx context.Context, k client.ObjectKey) string {
+	cluster, ok := logicalcluster.ClusterFromContext(ctx)
+	if ok {
+		return kcpcache.ToClusterAwareKey(cluster.String(), k.Namespace, k.Name)
+	}
+
 	if k.Namespace == "" {
 		return k.Name
 	}

pkg/handler/enqueue.go

@@ -17,8 +17,10 @@ limitations under the License.
 package handler
 
 import (
+	"github.com/kcp-dev/logicalcluster"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/util/workqueue"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/event"
 	logf "sigs.k8s.io/controller-runtime/pkg/internal/log"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
@@ -41,25 +43,16 @@ func (e *EnqueueRequestForObject) Create(evt event.CreateEvent, q workqueue.Rate
 		enqueueLog.Error(nil, "CreateEvent received with no metadata", "event", evt)
 		return
 	}
-	q.Add(reconcile.Request{NamespacedName: types.NamespacedName{
-		Name:      evt.Object.GetName(),
-		Namespace: evt.Object.GetNamespace(),
-	}})
+	q.Add(request(evt.Object))
 }
 
 // Update implements EventHandler.
 func (e *EnqueueRequestForObject) Update(evt event.UpdateEvent, q workqueue.RateLimitingInterface) {
 	switch {
 	case evt.ObjectNew != nil:
-		q.Add(reconcile.Request{NamespacedName: types.NamespacedName{
-			Name:      evt.ObjectNew.GetName(),
-			Namespace: evt.ObjectNew.GetNamespace(),
-		}})
+		q.Add(request(evt.ObjectNew))
 	case evt.ObjectOld != nil:
-		q.Add(reconcile.Request{NamespacedName: types.NamespacedName{
-			Name:      evt.ObjectOld.GetName(),
-			Namespace: evt.ObjectOld.GetNamespace(),
-		}})
+		q.Add(request(evt.ObjectOld))
 	default:
 		enqueueLog.Error(nil, "UpdateEvent received with no metadata", "event", evt)
 	}
@@ -71,10 +64,7 @@ func (e *EnqueueRequestForObject) Delete(evt event.DeleteEvent, q workqueue.Rate
 		enqueueLog.Error(nil, "DeleteEvent received with no metadata", "event", evt)
 		return
 	}
-	q.Add(reconcile.Request{NamespacedName: types.NamespacedName{
-		Name:      evt.Object.GetName(),
-		Namespace: evt.Object.GetNamespace(),
-	}})
+	q.Add(request(evt.Object))
 }
 
 // Generic implements EventHandler.
@@ -83,8 +73,16 @@ func (e *EnqueueRequestForObject) Generic(evt event.GenericEvent, q workqueue.Ra
 		enqueueLog.Error(nil, "GenericEvent received with no metadata", "event", evt)
 		return
 	}
-	q.Add(reconcile.Request{NamespacedName: types.NamespacedName{
-		Name:      evt.Object.GetName(),
-		Namespace: evt.Object.GetNamespace(),
-	}})
+	q.Add(request(evt.Object))
+}
+
+func request(obj client.Object) reconcile.Request {
+	return reconcile.Request{
+		// TODO(kcp) Need to implement a non-kcp-specific way to support this
+		ClusterName: logicalcluster.From(obj).String(),
+		NamespacedName: types.NamespacedName{
+			Namespace: obj.GetNamespace(),
+			Name:      obj.GetName(),
+		},
+	}
 }

pkg/handler/enqueue_owner.go

@@ -19,6 +19,7 @@ package handler
 import (
 	"fmt"
 
+	"github.com/kcp-dev/logicalcluster"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -134,9 +135,12 @@ func (e *EnqueueRequestForOwner) getOwnerReconcileRequest(object metav1.Object,
 		// object in the event.
 		if ref.Kind == e.groupKind.Kind && refGV.Group == e.groupKind.Group {
 			// Match found - add a Request for the object referred to in the OwnerReference
-			request := reconcile.Request{NamespacedName: types.NamespacedName{
-				Name: ref.Name,
-			}}
+			request := reconcile.Request{
+				ClusterName: logicalcluster.From(object).String(),
+				NamespacedName: types.NamespacedName{
+					Name: ref.Name,
+				},
+			}
 
 			// if owner is not namespaced then we should set the namespace to the empty
 			mapping, err := e.mapper.RESTMapping(e.groupKind, refGV.Version)

pkg/reconcile/reconcile.go

@@ -47,6 +47,10 @@ func (r *Result) IsZero() bool {
 type Request struct {
 	// NamespacedName is the name and namespace of the object to reconcile.
 	types.NamespacedName
+
+	// ClusterName can be used for reconciling requests across multiple clusters,
+	// to prevent objects with the same name and namespace from conflicting
+	ClusterName string
 }
 
 /*