test: LoadTest Setup

hack/loadtest/main.go

@@ -22,42 +22,41 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/config"
-	workv1alpha1 "sigs.k8s.io/work-api/pkg/apis/v1alpha1"
 
-	fleetv1alpha1 "go.goms.io/fleet/apis/v1alpha1"
+	fleetv1beta1 "go.goms.io/fleet/apis/placement/v1beta1"
 	"go.goms.io/fleet/hack/loadtest/util"
 )
 
 var (
-	scheme = runtime.NewScheme()
+	scheme  = runtime.NewScheme()
+	crpFile = "test-crp.yaml"
 )
 
 var (
 	placementDeadline   = flag.Int("placement-deadline-second", 300, "The deadline for a placement to be applied (in seconds)")
 	pollInterval        = flag.Int("poll-interval-millisecond", 250, "The poll interval for verification (in milli-second)")
 	maxCurrentPlacement = flag.Int("max-current-placement", 10, "The number of current placement load.")
-	loadTestLength      = flag.Int("load-test-length-minute", 15, "The length of the load test in miniutes.")
-	clusterNames        util.ClusterNames
+	loadTestLength      = flag.Int("load-test-length-minute", 15, "The length of the load test in minutes.")
+	clusterNames        util.ClusterNames //will be used for PickFixed scenario, otherwise will apply to all clusters
 )
 
 func init() {
 	klog.InitFlags(nil)
+	flag.StringVar(&crpFile, "crp-file", "test-crp.yaml", "The CRP yaml file.")
 	utilrand.Seed(time.Now().UnixNano())
 
 	utilruntime.Must(clientgoscheme.AddToScheme(scheme))
-	utilruntime.Must(fleetv1alpha1.AddToScheme(scheme))
-	utilruntime.Must(workv1alpha1.AddToScheme(scheme))
+	utilruntime.Must(fleetv1beta1.AddToScheme(scheme))
 	//+kubebuilder:scaffold:scheme
 }
 
 func main() {
-	flag.Var(&clusterNames, "cluster", "The name of a member cluster")
 	flag.Parse()
 	defer klog.Flush()
 
-	klog.InfoS("start to run placement load test", "pollInterval", *pollInterval, "placementDeadline", *placementDeadline, "maxCurrentPlacement", *maxCurrentPlacement, "clusterNames", clusterNames)
+	klog.InfoS("start to run placement load test", "crpFile", crpFile, "pollInterval", *pollInterval, "placementDeadline", *placementDeadline, "maxCurrentPlacement", *maxCurrentPlacement, "clusterNames", clusterNames)
 	config := config.GetConfigOrDie()
-	config.QPS, config.Burst = float32(100), 500
+	config.QPS, config.Burst = float32(100), 500 //queries per second, max # of queries queued at once
 	hubClient, err := client.New(config, client.Options{
 		Scheme: scheme,
 	})
@@ -75,7 +74,7 @@ func main() {
 	// setup prometheus server
 	http.Handle("/metrics", promhttp.Handler())
 	/* #nosec */
-	if err = http.ListenAndServe(":4848", http.TimeoutHandler(nil, 10*time.Second, "request time out")); err != nil {
+	if err = http.ListenAndServe(":4848", nil); err != nil {
 		panic(err)
 	}
 }
@@ -101,7 +100,7 @@ func runLoadTest(ctx context.Context, config *rest.Config) {
 				case <-ctx.Done():
 					return
 				default:
-					if err = util.MeasureOnePlacement(ctx, hubClient, time.Duration(*placementDeadline)*time.Second, time.Duration(*pollInterval)*time.Millisecond, *maxCurrentPlacement, clusterNames); err != nil {
+					if err = util.MeasureOnePlacement(ctx, hubClient, time.Duration(*placementDeadline)*time.Second, time.Duration(*pollInterval)*time.Millisecond, *maxCurrentPlacement, clusterNames, crpFile); err != nil {
 						klog.ErrorS(err, "placement load test failed")
 					}
 				}

hack/loadtest/test-crp.yaml

@@ -0,0 +1,12 @@
+apiVersion: placement.kubernetes-fleet.io/v1beta1
+kind: ClusterResourcePlacement
+metadata:
+  name: crp
+spec:
+  strategy:
+    type: RollingUpdate
+    rollingUpdate:
+      maxUnavailable: 15%
+      maxSurge: 35%
+      unavailablePeriodSeconds: 45
+    revisionHistoryLimit: 15

hack/loadtest/test-crp1.yaml

@@ -0,0 +1,15 @@
+apiVersion: placement.kubernetes-fleet.io/v1beta1
+kind: ClusterResourcePlacement
+metadata:
+  name: crp
+spec:
+  resourceSelectors:
+    - group: ""
+      kind: Namespace
+      version: v1
+      name: load-test-ns-testing
+  policy:
+    placementType: PickFixed
+    clusterNames:
+      - cluster-2
+      - cluster-4

hack/loadtest/test-crp2.yaml

@@ -0,0 +1,31 @@
+apiVersion: placement.kubernetes-fleet.io/v1beta1
+kind: ClusterResourcePlacement
+metadata:
+  name: crp
+spec:
+  resourceSelectors:
+    - group: apiextensions.k8s.io
+      kind: CustomResourceDefinition
+      name: clonesets.apps.kruise.io
+      version: v1
+  policy:
+    placementType: PickN
+    numberOfClusters: 3
+    affinity:
+      clusterAffinity:
+        preferredDuringSchedulingIgnoredDuringExecution:
+        - weight: 20
+          preference:
+            labelSelector:
+              matchExpressions:
+                - key: system
+                  operator: DoesNotExist
+        - weight: -20
+          preference:
+            labelSelector:
+              matchExpressions:
+                - key: system
+                  operator: In
+                  values:
+                    - critical
+                    - standard

hack/loadtest/test-crp3.yaml

@@ -0,0 +1,25 @@
+apiVersion: placement.kubernetes-fleet.io/v1beta1
+kind: ClusterResourcePlacement
+metadata:
+  name: crp
+spec:
+  resourceSelectors:
+    - group: apiextensions.k8s.io
+      kind: CustomResourceDefinition
+      name: clonesets.apps.kruise.io
+      version: v1
+  policy:
+    placementType: PickAll
+    affinity:
+      clusterAffinity:
+        requiredDuringSchedulingIgnoredDuringExecution:
+          clusterSelectorTerms:
+            - labelSelector:
+                matchLabels:
+                  region: east
+                matchExpressions:
+                  - key: system
+                    operator: In
+                    values:
+                      - critical
+                      - standard

hack/loadtest/test-crp4.yaml

@@ -0,0 +1,17 @@
+apiVersion: placement.kubernetes-fleet.io/v1beta1
+kind: ClusterResourcePlacement
+metadata:
+  name: crp
+spec:
+  resourceSelectors:
+    - group: apiextensions.k8s.io
+      kind: CustomResourceDefinition
+      name: clonesets.apps.kruise.io
+      version: v1
+  policy:
+    placementType: PickN
+    numberOfClusters: 3
+    topologySpreadConstraints:
+      - maxSkew: 2
+        topologyKey: env
+        whenUnsatisfiable: DoNotSchedule

hack/loadtest/util/help.go

@@ -2,18 +2,21 @@ package util
 
 import (
 	"context"
+	"fmt"
 	"os"
 	"strings"
 
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/apimachinery/pkg/util/yaml"
 	"k8s.io/klog/v2"
+	"k8s.io/utils/pointer"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
-	"go.goms.io/fleet/apis/v1alpha1"
+	"go.goms.io/fleet/apis/placement/v1beta1"
 )
 
 type ClusterNames []string
@@ -155,6 +158,17 @@ func deleteNamespace(ctx context.Context, hubClient client.Client, namespaceName
 }
 
 func CleanupAll(hubClient client.Client) error {
+	var crps v1beta1.ClusterResourcePlacementList
+	if err := hubClient.List(context.Background(), &crps); err != nil {
+		klog.ErrorS(err, "failed to list namespace")
+		return err
+	}
+	for index, crp := range crps.Items {
+		if err := hubClient.Delete(context.Background(), &crps.Items[index]); err != nil {
+			klog.ErrorS(err, "failed to delete crp", "crp", crp.Name)
+		}
+	}
+
 	var namespaces corev1.NamespaceList
 	if err := hubClient.List(context.Background(), &namespaces); err != nil {
 		klog.ErrorS(err, "failed to list namespace")
@@ -167,15 +181,116 @@ func CleanupAll(hubClient client.Client) error {
 			}
 		}
 	}
-	var crps v1alpha1.ClusterResourcePlacementList
-	if err := hubClient.List(context.Background(), &crps); err != nil {
-		klog.ErrorS(err, "failed to list namespace")
+	return nil
+}
+
+func executeStrategy(crp *v1beta1.ClusterResourcePlacement, obj *unstructured.Unstructured) error {
+	strategy, foundType, err := unstructured.NestedString(obj.Object, "spec", "strategy", "type")
+	if err != nil {
+		klog.ErrorS(err, "Failed to get strategy.")
 		return err
 	}
-	for index, crp := range crps.Items {
-		if err := hubClient.Delete(context.Background(), &crps.Items[index]); err != nil {
-			klog.ErrorS(err, "failed to delete crp", "crp", crp.Name)
-		}
+	if !foundType {
+		return nil
+	}
+	crp.Spec.Strategy.Type = v1beta1.RolloutStrategyType(strategy)
+
+	rollingUpdate, found, err := unstructured.NestedMap(obj.Object, "spec", "strategy", "rollingUpdate")
+	if err != nil {
+		klog.ErrorS(err, "Failed to get RollingUpdate.")
+		return err
+	}
+	if !found {
+		klog.Info("RollingUpdate not found in file.")
+		return nil
+	}
+	crp.Spec.Strategy.RollingUpdate = &v1beta1.RollingUpdateConfig{}
+
+	if maxUnavailable, found := rollingUpdate["maxUnavailable"].(string); found {
+		maxUnvailable := intstr.FromString(maxUnavailable)
+		crp.Spec.Strategy.RollingUpdate.MaxUnavailable = &maxUnvailable
+	}
+	if maxSurge, found := rollingUpdate["maxSurge"].(string); found {
+		maxSurge := intstr.FromString(maxSurge)
+		crp.Spec.Strategy.RollingUpdate.MaxSurge = &maxSurge
+	}
+	if unavailablePeriodSeconds, found := rollingUpdate["unavailablePeriodSeconds"].(int64); found {
+		crp.Spec.Strategy.RollingUpdate.UnavailablePeriodSeconds = pointer.Int(int(unavailablePeriodSeconds))
 	}
 	return nil
 }
+
+func executeRevisionHistoryLimit(crp *v1beta1.ClusterResourcePlacement, obj *unstructured.Unstructured) error {
+	revisionLimit, found, err := unstructured.NestedInt64(obj.Object, "spec", "revisionHistoryLimit")
+	if err != nil {
+		klog.ErrorS(err, "Failed to get RevisionHistoryLimit.")
+		return err
+	}
+	if !found {
+		return nil
+	}
+	crp.Spec.RevisionHistoryLimit = pointer.Int32(int32(revisionLimit))
+	return nil
+}
+
+func executePickFixedPolicy(crp *v1beta1.ClusterResourcePlacement, obj *unstructured.Unstructured, clusterNames ClusterNames) (ClusterNames, error) {
+	names, found, err := unstructured.NestedStringSlice(obj.Object, "spec", "policy", "clusterNames")
+	if err != nil {
+		klog.ErrorS(err, "Failed to get cluster names.")
+		return clusterNames, err
+	}
+	if !found {
+		klog.Infof("Cluster Names not found in file.")
+		return clusterNames, nil
+	}
+
+	for _, name := range names {
+		if err := clusterNames.Set(name); err != nil {
+			klog.ErrorS(err, "Error setting cluster name: %v")
+			return clusterNames, err
+		}
+	}
+
+	crp.Spec.Policy = &v1beta1.PlacementPolicy{
+		PlacementType: v1beta1.PickFixedPlacementType,
+		ClusterNames:  names,
+	}
+	return clusterNames, nil
+}
+
+func applyCRP(crp *v1beta1.ClusterResourcePlacement, crpFile string, nsName string, clusterNames ClusterNames) error {
+	obj, err := readObjFromFile(fmt.Sprintf("hack/loadtest/%s", crpFile), nsName)
+	if err != nil {
+		klog.ErrorS(err, "Failed to read object from file.")
+		return err
+	}
+
+	placementType, found, err := unstructured.NestedString(obj.Object, "spec", "policy", "placementType")
+	if err != nil {
+		klog.ErrorS(err, "Failed to get placement type.")
+		return err
+	}
+	if !found {
+		placementType = "default"
+	}
+
+	switch placementType {
+	case "PickFixed":
+		if clusterNames, err = executePickFixedPolicy(crp, obj, clusterNames); err != nil {
+			return err
+		}
+	default:
+		for i := 1; i <= 4; i++ {
+			if err := clusterNames.Set(fmt.Sprintf("cluster-%d", i)); err != nil {
+				klog.ErrorS(err, "Error setting cluster name: %v")
+				return err
+			}
+		}
+	}
+
+	if err := executeStrategy(crp, obj); err != nil {
+		return err
+	}
+
+	return executeRevisionHistoryLimit(crp, obj)
+}