issue events during synchronization

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

Author: xrstf

internal/controller/sync/controller.go

@@ -215,6 +215,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, request mcreconcile.Request)
 	if r.pubRes.Spec.EnableWorkspacePaths {
 		lc := &kcpdevcorev1alpha1.LogicalCluster{}
 		if err := vwClient.Get(ctx, types.NamespacedName{Name: kcpdevcorev1alpha1.LogicalClusterName}, lc); err != nil {
+			recorder.Event(remoteObj, corev1.EventTypeWarning, "ReconcilingError", "Failed to retrieve workspace path, cannot process object.")
 			return reconcile.Result{}, fmt.Errorf("failed to retrieve remote logicalcluster: %w", err)
 		}
 
@@ -225,11 +226,13 @@ func (r *Reconciler) Reconcile(ctx context.Context, request mcreconcile.Request)
 	// sync main object
 	syncer, err := sync.NewResourceSyncer(log, r.localClient, vwClient, r.pubRes, r.localCRD, mutation.NewMutator, r.stateNamespace, r.agentName)
 	if err != nil {
+		recorder.Event(remoteObj, corev1.EventTypeWarning, "ReconcilingError", "Failed to process object: a provider-side issue has occurred.")
 		return reconcile.Result{}, fmt.Errorf("failed to create syncer: %w", err)
 	}
 
 	requeue, err := syncer.Process(ctx, remoteObj)
 	if err != nil {
+		recorder.Event(remoteObj, corev1.EventTypeWarning, "ReconcilingError", "Failed to process object: a provider-side issue has occurred.")
 		return reconcile.Result{}, err
 	}
 

internal/sync/context.go

@@ -58,6 +58,8 @@ func workspacePathFromContext(ctx context.Context) logicalcluster.Path {
 	return path
 }
 
+// WithEventRecorder adds a event recorder to the context. The recorder must be configured
+// for the kcp side of the synchronization as the agent never issues events on the local cluster.
 func WithEventRecorder(ctx context.Context, recorder record.EventRecorder) context.Context {
 	return context.WithValue(ctx, recorderContextKey, recorder)
 }

internal/sync/object_syncer.go

@@ -31,6 +31,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	ctrlruntimeclient "sigs.k8s.io/controller-runtime/pkg/client"
 )
@@ -57,15 +58,39 @@ type objectSyncer struct {
 	mutator mutation.Mutator
 	// stateStore is capable of remembering the state of a Kubernetes object
 	stateStore ObjectStateStore
+	// eventObjSide is configuring whether the source or destination object will
+	// receive events. Since these objects might be created during the sync,
+	// they cannot be specified here directly.
+	eventObjSide syncSideType
 }
 
+type syncSideType int
+
+const (
+	syncSideSource syncSideType = iota
+	syncSideDestination
+)
+
 type syncSide struct {
 	clusterName   logicalcluster.Name
 	workspacePath logicalcluster.Path
 	client        ctrlruntimeclient.Client
 	object        *unstructured.Unstructured
 }
 
