Merge pull request #6674 from zhzhuang-zju/changebykarmada

fix the issue that the relevant fields in rb and pp are inconsistent

Author: karmada-bot

pkg/detector/detector.go

@@ -343,13 +343,26 @@ func (d *ResourceDetector) OnUpdate(oldObj, newObj interface{}) {
 		return
 	}
 
-	resourceChangeByKarmada := eventfilter.ResourceChangeByKarmada(unstructuredOldObj, unstructuredNewObj)
+	isLazyActivation, err := d.isClaimedByLazyPolicy(unstructuredNewObj)
+	if err != nil {
+		// should never come here
+		klog.Errorf("Failed to check if the object (kind=%s, %s/%s) is bound by lazy policy. err: %v", unstructuredNewObj.GetKind(), unstructuredNewObj.GetNamespace(), unstructuredNewObj.GetName(), err)
+	}
 
-	resourceItem := ResourceItem{
-		Obj:                     newRuntimeObj,
-		ResourceChangeByKarmada: resourceChangeByKarmada,
+	if isLazyActivation {
+		resourceItem := ResourceItem{
+			Obj:                     newRuntimeObj,
+			ResourceChangeByKarmada: eventfilter.ResourceChangeByKarmada(unstructuredOldObj, unstructuredNewObj),
+		}
+
+		d.Processor.Enqueue(resourceItem)
+		return
 	}
 
+	// For non-lazy policies, it is no need to distinguish whether the change is from Karmada or not.
+	resourceItem := ResourceItem{
+		Obj: newRuntimeObj,
+	}
 	d.Processor.Enqueue(resourceItem)
 }
 
@@ -1208,7 +1221,7 @@ func (d *ResourceDetector) HandlePropagationPolicyCreationOrUpdate(policy *polic
 		if err != nil {
 			return err
 		}
-		d.Processor.Add(keys.ClusterWideKeyWithConfig{ClusterWideKey: resourceKey, ResourceChangeByKarmada: true})
+		d.enqueueResourceTemplateForPolicyChange(resourceKey, policy.Spec.ActivationPreference)
 	}
 
 	// check whether there are matched RT in waiting list, is so, add it to processor
