Add DependencyPolicyConflict and DependencyPolicyAggregated event

Author: Kexin2000

pkg/dependenciesdistributor/dependencies_distributor.go

@@ -20,6 +20,8 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"sort"
+	"strings"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/equality"
@@ -49,6 +51,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/source"
 
 	configv1alpha1 "github.com/karmada-io/karmada/pkg/apis/config/v1alpha1"
+	policyv1alpha1 "github.com/karmada-io/karmada/pkg/apis/policy/v1alpha1"
 	workv1alpha2 "github.com/karmada-io/karmada/pkg/apis/work/v1alpha2"
 	"github.com/karmada-io/karmada/pkg/events"
 	"github.com/karmada-io/karmada/pkg/resourceinterpreter"
@@ -585,20 +588,59 @@ func (d *DependenciesDistributor) createOrUpdateAttachedBinding(attachedBinding
 				"try again later after binding is garbage collected, see https://github.com/karmada-io/karmada/issues/6034", bindingKey)
 		}
 
+		oldConflictResolution := effectiveConflictResolution(existBinding.Spec.ConflictResolution)
+		oldPreserve := ptr.Deref(existBinding.Spec.PreserveResourcesOnDeletion, false)
+		oldRequiredByLen := len(existBinding.Spec.RequiredBy)
+
+		mergedRequiredBy := mergeBindingSnapshot(existBinding.Spec.RequiredBy, attachedBinding.Spec.RequiredBy)
+		parentsForEvent := formatBindingParents(mergedRequiredBy)
+		multiParentAfter := len(mergedRequiredBy) > 1
+
+		finalConflictResolution := effectiveConflictResolution(attachedBinding.Spec.ConflictResolution)
+		finalPreserve := ptr.Deref(attachedBinding.Spec.PreserveResourcesOnDeletion, false)
+
+		var conflictDetails []string
+		conflictDetected := false
+		shouldEmitAggregationEvent := false
+
 		// If the spec.Placement is nil, this means that existBinding is generated by the dependency mechanism.
 		// If the spec.Placement is not nil, then it must be generated by PropagationPolicy.
 		if existBinding.Spec.Placement == nil {
+			if oldRequiredByLen > 0 && oldConflictResolution != finalConflictResolution {
+				conflictDetected = true
+				conflictDetails = append(conflictDetails, fmt.Sprintf("conflictResolution existing=%s incoming=%s",
+					oldConflictResolution, finalConflictResolution))
+			}
+			if oldRequiredByLen > 0 && oldPreserve != finalPreserve {
+				conflictDetected = true
+				conflictDetails = append(conflictDetails, fmt.Sprintf("preserveResourcesOnDeletion existing=%t incoming=%t",
+					oldPreserve, finalPreserve))
+			}
 			existBinding.Spec.ConflictResolution = attachedBinding.Spec.ConflictResolution
+			shouldEmitAggregationEvent = multiParentAfter
 		}
-		existBinding.Spec.RequiredBy = mergeBindingSnapshot(existBinding.Spec.RequiredBy, attachedBinding.Spec.RequiredBy)
+
+		existBinding.Spec.RequiredBy = mergedRequiredBy
 		existBinding.Labels = util.DedupeAndMergeLabels(existBinding.Labels, attachedBinding.Labels)
 		existBinding.Spec.Resource = attachedBinding.Spec.Resource
 		existBinding.Spec.PreserveResourcesOnDeletion = attachedBinding.Spec.PreserveResourcesOnDeletion
 
+		if conflictDetected {
+			message := fmt.Sprintf("[dep-agg] conflict on %s/%s from parents %s: %s",
+				existBinding.Namespace, existBinding.Name, strings.Join(parentsForEvent, ","), strings.Join(conflictDetails, "; "))
+			d.EventRecorder.Eventf(existBinding, corev1.EventTypeWarning, events.EventReasonDependencyPolicyConflict, message)
+		}
+
 		if err := d.Client.Update(context.TODO(), existBinding); err != nil {
 			klog.Errorf("Failed to update resourceBinding(%s): %v", bindingKey, err)
 			return err
 		}
+
+		if shouldEmitAggregationEvent {
+			message := fmt.Sprintf("[dep-agg] aggregated policy for %s/%s from parents %s: conflictResolution=%s preserveResourcesOnDeletion=%t",
+				existBinding.Namespace, existBinding.Name, strings.Join(parentsForEvent, ","), finalConflictResolution, finalPreserve)
+			d.EventRecorder.Eventf(existBinding, corev1.EventTypeNormal, events.EventReasonDependencyPolicyAggregated, message)
+		}
 		return nil
 	}
 
