feat: check in performance test related code/utilities/scripts (2 of N) (kubefleet-dev#543)

Author: michaelawyu

hack/perftest/1000placements/README.md

@@ -0,0 +1,35 @@
+# KubeFleet Performance/Scalability Test Utility: Creating 1K Placements Concurrently
+
+This directory contains a utility program that creates 1K placements that run concurrently and polls
+until their full completion. 
+
+The program is added for the purpose of testing the performance and scalability of KubeFleet.
+
+## Before you begin
+
+* Set up a KubeFleet deployment.
+* Make sure that all member clusters are labelled with `placement-group=N`, where N ranges from 0 to 9.
+Placements created by this utility program will be each assigned an index X, ranging from 0 to 999,
+and a placement of index X will select all member clusters that are labelled with `placement-group=X%10`.
+* The program requires the following tools (aside from the Go runtime) to be installed:
+    * `curl` (for retrieving pprof data)
+
+> If you have followed the instructions in `../../README.md` and have used the given scripts and utility programs
+> to run the performance/scalability test, all the steps above should have been done for you already.
+
+## Running the utility program
+
+Run the commands below to run the utility program:
+
+```bash
+RUN_NAME=example go run main.go
+```
+
+With the default setup, it might take ~1 hour before all 1K placements are fully completed. The program
+reports the progress in the output.
+
+After the program completes, run the commands below to clean up the created 1K placements:
+
+```bash
+CLEANUP=set go run main.go
+```

hack/perftest/1000placements/example-placement.yaml

@@ -0,0 +1,24 @@
+apiVersion: placement.kubernetes-fleet.io/v1beta1
+kind: ClusterResourcePlacement
+metadata:
+  name: crp-0
+spec:
+  policy:
+    placementType: PickAll
+    affinity:
+      clusterAffinity:
+        requiredDuringSchedulingIgnoredDuringExecution:
+          clusterSelectorTerms:
+            - labelSelector:
+                matchLabels:
+                  placement-group: "0"
+  resourceSelectors:
+    - group: ""
+      kind: Namespace
+      version: v1
+      name: work-0
+  strategy:
+    type: RollingUpdate
+    rollingUpdate:
+      maxUnavailable: 100%
+      unavailablePeriodSeconds: 1

hack/perftest/1000placements/main.go

@@ -0,0 +1,116 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"time"
+
+	"k8s.io/apimachinery/pkg/util/wait"
+
+	"github.com/kubefleet-dev/kubefleet/hack/perftest/1000placements/utils"
+)
+
+const (
+	resourceSetupWorkerCount       = 15
+	longPollingWorkerCount         = 15
+	resourceCreationCoolDownPeriod = time.Second * 1
+	longPollingCoolDownPeriod      = time.Second * 2
+	betweenStageCoolDownPeriod     = time.Second * 30
+
+	maxCRPToCreateCount = 1000
+
+	configMapDataByteCount = 1024 // 1 KB.
+)
+
+var (
+	// Add trigger points for dumping pprof profiles when a specific # of placement
+	// have been created.
+	triggerPtsForMemProfileDumping = map[int]bool{}
+)
+
+var (
+	retryOpsBackoff = wait.Backoff{
+		Steps:    4,
+		Duration: 4 * time.Second,
+		Factor:   2.0,
+		Jitter:   0.1,
+	}
+)
+
+func main() {
+	ctx := context.Background()
+
+	// Read the arguments.
+	doCleanUp := false
+	cleanUpFlag := os.Getenv("CLEANUP")
+	if len(cleanUpFlag) != 0 {
+		doCleanUp = true
+	}
+
+	runName := os.Getenv("RUN_NAME")
+	if len(runName) == 0 {
+		panic("RUN_NAME environment variable is not set")
+	}
+
+	retrievePprofData := false
+	if s := os.Getenv("RETRIEVE_PPROF_DATA"); len(s) != 0 {
+		retrievePprofData = true
+	}
+	var pProfEndpoint string
+	if retrievePprofData {
+		if pProfEndpoint = os.Getenv("PPROF_ENDPOINT"); len(pProfEndpoint) == 0 {
+			panic("PPROF_ENDPOINT environment variable is not set")
+		}
+	}
+
+	runner := utils.New(
+		runName,
+		resourceSetupWorkerCount,
+		longPollingWorkerCount,
+		betweenStageCoolDownPeriod,
+		resourceCreationCoolDownPeriod,
+		longPollingCoolDownPeriod,
+		configMapDataByteCount,
+		maxCRPToCreateCount,
+		triggerPtsForMemProfileDumping,
+		pProfEndpoint,
+		retryOpsBackoff,
+	)
+
+	if doCleanUp {
+		runner.CleanUp(ctx)
+		return
+	}
+
+	fmt.Println("Preparing the resources...")
+	runner.CreateResources(ctx)
+
+	// Cool down.
+	fmt.Println("Cooling down...")
+	runner.CoolDown()
+
+	fmt.Println("Creating the placements...")
+	runner.CreatePlacements(ctx)
+
+	// Cool down.
+	fmt.Println("Cooling down...")
+	runner.CoolDown()
+
+	fmt.Println("Long polling the placements...")
+	runner.LongPollPlacements(ctx)
+
+	fmt.Println("Tracking latency")
+	runner.TrackLatency(ctx)
+
+	if retrievePprofData {
+		fmt.Println("retrieving final pprof data")
+		runner.RetrievePProfProfile(maxCRPToCreateCount + 1)
+	}
+
+	// Tally the latency quantiles.
+	fmt.Println("Tallying latency quantiles...")
+	runner.TallyLatencyQuantiles()
+
+	fmt.Println("All placements have been completed, exiting.")
+}

