diff --git a/contrib/cmd/runkperf/commands/data/client.go b/contrib/cmd/runkperf/commands/data/client.go
index 47e6df2b..a4473f4b 100644
--- a/contrib/cmd/runkperf/commands/data/client.go
+++ b/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
+}
diff --git a/contrib/cmd/runkperf/commands/data/configmaps/configmap.go b/contrib/cmd/runkperf/commands/data/configmaps/configmap.go
index 6a5b1d32..c44149ac 100644
--- a/contrib/cmd/runkperf/commands/data/configmaps/configmap.go
+++ b/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{})
diff --git a/contrib/cmd/runkperf/commands/data/daemonsets/daemonset.go b/contrib/cmd/runkperf/commands/data/daemonsets/daemonset.go
index 9c7b79f1..79d3b060 100644
--- a/contrib/cmd/runkperf/commands/data/daemonsets/daemonset.go
+++ b/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{
diff --git a/contrib/cmd/runkperf/commands/data/events/event.go b/contrib/cmd/runkperf/commands/data/events/event.go
new file mode 100644
index 00000000..ef33756b
--- /dev/null
+++ b/contrib/cmd/runkperf/commands/data/events/event.go
@@ -0,0 +1,378 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+package events
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"strings"
+	"text/tabwriter"
+	"time"
+
+	"golang.org/x/sync/errgroup"
+
+	"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"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+)
+
+var Command = cli.Command{
+	Name:      "event",
+	ShortName: "ev",
+	Usage:     "Manage events",
+	Flags: []cli.Flag{
+		cli.StringFlag{
+			Name:  "kubeconfig",
+			Usage: "Path to the kubeconfig file",
+			Value: utils.DefaultKubeConfigPath,
+		},
+		cli.StringFlag{
+			Name:  "namespace",
+			Usage: "Namespace to use with commands. If the namespace does not exist, it will be created when executing add subcommand.",
+			Value: "default",
+		},
+	},
+	Subcommands: []cli.Command{
+		eventAddCommand,
+		eventDelCommand,
+		eventListCommand,
+	},
+}
+
+var eventAddCommand = cli.Command{
+	Name:      "add",
+	Usage:     "Add events set",
+	ArgsUsage: "NAME of the events set",
+	Flags: []cli.Flag{
+		cli.IntFlag{
+			Name:  "total",
+			Usage: "Total number of events to create",
+			Value: 10,
+		},
+		cli.IntFlag{
+			Name:  "group-size",
+			Usage: "Number of events to create in parallel per batch",
+			Value: 30,
+		},
+		cli.StringFlag{
+			Name:  "reason",
+			Usage: "The reason string for the events (e.g. ScaleUp, FailedScheduling)",
+			Value: "BenchmarkEvent",
+		},
+		cli.StringFlag{
+			Name:  "type",
+			Usage: "The type of the events: Normal or Warning",
+			Value: "Normal",
+		},
+		cli.IntFlag{
+			Name:  "size",
+			Usage: "The size of the event message in bytes (default targets ~5KB total event size; k8s metadata overhead is ~500-700 bytes)",
+			Value: 5000,
+		},
+		cli.StringFlag{
+			Name:  "involved-object-kind",
+			Usage: "The kind of the involved object (e.g. Pod, Node, Deployment)",
+			Value: "Pod",
+		},
+		cli.StringFlag{
+			Name:  "involved-object-name",
+			Usage: "The name of the involved object. If empty, a name will be generated.",
+			Value: "",
+		},
+		cli.Float64Flag{
+			Name:  "qps",
+			Usage: "QPS for the Kubernetes client rate limiter to control event operations",
+			Value: 100,
+		},
+		cli.IntFlag{
+			Name:  "burst",
+			Usage: "Burst for the Kubernetes client rate limiter to control event operations",
+			Value: 200,
+		},
+	},
+	Action: func(cliCtx *cli.Context) error {
+		if cliCtx.NArg() != 1 {
+			return fmt.Errorf("required only one argument as events set name: %v", cliCtx.Args())
+		}
+		eventSetName := strings.TrimSpace(cliCtx.Args().Get(0))
+		if len(eventSetName) == 0 {
+			return fmt.Errorf("required non-empty event set name")
+		}
+
+		kubeCfgPath := cliCtx.GlobalString("kubeconfig")
+		namespace := cliCtx.GlobalString("namespace")
+		total := cliCtx.Int("total")
+		groupSize := cliCtx.Int("group-size")
+		reason := cliCtx.String("reason")
+		eventType := cliCtx.String("type")
+		messageSize := cliCtx.Int("size")
+		involvedObjectKind := cliCtx.String("involved-object-kind")
+		involvedObjectName := cliCtx.String("involved-object-name")
+		qps := float32(cliCtx.Float64("qps"))
+		burst := cliCtx.Int("burst")
+
+		if total <= 0 {
+			return fmt.Errorf("total must be greater than 0")
+		}
+		if groupSize <= 0 {
+			return fmt.Errorf("group-size must be greater than 0")
+		}
+		if eventType != "Normal" && eventType != "Warning" {
+			return fmt.Errorf("type must be either Normal or Warning")
+		}
+		if messageSize <= 0 {
+			return fmt.Errorf("size must be greater than 0")
+		}
+
+		clientset, err := data.NewClientsetWithRateLimiter(kubeCfgPath, qps, burst)
+		if err != nil {
+			return err
+		}
+
+		err = data.PrepareNamespace(clientset, namespace)
+		if err != nil {
+			return err
+		}
+
+		err = createEvents(clientset, createOptions{
+			Namespace:          namespace,
+			EventSetName:       eventSetName,
+			Total:              total,
+			GroupSize:          groupSize,
+			Reason:             reason,
+			EventType:          eventType,
+			MessageSize:        messageSize,
+			InvolvedObjectKind: involvedObjectKind,
+			InvolvedObjectName: involvedObjectName,
+		})
+		if err != nil {
+			return err
+		}
+		fmt.Printf("Created %d events (set=%s, reason=%s, type=%s) in namespace %s\n",
+			total, eventSetName, reason, eventType, namespace)
+		return nil
+	},
+}
+
+var eventDelCommand = cli.Command{
+	Name:      "delete",
+	ShortName: "del",
+	ArgsUsage: "NAME",
+	Usage:     "Delete an events set",
+	Flags: []cli.Flag{
+		cli.Float64Flag{
+			Name:  "qps",
+			Usage: "QPS for the Kubernetes client rate limiter to control event deletion",
+			Value: 100,
+		},
+		cli.IntFlag{
+			Name:  "burst",
+			Usage: "Burst for the Kubernetes client rate limiter to control event deletion",
+			Value: 200,
+		},
+		cli.IntFlag{
+			Name:  "group-size",
+			Usage: "Number of events to delete in parallel per batch",
+			Value: 30,
+		},
+	},
+	Action: func(cliCtx *cli.Context) error {
+		if cliCtx.NArg() != 1 {
+			return fmt.Errorf("required only one events set name")
+		}
+		eventSetName := strings.TrimSpace(cliCtx.Args().Get(0))
+		if len(eventSetName) == 0 {
+			return fmt.Errorf("required non-empty events set name")
+		}
+
+		namespace := cliCtx.GlobalString("namespace")
+		kubeCfgPath := cliCtx.GlobalString("kubeconfig")
+		qps := float32(cliCtx.Float64("qps"))
+		burst := cliCtx.Int("burst")
+		groupSize := cliCtx.Int("group-size")
+		if groupSize <= 0 {
+			return fmt.Errorf("group-size must be greater than 0")
+		}
+		labelSelector := fmt.Sprintf("app=%s,eventSetName=%s", data.AppLabel, eventSetName)
+
+		clientset, err := data.NewClientsetWithRateLimiter(kubeCfgPath, qps, burst)
+		if err != nil {
+			return err
+		}
+
+		err = deleteEvents(clientset, labelSelector, namespace, groupSize)
+		if err != nil {
+			return err
+		}
+
+		fmt.Printf("Deleted events set %s in %s namespace\n", eventSetName, namespace)
+		return nil
+	},
+}
+
+var eventListCommand = cli.Command{
+	Name:      "list",
+	Usage:     "List generated event sets. Lists all if no arguments are given; otherwise, provide event set names separated by spaces.",
+	ArgsUsage: "[NAME ...]",
+	Action: func(cliCtx *cli.Context) error {
+		namespace := cliCtx.GlobalString("namespace")
+		kubeCfgPath := cliCtx.GlobalString("kubeconfig")
+		clientset, err := data.NewClientset(kubeCfgPath)
+		if err != nil {
+			return err
+		}
+
+		const (
+			minWidth = 1
+			tabWidth = 12
+			padding  = 3
+			padChar  = ' '
+			flags    = 0
+		)
+		tw := tabwriter.NewWriter(os.Stdout, minWidth, tabWidth, padding, padChar, flags)
+		fmt.Fprintln(tw, "NAME\tTYPE\tREASON\tTOTAL\t")
+
+		var labelSelector string
+		if cliCtx.NArg() == 0 {
+			labelSelector = fmt.Sprintf("app=%s", data.AppLabel)
+		} else {
+			args := cliCtx.Args()
+			namesStr := strings.Join(args, ",")
+			labelSelector = fmt.Sprintf("app=%s, eventSetName in (%s)", data.AppLabel, namesStr)
+		}
+
+		evMap := make(map[string]*eventSetInfo)
+		err = listEvents(clientset, labelSelector, namespace, data.DefaultPageSize, func(ev corev1.Event) error {
+			name, ok := ev.Labels["eventSetName"]
+			if !ok {
+				return nil
+			}
+
+			info, ok := evMap[name]
+			if !ok {
+				info = &eventSetInfo{
+					eventType: ev.Type,
+					reason:    ev.Reason,
+				}
+				evMap[name] = info
+			}
+			info.total++
+			return nil
+		})
+		if err != nil {
+			return err
+		}
+
+		for name, info := range evMap {
+			fmt.Fprintf(tw, "%s\t%s\t%s\t%d\n",
+				name,
+				info.eventType,
+				info.reason,
+				info.total,
+			)
+		}
+		return tw.Flush()
+	},
+}
+
+type eventSetInfo struct {
+	eventType string
+	reason    string
+	total     int
+}
+
+type createOptions struct {
+	Namespace          string
+	EventSetName       string
+	Total              int
+	GroupSize          int
+	Reason             string
+	EventType          string
+	MessageSize        int
+	InvolvedObjectKind string
+	InvolvedObjectName string
+}
+
+func createEvents(clientset kubernetes.Interface, opts createOptions) error {
+	now := time.Now()
+
+	for i := 0; i < opts.Total; i += opts.GroupSize {
+		g := new(errgroup.Group)
+		for j := i; j < i+opts.GroupSize && j < opts.Total; j++ {
+			idx := j
+			g.Go(func() error {
+				name := fmt.Sprintf("%s-ev-%s-%d", data.AppLabel, opts.EventSetName, idx)
+
+				message := request.RandomPayload(opts.MessageSize)
+
+				objName := opts.InvolvedObjectName
+				if objName == "" {
+					objName = fmt.Sprintf("%s-obj-%d", opts.EventSetName, idx)
+				}
+
+				event := &corev1.Event{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      name,
+						Namespace: opts.Namespace,
+						Labels: map[string]string{
+							"app":          data.AppLabel,
+							"eventSetName": opts.EventSetName,
+						},
+					},
+					InvolvedObject: corev1.ObjectReference{
+						Kind:      opts.InvolvedObjectKind,
+						Name:      objName,
+						Namespace: opts.Namespace,
+					},
+					Reason:         opts.Reason,
+					Message:        message,
+					Type:           opts.EventType,
+					Count:          1,
+					FirstTimestamp: metav1.NewTime(now),
+					LastTimestamp:  metav1.NewTime(now),
+					Source: corev1.EventSource{
+						Component: "runkperf-bench",
+					},
+				}
+
+				_, err := clientset.CoreV1().Events(opts.Namespace).Create(context.TODO(), event, metav1.CreateOptions{})
+				if err != nil {
+					return fmt.Errorf("failed to create event %s: %v", name, err)
+				}
+				return nil
+			})
+		}
+		if err := g.Wait(); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func deleteEvents(clientset kubernetes.Interface, labelSelector, namespace string, groupSize int) error {
+	return data.BatchDelete("event", namespace, groupSize, func(visit func(string) error) error {
+		return listEvents(clientset, labelSelector, namespace, data.DefaultPageSize, func(ev corev1.Event) error {
+			return visit(ev.Name)
+		})
+	}, func(name string) error {
+		return clientset.CoreV1().Events(namespace).Delete(context.TODO(), name, metav1.DeleteOptions{})
+	})
+}
+
+func listEvents(clientset kubernetes.Interface, labelSelector, namespace string, limit int64, fn func(corev1.Event) error) error {
+	return data.ListPaginated(metav1.ListOptions{LabelSelector: labelSelector, Limit: limit}, func(opts metav1.ListOptions) ([]corev1.Event, string, error) {
+		eventList, err := clientset.CoreV1().Events(namespace).List(context.TODO(), opts)
+		if err != nil {
+			return nil, "", fmt.Errorf("failed to list events: %w", err)
+		}
+		return eventList.Items, eventList.Continue, nil
+	}, fn)
+}
diff --git a/contrib/cmd/runkperf/commands/data/events/event_test.go b/contrib/cmd/runkperf/commands/data/events/event_test.go
new file mode 100644
index 00000000..ef6e771f
--- /dev/null
+++ b/contrib/cmd/runkperf/commands/data/events/event_test.go
@@ -0,0 +1,114 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+package events
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/Azure/kperf/contrib/cmd/runkperf/commands/data"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	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/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/kubernetes/fake"
+	ktesting "k8s.io/client-go/testing"
+)
+
+func TestCreateAndDeleteEvents(t *testing.T) {
+	client := fake.NewClientset()
+	opts := createOptions{
+		Namespace: "default", EventSetName: "set-a", Total: 5, GroupSize: 2,
+		Reason: "Benchmark", EventType: "Warning", MessageSize: 32, InvolvedObjectKind: "Pod",
+	}
+	require.NoError(t, createEvents(client, opts))
+	other := opts
+	other.EventSetName = "set-b"
+	other.Total = 1
+	other.InvolvedObjectName = "explicit"
+	require.NoError(t, createEvents(client, other))
+	events, err := client.CoreV1().Events(opts.Namespace).List(context.TODO(), metav1.ListOptions{})
+	require.NoError(t, err)
+	require.Len(t, events.Items, 6)
+	for _, event := range events.Items {
+		assert.Equal(t, data.AppLabel, event.Labels["app"])
+		assert.Equal(t, opts.Namespace, event.Namespace)
+		assert.Equal(t, opts.Namespace, event.InvolvedObject.Namespace)
+		assert.Equal(t, opts.InvolvedObjectKind, event.InvolvedObject.Kind)
+		assert.Equal(t, opts.Reason, event.Reason)
+		assert.Equal(t, opts.EventType, event.Type)
+		assert.Len(t, event.Message, opts.MessageSize)
+		assert.Equal(t, int32(1), event.Count)
+		assert.Equal(t, "runkperf-bench", event.Source.Component)
+		assert.False(t, event.FirstTimestamp.IsZero())
+		assert.Equal(t, event.FirstTimestamp, event.LastTimestamp)
+		if event.Labels["eventSetName"] == "set-b" {
+			assert.Equal(t, "explicit", event.InvolvedObject.Name)
+		} else {
+			assert.Contains(t, event.InvolvedObject.Name, "set-a-obj-")
+		}
+	}
+	require.NoError(t, deleteEvents(client, "app=runkperf,eventSetName=set-a", opts.Namespace, 2))
+	events, err = client.CoreV1().Events(opts.Namespace).List(context.TODO(), metav1.ListOptions{})
+	require.NoError(t, err)
+	require.Len(t, events.Items, 1)
+	assert.Equal(t, "set-b", events.Items[0].Labels["eventSetName"])
+}
+
+func TestDeleteEventsPages(t *testing.T) {
+	client := fake.NewClientset()
+	calls := 0
+	client.PrependReactor("list", "events", func(action ktesting.Action) (bool, runtime.Object, error) {
+		opts := action.(ktesting.ListActionImpl).GetListOptions()
+		assert.Equal(t, "default", action.GetNamespace())
+		assert.Equal(t, "app=runkperf", opts.LabelSelector)
+		assert.Equal(t, data.DefaultPageSize, opts.Limit)
+		list := &corev1.EventList{}
+		if calls == 0 {
+			assert.Empty(t, opts.Continue)
+			list.Continue = "next"
+			list.Items = []corev1.Event{{ObjectMeta: metav1.ObjectMeta{Name: "first", Labels: map[string]string{"app": "runkperf"}}}}
+		} else {
+			assert.Equal(t, 1, calls)
+			assert.Equal(t, "next", opts.Continue)
+			list.Items = []corev1.Event{{ObjectMeta: metav1.ObjectMeta{Name: "second", Labels: map[string]string{"app": "runkperf"}}}}
+		}
+		calls++
+		return true, list, nil
+	})
+	var deleted []string
+	client.PrependReactor("delete", "events", func(action ktesting.Action) (bool, runtime.Object, error) {
+		assert.Equal(t, 2, calls, "all pages must be fetched before deletion")
+		name := action.(ktesting.DeleteAction).GetName()
+		deleted = append(deleted, name)
+		return true, nil, apierrors.NewNotFound(schema.GroupResource{Resource: "events"}, name)
+	})
+	require.NoError(t, deleteEvents(client, "app=runkperf", "default", 1))
+	assert.Equal(t, []string{"first", "second"}, deleted)
+}
+
+func TestEventOperationErrors(t *testing.T) {
+	for _, verb := range []string{"list", "delete"} {
+		t.Run(verb, func(t *testing.T) {
+			client := fake.NewClientset(&corev1.Event{ObjectMeta: metav1.ObjectMeta{Name: "event", Namespace: "default"}})
+			sentinel := errors.New("denied")
+			client.PrependReactor(verb, "events", func(ktesting.Action) (bool, runtime.Object, error) {
+				return true, nil, sentinel
+			})
+			err := deleteEvents(client, "", "default", 1)
+			require.Error(t, err)
+			assert.ErrorIs(t, err, sentinel)
+			assert.Contains(t, err.Error(), "failed to "+verb+" event")
+			if verb == "list" {
+				for _, action := range client.Actions() {
+					assert.NotEqual(t, "delete", action.GetVerb())
+				}
+			}
+		})
+	}
+}
diff --git a/contrib/cmd/runkperf/commands/data/operations.go b/contrib/cmd/runkperf/commands/data/operations.go
new file mode 100644
index 00000000..29c67fad
--- /dev/null
+++ b/contrib/cmd/runkperf/commands/data/operations.go
@@ -0,0 +1,74 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+package data
+
+import (
+	"fmt"
+
+	"golang.org/x/sync/errgroup"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// DefaultPageSize is the list page size for Event, Secret, and Pod data commands.
+const DefaultPageSize int64 = 300
+
+// ListPaginated visits items from successive pages until the continuation token
+// is empty. List and visit errors stop traversal immediately.
+func ListPaginated[T any](opts metav1.ListOptions, list func(metav1.ListOptions) ([]T, string, error), visit func(T) error) error {
+	if opts.Limit <= 0 {
+		opts.Limit = DefaultPageSize
+	}
+	for {
+		items, next, err := list(opts)
+		if err != nil {
+			return err
+		}
+		for _, item := range items {
+			if err := visit(item); err != nil {
+				return err
+			}
+		}
+		if next == "" {
+			return nil
+		}
+		opts.Continue = next
+	}
+}
+
+// BatchDelete collects all names before deleting in fixed concurrent batches.
+// resource is the singular resource name used in messages. Missing objects are
+// ignored; other errors stop deletion after the current batch finishes.
+func BatchDelete(resource, namespace string, groupSize int, list func(func(string) error) error, deleteFn func(string) error) error {
+	if groupSize <= 0 {
+		return fmt.Errorf("group-size must be greater than 0")
+	}
+	var names []string
+	if err := list(func(name string) error {
+		names = append(names, name)
+		return nil
+	}); err != nil {
+		return err
+	}
+	if len(names) == 0 {
+		fmt.Printf("No %ss found in namespace %s, nothing to delete\n", resource, namespace)
+		return nil
+	}
+	for start := 0; start < len(names); start += groupSize {
+		g := new(errgroup.Group)
+		for i := start; i < len(names) && i-start < groupSize; i++ {
+			name := names[i]
+			g.Go(func() error {
+				if err := deleteFn(name); err != nil && !errors.IsNotFound(err) {
+					return fmt.Errorf("failed to delete %s %s: %w", resource, name, err)
+				}
+				return nil
+			})
+		}
+		if err := g.Wait(); err != nil {
+			return err
+		}
+	}
+	return nil
+}
diff --git a/contrib/cmd/runkperf/commands/data/operations_test.go b/contrib/cmd/runkperf/commands/data/operations_test.go
new file mode 100644
index 00000000..98544f91
--- /dev/null
+++ b/contrib/cmd/runkperf/commands/data/operations_test.go
@@ -0,0 +1,202 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+package data
+
+import (
+	"context"
+	"errors"
+	"strconv"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+func TestListPaginated(t *testing.T) {
+	for _, limit := range []int64{-1, 0, 2} {
+		t.Run(strconv.FormatInt(limit, 10), func(t *testing.T) {
+			expectedLimit := limit
+			if limit <= 0 {
+				expectedLimit = DefaultPageSize
+			}
+			pages := []struct {
+				items []int
+				next  string
+			}{{[]int{1, 2}, "second"}, {nil, "third"}, {[]int{3}, ""}}
+			tokens := []string{"", "second", "third"}
+			calls := 0
+			var visited []int
+			err := ListPaginated(metav1.ListOptions{LabelSelector: "app=runkperf", Limit: limit},
+				func(opts metav1.ListOptions) ([]int, string, error) {
+					require.Less(t, calls, len(pages))
+					assert.Equal(t, "app=runkperf", opts.LabelSelector)
+					assert.Equal(t, expectedLimit, opts.Limit)
+					assert.Equal(t, tokens[calls], opts.Continue)
+					page := pages[calls]
+					calls++
+					return page.items, page.next, nil
+				}, func(item int) error {
+					visited = append(visited, item)
+					return nil
+				})
+			require.NoError(t, err)
+			assert.Equal(t, 3, calls)
+			assert.Equal(t, []int{1, 2, 3}, visited)
+		})
+	}
+}
+
+func TestListPaginatedErrors(t *testing.T) {
+	sentinel := errors.New("failed")
+	t.Run("list", func(t *testing.T) {
+		err := ListPaginated(metav1.ListOptions{}, func(metav1.ListOptions) ([]int, string, error) {
+			return nil, "", sentinel
+		}, func(int) error {
+			t.Fatal("must not visit items after a list error")
+			return nil
+		})
+		assert.ErrorIs(t, err, sentinel)
+	})
+	t.Run("visit", func(t *testing.T) {
+		calls, visits := 0, 0
+		err := ListPaginated(metav1.ListOptions{}, func(metav1.ListOptions) ([]int, string, error) {
+			calls++
+			return []int{1, 2}, "next", nil
+		}, func(int) error {
+			visits++
+			return sentinel
+		})
+		assert.ErrorIs(t, err, sentinel)
+		assert.Equal(t, 1, calls)
+		assert.Equal(t, 1, visits)
+	})
+}
+
+func enumerateNames(names ...string) func(func(string) error) error {
+	return func(visit func(string) error) error {
+		for _, name := range names {
+			if err := visit(name); err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+}
+
+func TestBatchDeleteWaves(t *testing.T) {
+	for _, groupSize := range []int{1, 2, 10} {
+		t.Run(strconv.Itoa(groupSize), func(t *testing.T) {
+			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+			defer cancel()
+			names := []string{"0", "1", "2", "3", "4"}
+			releases := make([]chan struct{}, (len(names)+groupSize-1)/groupSize)
+			for i := range releases {
+				releases[i] = make(chan struct{})
+			}
+			started := make(chan int, len(names))
+			done := make(chan error, 1)
+			var completed, active atomic.Int64
+			listed := false
+			go func() {
+				done <- BatchDelete("pod", "default", groupSize, func(visit func(string) error) error {
+					err := enumerateNames(names...)(visit)
+					listed = true
+					return err
+				}, func(name string) error {
+					assert.True(t, listed, "listing must finish before deletion")
+					idx, err := strconv.Atoi(name)
+					if err != nil {
+						return err
+					}
+					wave := idx / groupSize
+					assert.GreaterOrEqual(t, completed.Load(), int64(wave*groupSize), "previous wave must finish")
+					assert.LessOrEqual(t, active.Add(1), int64(groupSize))
+					defer active.Add(-1)
+					started <- idx
+					select {
+					case <-releases[wave]:
+						completed.Add(1)
+						return nil
+					case <-ctx.Done():
+						return ctx.Err()
+					}
+				})
+			}()
+			for wave := range releases {
+				count := groupSize
+				if remaining := len(names) - wave*groupSize; remaining < count {
+					count = remaining
+				}
+				for i := 0; i < count; i++ {
+					select {
+					case idx := <-started:
+						assert.Equal(t, wave, idx/groupSize)
+					case <-ctx.Done():
+						t.Fatal("timed out waiting for deletion batch")
+					}
+				}
+				close(releases[wave])
+			}
+			require.NoError(t, <-done)
+			assert.Equal(t, int64(len(names)), completed.Load())
+		})
+	}
+}
+
+func TestBatchDeleteErrors(t *testing.T) {
+	sentinel := errors.New("denied")
+	for _, groupSize := range []int{-1, 0} {
+		err := BatchDelete("pod", "default", groupSize, func(func(string) error) error {
+			t.Fatal("invalid batch size must be rejected before listing")
+			return nil
+		}, nil)
+		assert.EqualError(t, err, "group-size must be greater than 0")
+	}
+	t.Run("list failure", func(t *testing.T) {
+		err := BatchDelete("pod", "default", 1, func(visit func(string) error) error {
+			require.NoError(t, visit("partial"))
+			return sentinel
+		}, func(string) error {
+			t.Fatal("must not delete after a list failure")
+			return nil
+		})
+		assert.ErrorIs(t, err, sentinel)
+	})
+	t.Run("empty", func(t *testing.T) {
+		require.NoError(t, BatchDelete("pod", "default", 1, enumerateNames(), func(string) error {
+			t.Fatal("must not delete when the list is empty")
+			return nil
+		}))
+	})
+	t.Run("missing", func(t *testing.T) {
+		var calls int
+		require.NoError(t, BatchDelete("pod", "default", 1, enumerateNames("gone", "present"), func(name string) error {
+			calls++
+			if name == "gone" {
+				return apierrors.NewNotFound(schema.GroupResource{Resource: "pods"}, name)
+			}
+			return nil
+		}))
+		assert.Equal(t, 2, calls)
+	})
+	t.Run("delete failure finishes current wave", func(t *testing.T) {
+		var calls atomic.Int32
+		err := BatchDelete("pod", "default", 2, enumerateNames("bad", "current", "later"), func(name string) error {
+			calls.Add(1)
+			assert.NotEqual(t, "later", name)
+			if name == "bad" {
+				return sentinel
+			}
+			return nil
+		})
+		assert.ErrorIs(t, err, sentinel)
+		assert.EqualError(t, err, "failed to delete pod bad: denied")
+		assert.Equal(t, int32(2), calls.Load())
+	})
+}
diff --git a/contrib/cmd/runkperf/commands/data/pod/pod.go b/contrib/cmd/runkperf/commands/data/pod/pod.go
new file mode 100644
index 00000000..f553cb40
--- /dev/null
+++ b/contrib/cmd/runkperf/commands/data/pod/pod.go
@@ -0,0 +1,332 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+package pod
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"strings"
+	"text/tabwriter"
+
+	"golang.org/x/sync/errgroup"
+
+	"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"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+)
+
+// paddingAnnotation holds random bytes used to inflate each pod's object size.
+const paddingAnnotation = "runkperf.io/padding"
+
+// Use an unhandled scheduler name to keep ingestion pods unscheduled.
+const dataOnlySchedulerName = "runkperf-data-only"
+
+// pauseImage is the placeholder image used for the bare ingestion pods.
+const pauseImage = "registry.k8s.io/pause:3.9"
+
+var Command = cli.Command{
+	Name:      "pod",
+	ShortName: "po",
+	Usage:     "Manage pods",
+	Flags: []cli.Flag{
+		cli.StringFlag{
+			Name:  "kubeconfig",
+			Usage: "Path to the kubeconfig file",
+			Value: utils.DefaultKubeConfigPath,
+		},
+		cli.StringFlag{
+			Name:  "namespace",
+			Usage: "Namespace to use with commands. If the namespace does not exist, it will be created when executing add subcommand.",
+			Value: "default",
+		},
+	},
+	Subcommands: []cli.Command{
+		podAddCommand,
+		podDelCommand,
+		podListCommand,
+	},
+}
+
+var podAddCommand = cli.Command{
+	Name:      "add",
+	Usage:     "Add pods set",
+	ArgsUsage: "NAME of the pods set",
+	Flags: []cli.Flag{
+		cli.IntFlag{
+			Name:  "total",
+			Usage: "Total number of pods to create",
+			Value: 10,
+		},
+		cli.IntFlag{
+			Name:  "group-size",
+			Usage: "Number of pods to create in parallel per batch",
+			Value: 30,
+		},
+		cli.IntFlag{
+			Name:  "size",
+			Usage: "The size of the padding stored in each pod annotation (Unit: Byte)",
+			Value: 10240,
+		},
+		cli.Float64Flag{
+			Name:  "qps",
+			Usage: "QPS for the Kubernetes client rate limiter to control pod operations",
+			Value: 100,
+		},
+		cli.IntFlag{
+			Name:  "burst",
+			Usage: "Burst for the Kubernetes client rate limiter to control pod operations",
+			Value: 200,
+		},
+	},
+	Action: func(cliCtx *cli.Context) error {
+		if cliCtx.NArg() != 1 {
+			return fmt.Errorf("required only one argument as pods set name: %v", cliCtx.Args())
+		}
+		podSetName := strings.TrimSpace(cliCtx.Args().Get(0))
+		if len(podSetName) == 0 {
+			return fmt.Errorf("required non-empty pod set name")
+		}
+
+		kubeCfgPath := cliCtx.GlobalString("kubeconfig")
+		namespace := cliCtx.GlobalString("namespace")
+		total := cliCtx.Int("total")
+		groupSize := cliCtx.Int("group-size")
+		size := cliCtx.Int("size")
+		qps := float32(cliCtx.Float64("qps"))
+		burst := cliCtx.Int("burst")
+
+		if err := checkPodParams(size, groupSize, total); err != nil {
+			return err
+		}
+
+		clientset, err := data.NewClientsetWithRateLimiter(kubeCfgPath, qps, burst)
+		if err != nil {
+			return err
+		}
+
+		err = data.PrepareNamespace(clientset, namespace)
+		if err != nil {
+			return err
+		}
+
+		err = createPods(clientset, namespace, podSetName, total, groupSize, size)
+		if err != nil {
+			return err
+		}
+		fmt.Printf("Created %d pods (set=%s, size=%d B) in namespace %s\n",
+			total, podSetName, size, namespace)
+		return nil
+	},
+}
+
+var podDelCommand = cli.Command{
+	Name:      "delete",
+	ShortName: "del",
+	ArgsUsage: "NAME",
+	Usage:     "Delete a pods set",
+	Flags: []cli.Flag{
+		cli.Float64Flag{
+			Name:  "qps",
+			Usage: "QPS for the Kubernetes client rate limiter to control pod deletion",
+			Value: 100,
+		},
+		cli.IntFlag{
+			Name:  "burst",
+			Usage: "Burst for the Kubernetes client rate limiter to control pod deletion",
+			Value: 200,
+		},
+		cli.IntFlag{
+			Name:  "group-size",
+			Usage: "Number of pods to delete in parallel per batch",
+			Value: 30,
+		},
+	},
+	Action: func(cliCtx *cli.Context) error {
+		if cliCtx.NArg() != 1 {
+			return fmt.Errorf("required only one pods set name")
+		}
+		podSetName := strings.TrimSpace(cliCtx.Args().Get(0))
+		if len(podSetName) == 0 {
+			return fmt.Errorf("required non-empty pods set name")
+		}
+
+		namespace := cliCtx.GlobalString("namespace")
+		kubeCfgPath := cliCtx.GlobalString("kubeconfig")
+		qps := float32(cliCtx.Float64("qps"))
+		burst := cliCtx.Int("burst")
+		groupSize := cliCtx.Int("group-size")
+		if groupSize <= 0 {
+			return fmt.Errorf("group-size must be greater than 0")
+		}
+		labelSelector := fmt.Sprintf("app=%s,podSetName=%s", data.AppLabel, podSetName)
+
+		clientset, err := data.NewClientsetWithRateLimiter(kubeCfgPath, qps, burst)
+		if err != nil {
+			return err
+		}
+
+		err = deletePods(clientset, labelSelector, namespace, groupSize)
+		if err != nil {
+			return err
+		}
+
+		fmt.Printf("Deleted pods set %s in %s namespace\n", podSetName, namespace)
+		return nil
+	},
+}
+
+var podListCommand = cli.Command{
+	Name:      "list",
+	Usage:     "List generated pod sets. Lists all if no arguments are given; otherwise, provide pod set names separated by spaces.",
+	ArgsUsage: "[NAME ...]",
+	Action: func(cliCtx *cli.Context) error {
+		namespace := cliCtx.GlobalString("namespace")
+		kubeCfgPath := cliCtx.GlobalString("kubeconfig")
+		clientset, err := data.NewClientset(kubeCfgPath)
+		if err != nil {
+			return err
+		}
+
+		const (
+			minWidth = 1
+			tabWidth = 12
+			padding  = 3
+			padChar  = ' '
+			flags    = 0
+		)
+		tw := tabwriter.NewWriter(os.Stdout, minWidth, tabWidth, padding, padChar, flags)
+		fmt.Fprintln(tw, "NAME\tSIZE\tTOTAL\t")
+
+		var labelSelector string
+		if cliCtx.NArg() == 0 {
+			labelSelector = fmt.Sprintf("app=%s", data.AppLabel)
+		} else {
+			args := cliCtx.Args()
+			namesStr := strings.Join(args, ",")
+			labelSelector = fmt.Sprintf("app=%s, podSetName in (%s)", data.AppLabel, namesStr)
+		}
+
+		podMap := make(map[string]*podSetInfo)
+		err = listPods(clientset, labelSelector, namespace, data.DefaultPageSize, func(pod corev1.Pod) error {
+			name, ok := pod.Labels["podSetName"]
+			if !ok {
+				return nil
+			}
+
+			info, ok := podMap[name]
+			if !ok {
+				info = &podSetInfo{
+					size: len(pod.Annotations[paddingAnnotation]),
+				}
+				podMap[name] = info
+			}
+			info.total++
+			return nil
+		})
+		if err != nil {
+			return err
+		}
+
+		for name, info := range podMap {
+			fmt.Fprintf(tw, "%s\t%d\t%d\n",
+				name,
+				info.size,
+				info.total,
+			)
+		}
+		return tw.Flush()
+	},
+}
+
+type podSetInfo struct {
+	size  int
+	total int
+}
+
+func checkPodParams(size, groupSize, total int) error {
+	if size <= 0 {
+		return fmt.Errorf("size must be greater than 0")
+	}
+	if groupSize <= 0 {
+		return fmt.Errorf("group-size must be greater than 0")
+	}
+	if total <= 0 {
+		return fmt.Errorf("total must be greater than 0")
+	}
+	return nil
+}
+
+func createPods(clientset kubernetes.Interface, namespace, podSetName string, total, groupSize, size int) error {
+	for i := 0; i < total; i += groupSize {
+		g := new(errgroup.Group)
+		for j := i; j < i+groupSize && j < total; j++ {
+			idx := j
+			g.Go(func() error {
+				name := fmt.Sprintf("%s-pod-%s-%d", data.AppLabel, podSetName, idx)
+
+				padding := request.RandomPayload(size)
+
+				pod := &corev1.Pod{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      name,
+						Namespace: namespace,
+						Labels: map[string]string{
+							"app":        data.AppLabel,
+							"podSetName": podSetName,
+						},
+						Annotations: map[string]string{
+							paddingAnnotation: padding,
+						},
+					},
+					Spec: corev1.PodSpec{
+						SchedulerName: dataOnlySchedulerName,
+						Containers: []corev1.Container{
+							{
+								Name:  "main",
+								Image: pauseImage,
+							},
+						},
+					},
+				}
+
+				_, err := clientset.CoreV1().Pods(namespace).Create(context.TODO(), pod, metav1.CreateOptions{})
+				if err != nil {
+					return fmt.Errorf("failed to create pod %s: %v", name, err)
+				}
+				return nil
+			})
+		}
+		if err := g.Wait(); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func deletePods(clientset kubernetes.Interface, labelSelector, namespace string, groupSize int) error {
+	return data.BatchDelete("pod", namespace, groupSize, func(visit func(string) error) error {
+		return listPods(clientset, labelSelector, namespace, data.DefaultPageSize, func(pod corev1.Pod) error {
+			return visit(pod.Name)
+		})
+	}, func(name string) error {
+		return clientset.CoreV1().Pods(namespace).Delete(context.TODO(), name, metav1.DeleteOptions{})
+	})
+}
+
+func listPods(clientset kubernetes.Interface, labelSelector, namespace string, limit int64, fn func(corev1.Pod) error) error {
+	return data.ListPaginated(metav1.ListOptions{LabelSelector: labelSelector, Limit: limit}, func(opts metav1.ListOptions) ([]corev1.Pod, string, error) {
+		podList, err := clientset.CoreV1().Pods(namespace).List(context.TODO(), opts)
+		if err != nil {
+			return nil, "", fmt.Errorf("failed to list pods: %w", err)
+		}
+		return podList.Items, podList.Continue, nil
+	}, fn)
+}
diff --git a/contrib/cmd/runkperf/commands/data/pod/pod_test.go b/contrib/cmd/runkperf/commands/data/pod/pod_test.go
new file mode 100644
index 00000000..db33eaca
--- /dev/null
+++ b/contrib/cmd/runkperf/commands/data/pod/pod_test.go
@@ -0,0 +1,160 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+package pod
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	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/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/kubernetes/fake"
+	ktesting "k8s.io/client-go/testing"
+
+	"github.com/Azure/kperf/contrib/cmd/runkperf/commands/data"
+)
+
+func TestCheckPodParams(t *testing.T) {
+	assert.NoError(t, checkPodParams(1, 1, 1))
+	assert.Error(t, checkPodParams(0, 1, 1))
+	assert.Error(t, checkPodParams(1, 0, 1))
+	assert.Error(t, checkPodParams(1, 1, 0))
+}
+
+func TestCreatePods(t *testing.T) {
+	clientset := fake.NewClientset()
+	const (
+		namespace  = "default"
+		podSetName = "set-a"
+		total      = 7
+		groupSize  = 3
+		size       = 128
+	)
+
+	err := createPods(clientset, namespace, podSetName, total, groupSize, size)
+	require.NoError(t, err)
+
+	pods, err := clientset.CoreV1().Pods(namespace).List(context.TODO(), metav1.ListOptions{})
+	require.NoError(t, err)
+	require.Len(t, pods.Items, total)
+
+	for _, pod := range pods.Items {
+		assert.Equal(t, data.AppLabel, pod.Labels["app"])
+		assert.Equal(t, podSetName, pod.Labels["podSetName"])
+		assert.Len(t, pod.Annotations[paddingAnnotation], size)
+		assert.Equal(t, "runkperf-data-only", pod.Spec.SchedulerName)
+		require.Len(t, pod.Spec.Containers, 1)
+		assert.Equal(t, pauseImage, pod.Spec.Containers[0].Image)
+	}
+}
+
+func TestListPods(t *testing.T) {
+	clientset := fake.NewClientset()
+	const namespace = "default"
+
+	require.NoError(t, createPods(clientset, namespace, "set-a", 5, 3, 64))
+	require.NoError(t, createPods(clientset, namespace, "set-b", 2, 3, 32))
+
+	podMap := make(map[string]*podSetInfo)
+	selector := "app=" + data.AppLabel
+	err := listPods(clientset, selector, namespace, data.DefaultPageSize, func(pod corev1.Pod) error {
+		name := pod.Labels["podSetName"]
+		info, ok := podMap[name]
+		if !ok {
+			info = &podSetInfo{size: len(pod.Annotations[paddingAnnotation])}
+			podMap[name] = info
+		}
+		info.total++
+		return nil
+	})
+	require.NoError(t, err)
+
+	require.Contains(t, podMap, "set-a")
+	require.Contains(t, podMap, "set-b")
+	assert.Equal(t, 5, podMap["set-a"].total)
+	assert.Equal(t, 64, podMap["set-a"].size)
+	assert.Equal(t, 2, podMap["set-b"].total)
+	assert.Equal(t, 32, podMap["set-b"].size)
+}
+
+func TestDeletePods(t *testing.T) {
+	clientset := fake.NewClientset()
+	const namespace = "default"
+
+	require.NoError(t, createPods(clientset, namespace, "set-a", 4, 2, 16))
+	require.NoError(t, createPods(clientset, namespace, "set-b", 3, 2, 16))
+
+	selectorA := "app=" + data.AppLabel + ",podSetName=set-a"
+	require.NoError(t, deletePods(clientset, selectorA, namespace, 2))
+
+	remaining, err := clientset.CoreV1().Pods(namespace).List(context.TODO(), metav1.ListOptions{})
+	require.NoError(t, err)
+	assert.Len(t, remaining.Items, 3)
+	for _, pod := range remaining.Items {
+		assert.Equal(t, "set-b", pod.Labels["podSetName"])
+	}
+
+	// Deleting a set that does not exist is a no-op and returns no error.
+	require.NoError(t, deletePods(clientset, "app="+data.AppLabel+",podSetName=missing", namespace, 2))
+}
+
+func TestDeletePodsPages(t *testing.T) {
+	client := fake.NewClientset()
+	calls := 0
+	client.PrependReactor("list", "pods", func(action ktesting.Action) (bool, runtime.Object, error) {
+		opts := action.(ktesting.ListActionImpl).GetListOptions()
+		assert.Equal(t, "default", action.GetNamespace())
+		assert.Equal(t, "app=runkperf", opts.LabelSelector)
+		assert.Equal(t, data.DefaultPageSize, opts.Limit)
+		list := &corev1.PodList{}
+		if calls == 0 {
+			assert.Empty(t, opts.Continue)
+			list.Continue = "next"
+			list.Items = []corev1.Pod{{ObjectMeta: metav1.ObjectMeta{Name: "first", Labels: map[string]string{"app": "runkperf"}}}}
+		} else {
+			assert.Equal(t, 1, calls)
+			assert.Equal(t, "next", opts.Continue)
+			list.Items = []corev1.Pod{{ObjectMeta: metav1.ObjectMeta{Name: "second", Labels: map[string]string{"app": "runkperf"}}}}
+		}
+		calls++
+		return true, list, nil
+	})
+	var deleted []string
+	client.PrependReactor("delete", "pods", func(action ktesting.Action) (bool, runtime.Object, error) {
+		assert.Equal(t, 2, calls, "all pages must be fetched before deletion")
+		name := action.(ktesting.DeleteAction).GetName()
+		deleted = append(deleted, name)
+		return true, nil, apierrors.NewNotFound(schema.GroupResource{Resource: "pods"}, name)
+	})
+	require.NoError(t, deletePods(client, "app=runkperf", "default", 1))
+	assert.Equal(t, []string{"first", "second"}, deleted)
+}
+
+func TestPodOperationErrors(t *testing.T) {
+	for _, verb := range []string{"list", "delete"} {
+		t.Run(verb, func(t *testing.T) {
+			client := fake.NewClientset(&corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod", Namespace: "default"}})
+			sentinel := errors.New("denied")
+			client.PrependReactor(verb, "pods", func(ktesting.Action) (bool, runtime.Object, error) {
+				return true, nil, sentinel
+			})
+			err := deletePods(client, "", "default", 1)
+			require.Error(t, err)
+			assert.ErrorIs(t, err, sentinel)
+			assert.Contains(t, err.Error(), "failed to "+verb+" pod")
+			if verb == "list" {
+				for _, action := range client.Actions() {
+					assert.NotEqual(t, "delete", action.GetVerb())
+				}
+			}
+		})
+	}
+}
diff --git a/contrib/cmd/runkperf/commands/data/secrets/secret.go b/contrib/cmd/runkperf/commands/data/secrets/secret.go
new file mode 100644
index 00000000..d44a9d9d
--- /dev/null
+++ b/contrib/cmd/runkperf/commands/data/secrets/secret.go
@@ -0,0 +1,339 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+package secrets
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"strconv"
+	"strings"
+	"text/tabwriter"
+
+	"golang.org/x/sync/errgroup"
+
+	"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"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+)
+
+var Command = cli.Command{
+	Name:      "secret",
+	ShortName: "sec",
+	Usage:     "Manage secrets",
+	Flags: []cli.Flag{
+		cli.StringFlag{
+			Name:  "kubeconfig",
+			Usage: "Path to the kubeconfig file",
+			Value: utils.DefaultKubeConfigPath,
+		},
+		cli.StringFlag{
+			Name:  "namespace",
+			Usage: "Namespace to use with commands. If the namespace does not exist, it will be created when executing add subcommand",
+			Value: "default",
+		},
+	},
+	Subcommands: []cli.Command{
+		secretAddCommand,
+		secretDelCommand,
+		secretListCommand,
+	},
+}
+
+var secretAddCommand = cli.Command{
+	Name:      "add",
+	Usage:     "Add secrets set",
+	ArgsUsage: "NAME of the secrets set",
+	Flags: []cli.Flag{
+		cli.IntFlag{
+			Name:  "size",
+			Usage: "The size of each secret value (Unit: Byte)",
+			Value: 100,
+		},
+		cli.IntFlag{
+			Name:  "group-size",
+			Usage: "The number of secrets to create in parallel per batch",
+			Value: 30,
+		},
+		cli.IntFlag{
+			Name:  "total",
+			Usage: "Total number of secrets to create",
+			Value: 10,
+		},
+		cli.Float64Flag{
+			Name:  "qps",
+			Usage: "QPS for the Kubernetes client rate limiter to control secret operations",
+			Value: 100,
+		},
+		cli.IntFlag{
+			Name:  "burst",
+			Usage: "Burst for the Kubernetes client rate limiter to control secret operations",
+			Value: 200,
+		},
+	},
+	Action: func(cliCtx *cli.Context) error {
+		if cliCtx.NArg() != 1 {
+			return fmt.Errorf("required only one argument as secrets set name: %v", cliCtx.Args())
+		}
+		secName := strings.TrimSpace(cliCtx.Args().Get(0))
+		if len(secName) == 0 {
+			return fmt.Errorf("required non-empty secret set name")
+		}
+
+		kubeCfgPath := cliCtx.GlobalString("kubeconfig")
+		size := cliCtx.Int("size")
+		groupSize := cliCtx.Int("group-size")
+		total := cliCtx.Int("total")
+
+		err := checkSecretParams(size, groupSize, total)
+		if err != nil {
+			return err
+		}
+
+		namespace := cliCtx.GlobalString("namespace")
+		qps := float32(cliCtx.Float64("qps"))
+		burst := cliCtx.Int("burst")
+
+		clientset, err := data.NewClientsetWithRateLimiter(kubeCfgPath, qps, burst)
+		if err != nil {
+			return err
+		}
+
+		err = data.PrepareNamespace(clientset, namespace)
+		if err != nil {
+			return err
+		}
+
+		err = createSecrets(clientset, namespace, secName, size, groupSize, total)
+		if err != nil {
+			return err
+		}
+		fmt.Printf("Created secret set %s with size %d B, group-size %d, total %d\n", secName, size, groupSize, total)
+		return nil
+	},
+}
+
+var secretDelCommand = cli.Command{
+	Name:      "delete",
+	ShortName: "del",
+	ArgsUsage: "NAME",
+	Usage:     "Delete a secrets set",
+	Flags: []cli.Flag{
+		cli.Float64Flag{
+			Name:  "qps",
+			Usage: "QPS for the Kubernetes client rate limiter to control secret deletion",
+			Value: 100,
+		},
+		cli.IntFlag{
+			Name:  "burst",
+			Usage: "Burst for the Kubernetes client rate limiter to control secret deletion",
+			Value: 200,
+		},
+		cli.IntFlag{
+			Name:  "group-size",
+			Usage: "Number of secrets to delete in parallel per batch",
+			Value: 30,
+		},
+	},
+	Action: func(cliCtx *cli.Context) error {
+		if cliCtx.NArg() != 1 {
+			return fmt.Errorf("required only one secrets set name")
+		}
+		secName := strings.TrimSpace(cliCtx.Args().Get(0))
+		if len(secName) == 0 {
+			return fmt.Errorf("required non-empty secrets set name")
+		}
+
+		namespace := cliCtx.GlobalString("namespace")
+		kubeCfgPath := cliCtx.GlobalString("kubeconfig")
+		qps := float32(cliCtx.Float64("qps"))
+		burst := cliCtx.Int("burst")
+		groupSize := cliCtx.Int("group-size")
+		if groupSize <= 0 {
+			return fmt.Errorf("group-size must be greater than 0")
+		}
+		labelSelector := fmt.Sprintf("app=%s,secName=%s", data.AppLabel, secName)
+
+		clientset, err := data.NewClientsetWithRateLimiter(kubeCfgPath, qps, burst)
+		if err != nil {
+			return err
+		}
+
+		err = deleteSecrets(clientset, labelSelector, namespace, groupSize)
+		if err != nil {
+			return err
+		}
+
+		fmt.Printf("Deleted secrets set %s in %s namespace\n", secName, namespace)
+		return nil
+	},
+}
+
+var secretListCommand = cli.Command{
+	Name:      "list",
+	Usage:     "List generated secrets. Lists all if no arguments are given; otherwise, provide secret set names separated by spaces.",
+	ArgsUsage: "[NAME ...]",
+	Action: func(cliCtx *cli.Context) error {
+		namespace := cliCtx.GlobalString("namespace")
+		kubeCfgPath := cliCtx.GlobalString("kubeconfig")
+		clientset, err := data.NewClientset(kubeCfgPath)
+		if err != nil {
+			return err
+		}
+
+		const (
+			minWidth = 1
+			tabWidth = 12
+			padding  = 3
+			padChar  = ' '
+			flags    = 0
+		)
+		tw := tabwriter.NewWriter(os.Stdout, minWidth, tabWidth, padding, padChar, flags)
+		fmt.Fprintln(tw, "NAME\tSIZE\tGROUP_SIZE\tTOTAL\t")
+
+		var labelSelector string
+		if cliCtx.NArg() == 0 {
+			labelSelector = fmt.Sprintf("app=%s", data.AppLabel)
+		} else {
+			args := cliCtx.Args()
+			namesStr := strings.Join(args, ",")
+			labelSelector = fmt.Sprintf("app=%s, secName in (%s)", data.AppLabel, namesStr)
+		}
+
+		type secretSetInfo struct {
+			size    int
+			total   int
+			ownerID map[string]int // ownerID -> count of secrets in that group
+		}
+		secMap := make(map[string]*secretSetInfo)
+		err = listSecrets(clientset, labelSelector, namespace, data.DefaultPageSize, func(sec corev1.Secret) error {
+			name, ok := sec.Labels["secName"]
+			if !ok {
+				return fmt.Errorf("failed to find the secName of secret %s", sec.Name)
+			}
+
+			info, ok := secMap[name]
+			if !ok {
+				info = &secretSetInfo{
+					ownerID: make(map[string]int),
+				}
+				if d, exists := sec.Data["data"]; exists {
+					info.size = len(d)
+				}
+				secMap[name] = info
+			}
+
+			info.total++
+
+			ownerID, ok := sec.Labels["ownerID"]
+			if !ok {
+				return fmt.Errorf("failed to find the ownerID of secret %s", sec.Name)
+			}
+			info.ownerID[ownerID]++
+			return nil
+		})
+		if err != nil {
+			return err
+		}
+
+		for name, info := range secMap {
+			// Group size should be the max count across all ownerID groups
+			groupSize := 0
+			for _, count := range info.ownerID {
+				if count > groupSize {
+					groupSize = count
+				}
+			}
+			fmt.Fprintf(tw, "%s\t%d\t%d\t%d\n",
+				name,
+				info.size,
+				groupSize,
+				info.total,
+			)
+		}
+		return tw.Flush()
+	},
+}
+
+func checkSecretParams(size int, groupSize int, total int) error {
+	if size <= 0 {
+		return fmt.Errorf("size must be greater than 0")
+	}
+	if groupSize <= 0 {
+		return fmt.Errorf("group-size must be greater than 0")
+	}
+	if total <= 0 {
+		return fmt.Errorf("total amount must be greater than 0")
+	}
+	return nil
+}
+
+func createSecrets(clientset kubernetes.Interface, namespace string, secName string, size int, groupSize int, total int) error {
+	for i := 0; i < total; i += groupSize {
+		ownerID := i
+		g := new(errgroup.Group)
+		for j := i; j < i+groupSize && j < total; j++ {
+			idx := j
+			g.Go(func() error {
+				name := fmt.Sprintf("%s-sec-%s-%d", data.AppLabel, secName, idx)
+
+				randData, err := request.RandBytes(size)
+				if err != nil {
+					return fmt.Errorf("failed to generate random data for secret %s: %v", name, err)
+				}
+
+				secret := &corev1.Secret{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: name,
+						Labels: map[string]string{
+							"ownerID": strconv.Itoa(ownerID),
+							"app":     data.AppLabel,
+							"secName": secName,
+						},
+					},
+					Type: corev1.SecretTypeOpaque,
+					Data: map[string][]byte{
+						"data": randData,
+					},
+				}
+
+				_, err = clientset.CoreV1().Secrets(namespace).Create(context.TODO(), secret, metav1.CreateOptions{})
+				if err != nil {
+					return fmt.Errorf("failed to create secret %s: %v", name, err)
+				}
+				return nil
+			})
+		}
+		if err := g.Wait(); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func deleteSecrets(clientset kubernetes.Interface, labelSelector string, namespace string, groupSize int) error {
+	return data.BatchDelete("secret", namespace, groupSize, func(visit func(string) error) error {
+		return listSecrets(clientset, labelSelector, namespace, data.DefaultPageSize, func(sec corev1.Secret) error {
+			return visit(sec.Name)
+		})
+	}, func(name string) error {
+		return clientset.CoreV1().Secrets(namespace).Delete(context.TODO(), name, metav1.DeleteOptions{})
+	})
+}
+
+func listSecrets(clientset kubernetes.Interface, labelSelector string, namespace string, limit int64, fn func(corev1.Secret) error) error {
+	return data.ListPaginated(metav1.ListOptions{LabelSelector: labelSelector, Limit: limit}, func(opts metav1.ListOptions) ([]corev1.Secret, string, error) {
+		secrets, err := clientset.CoreV1().Secrets(namespace).List(context.TODO(), opts)
+		if err != nil {
+			return nil, "", fmt.Errorf("failed to list secrets: %w", err)
+		}
+		return secrets.Items, secrets.Continue, nil
+	}, fn)
+}
diff --git a/contrib/cmd/runkperf/commands/data/secrets/secret_test.go b/contrib/cmd/runkperf/commands/data/secrets/secret_test.go
new file mode 100644
index 00000000..7a0dfc99
--- /dev/null
+++ b/contrib/cmd/runkperf/commands/data/secrets/secret_test.go
@@ -0,0 +1,101 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+package secrets
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/Azure/kperf/contrib/cmd/runkperf/commands/data"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	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/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/kubernetes/fake"
+	ktesting "k8s.io/client-go/testing"
+)
+
+func TestCreateAndDeleteSecrets(t *testing.T) {
+	client := fake.NewClientset()
+	require.NoError(t, createSecrets(client, "default", "set-a", 32, 2, 5))
+	require.NoError(t, createSecrets(client, "default", "set-b", 16, 10, 1))
+	secrets, err := client.CoreV1().Secrets("default").List(context.TODO(), metav1.ListOptions{})
+	require.NoError(t, err)
+	require.Len(t, secrets.Items, 6)
+	groups := map[string]int{}
+	for _, secret := range secrets.Items {
+		assert.Equal(t, data.AppLabel, secret.Labels["app"])
+		assert.Equal(t, corev1.SecretTypeOpaque, secret.Type)
+		if secret.Labels["secName"] == "set-a" {
+			assert.Len(t, secret.Data["data"], 32)
+			groups[secret.Labels["ownerID"]]++
+		} else {
+			assert.Equal(t, "set-b", secret.Labels["secName"])
+			assert.Len(t, secret.Data["data"], 16)
+		}
+	}
+	assert.Equal(t, map[string]int{"0": 2, "2": 2, "4": 1}, groups)
+	require.NoError(t, deleteSecrets(client, "app=runkperf,secName=set-a", "default", 2))
+	secrets, err = client.CoreV1().Secrets("default").List(context.TODO(), metav1.ListOptions{})
+	require.NoError(t, err)
+	require.Len(t, secrets.Items, 1)
+	assert.Equal(t, "set-b", secrets.Items[0].Labels["secName"])
+}
+
+func TestDeleteSecretsPages(t *testing.T) {
+	client := fake.NewClientset()
+	calls := 0
+	client.PrependReactor("list", "secrets", func(action ktesting.Action) (bool, runtime.Object, error) {
+		opts := action.(ktesting.ListActionImpl).GetListOptions()
+		assert.Equal(t, "default", action.GetNamespace())
+		assert.Equal(t, "app=runkperf", opts.LabelSelector)
+		assert.Equal(t, data.DefaultPageSize, opts.Limit)
+		list := &corev1.SecretList{}
+		if calls == 0 {
+			assert.Empty(t, opts.Continue)
+			list.Continue = "next"
+			list.Items = []corev1.Secret{{ObjectMeta: metav1.ObjectMeta{Name: "first", Labels: map[string]string{"app": "runkperf"}}}}
+		} else {
+			assert.Equal(t, 1, calls)
+			assert.Equal(t, "next", opts.Continue)
+			list.Items = []corev1.Secret{{ObjectMeta: metav1.ObjectMeta{Name: "second", Labels: map[string]string{"app": "runkperf"}}}}
+		}
+		calls++
+		return true, list, nil
+	})
+	var deleted []string
+	client.PrependReactor("delete", "secrets", func(action ktesting.Action) (bool, runtime.Object, error) {
+		assert.Equal(t, 2, calls, "all pages must be fetched before deletion")
+		name := action.(ktesting.DeleteAction).GetName()
+		deleted = append(deleted, name)
+		return true, nil, apierrors.NewNotFound(schema.GroupResource{Resource: "secrets"}, name)
+	})
+	require.NoError(t, deleteSecrets(client, "app=runkperf", "default", 1))
+	assert.Equal(t, []string{"first", "second"}, deleted)
+}
+
+func TestSecretOperationErrors(t *testing.T) {
+	for _, verb := range []string{"list", "delete"} {
+		t.Run(verb, func(t *testing.T) {
+			client := fake.NewClientset(&corev1.Secret{ObjectMeta: metav1.ObjectMeta{Name: "secret", Namespace: "default"}})
+			sentinel := errors.New("denied")
+			client.PrependReactor(verb, "secrets", func(ktesting.Action) (bool, runtime.Object, error) {
+				return true, nil, sentinel
+			})
+			err := deleteSecrets(client, "", "default", 1)
+			require.Error(t, err)
+			assert.ErrorIs(t, err, sentinel)
+			assert.Contains(t, err.Error(), "failed to "+verb+" secret")
+			if verb == "list" {
+				for _, action := range client.Actions() {
+					assert.NotEqual(t, "delete", action.GetVerb())
+				}
+			}
+		})
+	}
+}
diff --git a/contrib/cmd/runkperf/commands/root.go b/contrib/cmd/runkperf/commands/root.go
index 50796d1f..bc3342d9 100644
--- a/contrib/cmd/runkperf/commands/root.go
+++ b/contrib/cmd/runkperf/commands/root.go
@@ -13,6 +13,9 @@ import (
 	"github.com/Azure/kperf/contrib/cmd/runkperf/commands/data"
 	"github.com/Azure/kperf/contrib/cmd/runkperf/commands/data/configmaps"
 	"github.com/Azure/kperf/contrib/cmd/runkperf/commands/data/daemonsets"
+	"github.com/Azure/kperf/contrib/cmd/runkperf/commands/data/events"
+	"github.com/Azure/kperf/contrib/cmd/runkperf/commands/data/pod"
+	"github.com/Azure/kperf/contrib/cmd/runkperf/commands/data/secrets"
 	"github.com/Azure/kperf/contrib/cmd/runkperf/commands/warmup"
 
 	"github.com/urfave/cli"
@@ -23,6 +26,9 @@ func init() {
 	data.RegisterSubcommands(
 		configmaps.Command,
 		daemonsets.Command,
+		events.Command,
+		pod.Command,
+		secrets.Command,
 	)
 }
 
diff --git a/request/random.go b/request/random.go
index 070133fa..f5d1967f 100644
--- a/request/random.go
+++ b/request/random.go
@@ -386,7 +386,7 @@ func renderPatchBody(raw string) []byte {
 	if !strings.Contains(raw, patchRandomPlaceholder) {
 		return []byte(raw)
 	}
-	return []byte(strings.ReplaceAll(raw, patchRandomPlaceholder, randomPayload(8)))
+	return []byte(strings.ReplaceAll(raw, patchRandomPlaceholder, RandomPayload(8)))
 }
 
 func newRequestPatchBuilder(src *types.RequestPatch, resourceVersion string, maxRetries int) *requestPatchBuilder {
@@ -526,7 +526,7 @@ func (b *requestPostDelBuilder) Build(cli rest.Interface) Requester {
 	body, err := utils.RenderTemplate(b.resource, map[string]interface{}{
 		"namePattern":  name,
 		"namespace":    b.namespace,
-		"payload":      randomPayload(b.payloadSize),
+		"payload":      RandomPayload(b.payloadSize),
 		"nodeSelector": b.nodeSelector,
 	})
 	if err != nil {
@@ -624,7 +624,7 @@ func (b *requestPutBuilder) Build(cli rest.Interface) Requester {
 			Name:      finalName,
 			Namespace: b.namespace,
 		},
-		Data: map[string]string{"payload": randomPayload(b.payloadSize)},
+		Data: map[string]string{"payload": RandomPayload(b.payloadSize)},
 	}
 	body, _ := json.Marshal(cm)
 
@@ -636,13 +636,26 @@ func (b *requestPutBuilder) Build(cli rest.Interface) Requester {
 	}
 }
 
-// randomPayload returns a string of exactly n bytes, uniformly sampled from
+// RandomPayload returns a string of exactly n bytes, uniformly sampled from
 // [a-zA-Z0-9] via crypto/rand with rejection sampling to avoid modulo bias.
-// Returns "" when n <= 0.
-func randomPayload(n int) string {
+// It returns "" when n <= 0 and panics if reading random bytes fails.
+func RandomPayload(n int) string {
 	if n <= 0 {
 		return ""
 	}
+	b, err := RandBytes(n)
+	if err != nil {
+		panic(err)
+	}
+	return string(b)
+}
+
+// RandBytes returns n uniformly sampled alphanumeric bytes using crypto/rand.
+// It returns an error for non-positive lengths or if reading random bytes fails.
+func RandBytes(n int) ([]byte, error) {
+	if n <= 0 {
+		return nil, fmt.Errorf("length must be positive")
+	}
 	const alphabet = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
 	// Only use byte values in [0, threshold). Since len(alphabet)=62 and
 	// 248 is the largest multiple of 62 below 256, mapping b%62 is uniform
@@ -659,7 +672,7 @@ func randomPayload(n int) string {
 
 	for i := 0; i < n; {
 		if _, err := rand.Read(buf); err != nil {
-			panic(fmt.Errorf("failed to read random bytes: %w", err))
+			return nil, fmt.Errorf("failed to read random bytes: %w", err)
 		}
 
 		for _, b := range buf {
@@ -669,10 +682,10 @@ func randomPayload(n int) string {
 			out[i] = alphabet[int(b)%len(alphabet)]
 			i++
 			if i == n {
-				return string(out)
+				return out, nil
 			}
 		}
 	}
 
-	return string(out)
+	return out, nil
 }
diff --git a/request/random_test.go b/request/random_test.go
index 865958f3..81e6996b 100644
--- a/request/random_test.go
+++ b/request/random_test.go
@@ -4,6 +4,7 @@
 package request
 
 import (
+	"strconv"
 	"sync"
 	"testing"
 
@@ -34,17 +35,38 @@ func TestRenderPatchBody(t *testing.T) {
 }
 
 func TestRandomPayload(t *testing.T) {
-	const n = 8
-	s := randomPayload(n)
-	assert.Len(t, s, n)
-	for _, r := range s {
-		isAlnum := (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9')
-		assert.True(t, isAlnum, "unexpected char %q", r)
+	for _, n := range []int{-1, 0, 1, 8, 62, 4095, 4096, 4097, 16384} {
+		t.Run(strconv.Itoa(n), func(t *testing.T) {
+			t.Parallel()
+			s := RandomPayload(n)
+			if n <= 0 {
+				assert.Empty(t, s)
+				return
+			}
+			assert.Len(t, s, n)
+			assert.Regexp(t, `^[a-zA-Z0-9]+$`, s)
+		})
+	}
+	assert.NotEqual(t, RandomPayload(32), RandomPayload(32))
+}
+
+func TestRandBytes(t *testing.T) {
+	for _, n := range []int{-1, 0, 1, 8, 62, 4095, 4096, 4097, 16384} {
+		t.Run(strconv.Itoa(n), func(t *testing.T) {
+			t.Parallel()
+			b, err := RandBytes(n)
+			if n <= 0 {
+				assert.EqualError(t, err, "length must be positive")
+				assert.Nil(t, b)
+				return
+			}
+			if !assert.NoError(t, err) {
+				return
+			}
+			assert.Len(t, b, n)
+			assert.Regexp(t, `^[a-zA-Z0-9]+$`, string(b))
+		})
 	}
-	// n <= 0 returns an empty string.
-	assert.Equal(t, "", randomPayload(0))
-	// Two calls should not collide.
-	assert.NotEqual(t, randomPayload(n), randomPayload(n))
 }
 
 func TestRequestPatchBuilderRendersUniqueBodies(t *testing.T) {