+func (s *objectSyncer) recordEvent(ctx context.Context, source, dest syncSide, eventtype, reason, msg string, args ...any) {
+	recorder := recorderFromContext(ctx)
+
+	var obj runtime.Object
+	if s.eventObjSide == syncSideDestination {
+		obj = dest.object
+	} else {
+		obj = source.object
+	}
+
+	recorder.Eventf(obj, eventtype, reason, msg, args...)
+}
+
 func (s *objectSyncer) Sync(ctx context.Context, log *zap.SugaredLogger, source, dest syncSide) (requeue bool, err error) {
 	// handle deletion: if source object is in deletion, delete the destination object (the clone)
 	if source.object.GetDeletionTimestamp() != nil {
@@ -81,6 +106,7 @@ func (s *objectSyncer) Sync(ctx context.Context, log *zap.SugaredLogger, source,
 
 		// the patch above would trigger a new reconciliation anyway
 		if updated {
+			s.recordEvent(ctx, source, dest, corev1.EventTypeNormal, "ObjectAccepted", "Object has been seen by the service provider.")
 			return true, nil
 		}
 	}
@@ -103,6 +129,7 @@ func (s *objectSyncer) Sync(ctx context.Context, log *zap.SugaredLogger, source,
 
 		// The function above either created a new destination object or patched-in the missing labels,
 		// in both cases do we want to requeue.
+		s.recordEvent(ctx, source, dest, corev1.EventTypeNormal, "ObjectPlaced", "Object has been placed.")
 		return true, nil
 	}
 
@@ -254,6 +281,8 @@ func (s *objectSyncer) syncObjectSpec(ctx context.Context, log *zap.SugaredLogge
 	}
 
 	if requeue {
+		s.recordEvent(ctx, source, dest, corev1.EventTypeNormal, "ObjectSynced", "The current desired state of the object has been synchronized.")
+
 		// remember this object state for the next reconciliation (this will strip any syncer-related
 		// metadata the 3-way diff may have added above)
 		if err := s.stateStore.Put(ctx, sourceObjCopy, source.clusterName, s.subresources); err != nil {
@@ -282,6 +311,8 @@ func (s *objectSyncer) syncObjectStatus(ctx context.Context, log *zap.SugaredLog
 		if err := source.client.Status().Update(ctx, source.object); err != nil {
 			return false, fmt.Errorf("failed to update source object status: %w", err)
 		}
+
+		s.recordEvent(ctx, source, dest, corev1.EventTypeNormal, "ObjectStatusSynced", "The current object status has been updated.")
 	}
 
 	// always return false; there is no need to requeue the source object when we changed its status
@@ -406,6 +437,7 @@ func (s *objectSyncer) handleDeletion(ctx context.Context, log *zap.SugaredLogge
 	if dest.object != nil {
 		if dest.object.GetDeletionTimestamp() == nil {
 			log.Debugw("Deleting destination object…", "dest-object", newObjectKey(dest.object, dest.clusterName, logicalcluster.None))
+			s.recordEvent(ctx, source, dest, corev1.EventTypeNormal, "ObjectCleanup", "Object deletion has been started and will progress in the background.")
 			if err := dest.client.Delete(ctx, dest.object); err != nil {
 				return false, fmt.Errorf("failed to delete destination object: %w", err)
 			}
@@ -422,6 +454,7 @@ func (s *objectSyncer) handleDeletion(ctx context.Context, log *zap.SugaredLogge
 
 	// if we just removed the finalizer, we can requeue the source object
 	if updated {
+		s.recordEvent(ctx, source, dest, corev1.EventTypeNormal, "ObjectDeleted", "Object deletion has been completed, finalizer has been removed.")
 		return true, nil
 	}
 

internal/sync/syncer.go

@@ -194,6 +194,7 @@ func (s *ResourceSyncer) Process(ctx context.Context, remoteObj *unstructured.Un
 		// (i.e. on the service cluster), so that the original and copy are linked
 		// together and can be found.
 		metadataOnDestination: true,
+		eventObjSide:          syncSideSource,
 	}
 
 	requeue, err = syncer.Sync(ctx, log, sourceSide, destSide)

internal/sync/syncer_related.go

@@ -64,16 +64,19 @@ type relatedObjectAnnotation struct {
 func (s *ResourceSyncer) processRelatedResource(ctx context.Context, log *zap.SugaredLogger, stateStore ObjectStateStore, remote, local syncSide, relRes syncagentv1alpha1.RelatedResourceSpec) (requeue bool, err error) {
 	// decide what direction to sync (local->remote vs. remote->local)
 	var (
-		origin syncSide
-		dest   syncSide
+		origin       syncSide
+		dest         syncSide
+		eventObjSide syncSideType
 	)
 
 	if relRes.Origin == syncagentv1alpha1.RelatedResourceOriginService {
 		origin = local
 		dest = remote
+		eventObjSide = syncSideDestination
 	} else {
 		origin = remote
 		dest = local
+		eventObjSide = syncSideSource
 	}
 
 	// find the all objects on the origin side that match the given criteria
@@ -143,6 +146,8 @@ func (s *ResourceSyncer) processRelatedResource(ctx context.Context, log *zap.Su
 			mutator: s.relatedMutators[relRes.Identifier],
 			// we never want to store sync-related metadata inside kcp
 			metadataOnDestination: false,
+			// events are always created on the kcp side
+			eventObjSide: eventObjSide,
 		}
 
 		req, err := syncer.Sync(ctx, log, sourceSide, destSide)