use context to transport cluster, workspace and now recorder

On-behalf-of: @SAP [email protected]

Author: xrstf

internal/controller/sync/controller.go

@@ -205,9 +205,12 @@ func (r *Reconciler) Reconcile(ctx context.Context, request mcreconcile.Request)
 		return reconcile.Result{}, nil
 	}
 
-	cInfo := sync.NewClusterInfo(logicalcluster.Name(request.ClusterName))
+	recorder := cl.GetEventRecorderFor(ControllerName)
 
-	// if desired, fetch the cluster path as well (some downstream service providers might make use of it,
+	ctx = sync.WithClusterName(ctx, logicalcluster.Name(request.ClusterName))
+	ctx = sync.WithEventRecorder(ctx, recorder)
+
+	// if desired, fetch the workspace path as well (some downstream service providers might make use of it,
 	// but since it requires an additional permission claim, it's optional)
 	if r.pubRes.Spec.EnableWorkspacePaths {
 		lc := &kcpdevcorev1alpha1.LogicalCluster{}
@@ -216,7 +219,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, request mcreconcile.Request)
 		}
 
 		path := lc.Annotations[kcpcore.LogicalClusterPathAnnotationKey]
-		cInfo = cInfo.WithWorkspacePath(logicalcluster.NewPath(path))
+		ctx = sync.WithWorkspacePath(ctx, logicalcluster.NewPath(path))
 	}
 
 	// sync main object
@@ -225,7 +228,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, request mcreconcile.Request)
 		return reconcile.Result{}, fmt.Errorf("failed to create syncer: %w", err)
 	}
 
-	requeue, err := syncer.Process(ctx, cInfo, remoteObj)
+	requeue, err := syncer.Process(ctx, remoteObj)
 	if err != nil {
 		return reconcile.Result{}, err
 	}

internal/sync/context.go

@@ -17,23 +17,56 @@ limitations under the License.
 package sync
 
 import (
+	"context"
+
 	"github.com/kcp-dev/logicalcluster/v3"
+
+	"k8s.io/client-go/tools/record"
+)
+
+type contextKey int
+
+const (
+	clusterContextKey contextKey = iota
+	workspaceContextKey
+	recorderContextKey
 )
 