@@ -1226,7 +1239,7 @@ func (d *ResourceDetector) HandlePropagationPolicyCreationOrUpdate(policy *polic
 
 	for _, key := range matchedKeys {
 		d.RemoveWaiting(key)
-		d.Processor.Add(keys.ClusterWideKeyWithConfig{ClusterWideKey: key, ResourceChangeByKarmada: true})
+		d.enqueueResourceTemplateForPolicyChange(key, policy.Spec.ActivationPreference)
 	}
 
 	// If preemption is enabled, handle the preemption process.
@@ -1275,14 +1288,14 @@ func (d *ResourceDetector) HandleClusterPropagationPolicyCreationOrUpdate(policy
 		if err != nil {
 			return err
 		}
-		d.Processor.Add(keys.ClusterWideKeyWithConfig{ClusterWideKey: resourceKey, ResourceChangeByKarmada: true})
+		d.enqueueResourceTemplateForPolicyChange(resourceKey, policy.Spec.ActivationPreference)
 	}
 	for _, crb := range clusterResourceBindings.Items {
 		resourceKey, err := helper.ConstructClusterWideKey(crb.Spec.Resource)
 		if err != nil {
 			return err
 		}
-		d.Processor.Add(keys.ClusterWideKeyWithConfig{ClusterWideKey: resourceKey, ResourceChangeByKarmada: true})
+		d.enqueueResourceTemplateForPolicyChange(resourceKey, policy.Spec.ActivationPreference)
 	}
 
 	matchedKeys := d.GetMatching(policy.Spec.ResourceSelectors)
@@ -1299,7 +1312,7 @@ func (d *ResourceDetector) HandleClusterPropagationPolicyCreationOrUpdate(policy
 
 	for _, key := range matchedKeys {
 		d.RemoveWaiting(key)
-		d.Processor.Add(keys.ClusterWideKeyWithConfig{ClusterWideKey: key, ResourceChangeByKarmada: true})
+		d.enqueueResourceTemplateForPolicyChange(key, policy.Spec.ActivationPreference)
 	}
 
 	// If preemption is enabled, handle the preemption process.
@@ -1472,3 +1485,21 @@ func (d *ResourceDetector) applyReplicaInterpretation(object *unstructured.Unstr
 
 	return nil
 }
+
+// enqueueResourceTemplateForPolicyChange enqueues a resource template key for reconciliation in response to a
+// PropagationPolicy or ClusterPropagationPolicy change. If the policy's ActivationPreference is set to Lazy,
+// the ResourceChangeByKarmada flag is set to true, indicating that the resource template is being enqueued
+// due to a policy change and should not be propagated to member clusters. For non-lazy policies, this flag
+// is omitted as the distinction is unnecessary.
+//
+// Note: Setting ResourceChangeByKarmada changes the effective queue key. Mixing both true/false for the same
+// resource may result in two different queue keys being processed concurrently, which can cause race conditions.
+// Therefore, only set ResourceChangeByKarmada in lazy activation mode.
+// For more details, see: https://github.com/karmada-io/karmada/issues/5996.
+func (d *ResourceDetector) enqueueResourceTemplateForPolicyChange(key keys.ClusterWideKey, pref policyv1alpha1.ActivationPreference) {
+	if util.IsLazyActivationEnabled(pref) {
+		d.Processor.Add(keys.ClusterWideKeyWithConfig{ClusterWideKey: key, ResourceChangeByKarmada: true})
+		return
+	}
+	d.Processor.Add(keys.ClusterWideKeyWithConfig{ClusterWideKey: key})
+}

pkg/detector/detector_test.go

@@ -429,7 +429,6 @@ func TestOnUpdate(t *testing.T) {
 		oldObj                    interface{}
 		newObj                    interface{}
 		expectedEnqueue           bool
-		expectedChangeByKarmada   bool
 		expectToUnstructuredError bool
 	}{
 		{
@@ -460,8 +459,7 @@ func TestOnUpdate(t *testing.T) {
 					},
 				},
 			},
-			expectedEnqueue:         true,
-			expectedChangeByKarmada: false,
+			expectedEnqueue: true,
 		},
 		{
 			name: "update without changes",
@@ -524,8 +522,7 @@ func TestOnUpdate(t *testing.T) {
 					},
 				},
 			},
-			expectedEnqueue:         true,
-			expectedChangeByKarmada: true,
+			expectedEnqueue: true,
 		},
 		{
 			name: "core v1 object",
@@ -573,7 +570,6 @@ func TestOnUpdate(t *testing.T) {
 				assert.IsType(t, ResourceItem{}, mockProcessor.lastEnqueued, "Enqueued item should be of type ResourceItem")
 				enqueued := mockProcessor.lastEnqueued.(ResourceItem)
 				assert.Equal(t, tt.newObj, enqueued.Obj, "Enqueued object should match the new object")
-				assert.Equal(t, tt.expectedChangeByKarmada, enqueued.ResourceChangeByKarmada, "ResourceChangeByKarmada flag should match expected value")
 			} else {
 				assert.Equal(t, 0, mockProcessor.enqueueCount, "Object should not be enqueued")
 			}
@@ -967,6 +963,71 @@ func TestApplyClusterPolicy(t *testing.T) {
 	}
 }
 