hack/perftest/1000placements/utils/cleanup.go

@@ -0,0 +1,130 @@
+package utils
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/util/retry"
+
+	placementv1beta1 "github.com/kubefleet-dev/kubefleet/apis/placement/v1beta1"
+)
+
+func (r *Runner) CleanUp(ctx context.Context) {
+	wg := sync.WaitGroup{}
+
+	// Run the producer.
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+
+		for i := 0; i < r.maxCRPToCreateCount; i++ {
+			select {
+			case r.toDeleteChan <- i:
+			case <-ctx.Done():
+				close(r.toDeleteChan)
+				return
+			}
+		}
+
+		close(r.toDeleteChan)
+	}()
+
+	// Run the workers.
+	for i := 0; i < r.resourceSetupWorkerCount; i++ {
+		wg.Add(1)
+		go func(workerIdx int) {
+			defer wg.Done()
+
+			for {
+				// Read from the channel.
+				var resIdx int
+				var readOk bool
+				select {
+				case resIdx, readOk = <-r.toDeleteChan:
+					if !readOk {
+						fmt.Printf("worker %d exits\n", workerIdx)
+						return
+					}
+				case <-ctx.Done():
+					return
+				}
+
+				// Delete the CRPs.
+				crp := placementv1beta1.ClusterResourcePlacement{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: fmt.Sprintf(placementNameFmt, resIdx),
+					},
+				}
+				errAfterRetries := retry.OnError(r.retryOpsBackoff, func(err error) bool {
+					return err != nil && !errors.IsNotFound(err)
+				}, func() error {
+					return r.hubClient.Delete(ctx, &crp)
+				})
+				if errAfterRetries != nil && !errors.IsNotFound(errAfterRetries) {
+					fmt.Printf("worker %d: failed to delete CRP %s after retries: %v\n", workerIdx, fmt.Sprintf(placementNameFmt, resIdx), errAfterRetries)
+					continue
+				}
+
+				// Wait until the CRP is deleted.
+				errAfterRetries = retry.OnError(r.retryOpsBackoff, func(err error) bool {
+					return err != nil && !errors.IsNotFound(err)
+				}, func() error {
+					crp := placementv1beta1.ClusterResourcePlacement{}
+					err := r.hubClient.Get(ctx, types.NamespacedName{Name: fmt.Sprintf(placementNameFmt, resIdx)}, &crp)
+					if err == nil {
+						return fmt.Errorf("CRP %s still exists", fmt.Sprintf(placementNameFmt, resIdx))
+					}
+					return err
+				})
+				if !errors.IsNotFound(errAfterRetries) {
+					fmt.Printf("worker %d: failed to wait for CRP %s to be deleted after retries: %v\n", workerIdx, fmt.Sprintf(placementNameFmt, resIdx), errAfterRetries)
+				} else {
+					fmt.Printf("worker %d: deleted CRP %s\n", workerIdx, fmt.Sprintf(placementNameFmt, resIdx))
+				}
+
+				// Delete the namespace if it exists.
+				namespace := corev1.Namespace{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: fmt.Sprintf(nsNameFmt, resIdx),
+					},
+				}
+				errAfterRetries = retry.OnError(r.retryOpsBackoff, func(err error) bool {
+					return err != nil && !errors.IsNotFound(err)
+				}, func() error {
+					return r.hubClient.Delete(ctx, &namespace)
+				})
+				if errAfterRetries != nil && !errors.IsNotFound(errAfterRetries) {
+					fmt.Printf("worker %d: failed to delete namespace %s after retries: %v\n", workerIdx, fmt.Sprintf(nsNameFmt, resIdx), errAfterRetries)
+					continue
+				}
+
+				// Wait until the namespace is deleted.
+				errAfterRetries = retry.OnError(r.retryOpsBackoff, func(err error) bool {
+					return err != nil && !errors.IsNotFound(err)
+				}, func() error {
+					namespace := corev1.Namespace{}
+					err := r.hubClient.Get(ctx, types.NamespacedName{Name: fmt.Sprintf(nsNameFmt, resIdx)}, &namespace)
+					if err == nil {
+						return fmt.Errorf("namespace %s still exists", fmt.Sprintf(nsNameFmt, resIdx))
+					}
+					return err
+				})
+				if errAfterRetries == nil || !errors.IsNotFound(errAfterRetries) {
+					fmt.Printf("worker %d: failed to wait for namespace %s to be deleted after retries: %v\n", workerIdx, fmt.Sprintf(nsNameFmt, resIdx), errAfterRetries)
+				} else {
+					fmt.Printf("worker %d: deleted namespace %s\n", workerIdx, fmt.Sprintf(nsNameFmt, resIdx))
+				}
+			}
+		}(i)
+	}
+	wg.Wait()
+
+	// Cool down.
+	time.Sleep(r.betweenStageCoolDownPeriod)
+}