-type clusterInfo struct {
-	clusterName   logicalcluster.Name
-	workspacePath logicalcluster.Path
+func WithClusterName(ctx context.Context, cluster logicalcluster.Name) context.Context {
+	return context.WithValue(ctx, clusterContextKey, cluster)
 }
 
-func NewClusterInfo(clusterName logicalcluster.Name) clusterInfo {
-	return clusterInfo{
-		clusterName: clusterName,
+func clusterFromContext(ctx context.Context) logicalcluster.Name {
+	cluster, ok := ctx.Value(clusterContextKey).(logicalcluster.Name)
+	if !ok {
+		return ""
 	}
+
+	return cluster
+}
+
+func WithWorkspacePath(ctx context.Context, path logicalcluster.Path) context.Context {
+	return context.WithValue(ctx, workspaceContextKey, path)
 }
 
-func (c *clusterInfo) WithWorkspacePath(path logicalcluster.Path) clusterInfo {
-	return clusterInfo{
-		clusterName:   c.clusterName,
-		workspacePath: path,
+func workspacePathFromContext(ctx context.Context) logicalcluster.Path {
+	path, ok := ctx.Value(workspaceContextKey).(logicalcluster.Path)
+	if !ok {
+		return logicalcluster.None
 	}
+
+	return path
+}
+
+func WithEventRecorder(ctx context.Context, recorder record.EventRecorder) context.Context {
+	return context.WithValue(ctx, recorderContextKey, recorder)
+}
+
+func recorderFromContext(ctx context.Context) record.EventRecorder {
+	cluster, ok := ctx.Value(recorderContextKey).(record.EventRecorder)
+	if !ok {
+		return nil
+	}
+
+	return cluster
 }

internal/sync/syncer.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/kcp-dev/logicalcluster/v3"
 	"go.uber.org/zap"
 
 	"github.com/kcp-dev/api-syncagent/internal/mutation"
@@ -136,11 +137,16 @@ func NewResourceSyncer(
 // Each of these steps can potentially end the current processing and return (true, nil). In this
 // case, the caller should re-fetch the remote object and call Process() again (most likely in the
 // next reconciliation). Only when (false, nil) is returned is the entire process finished.
-func (s *ResourceSyncer) Process(ctx context.Context, info clusterInfo, remoteObj *unstructured.Unstructured) (requeue bool, err error) {
-	log := s.log.With("source-object", newObjectKey(remoteObj, info.clusterName, info.workspacePath))
+// The context must contain a cluster name and event recorder, optionally a workspace path.
+func (s *ResourceSyncer) Process(ctx context.Context, remoteObj *unstructured.Unstructured) (requeue bool, err error) {
+	clusterName := clusterFromContext(ctx)
+	workspacePath := workspacePathFromContext(ctx)
+	objectKey := newObjectKey(remoteObj, clusterName, workspacePath)
+
+	log := s.log.With("source-object", objectKey)
 
 	// find the local equivalent object in the local service cluster
-	localObj, err := s.findLocalObject(ctx, info, remoteObj)
+	localObj, err := s.findLocalObject(ctx, objectKey)
 	if err != nil {
 		return false, fmt.Errorf("failed to find local equivalent: %w", err)
 	}
@@ -151,8 +157,8 @@ func (s *ResourceSyncer) Process(ctx context.Context, info clusterInfo, remoteOb
 	// Prepare object sync sides.
 
 	sourceSide := syncSide{
-		clusterName:   info.clusterName,
-		workspacePath: info.workspacePath,
+		clusterName:   clusterName,
+		workspacePath: workspacePath,
 		client:        s.remoteClient,
 		object:        remoteObj,
 	}
@@ -172,7 +178,7 @@ func (s *ResourceSyncer) Process(ctx context.Context, info clusterInfo, remoteOb
 		agentName:    s.agentName,
 		subresources: s.subresources,
 		// use the projection and renaming rules configured in the PublishedResource
-		destCreator: s.newLocalObjectCreator(info),
+		destCreator: s.newLocalObjectCreator(clusterName, workspacePath),
 		// for the main resource, status subresource handling is enabled (this
 		// means _allowing_ status back-syncing, it still depends on whether the
 		// status subresource even exists whether an update happens)
@@ -210,8 +216,8 @@ func (s *ResourceSyncer) Process(ctx context.Context, info clusterInfo, remoteOb
 	return s.processRelatedResources(ctx, log, stateStore, sourceSide, destSide)
 }
 
-func (s *ResourceSyncer) findLocalObject(ctx context.Context, info clusterInfo, remoteObj *unstructured.Unstructured) (*unstructured.Unstructured, error) {
-	localSelector := labels.SelectorFromSet(newObjectKey(remoteObj, info.clusterName, info.workspacePath).Labels())
+func (s *ResourceSyncer) findLocalObject(ctx context.Context, objectKey objectKey) (*unstructured.Unstructured, error) {
+	localSelector := labels.SelectorFromSet(objectKey.Labels())
 
 	localObjects := &unstructured.UnstructuredList{}
 	localObjects.SetAPIVersion(s.destDummy.GetAPIVersion())
@@ -234,7 +240,7 @@ func (s *ResourceSyncer) findLocalObject(ctx context.Context, info clusterInfo,
 	}
 }
 
-func (s *ResourceSyncer) newLocalObjectCreator(info clusterInfo) objectCreatorFunc {
+func (s *ResourceSyncer) newLocalObjectCreator(clusterName logicalcluster.Name, workspacePath logicalcluster.Path) objectCreatorFunc {
 	return func(remoteObj *unstructured.Unstructured) (*unstructured.Unstructured, error) {
 		// map from the remote API into the actual, local API group
 		destObj := remoteObj.DeepCopy()
@@ -244,7 +250,7 @@ func (s *ResourceSyncer) newLocalObjectCreator(info clusterInfo) objectCreatorFu
 		destScope := syncagentv1alpha1.ResourceScope(s.localCRD.Spec.Scope)
 
 		// map namespace/name
-		mappedName, err := templating.GenerateLocalObjectName(s.pubRes, remoteObj, info.clusterName, info.workspacePath)
+		mappedName, err := templating.GenerateLocalObjectName(s.pubRes, remoteObj, clusterName, workspacePath)
 		if err != nil {
 			return nil, fmt.Errorf("failed to generate local object name: %w", err)
 		}

internal/sync/syncer_test.go

@@ -37,6 +37,7 @@ import (
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	yamlutil "k8s.io/apimachinery/pkg/util/yaml"
+	"k8s.io/client-go/tools/record"
 	ctrlruntimeclient "sigs.k8s.io/controller-runtime/pkg/client"
 	fakectrlruntimeclient "sigs.k8s.io/controller-runtime/pkg/client/fake"
 )
@@ -910,7 +911,8 @@ func TestSyncerProcessingSingleResourceWithoutStatus(t *testing.T) {
 			}
 
 			ctx := t.Context()
-			cInfo := NewClusterInfo(clusterName)
+			ctx = WithClusterName(ctx, clusterName)
+			ctx = WithEventRecorder(ctx, record.NewFakeRecorder(99))
 
 			// setup a custom state backend that we can prime
 			var backend *kubernetesBackend
@@ -940,7 +942,7 @@ func TestSyncerProcessingSingleResourceWithoutStatus(t *testing.T) {
 						t.Fatalf("Detected potential infinite loop, stopping after %d requeues.", i)
 					}
 
-					requeue, err = syncer.Process(ctx, cInfo, target)
+					requeue, err = syncer.Process(ctx, target)
 					if err != nil {
 						break
 					}
@@ -960,7 +962,7 @@ func TestSyncerProcessingSingleResourceWithoutStatus(t *testing.T) {
 					}
 				}
 			} else {
-				requeue, err = syncer.Process(ctx, cInfo, testcase.remoteObject)
+				requeue, err = syncer.Process(ctx, testcase.remoteObject)
 			}
 
 			finalRemoteObject, getErr := getFinalObjectVersion(ctx, remoteClient, testcase.remoteObject, testcase.expectedRemoteObject)
@@ -1217,7 +1219,8 @@ func TestSyncerProcessingSingleResourceWithStatus(t *testing.T) {
 			}
 
 			ctx := t.Context()
-			cInfo := NewClusterInfo(clusterName)
+			ctx = WithClusterName(ctx, clusterName)
+			ctx = WithEventRecorder(ctx, record.NewFakeRecorder(99))
 
 			// setup a custom state backend that we can prime
 			var backend *kubernetesBackend
@@ -1247,7 +1250,7 @@ func TestSyncerProcessingSingleResourceWithStatus(t *testing.T) {
 						t.Fatalf("Detected potential infinite loop, stopping after %d requeues.", i)
 					}
 
-					requeue, err = syncer.Process(ctx, cInfo, target)
+					requeue, err = syncer.Process(ctx, target)
 					if err != nil {
 						break
 					}
@@ -1267,7 +1270,7 @@ func TestSyncerProcessingSingleResourceWithStatus(t *testing.T) {
 					}
 				}
 			} else {
-				requeue, err = syncer.Process(ctx, cInfo, testcase.remoteObject)
+				requeue, err = syncer.Process(ctx, testcase.remoteObject)
 			}
 
 			finalRemoteObject, getErr := getFinalObjectVersion(ctx, remoteClient, testcase.remoteObject, testcase.expectedRemoteObject)

internal/sync/templating/naming.go

@@ -40,11 +40,11 @@ type localObjectNamingContext struct {
 	ClusterPath logicalcluster.Path
 }
 
-func newLocalObjectNamingContext(object *unstructured.Unstructured, clusterName logicalcluster.Name, clusterPath logicalcluster.Path) localObjectNamingContext {
+func newLocalObjectNamingContext(object *unstructured.Unstructured, clusterName logicalcluster.Name, workspacePath logicalcluster.Path) localObjectNamingContext {
 	return localObjectNamingContext{
 		Object:      object.Object,
 		ClusterName: clusterName,
-		ClusterPath: clusterPath,
+		ClusterPath: workspacePath,
 	}
 }
 
@@ -53,7 +53,7 @@ var defaultNamingScheme = syncagentv1alpha1.ResourceNaming{
 	Name:      "{{ .Object.metadata.namespace | sha3short }}-{{ .Object.metadata.name | sha3short }}",
 }
 
-func GenerateLocalObjectName(pr *syncagentv1alpha1.PublishedResource, object *unstructured.Unstructured, clusterName logicalcluster.Name, clusterPath logicalcluster.Path) (types.NamespacedName, error) {
+func GenerateLocalObjectName(pr *syncagentv1alpha1.PublishedResource, object *unstructured.Unstructured, clusterName logicalcluster.Name, workspacePath logicalcluster.Path) (types.NamespacedName, error) {
 	naming := pr.Spec.Naming
 	if naming == nil {
 		naming = &syncagentv1alpha1.ResourceNaming{}
@@ -65,7 +65,7 @@ func GenerateLocalObjectName(pr *syncagentv1alpha1.PublishedResource, object *un
 	if pattern == "" {
 		pattern = defaultNamingScheme.Namespace
 	}
-	rendered, err := generateLocalObjectIdentifier(pattern, object, clusterName, clusterPath)
+	rendered, err := generateLocalObjectIdentifier(pattern, object, clusterName, workspacePath)
 	if err != nil {
 		return result, fmt.Errorf("invalid namespace naming: %w", err)
 	}
@@ -76,7 +76,7 @@ func GenerateLocalObjectName(pr *syncagentv1alpha1.PublishedResource, object *un
 	if pattern == "" {
 		pattern = defaultNamingScheme.Name
 	}
-	rendered, err = generateLocalObjectIdentifier(pattern, object, clusterName, clusterPath)
+	rendered, err = generateLocalObjectIdentifier(pattern, object, clusterName, workspacePath)
 	if err != nil {
 		return result, fmt.Errorf("invalid name naming: %w", err)
 	}
@@ -86,10 +86,10 @@ func GenerateLocalObjectName(pr *syncagentv1alpha1.PublishedResource, object *un
 	return result, nil
 }
 
-func generateLocalObjectIdentifier(pattern string, object *unstructured.Unstructured, clusterName logicalcluster.Name, clusterPath logicalcluster.Path) (string, error) {
+func generateLocalObjectIdentifier(pattern string, object *unstructured.Unstructured, clusterName logicalcluster.Name, workspacePath logicalcluster.Path) (string, error) {
 	// modern Go template style
 	if strings.Contains(pattern, "{{") {
-		return Render(pattern, newLocalObjectNamingContext(object, clusterName, clusterPath))
+		return Render(pattern, newLocalObjectNamingContext(object, clusterName, workspacePath))
 	}
 
 	// Legacy $variable style, does also not support clusterPath;