@@ -684,6 +726,32 @@ func (d *DependenciesDistributor) SetupWithManager(mgr controllerruntime.Manager
 	})
 }
 
+func effectiveConflictResolution(value policyv1alpha1.ConflictResolution) policyv1alpha1.ConflictResolution {
+	if value == "" {
+		return policyv1alpha1.ConflictAbort
+	}
+
+	return value
+}
+
+func formatBindingParents(snapshots []workv1alpha2.BindingSnapshot) []string {
+	if len(snapshots) == 0 {
+		return nil
+	}
+
+	parents := make([]string, 0, len(snapshots))
+	for _, snapshot := range snapshots {
+		if snapshot.Namespace != "" {
+			parents = append(parents, fmt.Sprintf("%s/%s", snapshot.Namespace, snapshot.Name))
+			continue
+		}
+		parents = append(parents, snapshot.Name)
+	}
+
+	sort.Strings(parents)
+	return parents
+}
+
 func generateBindingDependedLabels(bindingID, bindingNamespace, bindingName string) map[string]string {
 	labelKey := generateBindingDependedLabelKey(bindingNamespace, bindingName)
 	return map[string]string{labelKey: bindingID}

pkg/dependenciesdistributor/dependencies_distributor_test.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"reflect"
+	"strings"
 	"testing"
 	"time"
 
@@ -34,6 +35,7 @@ import (
 	"k8s.io/client-go/dynamic"
 	dynamicfake "k8s.io/client-go/dynamic/fake"
 	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/tools/record"
 	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
@@ -42,6 +44,7 @@ import (
 	configv1alpha1 "github.com/karmada-io/karmada/pkg/apis/config/v1alpha1"
 	policyv1alpha1 "github.com/karmada-io/karmada/pkg/apis/policy/v1alpha1"
 	workv1alpha2 "github.com/karmada-io/karmada/pkg/apis/work/v1alpha2"
+	"github.com/karmada-io/karmada/pkg/events"
 	"github.com/karmada-io/karmada/pkg/util"
 	"github.com/karmada-io/karmada/pkg/util/fedinformer/genericmanager"
 	"github.com/karmada-io/karmada/pkg/util/fedinformer/keys"
@@ -3432,3 +3435,117 @@ func Test_deleteBindingFromSnapshot(t *testing.T) {
 		})
 	}
 }