hack/perftest/1000placements/utils/latency.go

@@ -0,0 +1,51 @@
+package utils
+
+import (
+	"context"
+	"fmt"
+	"math"
+	"sort"
+	"sync"
+)
+
+func (r *Runner) TrackLatency(ctx context.Context) {
+	wg := sync.WaitGroup{}
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+
+		for {
+			var attempt latencyTrackAttempt
+			var readOK bool
+			select {
+			case attempt, readOK = <-r.toTrackLatencyChan:
+				if !readOK {
+					return
+				}
+				fmt.Printf("latency tracker: CRP %s has latency %v\n", fmt.Sprintf(placementNameFmt, attempt.resIdx), attempt.latency)
+				r.placementCompletionLatencyByName[fmt.Sprintf(placementNameFmt, attempt.resIdx)] = attempt.latency
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+	wg.Wait()
+}
+
+func (r *Runner) TallyLatencyQuantiles() {
+	latencies := make([]float64, 0, len(r.placementCompletionLatencyByName))
+	for _, latency := range r.placementCompletionLatencyByName {
+		latencies = append(latencies, float64(latency.Seconds()))
+	}
+	sort.Slice(latencies, func(i, j int) bool {
+		return latencies[i] < latencies[j]
+	})
+	q25 := int(math.Floor(float64(len(latencies)) * 0.25))
+	q50 := int(math.Floor(float64(len(latencies)) * 0.50))
+	q75 := int(math.Floor(float64(len(latencies)) * 0.75))
+	q90 := int(math.Floor(float64(len(latencies)) * 0.90))
+	q99 := int(math.Floor(float64(len(latencies)) * 0.99))
+	fmt.Printf("latencies: 25th=%v, 50th=%v, 75th=%v, 90th=%v, 99th=%v\n",
+		latencies[q25], latencies[q50], latencies[q75], latencies[q90], latencies[q99])
+}