Add commands for event/secret data ingestion

contrib/cmd/runkperf/commands/data/client.go

@@ -4,11 +4,20 @@
 package data
 
 import (
+	"context"
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/client-go/util/flowcontrol"
 )
 
+// default label value used to identify resources created by runkperf.
+const AppLabel = "runkperf"
+
 // NewClientset creates a Kubernetes clientset with default rate limiting.
 func NewClientset(kubeCfgPath string) (*kubernetes.Clientset, error) {
 	config, err := clientcmd.BuildConfigFromFlags("", kubeCfgPath)
@@ -29,3 +38,28 @@ func NewClientsetWithRateLimiter(kubeCfgPath string, qps float32, burst int) (*k
 	config.RateLimiter = flowcontrol.NewTokenBucketRateLimiter(qps, burst)
 	return kubernetes.NewForConfig(config)
 }
+
+// PrepareNamespace creates the namespace if it does not already exist.
+// It skips creation for the "default" namespace.
+func PrepareNamespace(clientset *kubernetes.Clientset, namespace string) error {
+	if namespace == "" {
+		return fmt.Errorf("namespace cannot be empty")
+	}
+
+	if namespace == "default" {
+		return nil
+	}
+
+	_, err := clientset.CoreV1().Namespaces().Create(context.TODO(), &corev1.Namespace{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: namespace,
+		},
+	}, metav1.CreateOptions{})
+	if err != nil {
+		if errors.IsAlreadyExists(err) {
+			return nil
+		}
+		return fmt.Errorf("failed to create namespace %s: %w", namespace, err)
+	}
+	return nil
+}

contrib/cmd/runkperf/commands/data/configmaps/configmap.go

