Add commands for event/secret data ingestion

.gitignore

@@ -32,3 +32,6 @@ tmp/
 
 # VSCode settings
 .vscode/
+
+# Claude Code
+.claude/

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

@@ -4,11 +4,21 @@
 package data
 
 import (
+	"context"
+	"fmt"
+	"math/rand"
+
+	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 +39,52 @@ 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
+}
+
+var letterRunes = []rune("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789")
+
+// RandBytes generates a random byte slice of length n using alphanumeric characters.
+func RandBytes(n int) ([]byte, error) {
+	if n <= 0 {
+		return nil, fmt.Errorf("length must be positive")
+	}
+
+	b := make([]byte, n)
+	for i := range b {
+		b[i] = byte(letterRunes[rand.Intn(len(letterRunes))]) //nolint:gosec // G404: intentionally using math/rand for benchmark data, cryptographic randomness not needed
+	}
+	return b, nil
+}
+
+// RandString generates a random string of length n using alphanumeric characters.
+func RandString(n int) (string, error) {
+	b, err := RandBytes(n)
+	if err != nil {
+		return "", err
+	}
+	return string(b), nil
+}

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

@@ -5,9 +5,7 @@ package configmaps
 
 import (
 	"context"
-	"crypto/rand"
 	"fmt"
-	"math/big"
 	"os"
 	"strconv"
 	"strings"
@@ -26,8 +24,6 @@ import (
 	"k8s.io/client-go/kubernetes"
 )
 
-var appLebel = "runkperf"
-
 var Command = cli.Command{
 	Name:      "configmap",
 	ShortName: "cm",
@@ -40,7 +36,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",
 		},
 	},
@@ -111,7 +107,7 @@ var configmapAddCommand = cli.Command{
 			return err
 		}
 
-		err = prepareNamespace(clientset, namespace)
+		err = data.PrepareNamespace(clientset, namespace)
 		if err != nil {
 			return err
 		}
@@ -141,7 +137,7 @@ var configmapDelCommand = cli.Command{
 
 		namespace := cliCtx.GlobalString("namespace")
 		kubeCfgPath := cliCtx.GlobalString("kubeconfig")
-		labelSelector := fmt.Sprintf("app=%s,cmName=%s", appLebel, cmName)
+		labelSelector := fmt.Sprintf("app=%s,cmName=%s", data.AppLabel, cmName)
 
 		clientset, err := data.NewClientset(kubeCfgPath)
 		if err != nil {
@@ -187,12 +183,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)
@@ -213,30 +209,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")
@@ -253,24 +225,6 @@ func checkConfigmapParams(size int, groupSize int, total int) error {
 	return nil
 }
 
-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
@@ -281,22 +235,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)
+				randData, err := data.RandString(size)
 				if err != nil {
 					return fmt.Errorf("failed to generate random string for configmap %s: %v", name, err)
 				}
 				cm.Data = map[string]string{
-					"data": data,
+					"data": randData,
 				}
 
 				_, err = cli.Create(context.TODO(), cm, metav1.CreateOptions{})

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{