Merge pull request #5204 from liangyuanpeng/statefulset_generation

Align federated StatefulSet's observedGeneration semantics with its native

pkg/resourceinterpreter/default/native/aggregatestatus.go

@@ -95,10 +95,9 @@ func aggregateDeploymentStatus(object *unstructured.Unstructured, aggregatedStat
 
 	// The 'observedGeneration' is mainly used by GitOps tools(like 'Argo CD') to assess the health status.
 	// For more details, please refer to https://argo-cd.readthedocs.io/en/stable/operator-manual/health/.
+	newStatus.ObservedGeneration = oldStatus.ObservedGeneration
 	if observedLatestResourceTemplateGenerationCount == len(aggregatedStatusItems) {
 		newStatus.ObservedGeneration = deploy.Generation
-	} else {
-		newStatus.ObservedGeneration = oldStatus.ObservedGeneration
 	}
 
 	if oldStatus.ObservedGeneration == newStatus.ObservedGeneration &&
@@ -311,12 +310,10 @@ func aggregateDaemonSetStatus(object *unstructured.Unstructured, aggregatedStatu
 
 	// The 'observedGeneration' is mainly used by GitOps tools(like 'Argo CD') to assess the health status.
 	// For more details, please refer to https://argo-cd.readthedocs.io/en/stable/operator-manual/health/.
+	newStatus.ObservedGeneration = oldStatus.ObservedGeneration
 	if observedLatestResourceTemplateGenerationCount == len(aggregatedStatusItems) {
 		newStatus.ObservedGeneration = daemonSet.Generation
-	} else {
-		newStatus.ObservedGeneration = oldStatus.ObservedGeneration
 	}
-
 	if equality.Semantic.DeepEqual(oldStatus, newStatus) {
 		klog.V(3).Infof("Ignore update daemonSet(%s/%s) status as up to date", daemonSet.Namespace, daemonSet.Name)
 		return object, nil
@@ -342,27 +339,38 @@ func aggregateStatefulSetStatus(object *unstructured.Unstructured, aggregatedSta
 	}
 	oldStatus := &statefulSet.Status
 	newStatus := &appsv1.StatefulSetStatus{}
+	observedLatestResourceTemplateGenerationCount := 0
 	for _, item := range aggregatedStatusItems {
 		if item.Status == nil {
 			continue
 		}
-		temp := &appsv1.StatefulSetStatus{}
+		member := &WrappedStatefulSetStatus{}
-		if err = json.Unmarshal(item.Status.Raw, temp); err != nil {
+		if err = json.Unmarshal(item.Status.Raw, member); err != nil {
 			return nil, err
 		}
 		klog.V(3).Infof("Grab statefulSet(%s/%s) status from cluster(%s), availableReplicas: %d, currentReplicas: %d, readyReplicas: %d, replicas: %d, updatedReplicas: %d",
-			statefulSet.Namespace, statefulSet.Name, item.ClusterName, temp.AvailableReplicas, temp.CurrentReplicas, temp.ReadyReplicas, temp.Replicas, temp.UpdatedReplicas)
+			statefulSet.Namespace, statefulSet.Name, item.ClusterName, member.AvailableReplicas, member.CurrentReplicas, member.ReadyReplicas, member.Replicas, member.UpdatedReplicas)
 
-		// always set 'observedGeneration' with current generation(.metadata.generation)
+		// `memberStatus.ObservedGeneration >= memberStatus.Generation` means the member's status corresponds the latest spec revision of the member statefulset.
-		// which is the generation Karmada 'observed'.
+		// `memberStatus.ResourceTemplateGeneration >= deploy.Generation` means the member statefulset has been aligned with the latest spec revision of federated statefulset.
-		// The 'observedGeneration' is mainly used by GitOps tools(like 'Argo CD') to assess the health status.
+		// If both conditions are met, we consider the member's status corresponds the latest spec revision of federated statefulset.
-		// For more details, please refer to https://argo-cd.readthedocs.io/en/stable/operator-manual/health/.
+		if member.ObservedGeneration >= member.Generation &&
+			member.ResourceTemplateGeneration >= member.Generation {
+			observedLatestResourceTemplateGenerationCount++
+		}
+
+		newStatus.AvailableReplicas += member.AvailableReplicas
+		newStatus.CurrentReplicas += member.CurrentReplicas
+		newStatus.ReadyReplicas += member.ReadyReplicas
+		newStatus.Replicas += member.Replicas
+		newStatus.UpdatedReplicas += member.UpdatedReplicas
+	}
+
+	// The 'observedGeneration' is mainly used by GitOps tools(like 'Argo CD') to assess the health status.
+	// For more details, please refer to https://argo-cd.readthedocs.io/en/stable/operator-manual/health/.
+	newStatus.ObservedGeneration = oldStatus.ObservedGeneration
+	if observedLatestResourceTemplateGenerationCount == len(aggregatedStatusItems) {
 		newStatus.ObservedGeneration = statefulSet.Generation
-		newStatus.AvailableReplicas += temp.AvailableReplicas
-		newStatus.CurrentReplicas += temp.CurrentReplicas
-		newStatus.ReadyReplicas += temp.ReadyReplicas
-		newStatus.Replicas += temp.Replicas
-		newStatus.UpdatedReplicas += temp.UpdatedReplicas
 	}
 
 	if oldStatus.ObservedGeneration == newStatus.ObservedGeneration &&

pkg/resourceinterpreter/default/native/reflectstatus.go

@@ -237,18 +237,32 @@ func reflectStatefulSetStatus(object *unstructured.Unstructured) (*runtime.RawEx
 		return nil, nil
 	}
 
+	resourceTemplateGenerationInt := int64(0)
+	resourceTemplateGenerationStr := util.GetAnnotationValue(object.GetAnnotations(), v1alpha2.ResourceTemplateGenerationAnnotationKey)
+	err = runtime.Convert_string_To_int64(&resourceTemplateGenerationStr, &resourceTemplateGenerationInt, nil)
+	if err != nil {
+		klog.Errorf("Failed to parse StatefulSet(%s/%s) generation from annotation(%s:%s): %v", object.GetNamespace(), object.GetName(), v1alpha2.ResourceTemplateGenerationAnnotationKey, resourceTemplateGenerationStr, err)
+		return nil, err
+	}
+
 	statefulSetStatus := &appsv1.StatefulSetStatus{}
 	err = helper.ConvertToTypedObject(statusMap, statefulSetStatus)
 	if err != nil {
 		return nil, fmt.Errorf("failed to convert StatefulSetStatus from map[string]interface{}: %v", err)
 	}
 
-	grabStatus := appsv1.StatefulSetStatus{
+	grabStatus := &WrappedStatefulSetStatus{
-		AvailableReplicas: statefulSetStatus.AvailableReplicas,
+		FederatedGeneration: FederatedGeneration{
-		CurrentReplicas:   statefulSetStatus.CurrentReplicas,
+			Generation:                 object.GetGeneration(),
-		ReadyReplicas:     statefulSetStatus.ReadyReplicas,
+			ResourceTemplateGeneration: resourceTemplateGenerationInt,
-		Replicas:          statefulSetStatus.Replicas,
+		},
-		UpdatedReplicas:   statefulSetStatus.UpdatedReplicas,
+		StatefulSetStatus: appsv1.StatefulSetStatus{
+			AvailableReplicas: statefulSetStatus.AvailableReplicas,
+			CurrentReplicas:   statefulSetStatus.CurrentReplicas,
+			ReadyReplicas:     statefulSetStatus.ReadyReplicas,
+			Replicas:          statefulSetStatus.Replicas,
+			UpdatedReplicas:   statefulSetStatus.UpdatedReplicas,
+		},
 	}
 	return helper.BuildStatusRawExtension(grabStatus)
 }

pkg/resourceinterpreter/default/native/status_type.go

@@ -39,3 +39,9 @@ type WrappedDaemonSetStatus struct {
 	FederatedGeneration    `json:",inline"`
 	appsv1.DaemonSetStatus `json:",inline"`
 }
+
+// WrappedStatefulSetStatus is a wrapper for appsv1.StatefulSetStatus.
+type WrappedStatefulSetStatus struct {
+	FederatedGeneration      `json:",inline"`
+	appsv1.StatefulSetStatus `json:",inline"`
+}