+
+func TestCreateOrUpdateAttachedBinding_EmitsDependencyEvents(t *testing.T) {
+	testScheme := runtime.NewScheme()
+	utilruntime.Must(workv1alpha2.AddToScheme(testScheme))
+
+	dependentObj := &unstructured.Unstructured{}
+	dependentObj.SetAPIVersion("v1")
+	dependentObj.SetKind("ConfigMap")
+	dependentObj.SetNamespace("default")
+	dependentObj.SetName("app-config")
+	dependentObj.SetUID(types.UID("cm-uid"))
+
+	parentOne := &workv1alpha2.ResourceBinding{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "workload-a",
+			Namespace: "default",
+			Labels: map[string]string{
+				workv1alpha2.ResourceBindingPermanentIDLabel: "rb-uid-a",
+			},
+		},
+		Spec: workv1alpha2.ResourceBindingSpec{
+			Clusters:                    []workv1alpha2.TargetCluster{{Name: "member1"}},
+			ConflictResolution:          policyv1alpha1.ConflictOverwrite,
+			PreserveResourcesOnDeletion: ptr.To(true),
+		},
+	}
+	attachedFromParentOne := buildAttachedBinding(parentOne, dependentObj)
+	// Simulate existing dependency binding already persisted in the cluster.
+	existing := attachedFromParentOne.DeepCopy()
+
+	fakeClient := fake.NewClientBuilder().WithScheme(testScheme).WithObjects(existing).Build()
+	recorder := record.NewFakeRecorder(10)
+	d := &DependenciesDistributor{
+		Client:        fakeClient,
+		EventRecorder: recorder,
+	}
+
+	parentTwo := &workv1alpha2.ResourceBinding{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "workload-b",
+			Namespace: "default",
+			Labels: map[string]string{
+				workv1alpha2.ResourceBindingPermanentIDLabel: "rb-uid-b",
+			},
+		},
+		Spec: workv1alpha2.ResourceBindingSpec{
+			Clusters:                    []workv1alpha2.TargetCluster{{Name: "member2"}},
+			ConflictResolution:          policyv1alpha1.ConflictAbort,
+			PreserveResourcesOnDeletion: ptr.To(false),
+		},
+	}
+	attachedFromParentTwo := buildAttachedBinding(parentTwo, dependentObj)
+
+	if err := d.createOrUpdateAttachedBinding(attachedFromParentTwo); err != nil {
+		t.Fatalf("createOrUpdateAttachedBinding() error = %v", err)
+	}
+
+	updated := &workv1alpha2.ResourceBinding{}
+	if err := fakeClient.Get(context.TODO(), client.ObjectKeyFromObject(existing), updated); err != nil {
+		t.Fatalf("failed to fetch updated binding: %v", err)
+	}
+	if len(updated.Spec.RequiredBy) != 2 {
+		t.Fatalf("expected 2 parents after aggregation, got %d", len(updated.Spec.RequiredBy))
+	}
+
+	eventsReceived := make([]string, 0, 2)
+	for len(eventsReceived) < 2 {
+		select {
+		case evt := <-recorder.Events:
+			eventsReceived = append(eventsReceived, evt)
+		case <-time.After(time.Second):
+			t.Fatalf("timed out waiting for dependency aggregation events, got %v", eventsReceived)
+		}
+	}
+
+	var conflictEvent, aggregatedEvent string
+	for _, evt := range eventsReceived {
+		if strings.Contains(evt, events.EventReasonDependencyPolicyConflict) {
+			conflictEvent = evt
+		}
+		if strings.Contains(evt, events.EventReasonDependencyPolicyAggregated) {
+			aggregatedEvent = evt
+		}
+	}
+
+	if conflictEvent == "" {
+		t.Fatalf("expected conflict event in %v", eventsReceived)
+	}
+	if !strings.Contains(conflictEvent, "[dep-agg] conflict") {
+		t.Errorf("unexpected conflict event message: %s", conflictEvent)
+	}
+	if !strings.Contains(conflictEvent, "conflictResolution existing=Overwrite incoming=Abort") {
+		t.Errorf("conflict event missing conflictResolution details: %s", conflictEvent)
+	}
+	if !strings.Contains(conflictEvent, "preserveResourcesOnDeletion existing=true incoming=false") {
+		t.Errorf("conflict event missing preserveResourcesOnDeletion details: %s", conflictEvent)
+	}
+
+	if aggregatedEvent == "" {
+		t.Fatalf("expected aggregation event in %v", eventsReceived)
+	}
+	if !strings.Contains(aggregatedEvent, "[dep-agg] aggregated policy") {
+		t.Errorf("unexpected aggregation event message: %s", aggregatedEvent)
+	}
+	if !strings.Contains(aggregatedEvent, "conflictResolution=Abort") {
+		t.Errorf("aggregation event missing conflictResolution result: %s", aggregatedEvent)
+	}
+	if !strings.Contains(aggregatedEvent, "preserveResourcesOnDeletion=false") {
+		t.Errorf("aggregation event missing preserveResourcesOnDeletion result: %s", aggregatedEvent)
+	}
+	if !strings.Contains(aggregatedEvent, "default/workload-a") || !strings.Contains(aggregatedEvent, "default/workload-b") {
+		t.Errorf("aggregation event missing parent list: %s", aggregatedEvent)
+	}
+}

pkg/events/events.go

@@ -66,6 +66,10 @@ const (
 	EventReasonSyncScheduleResultToDependenciesSucceed = "SyncScheduleResultToDependenciesSucceed"
 	// EventReasonSyncScheduleResultToDependenciesFailed indicates sync schedule result to attached bindings failed.
 	EventReasonSyncScheduleResultToDependenciesFailed = "SyncScheduleResultToDependenciesFailed"
+	// EventReasonDependencyPolicyConflict indicates conflict detected while aggregating dependency policies.
+	EventReasonDependencyPolicyConflict = "DependencyPolicyConflict"
+	// EventReasonDependencyPolicyAggregated indicates dependency policies aggregated successfully.
+	EventReasonDependencyPolicyAggregated = "DependencyPolicyAggregated"
 )
 
 // Define events for ResourceBinding, ClusterResourceBinding objects and their associated resources.