+func TestEnqueueResourceKeyWithActivationPref(t *testing.T) {
+	testClusterWideKey := keys.ClusterWideKey{
+		Group:     "foo",
+		Version:   "foo",
+		Kind:      "foo",
+		Namespace: "foo",
+		Name:      "foo",
+	}
+	tests := []struct {
+		name string
+		key  keys.ClusterWideKey
+		pref policyv1alpha1.ActivationPreference
+		want keys.ClusterWideKeyWithConfig
+	}{
+		{
+			name: "lazy pp and resourceChangeByKarmada is true",
+			key:  testClusterWideKey,
+			pref: policyv1alpha1.LazyActivation,
+			want: keys.ClusterWideKeyWithConfig{
+				ClusterWideKey:          testClusterWideKey,
+				ResourceChangeByKarmada: true,
+			},
+		},
+		{
+			name: "non-lazy ignores ResourceChangeByKarmada",
+			key:  testClusterWideKey,
+			pref: "",
+			want: keys.ClusterWideKeyWithConfig{
+				ClusterWideKey:          testClusterWideKey,
+				ResourceChangeByKarmada: false,
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctx, cancel := context.WithCancel(context.Background())
+			detector := ResourceDetector{
+				Processor: util.NewAsyncWorker(util.Options{
+					Name:    "resource detector",
+					KeyFunc: ResourceItemKeyFunc,
+					ReconcileFunc: func(key util.QueueKey) (err error) {
+						defer cancel()
+						defer func() {
+							assert.NoError(t, err)
+						}()
+						clusterWideKeyWithConfig, ok := key.(keys.ClusterWideKeyWithConfig)
+						if !ok {
+							err = fmt.Errorf("invalid key")
+							return err
+						}
+						if clusterWideKeyWithConfig != tt.want {
+							err = fmt.Errorf("unexpected key. want:%+v, got:%+v", tt.want, clusterWideKeyWithConfig)
+							return err
+						}
+						return nil
+					},
+				}),
+			}
+			detector.Processor.Run(ctx, 1)
+			detector.enqueueResourceTemplateForPolicyChange(tt.key, tt.pref)
+			<-ctx.Done()
+		})
+	}
+}
+
 // Helper Functions
 
 // setupTestScheme creates a runtime scheme with necessary types for testing

pkg/detector/policy.go

@@ -369,6 +369,45 @@ func (d *ResourceDetector) listCPPDerivedCRBs(policyID, policyName string) (*wor
 	return bindings, nil
 }
 
+func (d *ResourceDetector) isClaimedByLazyPolicy(obj *unstructured.Unstructured) (bool, error) {
+	policyAnnotations := obj.GetAnnotations()
+	policyLabels := obj.GetLabels()
+	policyNamespace := util.GetAnnotationValue(policyAnnotations, policyv1alpha1.PropagationPolicyNamespaceAnnotation)
+	policyName := util.GetAnnotationValue(policyAnnotations, policyv1alpha1.PropagationPolicyNameAnnotation)
+	claimedID := util.GetLabelValue(policyLabels, policyv1alpha1.PropagationPolicyPermanentIDLabel)
+	if policyNamespace != "" && policyName != "" && claimedID != "" {
+		matchedPropagationPolicy := &policyv1alpha1.PropagationPolicy{}
+		err := d.Client.Get(context.TODO(), client.ObjectKey{Namespace: policyNamespace, Name: policyName}, matchedPropagationPolicy)
+		if err != nil {
+			if apierrors.IsNotFound(err) {
+				return false, nil
+			}
+
+			return false, err
+		}
+
+		return util.IsLazyActivationEnabled(matchedPropagationPolicy.Spec.ActivationPreference), nil
+	}
+
+	policyName = util.GetAnnotationValue(policyAnnotations, policyv1alpha1.ClusterPropagationPolicyAnnotation)
+	claimedID = util.GetLabelValue(policyLabels, policyv1alpha1.ClusterPropagationPolicyPermanentIDLabel)
+	if policyName != "" && claimedID != "" {
+		matchedClusterPropagationPolicy := &policyv1alpha1.ClusterPropagationPolicy{}
+		err := d.Client.Get(context.TODO(), client.ObjectKey{Name: policyName}, matchedClusterPropagationPolicy)
+		if err != nil {
+			if apierrors.IsNotFound(err) {
+				return false, nil
+			}
+
+			return false, err
+		}
+
+		return util.IsLazyActivationEnabled(matchedClusterPropagationPolicy.Spec.ActivationPreference), nil
+	}
+
+	return false, nil
+}
+
 // excludeClusterPolicy excludes cluster propagation policy.
 // If propagation policy was claimed, cluster propagation policy should not exist.
 func excludeClusterPolicy(obj metav1.Object) (hasClaimedClusterPolicy bool) {