Merge pull request #6666 from liaolecheng/feature/stateful-workload-cluster-failover

Build gracefulEviction task with ClusterFailoverBehavior

Author: karmada-bot

pkg/controllers/applicationfailover/common.go

@@ -17,21 +17,19 @@ limitations under the License.
 package applicationfailover
 
 import (
-	"bytes"
-	"encoding/json"
 	"fmt"
 	"sync"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
-	"k8s.io/client-go/util/jsonpath"
 	"k8s.io/klog/v2"
 	"k8s.io/utils/ptr"
 
 	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/features"
+	"github.com/karmada-io/karmada/pkg/util/helper"
 )
 
 type workloadUnhealthyMap struct {
@@ -126,55 +124,6 @@ func distinguishUnhealthyClustersWithOthers(aggregatedStatusItems []workv1alpha2
 	return unhealthyClusters, others
 }
 
-func buildPreservedLabelState(statePreservation *policyv1alpha1.StatePreservation, rawStatus []byte) (map[string]string, error) {
-	results := make(map[string]string, len(statePreservation.Rules))
-	for _, rule := range statePreservation.Rules {
-		value, err := parseJSONValue(rawStatus, rule.JSONPath)
-		if err != nil {
-			klog.ErrorS(err, "Failed to parse value with jsonPath from status",
-				"jsonPath", rule.JSONPath,
-				"status", string(rawStatus))
-			return nil, err
-		}
-		results[rule.AliasLabelName] = value
-	}
-
-	return results, nil
-}
-
-func parseJSONValue(rawStatus []byte, jsonPath string) (string, error) {
-	template := jsonPath
-	j := jsonpath.New(jsonPath)
-	j.AllowMissingKeys(false)
-	err := j.Parse(template)
-	if err != nil {
-		return "", err
-	}
-
-	buf := new(bytes.Buffer)
-	unmarshalled := make(map[string]interface{})
-	_ = json.Unmarshal(rawStatus, &unmarshalled)
-	err = j.Execute(buf, unmarshalled)
-	if err != nil {
-		return "", err
-	}
-	return buf.String(), nil
-}
-
-func findTargetStatusItemByCluster(aggregatedStatusItems []workv1alpha2.AggregatedStatusItem, cluster string) (workv1alpha2.AggregatedStatusItem, bool) {
-	if len(aggregatedStatusItems) == 0 {
-		return workv1alpha2.AggregatedStatusItem{}, false
-	}
-
-	for index, statusItem := range aggregatedStatusItems {
-		if statusItem.ClusterName == cluster {
-			return aggregatedStatusItems[index], true
-		}
-	}
-
-	return workv1alpha2.AggregatedStatusItem{}, false
-}
-
 func getClusterNamesFromTargetClusters(targetClusters []workv1alpha2.TargetCluster) []string {
 	if targetClusters == nil {
 		return nil
@@ -195,11 +144,11 @@ func buildTaskOptions(failoverBehavior *policyv1alpha1.ApplicationFailoverBehavi
 
 	if features.FeatureGate.Enabled(features.StatefulFailoverInjection) {
 		if failoverBehavior.StatePreservation != nil && len(failoverBehavior.StatePreservation.Rules) != 0 {
-			targetStatusItem, exist := findTargetStatusItemByCluster(aggregatedStatus, cluster)
+			targetStatusItem, exist := helper.FindTargetStatusItemByCluster(aggregatedStatus, cluster)
 			if !exist || targetStatusItem.Status == nil || targetStatusItem.Status.Raw == nil {
 				return nil, fmt.Errorf("the application status has not yet been collected from Cluster(%s)", cluster)
 			}
-			preservedLabelState, err := buildPreservedLabelState(failoverBehavior.StatePreservation, targetStatusItem.Status.Raw)
+			preservedLabelState, err := helper.BuildPreservedLabelState(failoverBehavior.StatePreservation, targetStatusItem.Status.Raw)
 			if err != nil {
 				return nil, err
 			}

pkg/controllers/applicationfailover/common_test.go

@@ -19,7 +19,6 @@ package applicationfailover
 import (
 	"fmt"
 	"reflect"
-	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -187,123 +186,6 @@ func TestDistinguishUnhealthyClustersWithOthers(t *testing.T) {
 	}
 }
 
-func Test_parseJSONValue(t *testing.T) {
-	// This json value describes a DeploymentList object, it contains two deployment elements.
-	var deploymentListStrBytes = []byte(`{"apiVersion":"v1","items":[{"apiVersion":"apps/v1","kind":"Deployment","metadata":{"creationTimestamp":"2024-11-27T07:59:13Z","generation":2,"labels":{"app":"nginx","propagationpolicy.karmada.io/permanent-id":"89a95e21-57ec-4f5d-b8c6-15bc196c1449"},"name":"nginx-01","namespace":"default","resourceVersion":"1148","uid":"12cf1e9f-61fd-4e47-a14e-e844165c7f93"},"spec":{"progressDeadlineSeconds":600,"replicas":2,"revisionHistoryLimit":10,"selector":{"matchLabels":{"app":"nginx"}},"strategy":{"rollingUpdate":{"maxSurge":"25%","maxUnavailable":"25%"},"type":"RollingUpdate"},"template":{"metadata":{"creationTimestamp":null,"labels":{"app":"nginx"}},"spec":{"containers":[{"image":"nginx","imagePullPolicy":"Always","name":"nginx","resources":{},"terminationMessagePath":"/dev/termination-log","terminationMessagePolicy":"File"}],"dnsPolicy":"ClusterFirst","restartPolicy":"Always","schedulerName":"default-scheduler","securityContext":{},"terminationGracePeriodSeconds":30}}},"status":{"availableReplicas":2,"observedGeneration":2,"readyReplicas":2,"replicas":2,"updatedReplicas":2}},{"apiVersion":"apps/v1","kind":"Deployment","metadata":{"creationTimestamp":"2024-11-27T07:59:13Z","generation":2,"labels":{"app":"nginx","propagationpolicy.karmada.io/permanent-id":"89a95e21-57ec-4f5d-b8c6-15bc196c1449"},"name":"nginx-02","namespace":"default","resourceVersion":"1149","uid":"12cf1e9f-61fd-4e47-a14e-e844165c7f93"},"spec":{"progressDeadlineSeconds":600,"replicas":2,"revisionHistoryLimit":10,"selector":{"matchLabels":{"app":"nginx"}},"strategy":{"rollingUpdate":{"maxSurge":"25%","maxUnavailable":"25%"},"type":"RollingUpdate"},"template":{"metadata":{"creationTimestamp":null,"labels":{"app":"nginx"}},"spec":{"containers":[{"image":"nginx","imagePullPolicy":"Always","name":"nginx","resources":{},"terminationMessagePath":"/dev/termination-log","terminationMessagePolicy":"File"}],"dnsPolicy":"ClusterFirst","restartPolicy":"Always","schedulerName":"default-scheduler","securityContext":{},"terminationGracePeriodSeconds":30}}},"status":{"availableReplicas":2,"observedGeneration":2,"readyReplicas":2,"replicas":2,"updatedReplicas":2}}],"kind":"List","metadata":{"resourceVersion":""}}`)
-	type args struct {
-		rawStatus []byte
-		jsonPath  string
-	}
-	tests := []struct {
-		name    string
-		args    args
-		want    string
-		wantErr assert.ErrorAssertionFunc
-	}{
-		// Build the following test cases from the perspective of parsing application state
-		{
-			name: "target field not found",
-			args: args{
-				rawStatus: []byte(`{"readyReplicas": 2}`),
-				jsonPath:  "{ .replicas }",
-			},
-			wantErr: assert.Error,
-		},
-		{
-			name: "invalid jsonPath",
-			args: args{
-				rawStatus: []byte(`{"readyReplicas": 2}`),
-				jsonPath:  "{ %replicas }",
-			},
-			wantErr: assert.Error,
-		},
-		{
-			name: "success to parse",
-			args: args{
-				rawStatus: []byte(`{"replicas": 2}`),
-				jsonPath:  "{ .replicas }",
-			},
-			wantErr: assert.NoError,
-			want:    "2",
-		},
-		// Build the following test cases in terms of what the function supports (which we don't use now).
-		// Please refer to Function Support: https://kubernetes.io/docs/reference/kubectl/jsonpath/
-		{
-			name: "the current object parse",
-			args: args{
-				rawStatus: deploymentListStrBytes,
-				jsonPath:  "{ @ }",
-			},
-			wantErr: assert.NoError,
-			want:    string(deploymentListStrBytes),
-		},
-		{
-			name: "child operator parse",
-			args: args{
-				rawStatus: deploymentListStrBytes,
-				jsonPath:  "{ ['kind'] }",
-			},
-			wantErr: assert.NoError,
-			want:    "List",
-		},
-		{
-			name: "recursive descent parse",
-			args: args{
-				rawStatus: deploymentListStrBytes,
-				jsonPath:  "{ ..resourceVersion }",
-			},
-			wantErr: assert.NoError,
-			want:    "1148 1149",
-		},
-		{
-			name: "wildcard get all objects parse",
-			args: args{
-				rawStatus: deploymentListStrBytes,
-				jsonPath:  "{ .items[*].metadata.name }",
-			},
-			wantErr: assert.NoError,
-			want:    "nginx-01 nginx-02",
-		},
-		{
-			name: "subscript operator parse",
-			args: args{
-				rawStatus: deploymentListStrBytes,
-				jsonPath:  "{ .items[0].metadata.name }",
-			},
-			wantErr: assert.NoError,
-			want:    "nginx-01",
-		},
-		{
-			name: "filter parse",
-			args: args{
-				rawStatus: deploymentListStrBytes,
-				jsonPath:  "{ .items[?(@.metadata.name==\"nginx-01\")].metadata.name }",
-			},
-			wantErr: assert.NoError,
-			want:    "nginx-01",
-		},
-		{
-			name: "iterate list parse",
-			args: args{
-				rawStatus: deploymentListStrBytes,
-				jsonPath:  "{range .items[*]}[{.metadata.name}, {.metadata.namespace}] {end}",
-			},
-			wantErr: assert.NoError,
-			want:    "[nginx-01, default] [nginx-02, default]",
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			got, err := parseJSONValue(tt.args.rawStatus, tt.args.jsonPath)
-			if !tt.wantErr(t, err, fmt.Sprintf("parseJSONValue(%s, %v)", tt.args.rawStatus, tt.args.jsonPath)) {
-				return
-			}
-			got = strings.Trim(got, " ")
-			assert.Equalf(t, tt.want, got, "parseJSONValue(%s, %v)", tt.args.rawStatus, tt.args.jsonPath)
-		})
-	}
-}
-
 func Test_getClusterNamesFromTargetClusters(t *testing.T) {
 	type args struct {
 		targetClusters []workv1alpha2.TargetCluster
@@ -338,125 +220,6 @@ func Test_getClusterNamesFromTargetClusters(t *testing.T) {
 	}
 }
 
-func Test_findTargetStatusItemByCluster(t *testing.T) {
-	type args struct {
-		aggregatedStatusItems []workv1alpha2.AggregatedStatusItem
-		cluster               string
-	}
-	tests := []struct {
-		name      string
-		args      args
-		want      workv1alpha2.AggregatedStatusItem
-		wantExist bool
-	}{
-		{
-			name: "nil aggregatedStatusItems",
-			args: args{
-				aggregatedStatusItems: nil,
-				cluster:               "c1",
-			},
-			want:      workv1alpha2.AggregatedStatusItem{},
-			wantExist: false,
-		},
-		{
-			name: "cluster exist in the aggregatedStatusItems",
-			args: args{
-				aggregatedStatusItems: []workv1alpha2.AggregatedStatusItem{
-					{ClusterName: "c1"},
-					{ClusterName: "c2"},
-				},
-				cluster: "c1",
-			},
-			want:      workv1alpha2.AggregatedStatusItem{ClusterName: "c1"},
-			wantExist: true,
-		},
-		{
-			name: "cluster does not exist in the aggregatedStatusItems",
-			args: args{
-				aggregatedStatusItems: []workv1alpha2.AggregatedStatusItem{
-					{ClusterName: "c1"},
-					{ClusterName: "c2"},
-				},
-				cluster: "c?",
-			},
-			want:      workv1alpha2.AggregatedStatusItem{},
-			wantExist: false,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			got, got1 := findTargetStatusItemByCluster(tt.args.aggregatedStatusItems, tt.args.cluster)
-			assert.Equalf(t, tt.want, got, "findTargetStatusItemByCluster(%v, %v)", tt.args.aggregatedStatusItems, tt.args.cluster)
-			assert.Equalf(t, tt.wantExist, got1, "findTargetStatusItemByCluster(%v, %v)", tt.args.aggregatedStatusItems, tt.args.cluster)
-		})
-	}
-}
-
-func Test_buildPreservedLabelState(t *testing.T) {
-	type args struct {
-		statePreservation *policyv1alpha1.StatePreservation
-		rawStatus         []byte
-	}
-	tests := []struct {
-		name    string
-		args    args
-		want    map[string]string
-		wantErr assert.ErrorAssertionFunc
-	}{
-		{
-			name: "successful case",
-			args: args{
-				statePreservation: &policyv1alpha1.StatePreservation{
-					Rules: []policyv1alpha1.StatePreservationRule{
-						{AliasLabelName: "key-a", JSONPath: "{ .replicas }"},
-						{AliasLabelName: "key-b", JSONPath: "{ .health }"},
-					},
-				},
-				rawStatus: []byte(`{"replicas": 2, "health": true}`),
-			},
-			wantErr: assert.NoError,
-			want:    map[string]string{"key-a": "2", "key-b": "true"},
-		},
-		{
-			name: "one statePreservation rule exist not found field",
-			args: args{
-				statePreservation: &policyv1alpha1.StatePreservation{
-					Rules: []policyv1alpha1.StatePreservationRule{
-						{AliasLabelName: "key-a", JSONPath: "{ .replicas }"},
-						{AliasLabelName: "key-b", JSONPath: "{ .notfound }"},
-					},
-				},
-				rawStatus: []byte(`{"replicas": 2, "health": true}`),
-			},
-			wantErr: assert.Error,
-			want:    nil,
-		},
-		{
-			name: "one statePreservation rule has invalid jsonPath",
-			args: args{
-				statePreservation: &policyv1alpha1.StatePreservation{
-					Rules: []policyv1alpha1.StatePreservationRule{
-						{AliasLabelName: "key-a", JSONPath: "{ .replicas }"},
-						{AliasLabelName: "key-b", JSONPath: "{ %health }"},
-					},
-				},
-				rawStatus: []byte(`{"replicas": 2, "health": true}`),
-			},
-			wantErr: assert.Error,
-			want:    nil,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			got, err := buildPreservedLabelState(tt.args.statePreservation, tt.args.rawStatus)
-			if !tt.wantErr(t, err, fmt.Sprintf("buildPreservedLabelState(%v, %s)", tt.args.statePreservation, tt.args.rawStatus)) {
-				return
-			}
-			assert.Equalf(t, tt.want, got, "buildPreservedLabelState(%v, %s)", tt.args.statePreservation, tt.args.rawStatus)
-		})
-	}
-}
-
 func Test_buildTaskOptions(t *testing.T) {
 	type args struct {
 		failoverBehavior       *policyv1alpha1.ApplicationFailoverBehavior