@@ -5,10 +5,8 @@ package configmaps
 
 import (
 	"context"
-	"crypto/rand"
 	"fmt"
 	"math"
-	"math/big"
 	"os"
 	"strconv"
 	"strings"
@@ -20,6 +18,7 @@ import (
 
 	"github.com/Azure/kperf/cmd/kperf/commands/utils"
 	"github.com/Azure/kperf/contrib/cmd/runkperf/commands/data"
+	"github.com/Azure/kperf/request"
 
 	"github.com/urfave/cli"
 
@@ -31,8 +30,6 @@ import (
 	"k8s.io/client-go/kubernetes"
 )
 
-var appLebel = "runkperf"
-
 var Command = cli.Command{
 	Name:      "configmap",
 	ShortName: "cm",
@@ -45,7 +42,7 @@ var Command = cli.Command{
 		},
 		cli.StringFlag{
 			Name:  "namespace",
-			Usage: "Namespace to use with commands. If the namespace does not exist, it will be created.",
+			Usage: "Namespace to use with commands. If the namespace does not exist, it will be created when executing add subcommand",
 			Value: "default",
 		},
 	},
@@ -116,7 +113,7 @@ var configmapAddCommand = cli.Command{
 			return err
 		}
 
-		err = prepareNamespace(clientset, namespace)
+		err = data.PrepareNamespace(clientset, namespace)
 		if err != nil {
 			return err
 		}
@@ -222,12 +219,12 @@ var configmapListCommand = cli.Command{
 		// If args are provided, list all configmaps with the label app=runkperf and cmName in (args)
 		var labelSelector string
 		if cliCtx.NArg() == 0 {
-			labelSelector = fmt.Sprintf("app=%s", appLebel)
+			labelSelector = fmt.Sprintf("app=%s", data.AppLabel)
 
 		} else {
 			args := cliCtx.Args()
 			namesStr := strings.Join(args, ",")
-			labelSelector = fmt.Sprintf("app=%s, cmName in (%s)", appLebel, namesStr)
+			labelSelector = fmt.Sprintf("app=%s, cmName in (%s)", data.AppLabel, namesStr)
 		}
 		cmMap := make(map[string][]int)
 		err = listConfigmapsByName(clientset, labelSelector, namespace, cmMap)
@@ -248,30 +245,6 @@ var configmapListCommand = cli.Command{
 	},
 }
 
-func prepareNamespace(clientset *kubernetes.Clientset, namespace string) error {
-	if namespace == "" {
-		return fmt.Errorf("namespace cannot be empty")
-	}
-
-	if namespace == "default" {
-		return nil
-	}
-
-	_, err := clientset.CoreV1().Namespaces().Create(context.TODO(), &corev1.Namespace{
-		ObjectMeta: metav1.ObjectMeta{
-			Name: namespace,
-		},
-	}, metav1.CreateOptions{})
-	if err != nil {
-		// If the namespace already exists, ignore the error
-		if errors.IsAlreadyExists(err) {
-			return nil
-		}
-		return fmt.Errorf("failed to create namespace %s: %v", namespace, err)
-	}
-	return nil
-}
-
 func checkConfigmapParams(size int, groupSize int, total int) error {
 	if size <= 0 {
 		return fmt.Errorf("size must be greater than 0")
@@ -282,9 +255,6 @@ func checkConfigmapParams(size int, groupSize int, total int) error {
 	if total <= 0 {
 		return fmt.Errorf("total amount must be greater than 0")
 	}
-	if groupSize > total {
-		return fmt.Errorf("group-size must be less than or equal to total")
-	}
 	return nil
 }
 
@@ -342,24 +312,6 @@ func isRetriableError(err error) bool {
 		errors.IsServiceUnavailable(err)
 }
 
-var letterRunes = []rune("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ")
-
-func randString(n int) (string, error) {
-	if n <= 0 {
-		return "", fmt.Errorf("length must be positive")
-	}
-
-	b := make([]rune, n)
-	for i := range b {
-		random, err := rand.Int(rand.Reader, big.NewInt(int64(len(letterRunes))))
-		if err != nil {
-			return "", fmt.Errorf("error generating random number: %w", err)
-		}
-		b[i] = letterRunes[int(random.Int64())]
-	}
-	return string(b), nil
-}
-
 func createConfigmaps(clientset *kubernetes.Clientset, namespace string, cmName string, size int, groupSize int, total int) error {
 	// Generate configmaps in parallel with fixed group size
 	// and random data
@@ -370,25 +322,22 @@ func createConfigmaps(clientset *kubernetes.Clientset, namespace string, cmName
 			g.Go(func() error {
 				cli := clientset.CoreV1().ConfigMaps(namespace)
 
-				name := fmt.Sprintf("%s-cm-%s-%d", appLebel, cmName, j)
+				name := fmt.Sprintf("%s-cm-%s-%d", data.AppLabel, cmName, j)
 
 				cm := &corev1.ConfigMap{}
 				cm.Name = name
 				// Set the labels for the configmap to easily identify in delete or list commands
 				cm.Labels = map[string]string{
 					"ownerID": strconv.Itoa(ownerID),
-					"app":     appLebel,
+					"app":     data.AppLabel,
 					"cmName":  cmName,
 				}
-				data, err := randString(size)
-				if err != nil {
-					return fmt.Errorf("failed to generate random string for configmap %s: %v", name, err)
-				}
+				randData := request.RandomPayload(size)
 				cm.Data = map[string]string{
-					"data": data,
+					"data": randData,
 				}
 
-				err = retryOnTransient(func() error {
+				err := retryOnTransient(func() error {
 					_, createErr := cli.Create(context.TODO(), cm, metav1.CreateOptions{})
 					// A retry after a dropped connection may hit an object that the
 					// previous (lost) attempt already created; treat it as success.
@@ -431,7 +380,7 @@ func deleteConfigmapsByName(clientset *kubernetes.Clientset, namespace, cmName s
 		var succeeded int64
 		g := new(errgroup.Group)
 		for j := start; j < end; j++ {
-			name := fmt.Sprintf("%s-cm-%s-%d", appLebel, cmName, j)
+			name := fmt.Sprintf("%s-cm-%s-%d", data.AppLabel, cmName, j)
 			g.Go(func() error {
 				err := retryOnTransient(func() error {
 					return cmClient.Delete(context.TODO(), name, metav1.DeleteOptions{})

contrib/cmd/runkperf/commands/data/daemonsets/daemonset.go

@@ -11,8 +11,7 @@ import (
 	"text/tabwriter"
 
 	"github.com/Azure/kperf/cmd/kperf/commands/utils"
-	"k8s.io/client-go/tools/clientcmd"
-	"k8s.io/client-go/util/flowcontrol"
+	"github.com/Azure/kperf/contrib/cmd/runkperf/commands/data"
 
 	"github.com/urfave/cli"
 
@@ -76,12 +75,12 @@ var daemonsetAddCommand = cli.Command{
 			return fmt.Errorf("count must be greater than 0")
 		}
 
-		err := prepareNamespace(kubeCfgPath, namespace)
+		clientset, err := data.NewClientsetWithRateLimiter(kubeCfgPath, 30, 10)
 		if err != nil {
 			return err
 		}
 
-		clientset, err := newClientsetWithRateLimiter(kubeCfgPath, 30, 10)
+		err = data.PrepareNamespace(clientset, namespace)
 		if err != nil {
 			return err
 		}
@@ -113,7 +112,7 @@ var daemonsetDelCommand = cli.Command{
 		namespace := cliCtx.GlobalString("namespace")
 		kubeCfgPath := cliCtx.GlobalString("kubeconfig")
 
-		clientset, err := newClientsetWithRateLimiter(kubeCfgPath, 30, 10)
+		clientset, err := data.NewClientsetWithRateLimiter(kubeCfgPath, 30, 10)
 		if err != nil {
 			return err
 		}
@@ -139,7 +138,7 @@ var daemonsetListCommand = cli.Command{
 	Action: func(cliCtx *cli.Context) error {
 		namespace := cliCtx.GlobalString("namespace")
 		kubeCfgPath := cliCtx.GlobalString("kubeconfig")
-		clientset, err := newClientsetWithRateLimiter(kubeCfgPath, 30, 10)
+		clientset, err := data.NewClientsetWithRateLimiter(kubeCfgPath, 30, 10)
 		if err != nil {
 			return err
 		}
@@ -192,50 +191,6 @@ var daemonsetListCommand = cli.Command{
 	},
 }
 
-func prepareNamespace(kubeCfgPath string, namespace string) error {
-	if namespace == "" {
-		return fmt.Errorf("namespace cannot be empty")
-	}
-
-	if namespace == "default" {
-		return nil
-	}
-
-	clientset, err := newClientsetWithRateLimiter(kubeCfgPath, 30, 10)
-	if err != nil {
-		return err
-	}
-
-	_, err = clientset.CoreV1().Namespaces().Create(context.TODO(), &corev1.Namespace{
-		ObjectMeta: metav1.ObjectMeta{
-			Name: namespace,
-		},
-	}, metav1.CreateOptions{})
-	if err != nil {
-		// If the namespace already exists, ignore the error
-		if errors.IsAlreadyExists(err) {
-			return nil
-		}
-		return fmt.Errorf("failed to create namespace %s: %v", namespace, err)
-	}
-	return nil
-}
-
-func newClientsetWithRateLimiter(kubeCfgPath string, qps float32, burst int) (*kubernetes.Clientset, error) {
-	config, err := clientcmd.BuildConfigFromFlags("", kubeCfgPath)
-	if err != nil {
-		return nil, err
-	}
-
-	config.QPS = qps
-	config.RateLimiter = flowcontrol.NewTokenBucketRateLimiter(qps, burst)
-	clientset, err := kubernetes.NewForConfig(config)
-	if err != nil {
-		return nil, err
-	}
-	return clientset, nil
-}
-
 func createDaemonsets(clientset *kubernetes.Clientset, namespace string, dsName string, count int) error {
 	for i := 0; i < count; i++ {
 		ds := &appsv1.DaemonSet{