Add a new runkperf bench for listing configmaps

api/types/load_traffic.go

@@ -41,6 +41,8 @@ type LoadProfileSpec struct {
 	Rate float64 `json:"rate" yaml:"rate"`
 	// Total defines the total number of requests.
 	Total int `json:"total" yaml:"total"`
+	// Duration defines the running time in seconds.
+	Duration int `json:"duration" yaml:"duration"`
 	// Conns defines total number of long connections used for traffic.
 	Conns int `json:"conns" yaml:"conns"`
 	// Client defines total number of HTTP clients.
@@ -179,8 +181,8 @@ func (spec LoadProfileSpec) Validate() error {
 		return fmt.Errorf("rate requires >= 0: %v", spec.Rate)
 	}
 
-	if spec.Total <= 0 {
-		return fmt.Errorf("total requires > 0: %v", spec.Total)
+	if spec.Total <= 0 && spec.Duration <= 0 {
+		return fmt.Errorf("total requires > 0: %v or duration > 0s: %v", spec.Total, spec.Duration)
 	}
 
 	if spec.Client <= 0 {

cmd/kperf/commands/runner/runner.go

@@ -88,6 +88,11 @@ var runCommand = cli.Command{
 			Name:  "raw-data",
 			Usage: "show raw letencies data in result",
 		},
+		cli.IntFlag{
+			Name:  "duration",
+			Usage: "Duration of the benchmark in seconds. It will be ignored if --total is set.",
+			Value: 0,
+		},
 	},
 	Action: func(cliCtx *cli.Context) error {
 		kubeCfgPath := cliCtx.String("kubeconfig")
@@ -169,9 +174,20 @@ func loadConfig(cliCtx *cli.Context) (*types.LoadProfile, error) {
 	if v := "client"; cliCtx.IsSet(v) || profileCfg.Spec.Client == 0 {
 		profileCfg.Spec.Client = cliCtx.Int(v)
 	}
-	if v := "total"; cliCtx.IsSet(v) || profileCfg.Spec.Total == 0 {
+	if v := "total"; cliCtx.IsSet(v) {
 		profileCfg.Spec.Total = cliCtx.Int(v)
 	}
+	if v := "duration"; cliCtx.IsSet(v) {
+		profileCfg.Spec.Duration = cliCtx.Int(v)
+	}
+	if profileCfg.Spec.Total > 0 && profileCfg.Spec.Duration > 0 {
+		profileCfg.Spec.Duration = 0
+		fmt.Printf("Warning: both total:%v and duration:%v are set, duration will be ignored\n", profileCfg.Spec.Total, profileCfg.Spec.Duration)
+	}
+	if profileCfg.Spec.Total == 0 && profileCfg.Spec.Duration == 0 {
+		// Use default total value
+		profileCfg.Spec.Total = cliCtx.Int("total")
+	}
 	if v := "content-type"; cliCtx.IsSet(v) || profileCfg.Spec.ContentType == "" {
 		profileCfg.Spec.ContentType = types.ContentType(cliCtx.String(v))
 	}

contrib/cmd/runkperf/commands/bench/list_configmaps.go

@@ -0,0 +1,134 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+package bench
+
+import (
+	"context"
+	"fmt"
+
+	internaltypes "github.com/Azure/kperf/contrib/internal/types"
+	"github.com/Azure/kperf/contrib/log"
+	"github.com/Azure/kperf/contrib/utils"
+
+	"github.com/urfave/cli"
+)
+
+var benchListConfigmapsCase = cli.Command{
+	Name: "list_configmaps",
+	Usage: `
+
+The test suite is to generate configmaps in a namespace and list them. The load profile is fixed.
+	`,
+	Flags: []cli.Flag{
+		cli.IntFlag{
+			Name:  "size",
+			Usage: "The size of each configmap (Unit: KiB)",
+			Value: 100,
+		},
+		cli.IntFlag{
+			Name:  "group-size",
+			Usage: "The size of each configmap group",
+			Value: 100,
+		},
+		cli.IntFlag{
+			Name:  "configmap-amount",
+			Usage: "Total amount of configmaps",
+			Value: 1024,
+		},
+		cli.IntFlag{
+			Name:  "total",
+			Usage: "Total requests per runner (There are 10 runners totally and runner's rate is 10)",
+			Value: 1000,
+		},
+		cli.IntFlag{
+			Name:  "duration",
+			Usage: "Duration of the benchmark in seconds. It will be ignored if --total is set.",
+			Value: 0,
+		},
+	},
+	Action: func(cliCtx *cli.Context) error {
+		_, err := renderBenchmarkReportInterceptor(
+			addAPIServerCoresInfoInterceptor(benchListConfigmapsRun),
+		)(cliCtx)
+		return err
+	},
+}
+
+var benchConfigmapNamespace = "kperf-configmaps-bench"
+
+// benchfigmapsCase is for subcommand benchConfigmapsCase.
+func benchListConfigmapsRun(cliCtx *cli.Context) (*internaltypes.BenchmarkReport, error) {
+	ctx := context.Background()
+	kubeCfgPath := cliCtx.GlobalString("kubeconfig")
+
+	rgCfgFile, rgSpec, rgCfgFileDone, err := newLoadProfileFromEmbed(cliCtx,
+		"loadprofile/list_configmaps.yaml")
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = rgCfgFileDone() }()
+
+	// Create a namespace for the benchmark
+	cmAmount := cliCtx.Int("configmap-amount")
+	cmSize := cliCtx.Int("size")
+	cmGroupSize := cliCtx.Int("group-size")
+
+	err = utils.CreateConfigmaps(ctx, kubeCfgPath, cmAmount, cmSize, cmGroupSize, benchConfigmapNamespace, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	defer func() {
+		// Delete the configmaps after the benchmark
+		err = utils.DeleteConfigmaps(ctx, kubeCfgPath, benchConfigmapNamespace, 0)
+		if err != nil {
+			log.GetLogger(ctx).WithKeyValues("level", "error").
+				LogKV("msg", fmt.Sprintf("Failed to delete configmaps: %v", err))
+		}
+
+		// Delete the namespace after the benchmark
+		kr := utils.NewKubectlRunner(kubeCfgPath, benchConfigmapNamespace)
+		err := kr.DeleteNamespace(ctx, 0, benchConfigmapNamespace)
+		if err != nil {
+			log.GetLogger(ctx).WithKeyValues("level", "error").
+				LogKV("msg", fmt.Sprintf("Failed to delete namespace: %v", err))
+		}
+	}()
+
+	dpCtx, dpCancel := context.WithCancel(ctx)
+	defer dpCancel()
+
+	duration := cliCtx.Duration("duration")
+	if duration != 0 {
+		log.GetLogger(dpCtx).
+			WithKeyValues("level", "info").
+			LogKV("msg", fmt.Sprintf("Running for %v seconds", duration.Seconds()))
+	}
+
+	rgResult, derr := utils.DeployRunnerGroup(ctx,
+		cliCtx.GlobalString("kubeconfig"),
+		cliCtx.GlobalString("runner-image"),
+		rgCfgFile,
+		cliCtx.GlobalString("runner-flowcontrol"),
+		cliCtx.GlobalString("rg-affinity"),
+	)
+
+	if derr != nil {
+		return nil, derr
+	}
+
+	return &internaltypes.BenchmarkReport{
+		Description: fmt.Sprintf(`
+Environment: Generate %v configmaps with %v bytes each in a namespace.
+Workload: List all configmaps in the namespace and get the percentile latency.`,
+			cmAmount, cmSize),
+
+		LoadSpec: *rgSpec,
+		Result:   *rgResult,
+		Info: map[string]interface{}{
+			"configmapSizeInBytes": cmSize,
+			"runningTime":          duration.String(),
+		},
+	}, nil
+}

contrib/cmd/runkperf/commands/bench/root.go

@@ -59,6 +59,7 @@ var Command = cli.Command{
 		benchNode100Job1Pod3KCase,
 		benchNode100DeploymentNPod10KCase,
 		benchCiliumCustomResourceListCase,
+		benchListConfigmapsCase,
 	},
 }
 

contrib/cmd/runkperf/commands/bench/utils.go

@@ -150,6 +150,11 @@ func newLoadProfileFromEmbed(cliCtx *cli.Context, name string) (_name string, _s
 			if reqs < 0 {
 				return fmt.Errorf("invalid total-requests value: %v", reqs)
 			}
+			reqsTime := cliCtx.Int("duration")
+			if !cliCtx.IsSet("total") && reqsTime > 0 {
+				reqs = 0
+				spec.Profile.Spec.Duration = reqsTime
+			}
 
 			rgAffinity := cliCtx.GlobalString("rg-affinity")
 			affinityLabels, err := kperfcmdutils.KeyValuesMap([]string{rgAffinity})

contrib/internal/manifests/loadprofile/list_configmaps.yaml

@@ -0,0 +1,20 @@
+count: 10
+loadProfile:
+  version: 1
+  description: "list configmaps"
+  spec:
+    rate: 10
+    conns: 10
+    client: 10
+    contentType: json
+    disableHTTP2: false
+    maxRetries: 0
+    requests:
+      - staleList:
+          version: v1
+          resource: configmaps
+        shares: 100 # chance 100 / (100 + 100)
+      - quorumList:
+          version: v1
+          resource: configmaps
+        shares: 100 # chance 100 / (100 + 100)

contrib/utils/utils.go

@@ -482,3 +482,33 @@ func CreateTempFileWithContent(data []byte) (_name string, _cleanup func() error
 		return os.RemoveAll(fName)
 	}, nil
 }
+
+// Creates configmaps for benchmark.
+func CreateConfigmaps(ctx context.Context, kubeCfgPath string,
+	cmAmount int, cmSize int, cmGroupSize int, namespace string, timeout time.Duration) error {
+	args := []string{"data", "configmap"}
+	if kubeCfgPath != "" {
+		args = append(args, fmt.Sprintf("--kubeconfig=%s", kubeCfgPath))
+	}
+	args = append(args, fmt.Sprintf("--namespace=%s", namespace), "add", "runkperf-bench")
+	args = append(args, fmt.Sprintf("--total=%d", cmAmount))
+	args = append(args, fmt.Sprintf("--size=%d", cmSize))
+	args = append(args, fmt.Sprintf("--group-size=%d", cmGroupSize))
+
+	_, err := runCommand(ctx, timeout, "runkperf", args)
+	return err
+
+}
+
+// Delete configmaps for benchmark.
+func DeleteConfigmaps(ctx context.Context, kubeCfgPath string, namespace string, timeout time.Duration) error {
+	args := []string{"data", "configmap"}
+	if kubeCfgPath != "" {
+		args = append(args, "--kubeconfig=%s", kubeCfgPath)
+	}
+	args = append(args, fmt.Sprintf("--namespace=%s", namespace), "delete", "runkperf-bench")
+
+	_, err := runCommand(ctx, timeout, "runkperf", args)
+	return err
+
+}

request/random.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"math/big"
 	"sync"
+	"time"
 
 	"github.com/Azure/kperf/api/types"
 
@@ -78,7 +79,10 @@ func (r *WeightedRandomRequests) Run(ctx context.Context, total int) {
 	r.wg.Add(1)
 
 	sum := 0
-	for sum < total {
+	for {
+		if total > 0 && sum >= total {
+			break
+		}
 		builder := r.randomPick()
 		select {
 		case r.reqBuilderCh <- builder:
@@ -91,6 +95,25 @@ func (r *WeightedRandomRequests) Run(ctx context.Context, total int) {
 	}
 }
 
+// Run starts to random pick request in specific time duration.
+func (r *WeightedRandomRequests) RunForDuration(ctx context.Context, timeDuration time.Duration) {
+	end := time.Now().Add(timeDuration)
+
+	defer r.wg.Done()
+	r.wg.Add(1)
+
+	for time.Now().Before(end) {
+		builder := r.randomPick()
+		select {
+		case r.reqBuilderCh <- builder:
+		case <-r.ctx.Done():
+			return
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
 // Chan returns channel to get random request.
 func (r *WeightedRandomRequests) Chan() chan RESTRequestBuilder {
 	return r.reqBuilderCh

request/schedule.go

@@ -119,13 +119,21 @@ func Schedule(ctx context.Context, spec *types.LoadProfileSpec, restCli []rest.I
 		"connections", len(restCli),
 		"rate", qps,
 		"total", spec.Total,
+		"duration", spec.Duration,
 		"http2", !spec.DisableHTTP2,
 		"content-type", spec.ContentType,
 	)
 
 	start := time.Now()
 
+	if spec.Duration > 0 {
+		// If duration is set, we will run for duration.
+		var cancel context.CancelFunc
+		ctx, cancel = context.WithTimeout(ctx, time.Duration(spec.Duration)*time.Second)
+		defer cancel()
+	}
 	rndReqs.Run(ctx, spec.Total)
+
 	rndReqs.Stop()
 	wg.Wait()