From 8e8e9617073fa7f0a9a51789923afa96d4d90121 Mon Sep 17 00:00:00 2001 From: liyu-ma <17180634+liyu-ma@users.noreply.github.com> Date: Wed, 28 Oct 2026 23:00:23 +1000 Subject: [PATCH 1/5] Extend pod creation configuration --- api/types/load_traffic.go | 19 +- .../manifests/loadprofile/pod_create.yaml | 28 + .../manifests/workload/pods/templates/pod.tpl | 21 +- request/builder.go | 530 ++++++++++++++++++ request/builder_test.go | 147 +++++ request/random.go | 509 ----------------- request/random_test.go | 82 --- 7 files changed, 736 insertions(+), 600 deletions(-) create mode 100644 contrib/internal/manifests/loadprofile/pod_create.yaml create mode 100644 request/builder.go create mode 100644 request/builder_test.go diff --git a/api/types/load_traffic.go b/api/types/load_traffic.go index d648224c..6203c529 100644 --- a/api/types/load_traffic.go +++ b/api/types/load_traffic.go @@ -210,6 +210,12 @@ type RequestPostDel struct { } type PostDelPodOptions struct { + // Labels adds metadata labels to created Pods. + Labels map[string]string `json:"labels,omitempty" yaml:"labels,omitempty"` + // ServiceAccountName selects a pre-created ServiceAccount. + ServiceAccountName string `json:"serviceAccountName,omitempty" yaml:"serviceAccountName,omitempty"` + // SchedulerName selects a scheduler; an unhandled name keeps Pods unscheduled. + SchedulerName string `json:"schedulerName,omitempty" yaml:"schedulerName,omitempty"` // NodeSelector schedules workload Pods separately from runner Pods, for // example on a dedicated node pool: // @@ -440,14 +446,15 @@ func (r *RequestPostDel) Validate() error { return fmt.Errorf("payloadSize must >= 0: %v", r.PayloadSize) } - if r.Pod != nil && len(r.Pod.NodeSelector) > 0 { + if r.Pod != nil { if r.Group != "" || r.Version != "v1" || r.Resource != "pods" { - return fmt.Errorf("pod nodeSelector only supports core/v1/pods") + return fmt.Errorf("pod options only support core/v1/pods") } - - errs := metav1validation.ValidateLabels(r.Pod.NodeSelector, field.NewPath("pod", "nodeSelector")) - if len(errs) > 0 { - return fmt.Errorf("invalid pod nodeSelector: %v", errs.ToAggregate()) + if len(r.Pod.NodeSelector) > 0 { + errs := metav1validation.ValidateLabels(r.Pod.NodeSelector, field.NewPath("pod", "nodeSelector")) + if len(errs) > 0 { + return fmt.Errorf("invalid pod nodeSelector: %v", errs.ToAggregate()) + } } } diff --git a/contrib/internal/manifests/loadprofile/pod_create.yaml b/contrib/internal/manifests/loadprofile/pod_create.yaml new file mode 100644 index 00000000..afe6ce96 --- /dev/null +++ b/contrib/internal/manifests/loadprofile/pod_create.yaml @@ -0,0 +1,28 @@ +version: 1 +description: Pod CREATE baseline - 400 creates/sec for 10 minutes +specs: + # Rate limit per runner. + - rate: 400 + duration: 600 + total: 0 + client: 400 + conns: 20 + contentType: json + maxRetries: 0 + requests: + - shares: 1 + postDel: + group: "" + version: v1 + resource: pods + namespace: pod-benchmark + deleteRatio: 0 + # Env padding for an approximately 4 KiB Pod. + payloadSize: 3650 + pod: + labels: + example.com/workload: benchmark + benchmark: pod-create + benchmark-run: replace-with-unique-run-id + serviceAccountName: pod-benchmark + schedulerName: kperf-unscheduled diff --git a/contrib/internal/manifests/workload/pods/templates/pod.tpl b/contrib/internal/manifests/workload/pods/templates/pod.tpl index 27038892..74ed4fdc 100644 --- a/contrib/internal/manifests/workload/pods/templates/pod.tpl +++ b/contrib/internal/manifests/workload/pods/templates/pod.tpl @@ -2,18 +2,33 @@ {{- $namespace:= .Values.namespace }} {{- $payload:= .Values.payload }} {{- $nodeSelector:= .Values.nodeSelector }} +{{- $hasApp := false }} +{{- range $key, $value := .Values.labels }} +{{- if eq $key "app" }}{{ $hasApp = true }}{{ end }} +{{- end }} apiVersion: v1 kind: Pod metadata: - name: {{ $name }} - namespace: {{ $namespace }} + name: {{ printf "%q" $name }} + namespace: {{ printf "%q" $namespace }} labels: + {{- if not $hasApp }} app: fake-pod + {{- end }} + {{- range $key, $value := .Values.labels }} + {{ printf "%q" $key }}: {{ printf "%q" $value }} + {{- end }} spec: + {{- if .Values.serviceAccountName }} + serviceAccountName: {{ printf "%q" .Values.serviceAccountName }} + {{- end }} + {{- if .Values.schedulerName }} + schedulerName: {{ printf "%q" .Values.schedulerName }} + {{- end }} {{- if $nodeSelector }} nodeSelector: {{- range $key, $value := $nodeSelector }} - "{{ $key }}": "{{ $value }}" + {{ printf "%q" $key }}: {{ printf "%q" $value }} {{- end }} {{- end }} # Use pause image so pods reach Running with few events, avoiding an diff --git a/request/builder.go b/request/builder.go new file mode 100644 index 00000000..e5a9d889 --- /dev/null +++ b/request/builder.go @@ -0,0 +1,530 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +package request + +import ( + "context" + "crypto/rand" + "encoding/json" + "fmt" + "maps" + "math/big" + "strings" + "sync/atomic" + "time" + + "github.com/Azure/kperf/api/types" + "github.com/Azure/kperf/contrib/utils" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime/schema" + apitypes "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/kubernetes/scheme" + "k8s.io/client-go/rest" +) + +// RESTRequestBuilder is used to build rest.Request. +type RESTRequestBuilder interface { + Build(cli rest.Interface) Requester +} + +type requestGetBuilder struct { + version schema.GroupVersion + resource string + namespace string + name string + resourceVersion string + maxRetries int +} + +func newRequestGetBuilder(src *types.RequestGet, resourceVersion string, maxRetries int) *requestGetBuilder { + return &requestGetBuilder{ + version: schema.GroupVersion{ + Group: src.Group, + Version: src.Version, + }, + resource: src.Resource, + namespace: src.Namespace, + name: src.Name, + resourceVersion: resourceVersion, + maxRetries: maxRetries, + } +} + +// Build implements RequestBuilder.Build. +func (b *requestGetBuilder) Build(cli rest.Interface) Requester { + // https://kubernetes.io/docs/reference/using-api/#api-groups + comps := make([]string, 0, 5) + if b.version.Group == "" { + comps = append(comps, "api", b.version.Version) + } else { + comps = append(comps, "apis", b.version.Group, b.version.Version) + } + if b.namespace != "" { + comps = append(comps, "namespaces", b.namespace) + } + comps = append(comps, b.resource, b.name) + + return &DiscardRequester{ + BaseRequester: BaseRequester{ + method: "GET", + req: cli.Get().AbsPath(comps...). + SpecificallyVersionedParams( + &metav1.GetOptions{ResourceVersion: b.resourceVersion}, + scheme.ParameterCodec, + schema.GroupVersion{Version: "v1"}, + ).MaxRetries(b.maxRetries), + }, + } +} + +type requestListBuilder struct { + version schema.GroupVersion + resource string + namespace string + limit int64 + labelSelector string + fieldSelector string + resourceVersion string + maxRetries int +} + +func newRequestListBuilder(src *types.RequestList, resourceVersion string, maxRetries int) *requestListBuilder { + return &requestListBuilder{ + version: schema.GroupVersion{ + Group: src.Group, + Version: src.Version, + }, + resource: src.Resource, + namespace: src.Namespace, + limit: int64(src.Limit), + labelSelector: src.Selector, + fieldSelector: src.FieldSelector, + resourceVersion: resourceVersion, + maxRetries: maxRetries, + } +} + +// Build implements RequestBuilder.Build. +func (b *requestListBuilder) Build(cli rest.Interface) Requester { + // https://kubernetes.io/docs/reference/using-api/#api-groups + comps := make([]string, 0, 5) + if b.version.Group == "" { + comps = append(comps, "api", b.version.Version) + } else { + comps = append(comps, "apis", b.version.Group, b.version.Version) + } + if b.namespace != "" { + comps = append(comps, "namespaces", b.namespace) + } + comps = append(comps, b.resource) + + return &DiscardRequester{ + BaseRequester: BaseRequester{ + method: "LIST", + req: cli.Get().AbsPath(comps...). + SpecificallyVersionedParams( + &metav1.ListOptions{ + LabelSelector: b.labelSelector, + FieldSelector: b.fieldSelector, + ResourceVersion: b.resourceVersion, + Limit: b.limit, + }, + scheme.ParameterCodec, + schema.GroupVersion{Version: "v1"}, + ).MaxRetries(b.maxRetries), + }, + } +} + +type requestWatchListBuilder struct { + version schema.GroupVersion + resource string + namespace string + labelSelector string + fieldSelector string + maxRetries int +} + +func newRequestWatchListBuilder(src *types.RequestWatchList, maxRetries int) *requestWatchListBuilder { + return &requestWatchListBuilder{ + version: schema.GroupVersion{ + Group: src.Group, + Version: src.Version, + }, + resource: src.Resource, + namespace: src.Namespace, + labelSelector: src.Selector, + fieldSelector: src.FieldSelector, + maxRetries: maxRetries, + } +} + +// Build implements RequestBuilder.Build. +func (b *requestWatchListBuilder) Build(cli rest.Interface) Requester { + // https://kubernetes.io/docs/reference/using-api/#api-groups + comps := make([]string, 0, 5) + if b.version.Group == "" { + comps = append(comps, "api", b.version.Version) + } else { + comps = append(comps, "apis", b.version.Group, b.version.Version) + } + if b.namespace != "" { + comps = append(comps, "namespaces", b.namespace) + } + comps = append(comps, b.resource) + + return &WatchListRequester{ + BaseRequester: BaseRequester{ + method: "WATCHLIST", + req: cli.Get().AbsPath(comps...). + SpecificallyVersionedParams( + &metav1.ListOptions{ + LabelSelector: b.labelSelector, + FieldSelector: b.fieldSelector, + ResourceVersion: "", + Watch: true, + SendInitialEvents: toPtr(true), + ResourceVersionMatch: metav1.ResourceVersionMatchNotOlderThan, + AllowWatchBookmarks: true, + }, + scheme.ParameterCodec, + schema.GroupVersion{Version: "v1"}, + ).MaxRetries(b.maxRetries), + }, + } +} + +type requestGetPodLogBuilder struct { + namespace string + name string + container string + tailLines *int64 + limitBytes *int64 + maxRetries int +} + +func newRequestGetPodLogBuilder(src *types.RequestGetPodLog, maxRetries int) *requestGetPodLogBuilder { + b := &requestGetPodLogBuilder{ + namespace: src.Namespace, + name: src.Name, + container: src.Container, + maxRetries: maxRetries, + } + if src.TailLines != nil { + b.tailLines = toPtr(*src.TailLines) + } + if src.LimitBytes != nil { + b.limitBytes = toPtr(*src.LimitBytes) + } + return b +} + +// Build implements RequestBuilder.Build. +func (b *requestGetPodLogBuilder) Build(cli rest.Interface) Requester { + // https://kubernetes.io/docs/reference/using-api/#api-groups + apiPath, version := "api", "v1" + + comps := make([]string, 2, 7) + comps[0], comps[1] = apiPath, version + comps = append(comps, "namespaces", b.namespace) + comps = append(comps, "pods", b.name, "log") + + return &DiscardRequester{ + BaseRequester: BaseRequester{ + method: "POD_LOG", + req: cli.Get().AbsPath(comps...). + SpecificallyVersionedParams( + &corev1.PodLogOptions{ + Container: b.container, + TailLines: b.tailLines, + LimitBytes: b.limitBytes, + }, + scheme.ParameterCodec, + schema.GroupVersion{Version: "v1"}, + ).MaxRetries(b.maxRetries), + }, + } +} + +// When this placeholder is present in a patch body, it is +// replaced on every request with a random string. This guarantees each patch +// actually mutates the target object. +const patchRandomPlaceholder = "{{.Random}}" + +type requestPatchBuilder struct { + version schema.GroupVersion + resource string + resourceVersion string + namespace string + name string + keySpaceSize int + patchType apitypes.PatchType + body string + maxRetries int +} + +// renderPatchBody replaces every occurrence of patchRandomPlaceholder in raw +// with a random string. +func renderPatchBody(raw string) []byte { + if !strings.Contains(raw, patchRandomPlaceholder) { + return []byte(raw) + } + return []byte(strings.ReplaceAll(raw, patchRandomPlaceholder, RandomPayload(8))) +} + +func newRequestPatchBuilder(src *types.RequestPatch, resourceVersion string, maxRetries int) *requestPatchBuilder { + patchType, _ := types.GetPatchType(src.PatchType) + + return &requestPatchBuilder{ + version: schema.GroupVersion{ + Group: src.Group, + Version: src.Version, + }, + resource: src.Resource, + resourceVersion: resourceVersion, + namespace: src.Namespace, + name: src.Name, + keySpaceSize: src.KeySpaceSize, + patchType: patchType, + body: src.Body, + maxRetries: maxRetries, + } +} + +// Build implements RequestBuilder.Build. +func (b *requestPatchBuilder) Build(cli rest.Interface) Requester { + // https://kubernetes.io/docs/reference/using-api/#api-groups + comps := make([]string, 0, 5) + if b.version.Group == "" { + comps = append(comps, "api", b.version.Version) + } else { + comps = append(comps, "apis", b.version.Group, b.version.Version) + } + if b.namespace != "" { + comps = append(comps, "namespaces", b.namespace) + } + // Generate random suffix based on keySpaceSize + randomInt, _ := rand.Int(rand.Reader, big.NewInt(int64(b.keySpaceSize))) + suffix := randomInt.Int64() + + // Create final resource name: name-{suffix} + finalName := fmt.Sprintf("%s-%d", b.name, suffix) + comps = append(comps, b.resource, finalName) + + // Render a unique body per request if patchRandomPlaceholder exists. Bodies without the + // placeholder pass through unchanged. + reqBody := renderPatchBody(b.body) + + return &DiscardRequester{ + BaseRequester: BaseRequester{ + method: "PATCH", + req: cli.Patch(b.patchType).AbsPath(comps...). + Body(reqBody). + MaxRetries(b.maxRetries), + }, + } +} + +type requestPostDelBuilder struct { + version schema.GroupVersion + resource string + resourceVersion string + namespace string + deleteRatio float64 + payloadSize int + nodeSelector map[string]string + labels map[string]string + serviceAccountName string + schedulerName string + maxRetries int + + // Per-builder cache for created resources + cache *Cache + + // Per-builder atomic counter for unique ID generation + resourceCounter int64 +} + +func newRequestPostDelBuilder(src *types.RequestPostDel, resourceVersion string, maxRetries int) *requestPostDelBuilder { + b := &requestPostDelBuilder{ + version: schema.GroupVersion{Group: src.Group, Version: src.Version}, + resource: src.Resource, + resourceVersion: resourceVersion, + namespace: src.Namespace, + deleteRatio: src.DeleteRatio, + payloadSize: src.PayloadSize, + maxRetries: maxRetries, + cache: InitCache(), // Initialize the cache + } + if p := src.Pod; p != nil { + b.nodeSelector = maps.Clone(p.NodeSelector) + b.labels = maps.Clone(p.Labels) + b.serviceAccountName = p.ServiceAccountName + b.schedulerName = p.SchedulerName + } + return b +} + +// Build implements RequestBuilder.Build. +func (b *requestPostDelBuilder) Build(cli rest.Interface) Requester { + comps := make([]string, 0, 5) + if b.version.Group == "" { + comps = append(comps, "api", b.version.Version) + } else { + comps = append(comps, "apis", b.version.Group, b.version.Version) + } + if b.namespace != "" { + comps = append(comps, "namespaces", b.namespace) + } + + // Random pick operation DELETE or CREATE based on deleteRatio weight probability + randomInt, _ := rand.Int(rand.Reader, big.NewInt(1000)) + shouldDelete := float64(randomInt.Int64())/1000.0 < b.deleteRatio + + if shouldDelete { + // Try to get a name from cache + if name, ok := b.cache.Pop(); ok { + comps = append(comps, b.resource, name) + + return &PostDelDiscardRequester{ + builder: b, + name: name, + operation: "DELETE", + DiscardRequester: DiscardRequester{ + BaseRequester: BaseRequester{ + method: "DELETE", + req: cli.Delete().AbsPath(comps...). + MaxRetries(b.maxRetries), + }, + }, + } + } + // If cache is empty, fall through to POST + } + + // POST logic - create resource and add to cache if successful + comps = append(comps, b.resource) + + // Use builder's atomic counter for synchronized unique ID generation + counter := atomic.AddInt64(&b.resourceCounter, 1) + timestamp := time.Now().UnixNano() + name := fmt.Sprintf("%d-%d", timestamp, counter) + + body, err := utils.RenderTemplate(b.resource, map[string]interface{}{ + "namePattern": name, + "namespace": b.namespace, + "payload": RandomPayload(b.payloadSize), + "nodeSelector": b.nodeSelector, + "labels": b.labels, + "serviceAccountName": b.serviceAccountName, + "schedulerName": b.schedulerName, + }) + if err != nil { + panic(fmt.Errorf("failed to render %s template: %w", b.resource, err)) + } + + return &PostDelDiscardRequester{ + builder: b, + name: name, + operation: "POST", + DiscardRequester: DiscardRequester{ + BaseRequester: BaseRequester{ + method: "POST", + req: cli.Post().AbsPath(comps...).Body(body).MaxRetries(b.maxRetries), + }, + }, + } +} + +// PostDelDiscardRequester handles both POST and DELETE requests with cache management +type PostDelDiscardRequester struct { + builder *requestPostDelBuilder + name string + operation string // "POST" or "DELETE" + DiscardRequester +} + +func (reqr *PostDelDiscardRequester) Do(ctx context.Context) (bytes int64, err error) { + // Use DiscardRequester's Do method to discard response body + bytes, err = reqr.DiscardRequester.Do(ctx) + + switch reqr.operation { + case "POST": + // Create-only runs retain objects, but do not need a deletion-name cache. + if err == nil && reqr.builder.deleteRatio > 0 { + reqr.builder.cache.Push(reqr.name) + } + case "DELETE": + // If DELETE request failed, restore the item back to cache + // since the resource still exists in Kubernetes + if err != nil { + reqr.builder.cache.Push(reqr.name) + } + } + + return bytes, err +} + +func toPtr[T any](v T) *T { + return &v +} + +// requestPutBuilder builds PUT requests that overwrite an existing ConfigMap +// named `-` with a body whose `data.payload` field +// carries `payloadSize` random bytes. The target ConfigMap must already exist +// (e.g. created via `runkperf data cm add`); otherwise the apiserver returns +// 404 and the request is counted as a failure. Only `configmaps` is supported +// today — see RequestPut.Validate. +type requestPutBuilder struct { + version schema.GroupVersion + resource string + namespace string + name string + keySpaceSize int + payloadSize int + maxRetries int +} + +func newRequestPutBuilder(src *types.RequestPut, maxRetries int) *requestPutBuilder { + return &requestPutBuilder{ + version: schema.GroupVersion{ + Group: src.Group, + Version: src.Version, + }, + resource: src.Resource, + namespace: src.Namespace, + name: src.Name, + keySpaceSize: src.KeySpaceSize, + payloadSize: src.PayloadSize, + maxRetries: maxRetries, + } +} + +// Build implements RESTRequestBuilder.Build. +func (b *requestPutBuilder) Build(cli rest.Interface) Requester { + comps := []string{"api", b.version.Version, "namespaces", b.namespace, b.resource} + + randomInt, _ := rand.Int(rand.Reader, big.NewInt(int64(b.keySpaceSize))) + finalName := fmt.Sprintf("%s-%d", b.name, randomInt.Int64()) + comps = append(comps, finalName) + + cm := &corev1.ConfigMap{ + TypeMeta: metav1.TypeMeta{APIVersion: "v1", Kind: "ConfigMap"}, + ObjectMeta: metav1.ObjectMeta{ + Name: finalName, + Namespace: b.namespace, + }, + Data: map[string]string{"payload": RandomPayload(b.payloadSize)}, + } + body, _ := json.Marshal(cm) + + return &DiscardRequester{ + BaseRequester: BaseRequester{ + method: "PUT", + req: cli.Put().AbsPath(comps...).Body(body).MaxRetries(b.maxRetries), + }, + } +} diff --git a/request/builder_test.go b/request/builder_test.go new file mode 100644 index 00000000..539601ab --- /dev/null +++ b/request/builder_test.go @@ -0,0 +1,147 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +package request + +import ( + "context" + "encoding/json" + "io" + "net/http" + "strings" + "sync" + "testing" + + "github.com/Azure/kperf/api/types" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/client-go/kubernetes/scheme" + "k8s.io/client-go/rest/fake" +) + +func TestRenderPatchBody(t *testing.T) { + const tmpl = `{"metadata":{"labels":{"bench-tick":"{{.Random}}"}}}` + + // Placeholder is replaced with a random string, so the result no longer + // contains the placeholder and differs from the raw template. + rendered := string(renderPatchBody(tmpl)) + assert.NotContains(t, rendered, patchRandomPlaceholder) + assert.NotEqual(t, tmpl, rendered) + + // Each render produces a distinct body. + assert.NotEqual(t, string(renderPatchBody(tmpl)), string(renderPatchBody(tmpl))) + + // Every occurrence is replaced. + multi := string(renderPatchBody(`{"a":"{{.Random}}","b":"{{.Random}}"}`)) + assert.NotContains(t, multi, patchRandomPlaceholder) + + // A body without the placeholder is returned unchanged (backward compatible). + const static = `{"metadata":{"labels":{"bench-tick":"1"}}}` + assert.Equal(t, static, string(renderPatchBody(static))) +} + +func TestRequestPatchBuilderRendersUniqueBodies(t *testing.T) { + b := newRequestPatchBuilder(&types.RequestPatch{ + KubeGroupVersionResource: types.KubeGroupVersionResource{ + Version: "v1", + Resource: "configmaps", + }, + Namespace: "bench", + Name: "cm", + KeySpaceSize: 100, + PatchType: "merge", + Body: `{"metadata":{"labels":{"bench-tick":"{{.Random}}"}}}`, + }, "", 0) + + // Each call renders a fresh random body, producing a distinct result. + first := string(renderPatchBody(b.body)) + second := string(renderPatchBody(b.body)) + assert.NotEqual(t, first, second) + assert.NotContains(t, first, patchRandomPlaceholder) + assert.NotContains(t, second, patchRandomPlaceholder) +} + +func TestRequestPatchBuilderRendersConcurrencySafe(t *testing.T) { + b := newRequestPatchBuilder(&types.RequestPatch{ + KubeGroupVersionResource: types.KubeGroupVersionResource{ + Version: "v1", + Resource: "configmaps", + }, + Name: "cm", + PatchType: "merge", + Body: `{"n":"{{.Random}}"}`, + }, "", 0) + + const goroutines = 16 + const perGoroutine = 100 + + var wg sync.WaitGroup + var mu sync.Mutex + seen := make(map[string]struct{}, goroutines*perGoroutine) + + for i := 0; i < goroutines; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for j := 0; j < perGoroutine; j++ { + body := string(renderPatchBody(b.body)) + assert.NotContains(t, body, patchRandomPlaceholder) + mu.Lock() + seen[body] = struct{}{} + mu.Unlock() + } + }() + } + wg.Wait() + + // Random bodies are effectively unique across all requests. + assert.Equal(t, goroutines*perGoroutine, len(seen)) +} + +func TestPodOptions(t *testing.T) { + for _, tt := range []struct { + name string + options *types.PostDelPodOptions + }{ + {name: "defaults"}, + {name: "custom", options: &types.PostDelPodOptions{ + Labels: map[string]string{"app": ""}, NodeSelector: map[string]string{"agentpool": "benchmark"}, + ServiceAccountName: "custom-account", SchedulerName: "custom-scheduler", + }}, + } { + t.Run(tt.name, func(t *testing.T) { + builder := newRequestPostDelBuilder(&types.RequestPostDel{ + KubeGroupVersionResource: types.KubeGroupVersionResource{Version: "v1", Resource: "pods"}, + Namespace: "default", + Pod: tt.options, + }, "", 0) + var pod corev1.Pod + cli := &fake.RESTClient{ + GroupVersion: schema.GroupVersion{Version: "v1"}, + NegotiatedSerializer: scheme.Codecs.WithoutConversion(), + Client: fake.CreateHTTPClient(func(r *http.Request) (*http.Response, error) { + require.Equal(t, "POST", r.Method) + require.NoError(t, json.NewDecoder(r.Body).Decode(&pod)) + return &http.Response{ + StatusCode: http.StatusCreated, Header: http.Header{"Content-Type": {"application/json"}}, + Body: io.NopCloser(strings.NewReader("{}")), + }, nil + }), + } + _, err := builder.Build(cli).Do(context.Background()) + require.NoError(t, err) + require.Equal(t, "default", pod.Namespace) + options := tt.options + if options == nil { + options = &types.PostDelPodOptions{Labels: map[string]string{"app": "fake-pod"}} + } + require.Equal(t, options.Labels, pod.Labels) + require.Equal(t, options.NodeSelector, pod.Spec.NodeSelector) + require.Equal(t, options.ServiceAccountName, pod.Spec.ServiceAccountName) + require.Equal(t, options.SchedulerName, pod.Spec.SchedulerName) + }) + } +} diff --git a/request/random.go b/request/random.go index f5d1967f..30a593c1 100644 --- a/request/random.go +++ b/request/random.go @@ -6,23 +6,11 @@ package request import ( "context" "crypto/rand" - "encoding/json" "fmt" "math/big" - "strings" "sync" - "sync/atomic" - "time" "github.com/Azure/kperf/api/types" - "github.com/Azure/kperf/contrib/utils" - - corev1 "k8s.io/api/core/v1" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/runtime/schema" - apitypes "k8s.io/apimachinery/pkg/types" - "k8s.io/client-go/kubernetes/scheme" - "k8s.io/client-go/rest" ) // WeightedRandomRequests is used to generate requests based on LoadProfileSpec. @@ -139,503 +127,6 @@ func (r *WeightedRandomRequests) Stop() { }) } -// RESTRequestBuilder is used to build rest.Request. -type RESTRequestBuilder interface { - Build(cli rest.Interface) Requester -} - -type requestGetBuilder struct { - version schema.GroupVersion - resource string - namespace string - name string - resourceVersion string - maxRetries int -} - -func newRequestGetBuilder(src *types.RequestGet, resourceVersion string, maxRetries int) *requestGetBuilder { - return &requestGetBuilder{ - version: schema.GroupVersion{ - Group: src.Group, - Version: src.Version, - }, - resource: src.Resource, - namespace: src.Namespace, - name: src.Name, - resourceVersion: resourceVersion, - maxRetries: maxRetries, - } -} - -// Build implements RequestBuilder.Build. -func (b *requestGetBuilder) Build(cli rest.Interface) Requester { - // https://kubernetes.io/docs/reference/using-api/#api-groups - comps := make([]string, 0, 5) - if b.version.Group == "" { - comps = append(comps, "api", b.version.Version) - } else { - comps = append(comps, "apis", b.version.Group, b.version.Version) - } - if b.namespace != "" { - comps = append(comps, "namespaces", b.namespace) - } - comps = append(comps, b.resource, b.name) - - return &DiscardRequester{ - BaseRequester: BaseRequester{ - method: "GET", - req: cli.Get().AbsPath(comps...). - SpecificallyVersionedParams( - &metav1.GetOptions{ResourceVersion: b.resourceVersion}, - scheme.ParameterCodec, - schema.GroupVersion{Version: "v1"}, - ).MaxRetries(b.maxRetries), - }, - } -} - -type requestListBuilder struct { - version schema.GroupVersion - resource string - namespace string - limit int64 - labelSelector string - fieldSelector string - resourceVersion string - maxRetries int -} - -func newRequestListBuilder(src *types.RequestList, resourceVersion string, maxRetries int) *requestListBuilder { - return &requestListBuilder{ - version: schema.GroupVersion{ - Group: src.Group, - Version: src.Version, - }, - resource: src.Resource, - namespace: src.Namespace, - limit: int64(src.Limit), - labelSelector: src.Selector, - fieldSelector: src.FieldSelector, - resourceVersion: resourceVersion, - maxRetries: maxRetries, - } -} - -// Build implements RequestBuilder.Build. -func (b *requestListBuilder) Build(cli rest.Interface) Requester { - // https://kubernetes.io/docs/reference/using-api/#api-groups - comps := make([]string, 0, 5) - if b.version.Group == "" { - comps = append(comps, "api", b.version.Version) - } else { - comps = append(comps, "apis", b.version.Group, b.version.Version) - } - if b.namespace != "" { - comps = append(comps, "namespaces", b.namespace) - } - comps = append(comps, b.resource) - - return &DiscardRequester{ - BaseRequester: BaseRequester{ - method: "LIST", - req: cli.Get().AbsPath(comps...). - SpecificallyVersionedParams( - &metav1.ListOptions{ - LabelSelector: b.labelSelector, - FieldSelector: b.fieldSelector, - ResourceVersion: b.resourceVersion, - Limit: b.limit, - }, - scheme.ParameterCodec, - schema.GroupVersion{Version: "v1"}, - ).MaxRetries(b.maxRetries), - }, - } -} - -type requestWatchListBuilder struct { - version schema.GroupVersion - resource string - namespace string - labelSelector string - fieldSelector string - maxRetries int -} - -func newRequestWatchListBuilder(src *types.RequestWatchList, maxRetries int) *requestWatchListBuilder { - return &requestWatchListBuilder{ - version: schema.GroupVersion{ - Group: src.Group, - Version: src.Version, - }, - resource: src.Resource, - namespace: src.Namespace, - labelSelector: src.Selector, - fieldSelector: src.FieldSelector, - maxRetries: maxRetries, - } -} - -// Build implements RequestBuilder.Build. -func (b *requestWatchListBuilder) Build(cli rest.Interface) Requester { - // https://kubernetes.io/docs/reference/using-api/#api-groups - comps := make([]string, 0, 5) - if b.version.Group == "" { - comps = append(comps, "api", b.version.Version) - } else { - comps = append(comps, "apis", b.version.Group, b.version.Version) - } - if b.namespace != "" { - comps = append(comps, "namespaces", b.namespace) - } - comps = append(comps, b.resource) - - return &WatchListRequester{ - BaseRequester: BaseRequester{ - method: "WATCHLIST", - req: cli.Get().AbsPath(comps...). - SpecificallyVersionedParams( - &metav1.ListOptions{ - LabelSelector: b.labelSelector, - FieldSelector: b.fieldSelector, - ResourceVersion: "", - Watch: true, - SendInitialEvents: toPtr(true), - ResourceVersionMatch: metav1.ResourceVersionMatchNotOlderThan, - AllowWatchBookmarks: true, - }, - scheme.ParameterCodec, - schema.GroupVersion{Version: "v1"}, - ).MaxRetries(b.maxRetries), - }, - } -} - -type requestGetPodLogBuilder struct { - namespace string - name string - container string - tailLines *int64 - limitBytes *int64 - maxRetries int -} - -func newRequestGetPodLogBuilder(src *types.RequestGetPodLog, maxRetries int) *requestGetPodLogBuilder { - b := &requestGetPodLogBuilder{ - namespace: src.Namespace, - name: src.Name, - container: src.Container, - maxRetries: maxRetries, - } - if src.TailLines != nil { - b.tailLines = toPtr(*src.TailLines) - } - if src.LimitBytes != nil { - b.limitBytes = toPtr(*src.LimitBytes) - } - return b -} - -// Build implements RequestBuilder.Build. -func (b *requestGetPodLogBuilder) Build(cli rest.Interface) Requester { - // https://kubernetes.io/docs/reference/using-api/#api-groups - apiPath, version := "api", "v1" - - comps := make([]string, 2, 7) - comps[0], comps[1] = apiPath, version - comps = append(comps, "namespaces", b.namespace) - comps = append(comps, "pods", b.name, "log") - - return &DiscardRequester{ - BaseRequester: BaseRequester{ - method: "POD_LOG", - req: cli.Get().AbsPath(comps...). - SpecificallyVersionedParams( - &corev1.PodLogOptions{ - Container: b.container, - TailLines: b.tailLines, - LimitBytes: b.limitBytes, - }, - scheme.ParameterCodec, - schema.GroupVersion{Version: "v1"}, - ).MaxRetries(b.maxRetries), - }, - } -} - -// When this placeholder is present in a patch body, it is -// replaced on every request with a random string. This guarantees each patch -// actually mutates the target object. -const patchRandomPlaceholder = "{{.Random}}" - -type requestPatchBuilder struct { - version schema.GroupVersion - resource string - resourceVersion string - namespace string - name string - keySpaceSize int - patchType apitypes.PatchType - body string - maxRetries int -} - -// renderPatchBody replaces every occurrence of patchRandomPlaceholder in raw -// with a random string. -func renderPatchBody(raw string) []byte { - if !strings.Contains(raw, patchRandomPlaceholder) { - return []byte(raw) - } - return []byte(strings.ReplaceAll(raw, patchRandomPlaceholder, RandomPayload(8))) -} - -func newRequestPatchBuilder(src *types.RequestPatch, resourceVersion string, maxRetries int) *requestPatchBuilder { - patchType, _ := types.GetPatchType(src.PatchType) - - return &requestPatchBuilder{ - version: schema.GroupVersion{ - Group: src.Group, - Version: src.Version, - }, - resource: src.Resource, - resourceVersion: resourceVersion, - namespace: src.Namespace, - name: src.Name, - keySpaceSize: src.KeySpaceSize, - patchType: patchType, - body: src.Body, - maxRetries: maxRetries, - } -} - -// Build implements RequestBuilder.Build. -func (b *requestPatchBuilder) Build(cli rest.Interface) Requester { - // https://kubernetes.io/docs/reference/using-api/#api-groups - comps := make([]string, 0, 5) - if b.version.Group == "" { - comps = append(comps, "api", b.version.Version) - } else { - comps = append(comps, "apis", b.version.Group, b.version.Version) - } - if b.namespace != "" { - comps = append(comps, "namespaces", b.namespace) - } - // Generate random suffix based on keySpaceSize - randomInt, _ := rand.Int(rand.Reader, big.NewInt(int64(b.keySpaceSize))) - suffix := randomInt.Int64() - - // Create final resource name: name-{suffix} - finalName := fmt.Sprintf("%s-%d", b.name, suffix) - comps = append(comps, b.resource, finalName) - - // Render a unique body per request if patchRandomPlaceholder exists. Bodies without the - // placeholder pass through unchanged. - reqBody := renderPatchBody(b.body) - - return &DiscardRequester{ - BaseRequester: BaseRequester{ - method: "PATCH", - req: cli.Patch(b.patchType).AbsPath(comps...). - Body(reqBody). - MaxRetries(b.maxRetries), - }, - } -} - -type requestPostDelBuilder struct { - version schema.GroupVersion - resource string - resourceVersion string - namespace string - deleteRatio float64 - payloadSize int - nodeSelector map[string]string - maxRetries int - - // Per-builder cache for created resources - cache *Cache - - // Per-builder atomic counter for unique ID generation - resourceCounter int64 -} - -func newRequestPostDelBuilder(src *types.RequestPostDel, resourceVersion string, maxRetries int) *requestPostDelBuilder { - var nodeSelector map[string]string - if src.Pod != nil { - nodeSelector = src.Pod.NodeSelector - } - - return &requestPostDelBuilder{ - version: schema.GroupVersion{Group: src.Group, Version: src.Version}, - resource: src.Resource, - resourceVersion: resourceVersion, - namespace: src.Namespace, - deleteRatio: src.DeleteRatio, - payloadSize: src.PayloadSize, - nodeSelector: nodeSelector, - maxRetries: maxRetries, - cache: InitCache(), // Initialize the cache - } -} - -// Build implements RequestBuilder.Build. -func (b *requestPostDelBuilder) Build(cli rest.Interface) Requester { - comps := make([]string, 0, 5) - if b.version.Group == "" { - comps = append(comps, "api", b.version.Version) - } else { - comps = append(comps, "apis", b.version.Group, b.version.Version) - } - if b.namespace != "" { - comps = append(comps, "namespaces", b.namespace) - } - - // Random pick operation DELETE or CREATE based on deleteRatio weight probability - randomInt, _ := rand.Int(rand.Reader, big.NewInt(1000)) - shouldDelete := float64(randomInt.Int64())/1000.0 < b.deleteRatio - - if shouldDelete { - // Try to get a name from cache - if name, ok := b.cache.Pop(); ok { - comps = append(comps, b.resource, name) - - return &PostDelDiscardRequester{ - builder: b, - name: name, - operation: "DELETE", - DiscardRequester: DiscardRequester{ - BaseRequester: BaseRequester{ - method: "DELETE", - req: cli.Delete().AbsPath(comps...). - MaxRetries(b.maxRetries), - }, - }, - } - } - // If cache is empty, fall through to POST - } - - // POST logic - create resource and add to cache if successful - comps = append(comps, b.resource) - - // Use builder's atomic counter for synchronized unique ID generation - counter := atomic.AddInt64(&b.resourceCounter, 1) - timestamp := time.Now().UnixNano() - name := fmt.Sprintf("%d-%d", timestamp, counter) - - body, err := utils.RenderTemplate(b.resource, map[string]interface{}{ - "namePattern": name, - "namespace": b.namespace, - "payload": RandomPayload(b.payloadSize), - "nodeSelector": b.nodeSelector, - }) - if err != nil { - panic(fmt.Errorf("failed to render %s template: %w", b.resource, err)) - } - - return &PostDelDiscardRequester{ - builder: b, - name: name, - operation: "POST", - DiscardRequester: DiscardRequester{ - BaseRequester: BaseRequester{ - method: "POST", - req: cli.Post().AbsPath(comps...).Body(body).MaxRetries(b.maxRetries), - }, - }, - } -} - -// PostDelDiscardRequester handles both POST and DELETE requests with cache management -type PostDelDiscardRequester struct { - builder *requestPostDelBuilder - name string - operation string // "POST" or "DELETE" - DiscardRequester -} - -func (reqr *PostDelDiscardRequester) Do(ctx context.Context) (bytes int64, err error) { - // Use DiscardRequester's Do method to discard response body - bytes, err = reqr.DiscardRequester.Do(ctx) - - switch reqr.operation { - case "POST": - // Only add to cache if POST request was successful - if err == nil { - reqr.builder.cache.Push(reqr.name) - } - case "DELETE": - // If DELETE request failed, restore the item back to cache - // since the resource still exists in Kubernetes - if err != nil { - reqr.builder.cache.Push(reqr.name) - } - } - - return bytes, err -} - -func toPtr[T any](v T) *T { - return &v -} - -// requestPutBuilder builds PUT requests that overwrite an existing ConfigMap -// named `-` with a body whose `data.payload` field -// carries `payloadSize` random bytes. The target ConfigMap must already exist -// (e.g. created via `runkperf data cm add`); otherwise the apiserver returns -// 404 and the request is counted as a failure. Only `configmaps` is supported -// today — see RequestPut.Validate. -type requestPutBuilder struct { - version schema.GroupVersion - resource string - namespace string - name string - keySpaceSize int - payloadSize int - maxRetries int -} - -func newRequestPutBuilder(src *types.RequestPut, maxRetries int) *requestPutBuilder { - return &requestPutBuilder{ - version: schema.GroupVersion{ - Group: src.Group, - Version: src.Version, - }, - resource: src.Resource, - namespace: src.Namespace, - name: src.Name, - keySpaceSize: src.KeySpaceSize, - payloadSize: src.PayloadSize, - maxRetries: maxRetries, - } -} - -// Build implements RESTRequestBuilder.Build. -func (b *requestPutBuilder) Build(cli rest.Interface) Requester { - comps := []string{"api", b.version.Version, "namespaces", b.namespace, b.resource} - - randomInt, _ := rand.Int(rand.Reader, big.NewInt(int64(b.keySpaceSize))) - finalName := fmt.Sprintf("%s-%d", b.name, randomInt.Int64()) - comps = append(comps, finalName) - - cm := &corev1.ConfigMap{ - TypeMeta: metav1.TypeMeta{APIVersion: "v1", Kind: "ConfigMap"}, - ObjectMeta: metav1.ObjectMeta{ - Name: finalName, - Namespace: b.namespace, - }, - Data: map[string]string{"payload": RandomPayload(b.payloadSize)}, - } - body, _ := json.Marshal(cm) - - return &DiscardRequester{ - BaseRequester: BaseRequester{ - method: "PUT", - req: cli.Put().AbsPath(comps...).Body(body).MaxRetries(b.maxRetries), - }, - } -} - // 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. // It returns "" when n <= 0 and panics if reading random bytes fails. diff --git a/request/random_test.go b/request/random_test.go index 81e6996b..7430fc81 100644 --- a/request/random_test.go +++ b/request/random_test.go @@ -5,35 +5,11 @@ package request import ( "strconv" - "sync" "testing" - "github.com/Azure/kperf/api/types" - "github.com/stretchr/testify/assert" ) -func TestRenderPatchBody(t *testing.T) { - const tmpl = `{"metadata":{"labels":{"bench-tick":"{{.Random}}"}}}` - - // Placeholder is replaced with a random string, so the result no longer - // contains the placeholder and differs from the raw template. - rendered := string(renderPatchBody(tmpl)) - assert.NotContains(t, rendered, patchRandomPlaceholder) - assert.NotEqual(t, tmpl, rendered) - - // Each render produces a distinct body. - assert.NotEqual(t, string(renderPatchBody(tmpl)), string(renderPatchBody(tmpl))) - - // Every occurrence is replaced. - multi := string(renderPatchBody(`{"a":"{{.Random}}","b":"{{.Random}}"}`)) - assert.NotContains(t, multi, patchRandomPlaceholder) - - // A body without the placeholder is returned unchanged (backward compatible). - const static = `{"metadata":{"labels":{"bench-tick":"1"}}}` - assert.Equal(t, static, string(renderPatchBody(static))) -} - func TestRandomPayload(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) { @@ -68,61 +44,3 @@ func TestRandBytes(t *testing.T) { }) } } - -func TestRequestPatchBuilderRendersUniqueBodies(t *testing.T) { - b := newRequestPatchBuilder(&types.RequestPatch{ - KubeGroupVersionResource: types.KubeGroupVersionResource{ - Version: "v1", - Resource: "configmaps", - }, - Namespace: "bench", - Name: "cm", - KeySpaceSize: 100, - PatchType: "merge", - Body: `{"metadata":{"labels":{"bench-tick":"{{.Random}}"}}}`, - }, "", 0) - - // Each call renders a fresh random body, producing a distinct result. - first := string(renderPatchBody(b.body)) - second := string(renderPatchBody(b.body)) - assert.NotEqual(t, first, second) - assert.NotContains(t, first, patchRandomPlaceholder) - assert.NotContains(t, second, patchRandomPlaceholder) -} - -func TestRequestPatchBuilderRendersConcurrencySafe(t *testing.T) { - b := newRequestPatchBuilder(&types.RequestPatch{ - KubeGroupVersionResource: types.KubeGroupVersionResource{ - Version: "v1", - Resource: "configmaps", - }, - Name: "cm", - PatchType: "merge", - Body: `{"n":"{{.Random}}"}`, - }, "", 0) - - const goroutines = 16 - const perGoroutine = 100 - - var wg sync.WaitGroup - var mu sync.Mutex - seen := make(map[string]struct{}, goroutines*perGoroutine) - - for i := 0; i < goroutines; i++ { - wg.Add(1) - go func() { - defer wg.Done() - for j := 0; j < perGoroutine; j++ { - body := string(renderPatchBody(b.body)) - assert.NotContains(t, body, patchRandomPlaceholder) - mu.Lock() - seen[body] = struct{}{} - mu.Unlock() - } - }() - } - wg.Wait() - - // Random bodies are effectively unique across all requests. - assert.Equal(t, goroutines*perGoroutine, len(seen)) -} From 48dfc433bdc39986353c7b69f595f0ed912788f1 Mon Sep 17 00:00:00 2001 From: liyu-ma <17180634+liyu-ma@users.noreply.github.com> Date: Thu, 29 Oct 2026 13:12:16 +1000 Subject: [PATCH 2/5] Remove Pod CREATE example profile Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../manifests/loadprofile/pod_create.yaml | 28 ------------------- 1 file changed, 28 deletions(-) delete mode 100644 contrib/internal/manifests/loadprofile/pod_create.yaml diff --git a/contrib/internal/manifests/loadprofile/pod_create.yaml b/contrib/internal/manifests/loadprofile/pod_create.yaml deleted file mode 100644 index afe6ce96..00000000 --- a/contrib/internal/manifests/loadprofile/pod_create.yaml +++ /dev/null @@ -1,28 +0,0 @@ -version: 1 -description: Pod CREATE baseline - 400 creates/sec for 10 minutes -specs: - # Rate limit per runner. - - rate: 400 - duration: 600 - total: 0 - client: 400 - conns: 20 - contentType: json - maxRetries: 0 - requests: - - shares: 1 - postDel: - group: "" - version: v1 - resource: pods - namespace: pod-benchmark - deleteRatio: 0 - # Env padding for an approximately 4 KiB Pod. - payloadSize: 3650 - pod: - labels: - example.com/workload: benchmark - benchmark: pod-create - benchmark-run: replace-with-unique-run-id - serviceAccountName: pod-benchmark - schedulerName: kperf-unscheduled From 192af853c3f5539d9718681ad9746798e809db83 Mon Sep 17 00:00:00 2001 From: liyu-ma <17180634+liyu-ma@users.noreply.github.com> Date: Fri, 30 Oct 2026 10:24:56 +1000 Subject: [PATCH 3/5] Validate custom Pod labels before running load Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- api/types/load_traffic.go | 3 +++ api/types/load_traffic_test.go | 26 ++++++++++++++++++++++++++ 2 files changed, 29 insertions(+) diff --git a/api/types/load_traffic.go b/api/types/load_traffic.go index 6203c529..72584e5d 100644 --- a/api/types/load_traffic.go +++ b/api/types/load_traffic.go @@ -450,6 +450,9 @@ func (r *RequestPostDel) Validate() error { if r.Group != "" || r.Version != "v1" || r.Resource != "pods" { return fmt.Errorf("pod options only support core/v1/pods") } + if errs := metav1validation.ValidateLabels(r.Pod.Labels, field.NewPath("pod", "labels")); len(errs) > 0 { + return fmt.Errorf("invalid pod labels: %v", errs.ToAggregate()) + } if len(r.Pod.NodeSelector) > 0 { errs := metav1validation.ValidateLabels(r.Pod.NodeSelector, field.NewPath("pod", "nodeSelector")) if len(errs) > 0 { diff --git a/api/types/load_traffic_test.go b/api/types/load_traffic_test.go index fface1fa..490f1ff5 100644 --- a/api/types/load_traffic_test.go +++ b/api/types/load_traffic_test.go @@ -315,6 +315,32 @@ func TestRequestPostDelNodeSelectorValidation(t *testing.T) { } } +func TestRequestPostDelLabelsValidation(t *testing.T) { + for _, tt := range []struct { + name string + labels map[string]string + wantErr bool + }{ + {name: "omitted"}, + {name: "valid", labels: map[string]string{"example.com/workload": "benchmark", "app": ""}}, + {name: "invalid key", labels: map[string]string{"invalid key": "benchmark"}, wantErr: true}, + {name: "invalid value", labels: map[string]string{"app": "invalid value"}, wantErr: true}, + } { + t.Run(tt.name, func(t *testing.T) { + request := RequestPostDel{ + KubeGroupVersionResource: KubeGroupVersionResource{Version: "v1", Resource: "pods"}, + Pod: &PostDelPodOptions{Labels: tt.labels}, + } + err := request.Validate() + if tt.wantErr { + require.ErrorContains(t, err, "invalid pod labels") + } else { + require.NoError(t, err) + } + }) + } +} + func TestRequestPostDelNodeSelectorUnmarshalFromYAML(t *testing.T) { input := ` version: v1 From 085a5ee54408edc041348bb7ca1e019e174c9bdc Mon Sep 17 00:00:00 2001 From: liyu-ma <17180634+liyu-ma@users.noreply.github.com> Date: Fri, 30 Oct 2026 14:19:12 +1000 Subject: [PATCH 4/5] Simplify Pod app label fallback Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../manifests/workload/pods/templates/pod.tpl | 10 +++------- request/builder_test.go | 17 ++++++++++++----- 2 files changed, 15 insertions(+), 12 deletions(-) diff --git a/contrib/internal/manifests/workload/pods/templates/pod.tpl b/contrib/internal/manifests/workload/pods/templates/pod.tpl index 74ed4fdc..7137f9a0 100644 --- a/contrib/internal/manifests/workload/pods/templates/pod.tpl +++ b/contrib/internal/manifests/workload/pods/templates/pod.tpl @@ -2,22 +2,18 @@ {{- $namespace:= .Values.namespace }} {{- $payload:= .Values.payload }} {{- $nodeSelector:= .Values.nodeSelector }} -{{- $hasApp := false }} -{{- range $key, $value := .Values.labels }} -{{- if eq $key "app" }}{{ $hasApp = true }}{{ end }} -{{- end }} apiVersion: v1 kind: Pod metadata: name: {{ printf "%q" $name }} namespace: {{ printf "%q" $namespace }} labels: - {{- if not $hasApp }} - app: fake-pod - {{- end }} {{- range $key, $value := .Values.labels }} {{ printf "%q" $key }}: {{ printf "%q" $value }} {{- end }} + {{- if eq .Values.labels.app nil }} + app: fake-pod + {{- end }} spec: {{- if .Values.serviceAccountName }} serviceAccountName: {{ printf "%q" .Values.serviceAccountName }} diff --git a/request/builder_test.go b/request/builder_test.go index 539601ab..4573b235 100644 --- a/request/builder_test.go +++ b/request/builder_test.go @@ -105,12 +105,19 @@ func TestPodOptions(t *testing.T) { for _, tt := range []struct { name string options *types.PostDelPodOptions + labels map[string]string }{ - {name: "defaults"}, + {name: "defaults", labels: map[string]string{"app": "fake-pod"}}, + {name: "empty app", options: &types.PostDelPodOptions{ + Labels: map[string]string{"app": ""}, + }, labels: map[string]string{"app": ""}}, + {name: "other labels", options: &types.PostDelPodOptions{ + Labels: map[string]string{"team": "benchmark"}, + }, labels: map[string]string{"app": "fake-pod", "team": "benchmark"}}, {name: "custom", options: &types.PostDelPodOptions{ - Labels: map[string]string{"app": ""}, NodeSelector: map[string]string{"agentpool": "benchmark"}, + Labels: map[string]string{"app": "custom-app", "team": "benchmark"}, NodeSelector: map[string]string{"agentpool": "benchmark"}, ServiceAccountName: "custom-account", SchedulerName: "custom-scheduler", - }}, + }, labels: map[string]string{"app": "custom-app", "team": "benchmark"}}, } { t.Run(tt.name, func(t *testing.T) { builder := newRequestPostDelBuilder(&types.RequestPostDel{ @@ -136,9 +143,9 @@ func TestPodOptions(t *testing.T) { require.Equal(t, "default", pod.Namespace) options := tt.options if options == nil { - options = &types.PostDelPodOptions{Labels: map[string]string{"app": "fake-pod"}} + options = &types.PostDelPodOptions{} } - require.Equal(t, options.Labels, pod.Labels) + require.Equal(t, tt.labels, pod.Labels) require.Equal(t, options.NodeSelector, pod.Spec.NodeSelector) require.Equal(t, options.ServiceAccountName, pod.Spec.ServiceAccountName) require.Equal(t, options.SchedulerName, pod.Spec.SchedulerName) From 38567a6b70337e7efa3d287988ab36cfe47f9d04 Mon Sep 17 00:00:00 2001 From: liyu-ma <17180634+liyu-ma@users.noreply.github.com> Date: Fri, 30 Oct 2026 22:52:38 +1000 Subject: [PATCH 5/5] Address review comment Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../manifests/workload/pods/templates/pod.tpl | 12 +- request/builder.go | 530 ------------------ request/builder_test.go | 154 ----- request/random.go | 516 +++++++++++++++++ request/random_test.go | 147 +++++ 5 files changed, 669 insertions(+), 690 deletions(-) delete mode 100644 request/builder.go delete mode 100644 request/builder_test.go diff --git a/contrib/internal/manifests/workload/pods/templates/pod.tpl b/contrib/internal/manifests/workload/pods/templates/pod.tpl index 7137f9a0..08cb7701 100644 --- a/contrib/internal/manifests/workload/pods/templates/pod.tpl +++ b/contrib/internal/manifests/workload/pods/templates/pod.tpl @@ -5,26 +5,26 @@ apiVersion: v1 kind: Pod metadata: - name: {{ printf "%q" $name }} - namespace: {{ printf "%q" $namespace }} + name: {{ $name }} + namespace: {{ $namespace }} labels: {{- range $key, $value := .Values.labels }} - {{ printf "%q" $key }}: {{ printf "%q" $value }} + "{{ $key }}": "{{ $value }}" {{- end }} {{- if eq .Values.labels.app nil }} app: fake-pod {{- end }} spec: {{- if .Values.serviceAccountName }} - serviceAccountName: {{ printf "%q" .Values.serviceAccountName }} + serviceAccountName: "{{ .Values.serviceAccountName }}" {{- end }} {{- if .Values.schedulerName }} - schedulerName: {{ printf "%q" .Values.schedulerName }} + schedulerName: "{{ .Values.schedulerName }}" {{- end }} {{- if $nodeSelector }} nodeSelector: {{- range $key, $value := $nodeSelector }} - {{ printf "%q" $key }}: {{ printf "%q" $value }} + "{{ $key }}": "{{ $value }}" {{- end }} {{- end }} # Use pause image so pods reach Running with few events, avoiding an diff --git a/request/builder.go b/request/builder.go deleted file mode 100644 index e5a9d889..00000000 --- a/request/builder.go +++ /dev/null @@ -1,530 +0,0 @@ -// Copyright (c) Microsoft Corporation. -// Licensed under the MIT License. - -package request - -import ( - "context" - "crypto/rand" - "encoding/json" - "fmt" - "maps" - "math/big" - "strings" - "sync/atomic" - "time" - - "github.com/Azure/kperf/api/types" - "github.com/Azure/kperf/contrib/utils" - - corev1 "k8s.io/api/core/v1" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/runtime/schema" - apitypes "k8s.io/apimachinery/pkg/types" - "k8s.io/client-go/kubernetes/scheme" - "k8s.io/client-go/rest" -) - -// RESTRequestBuilder is used to build rest.Request. -type RESTRequestBuilder interface { - Build(cli rest.Interface) Requester -} - -type requestGetBuilder struct { - version schema.GroupVersion - resource string - namespace string - name string - resourceVersion string - maxRetries int -} - -func newRequestGetBuilder(src *types.RequestGet, resourceVersion string, maxRetries int) *requestGetBuilder { - return &requestGetBuilder{ - version: schema.GroupVersion{ - Group: src.Group, - Version: src.Version, - }, - resource: src.Resource, - namespace: src.Namespace, - name: src.Name, - resourceVersion: resourceVersion, - maxRetries: maxRetries, - } -} - -// Build implements RequestBuilder.Build. -func (b *requestGetBuilder) Build(cli rest.Interface) Requester { - // https://kubernetes.io/docs/reference/using-api/#api-groups - comps := make([]string, 0, 5) - if b.version.Group == "" { - comps = append(comps, "api", b.version.Version) - } else { - comps = append(comps, "apis", b.version.Group, b.version.Version) - } - if b.namespace != "" { - comps = append(comps, "namespaces", b.namespace) - } - comps = append(comps, b.resource, b.name) - - return &DiscardRequester{ - BaseRequester: BaseRequester{ - method: "GET", - req: cli.Get().AbsPath(comps...). - SpecificallyVersionedParams( - &metav1.GetOptions{ResourceVersion: b.resourceVersion}, - scheme.ParameterCodec, - schema.GroupVersion{Version: "v1"}, - ).MaxRetries(b.maxRetries), - }, - } -} - -type requestListBuilder struct { - version schema.GroupVersion - resource string - namespace string - limit int64 - labelSelector string - fieldSelector string - resourceVersion string - maxRetries int -} - -func newRequestListBuilder(src *types.RequestList, resourceVersion string, maxRetries int) *requestListBuilder { - return &requestListBuilder{ - version: schema.GroupVersion{ - Group: src.Group, - Version: src.Version, - }, - resource: src.Resource, - namespace: src.Namespace, - limit: int64(src.Limit), - labelSelector: src.Selector, - fieldSelector: src.FieldSelector, - resourceVersion: resourceVersion, - maxRetries: maxRetries, - } -} - -// Build implements RequestBuilder.Build. -func (b *requestListBuilder) Build(cli rest.Interface) Requester { - // https://kubernetes.io/docs/reference/using-api/#api-groups - comps := make([]string, 0, 5) - if b.version.Group == "" { - comps = append(comps, "api", b.version.Version) - } else { - comps = append(comps, "apis", b.version.Group, b.version.Version) - } - if b.namespace != "" { - comps = append(comps, "namespaces", b.namespace) - } - comps = append(comps, b.resource) - - return &DiscardRequester{ - BaseRequester: BaseRequester{ - method: "LIST", - req: cli.Get().AbsPath(comps...). - SpecificallyVersionedParams( - &metav1.ListOptions{ - LabelSelector: b.labelSelector, - FieldSelector: b.fieldSelector, - ResourceVersion: b.resourceVersion, - Limit: b.limit, - }, - scheme.ParameterCodec, - schema.GroupVersion{Version: "v1"}, - ).MaxRetries(b.maxRetries), - }, - } -} - -type requestWatchListBuilder struct { - version schema.GroupVersion - resource string - namespace string - labelSelector string - fieldSelector string - maxRetries int -} - -func newRequestWatchListBuilder(src *types.RequestWatchList, maxRetries int) *requestWatchListBuilder { - return &requestWatchListBuilder{ - version: schema.GroupVersion{ - Group: src.Group, - Version: src.Version, - }, - resource: src.Resource, - namespace: src.Namespace, - labelSelector: src.Selector, - fieldSelector: src.FieldSelector, - maxRetries: maxRetries, - } -} - -// Build implements RequestBuilder.Build. -func (b *requestWatchListBuilder) Build(cli rest.Interface) Requester { - // https://kubernetes.io/docs/reference/using-api/#api-groups - comps := make([]string, 0, 5) - if b.version.Group == "" { - comps = append(comps, "api", b.version.Version) - } else { - comps = append(comps, "apis", b.version.Group, b.version.Version) - } - if b.namespace != "" { - comps = append(comps, "namespaces", b.namespace) - } - comps = append(comps, b.resource) - - return &WatchListRequester{ - BaseRequester: BaseRequester{ - method: "WATCHLIST", - req: cli.Get().AbsPath(comps...). - SpecificallyVersionedParams( - &metav1.ListOptions{ - LabelSelector: b.labelSelector, - FieldSelector: b.fieldSelector, - ResourceVersion: "", - Watch: true, - SendInitialEvents: toPtr(true), - ResourceVersionMatch: metav1.ResourceVersionMatchNotOlderThan, - AllowWatchBookmarks: true, - }, - scheme.ParameterCodec, - schema.GroupVersion{Version: "v1"}, - ).MaxRetries(b.maxRetries), - }, - } -} - -type requestGetPodLogBuilder struct { - namespace string - name string - container string - tailLines *int64 - limitBytes *int64 - maxRetries int -} - -func newRequestGetPodLogBuilder(src *types.RequestGetPodLog, maxRetries int) *requestGetPodLogBuilder { - b := &requestGetPodLogBuilder{ - namespace: src.Namespace, - name: src.Name, - container: src.Container, - maxRetries: maxRetries, - } - if src.TailLines != nil { - b.tailLines = toPtr(*src.TailLines) - } - if src.LimitBytes != nil { - b.limitBytes = toPtr(*src.LimitBytes) - } - return b -} - -// Build implements RequestBuilder.Build. -func (b *requestGetPodLogBuilder) Build(cli rest.Interface) Requester { - // https://kubernetes.io/docs/reference/using-api/#api-groups - apiPath, version := "api", "v1" - - comps := make([]string, 2, 7) - comps[0], comps[1] = apiPath, version - comps = append(comps, "namespaces", b.namespace) - comps = append(comps, "pods", b.name, "log") - - return &DiscardRequester{ - BaseRequester: BaseRequester{ - method: "POD_LOG", - req: cli.Get().AbsPath(comps...). - SpecificallyVersionedParams( - &corev1.PodLogOptions{ - Container: b.container, - TailLines: b.tailLines, - LimitBytes: b.limitBytes, - }, - scheme.ParameterCodec, - schema.GroupVersion{Version: "v1"}, - ).MaxRetries(b.maxRetries), - }, - } -} - -// When this placeholder is present in a patch body, it is -// replaced on every request with a random string. This guarantees each patch -// actually mutates the target object. -const patchRandomPlaceholder = "{{.Random}}" - -type requestPatchBuilder struct { - version schema.GroupVersion - resource string - resourceVersion string - namespace string - name string - keySpaceSize int - patchType apitypes.PatchType - body string - maxRetries int -} - -// renderPatchBody replaces every occurrence of patchRandomPlaceholder in raw -// with a random string. -func renderPatchBody(raw string) []byte { - if !strings.Contains(raw, patchRandomPlaceholder) { - return []byte(raw) - } - return []byte(strings.ReplaceAll(raw, patchRandomPlaceholder, RandomPayload(8))) -} - -func newRequestPatchBuilder(src *types.RequestPatch, resourceVersion string, maxRetries int) *requestPatchBuilder { - patchType, _ := types.GetPatchType(src.PatchType) - - return &requestPatchBuilder{ - version: schema.GroupVersion{ - Group: src.Group, - Version: src.Version, - }, - resource: src.Resource, - resourceVersion: resourceVersion, - namespace: src.Namespace, - name: src.Name, - keySpaceSize: src.KeySpaceSize, - patchType: patchType, - body: src.Body, - maxRetries: maxRetries, - } -} - -// Build implements RequestBuilder.Build. -func (b *requestPatchBuilder) Build(cli rest.Interface) Requester { - // https://kubernetes.io/docs/reference/using-api/#api-groups - comps := make([]string, 0, 5) - if b.version.Group == "" { - comps = append(comps, "api", b.version.Version) - } else { - comps = append(comps, "apis", b.version.Group, b.version.Version) - } - if b.namespace != "" { - comps = append(comps, "namespaces", b.namespace) - } - // Generate random suffix based on keySpaceSize - randomInt, _ := rand.Int(rand.Reader, big.NewInt(int64(b.keySpaceSize))) - suffix := randomInt.Int64() - - // Create final resource name: name-{suffix} - finalName := fmt.Sprintf("%s-%d", b.name, suffix) - comps = append(comps, b.resource, finalName) - - // Render a unique body per request if patchRandomPlaceholder exists. Bodies without the - // placeholder pass through unchanged. - reqBody := renderPatchBody(b.body) - - return &DiscardRequester{ - BaseRequester: BaseRequester{ - method: "PATCH", - req: cli.Patch(b.patchType).AbsPath(comps...). - Body(reqBody). - MaxRetries(b.maxRetries), - }, - } -} - -type requestPostDelBuilder struct { - version schema.GroupVersion - resource string - resourceVersion string - namespace string - deleteRatio float64 - payloadSize int - nodeSelector map[string]string - labels map[string]string - serviceAccountName string - schedulerName string - maxRetries int - - // Per-builder cache for created resources - cache *Cache - - // Per-builder atomic counter for unique ID generation - resourceCounter int64 -} - -func newRequestPostDelBuilder(src *types.RequestPostDel, resourceVersion string, maxRetries int) *requestPostDelBuilder { - b := &requestPostDelBuilder{ - version: schema.GroupVersion{Group: src.Group, Version: src.Version}, - resource: src.Resource, - resourceVersion: resourceVersion, - namespace: src.Namespace, - deleteRatio: src.DeleteRatio, - payloadSize: src.PayloadSize, - maxRetries: maxRetries, - cache: InitCache(), // Initialize the cache - } - if p := src.Pod; p != nil { - b.nodeSelector = maps.Clone(p.NodeSelector) - b.labels = maps.Clone(p.Labels) - b.serviceAccountName = p.ServiceAccountName - b.schedulerName = p.SchedulerName - } - return b -} - -// Build implements RequestBuilder.Build. -func (b *requestPostDelBuilder) Build(cli rest.Interface) Requester { - comps := make([]string, 0, 5) - if b.version.Group == "" { - comps = append(comps, "api", b.version.Version) - } else { - comps = append(comps, "apis", b.version.Group, b.version.Version) - } - if b.namespace != "" { - comps = append(comps, "namespaces", b.namespace) - } - - // Random pick operation DELETE or CREATE based on deleteRatio weight probability - randomInt, _ := rand.Int(rand.Reader, big.NewInt(1000)) - shouldDelete := float64(randomInt.Int64())/1000.0 < b.deleteRatio - - if shouldDelete { - // Try to get a name from cache - if name, ok := b.cache.Pop(); ok { - comps = append(comps, b.resource, name) - - return &PostDelDiscardRequester{ - builder: b, - name: name, - operation: "DELETE", - DiscardRequester: DiscardRequester{ - BaseRequester: BaseRequester{ - method: "DELETE", - req: cli.Delete().AbsPath(comps...). - MaxRetries(b.maxRetries), - }, - }, - } - } - // If cache is empty, fall through to POST - } - - // POST logic - create resource and add to cache if successful - comps = append(comps, b.resource) - - // Use builder's atomic counter for synchronized unique ID generation - counter := atomic.AddInt64(&b.resourceCounter, 1) - timestamp := time.Now().UnixNano() - name := fmt.Sprintf("%d-%d", timestamp, counter) - - body, err := utils.RenderTemplate(b.resource, map[string]interface{}{ - "namePattern": name, - "namespace": b.namespace, - "payload": RandomPayload(b.payloadSize), - "nodeSelector": b.nodeSelector, - "labels": b.labels, - "serviceAccountName": b.serviceAccountName, - "schedulerName": b.schedulerName, - }) - if err != nil { - panic(fmt.Errorf("failed to render %s template: %w", b.resource, err)) - } - - return &PostDelDiscardRequester{ - builder: b, - name: name, - operation: "POST", - DiscardRequester: DiscardRequester{ - BaseRequester: BaseRequester{ - method: "POST", - req: cli.Post().AbsPath(comps...).Body(body).MaxRetries(b.maxRetries), - }, - }, - } -} - -// PostDelDiscardRequester handles both POST and DELETE requests with cache management -type PostDelDiscardRequester struct { - builder *requestPostDelBuilder - name string - operation string // "POST" or "DELETE" - DiscardRequester -} - -func (reqr *PostDelDiscardRequester) Do(ctx context.Context) (bytes int64, err error) { - // Use DiscardRequester's Do method to discard response body - bytes, err = reqr.DiscardRequester.Do(ctx) - - switch reqr.operation { - case "POST": - // Create-only runs retain objects, but do not need a deletion-name cache. - if err == nil && reqr.builder.deleteRatio > 0 { - reqr.builder.cache.Push(reqr.name) - } - case "DELETE": - // If DELETE request failed, restore the item back to cache - // since the resource still exists in Kubernetes - if err != nil { - reqr.builder.cache.Push(reqr.name) - } - } - - return bytes, err -} - -func toPtr[T any](v T) *T { - return &v -} - -// requestPutBuilder builds PUT requests that overwrite an existing ConfigMap -// named `-` with a body whose `data.payload` field -// carries `payloadSize` random bytes. The target ConfigMap must already exist -// (e.g. created via `runkperf data cm add`); otherwise the apiserver returns -// 404 and the request is counted as a failure. Only `configmaps` is supported -// today — see RequestPut.Validate. -type requestPutBuilder struct { - version schema.GroupVersion - resource string - namespace string - name string - keySpaceSize int - payloadSize int - maxRetries int -} - -func newRequestPutBuilder(src *types.RequestPut, maxRetries int) *requestPutBuilder { - return &requestPutBuilder{ - version: schema.GroupVersion{ - Group: src.Group, - Version: src.Version, - }, - resource: src.Resource, - namespace: src.Namespace, - name: src.Name, - keySpaceSize: src.KeySpaceSize, - payloadSize: src.PayloadSize, - maxRetries: maxRetries, - } -} - -// Build implements RESTRequestBuilder.Build. -func (b *requestPutBuilder) Build(cli rest.Interface) Requester { - comps := []string{"api", b.version.Version, "namespaces", b.namespace, b.resource} - - randomInt, _ := rand.Int(rand.Reader, big.NewInt(int64(b.keySpaceSize))) - finalName := fmt.Sprintf("%s-%d", b.name, randomInt.Int64()) - comps = append(comps, finalName) - - cm := &corev1.ConfigMap{ - TypeMeta: metav1.TypeMeta{APIVersion: "v1", Kind: "ConfigMap"}, - ObjectMeta: metav1.ObjectMeta{ - Name: finalName, - Namespace: b.namespace, - }, - Data: map[string]string{"payload": RandomPayload(b.payloadSize)}, - } - body, _ := json.Marshal(cm) - - return &DiscardRequester{ - BaseRequester: BaseRequester{ - method: "PUT", - req: cli.Put().AbsPath(comps...).Body(body).MaxRetries(b.maxRetries), - }, - } -} diff --git a/request/builder_test.go b/request/builder_test.go deleted file mode 100644 index 4573b235..00000000 --- a/request/builder_test.go +++ /dev/null @@ -1,154 +0,0 @@ -// Copyright (c) Microsoft Corporation. -// Licensed under the MIT License. - -package request - -import ( - "context" - "encoding/json" - "io" - "net/http" - "strings" - "sync" - "testing" - - "github.com/Azure/kperf/api/types" - - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" - corev1 "k8s.io/api/core/v1" - "k8s.io/apimachinery/pkg/runtime/schema" - "k8s.io/client-go/kubernetes/scheme" - "k8s.io/client-go/rest/fake" -) - -func TestRenderPatchBody(t *testing.T) { - const tmpl = `{"metadata":{"labels":{"bench-tick":"{{.Random}}"}}}` - - // Placeholder is replaced with a random string, so the result no longer - // contains the placeholder and differs from the raw template. - rendered := string(renderPatchBody(tmpl)) - assert.NotContains(t, rendered, patchRandomPlaceholder) - assert.NotEqual(t, tmpl, rendered) - - // Each render produces a distinct body. - assert.NotEqual(t, string(renderPatchBody(tmpl)), string(renderPatchBody(tmpl))) - - // Every occurrence is replaced. - multi := string(renderPatchBody(`{"a":"{{.Random}}","b":"{{.Random}}"}`)) - assert.NotContains(t, multi, patchRandomPlaceholder) - - // A body without the placeholder is returned unchanged (backward compatible). - const static = `{"metadata":{"labels":{"bench-tick":"1"}}}` - assert.Equal(t, static, string(renderPatchBody(static))) -} - -func TestRequestPatchBuilderRendersUniqueBodies(t *testing.T) { - b := newRequestPatchBuilder(&types.RequestPatch{ - KubeGroupVersionResource: types.KubeGroupVersionResource{ - Version: "v1", - Resource: "configmaps", - }, - Namespace: "bench", - Name: "cm", - KeySpaceSize: 100, - PatchType: "merge", - Body: `{"metadata":{"labels":{"bench-tick":"{{.Random}}"}}}`, - }, "", 0) - - // Each call renders a fresh random body, producing a distinct result. - first := string(renderPatchBody(b.body)) - second := string(renderPatchBody(b.body)) - assert.NotEqual(t, first, second) - assert.NotContains(t, first, patchRandomPlaceholder) - assert.NotContains(t, second, patchRandomPlaceholder) -} - -func TestRequestPatchBuilderRendersConcurrencySafe(t *testing.T) { - b := newRequestPatchBuilder(&types.RequestPatch{ - KubeGroupVersionResource: types.KubeGroupVersionResource{ - Version: "v1", - Resource: "configmaps", - }, - Name: "cm", - PatchType: "merge", - Body: `{"n":"{{.Random}}"}`, - }, "", 0) - - const goroutines = 16 - const perGoroutine = 100 - - var wg sync.WaitGroup - var mu sync.Mutex - seen := make(map[string]struct{}, goroutines*perGoroutine) - - for i := 0; i < goroutines; i++ { - wg.Add(1) - go func() { - defer wg.Done() - for j := 0; j < perGoroutine; j++ { - body := string(renderPatchBody(b.body)) - assert.NotContains(t, body, patchRandomPlaceholder) - mu.Lock() - seen[body] = struct{}{} - mu.Unlock() - } - }() - } - wg.Wait() - - // Random bodies are effectively unique across all requests. - assert.Equal(t, goroutines*perGoroutine, len(seen)) -} - -func TestPodOptions(t *testing.T) { - for _, tt := range []struct { - name string - options *types.PostDelPodOptions - labels map[string]string - }{ - {name: "defaults", labels: map[string]string{"app": "fake-pod"}}, - {name: "empty app", options: &types.PostDelPodOptions{ - Labels: map[string]string{"app": ""}, - }, labels: map[string]string{"app": ""}}, - {name: "other labels", options: &types.PostDelPodOptions{ - Labels: map[string]string{"team": "benchmark"}, - }, labels: map[string]string{"app": "fake-pod", "team": "benchmark"}}, - {name: "custom", options: &types.PostDelPodOptions{ - Labels: map[string]string{"app": "custom-app", "team": "benchmark"}, NodeSelector: map[string]string{"agentpool": "benchmark"}, - ServiceAccountName: "custom-account", SchedulerName: "custom-scheduler", - }, labels: map[string]string{"app": "custom-app", "team": "benchmark"}}, - } { - t.Run(tt.name, func(t *testing.T) { - builder := newRequestPostDelBuilder(&types.RequestPostDel{ - KubeGroupVersionResource: types.KubeGroupVersionResource{Version: "v1", Resource: "pods"}, - Namespace: "default", - Pod: tt.options, - }, "", 0) - var pod corev1.Pod - cli := &fake.RESTClient{ - GroupVersion: schema.GroupVersion{Version: "v1"}, - NegotiatedSerializer: scheme.Codecs.WithoutConversion(), - Client: fake.CreateHTTPClient(func(r *http.Request) (*http.Response, error) { - require.Equal(t, "POST", r.Method) - require.NoError(t, json.NewDecoder(r.Body).Decode(&pod)) - return &http.Response{ - StatusCode: http.StatusCreated, Header: http.Header{"Content-Type": {"application/json"}}, - Body: io.NopCloser(strings.NewReader("{}")), - }, nil - }), - } - _, err := builder.Build(cli).Do(context.Background()) - require.NoError(t, err) - require.Equal(t, "default", pod.Namespace) - options := tt.options - if options == nil { - options = &types.PostDelPodOptions{} - } - require.Equal(t, tt.labels, pod.Labels) - require.Equal(t, options.NodeSelector, pod.Spec.NodeSelector) - require.Equal(t, options.ServiceAccountName, pod.Spec.ServiceAccountName) - require.Equal(t, options.SchedulerName, pod.Spec.SchedulerName) - }) - } -} diff --git a/request/random.go b/request/random.go index 30a593c1..d2ce6c86 100644 --- a/request/random.go +++ b/request/random.go @@ -6,11 +6,23 @@ package request import ( "context" "crypto/rand" + "encoding/json" "fmt" "math/big" + "strings" "sync" + "sync/atomic" + "time" "github.com/Azure/kperf/api/types" + "github.com/Azure/kperf/contrib/utils" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime/schema" + apitypes "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/kubernetes/scheme" + "k8s.io/client-go/rest" ) // WeightedRandomRequests is used to generate requests based on LoadProfileSpec. @@ -127,6 +139,510 @@ func (r *WeightedRandomRequests) Stop() { }) } +// RESTRequestBuilder is used to build rest.Request. +type RESTRequestBuilder interface { + Build(cli rest.Interface) Requester +} + +type requestGetBuilder struct { + version schema.GroupVersion + resource string + namespace string + name string + resourceVersion string + maxRetries int +} + +func newRequestGetBuilder(src *types.RequestGet, resourceVersion string, maxRetries int) *requestGetBuilder { + return &requestGetBuilder{ + version: schema.GroupVersion{ + Group: src.Group, + Version: src.Version, + }, + resource: src.Resource, + namespace: src.Namespace, + name: src.Name, + resourceVersion: resourceVersion, + maxRetries: maxRetries, + } +} + +// Build implements RequestBuilder.Build. +func (b *requestGetBuilder) Build(cli rest.Interface) Requester { + // https://kubernetes.io/docs/reference/using-api/#api-groups + comps := make([]string, 0, 5) + if b.version.Group == "" { + comps = append(comps, "api", b.version.Version) + } else { + comps = append(comps, "apis", b.version.Group, b.version.Version) + } + if b.namespace != "" { + comps = append(comps, "namespaces", b.namespace) + } + comps = append(comps, b.resource, b.name) + + return &DiscardRequester{ + BaseRequester: BaseRequester{ + method: "GET", + req: cli.Get().AbsPath(comps...). + SpecificallyVersionedParams( + &metav1.GetOptions{ResourceVersion: b.resourceVersion}, + scheme.ParameterCodec, + schema.GroupVersion{Version: "v1"}, + ).MaxRetries(b.maxRetries), + }, + } +} + +type requestListBuilder struct { + version schema.GroupVersion + resource string + namespace string + limit int64 + labelSelector string + fieldSelector string + resourceVersion string + maxRetries int +} + +func newRequestListBuilder(src *types.RequestList, resourceVersion string, maxRetries int) *requestListBuilder { + return &requestListBuilder{ + version: schema.GroupVersion{ + Group: src.Group, + Version: src.Version, + }, + resource: src.Resource, + namespace: src.Namespace, + limit: int64(src.Limit), + labelSelector: src.Selector, + fieldSelector: src.FieldSelector, + resourceVersion: resourceVersion, + maxRetries: maxRetries, + } +} + +// Build implements RequestBuilder.Build. +func (b *requestListBuilder) Build(cli rest.Interface) Requester { + // https://kubernetes.io/docs/reference/using-api/#api-groups + comps := make([]string, 0, 5) + if b.version.Group == "" { + comps = append(comps, "api", b.version.Version) + } else { + comps = append(comps, "apis", b.version.Group, b.version.Version) + } + if b.namespace != "" { + comps = append(comps, "namespaces", b.namespace) + } + comps = append(comps, b.resource) + + return &DiscardRequester{ + BaseRequester: BaseRequester{ + method: "LIST", + req: cli.Get().AbsPath(comps...). + SpecificallyVersionedParams( + &metav1.ListOptions{ + LabelSelector: b.labelSelector, + FieldSelector: b.fieldSelector, + ResourceVersion: b.resourceVersion, + Limit: b.limit, + }, + scheme.ParameterCodec, + schema.GroupVersion{Version: "v1"}, + ).MaxRetries(b.maxRetries), + }, + } +} + +type requestWatchListBuilder struct { + version schema.GroupVersion + resource string + namespace string + labelSelector string + fieldSelector string + maxRetries int +} + +func newRequestWatchListBuilder(src *types.RequestWatchList, maxRetries int) *requestWatchListBuilder { + return &requestWatchListBuilder{ + version: schema.GroupVersion{ + Group: src.Group, + Version: src.Version, + }, + resource: src.Resource, + namespace: src.Namespace, + labelSelector: src.Selector, + fieldSelector: src.FieldSelector, + maxRetries: maxRetries, + } +} + +// Build implements RequestBuilder.Build. +func (b *requestWatchListBuilder) Build(cli rest.Interface) Requester { + // https://kubernetes.io/docs/reference/using-api/#api-groups + comps := make([]string, 0, 5) + if b.version.Group == "" { + comps = append(comps, "api", b.version.Version) + } else { + comps = append(comps, "apis", b.version.Group, b.version.Version) + } + if b.namespace != "" { + comps = append(comps, "namespaces", b.namespace) + } + comps = append(comps, b.resource) + + return &WatchListRequester{ + BaseRequester: BaseRequester{ + method: "WATCHLIST", + req: cli.Get().AbsPath(comps...). + SpecificallyVersionedParams( + &metav1.ListOptions{ + LabelSelector: b.labelSelector, + FieldSelector: b.fieldSelector, + ResourceVersion: "", + Watch: true, + SendInitialEvents: toPtr(true), + ResourceVersionMatch: metav1.ResourceVersionMatchNotOlderThan, + AllowWatchBookmarks: true, + }, + scheme.ParameterCodec, + schema.GroupVersion{Version: "v1"}, + ).MaxRetries(b.maxRetries), + }, + } +} + +type requestGetPodLogBuilder struct { + namespace string + name string + container string + tailLines *int64 + limitBytes *int64 + maxRetries int +} + +func newRequestGetPodLogBuilder(src *types.RequestGetPodLog, maxRetries int) *requestGetPodLogBuilder { + b := &requestGetPodLogBuilder{ + namespace: src.Namespace, + name: src.Name, + container: src.Container, + maxRetries: maxRetries, + } + if src.TailLines != nil { + b.tailLines = toPtr(*src.TailLines) + } + if src.LimitBytes != nil { + b.limitBytes = toPtr(*src.LimitBytes) + } + return b +} + +// Build implements RequestBuilder.Build. +func (b *requestGetPodLogBuilder) Build(cli rest.Interface) Requester { + // https://kubernetes.io/docs/reference/using-api/#api-groups + apiPath, version := "api", "v1" + + comps := make([]string, 2, 7) + comps[0], comps[1] = apiPath, version + comps = append(comps, "namespaces", b.namespace) + comps = append(comps, "pods", b.name, "log") + + return &DiscardRequester{ + BaseRequester: BaseRequester{ + method: "POD_LOG", + req: cli.Get().AbsPath(comps...). + SpecificallyVersionedParams( + &corev1.PodLogOptions{ + Container: b.container, + TailLines: b.tailLines, + LimitBytes: b.limitBytes, + }, + scheme.ParameterCodec, + schema.GroupVersion{Version: "v1"}, + ).MaxRetries(b.maxRetries), + }, + } +} + +// When this placeholder is present in a patch body, it is +// replaced on every request with a random string. This guarantees each patch +// actually mutates the target object. +const patchRandomPlaceholder = "{{.Random}}" + +type requestPatchBuilder struct { + version schema.GroupVersion + resource string + resourceVersion string + namespace string + name string + keySpaceSize int + patchType apitypes.PatchType + body string + maxRetries int +} + +// renderPatchBody replaces every occurrence of patchRandomPlaceholder in raw +// with a random string. +func renderPatchBody(raw string) []byte { + if !strings.Contains(raw, patchRandomPlaceholder) { + return []byte(raw) + } + return []byte(strings.ReplaceAll(raw, patchRandomPlaceholder, RandomPayload(8))) +} + +func newRequestPatchBuilder(src *types.RequestPatch, resourceVersion string, maxRetries int) *requestPatchBuilder { + patchType, _ := types.GetPatchType(src.PatchType) + + return &requestPatchBuilder{ + version: schema.GroupVersion{ + Group: src.Group, + Version: src.Version, + }, + resource: src.Resource, + resourceVersion: resourceVersion, + namespace: src.Namespace, + name: src.Name, + keySpaceSize: src.KeySpaceSize, + patchType: patchType, + body: src.Body, + maxRetries: maxRetries, + } +} + +// Build implements RequestBuilder.Build. +func (b *requestPatchBuilder) Build(cli rest.Interface) Requester { + // https://kubernetes.io/docs/reference/using-api/#api-groups + comps := make([]string, 0, 5) + if b.version.Group == "" { + comps = append(comps, "api", b.version.Version) + } else { + comps = append(comps, "apis", b.version.Group, b.version.Version) + } + if b.namespace != "" { + comps = append(comps, "namespaces", b.namespace) + } + // Generate random suffix based on keySpaceSize + randomInt, _ := rand.Int(rand.Reader, big.NewInt(int64(b.keySpaceSize))) + suffix := randomInt.Int64() + + // Create final resource name: name-{suffix} + finalName := fmt.Sprintf("%s-%d", b.name, suffix) + comps = append(comps, b.resource, finalName) + + // Render a unique body per request if patchRandomPlaceholder exists. Bodies without the + // placeholder pass through unchanged. + reqBody := renderPatchBody(b.body) + + return &DiscardRequester{ + BaseRequester: BaseRequester{ + method: "PATCH", + req: cli.Patch(b.patchType).AbsPath(comps...). + Body(reqBody). + MaxRetries(b.maxRetries), + }, + } +} + +type requestPostDelBuilder struct { + version schema.GroupVersion + resource string + resourceVersion string + namespace string + deleteRatio float64 + payloadSize int + nodeSelector map[string]string + labels map[string]string + serviceAccountName string + schedulerName string + maxRetries int + + // Per-builder cache for created resources + cache *Cache + + // Per-builder atomic counter for unique ID generation + resourceCounter int64 +} + +func newRequestPostDelBuilder(src *types.RequestPostDel, resourceVersion string, maxRetries int) *requestPostDelBuilder { + b := &requestPostDelBuilder{ + version: schema.GroupVersion{Group: src.Group, Version: src.Version}, + resource: src.Resource, + resourceVersion: resourceVersion, + namespace: src.Namespace, + deleteRatio: src.DeleteRatio, + payloadSize: src.PayloadSize, + maxRetries: maxRetries, + cache: InitCache(), // Initialize the cache + } + if p := src.Pod; p != nil { + b.nodeSelector = p.NodeSelector + b.labels = p.Labels + b.serviceAccountName = p.ServiceAccountName + b.schedulerName = p.SchedulerName + } + return b +} + +// Build implements RequestBuilder.Build. +func (b *requestPostDelBuilder) Build(cli rest.Interface) Requester { + comps := make([]string, 0, 5) + if b.version.Group == "" { + comps = append(comps, "api", b.version.Version) + } else { + comps = append(comps, "apis", b.version.Group, b.version.Version) + } + if b.namespace != "" { + comps = append(comps, "namespaces", b.namespace) + } + + // Random pick operation DELETE or CREATE based on deleteRatio weight probability + randomInt, _ := rand.Int(rand.Reader, big.NewInt(1000)) + shouldDelete := float64(randomInt.Int64())/1000.0 < b.deleteRatio + + if shouldDelete { + // Try to get a name from cache + if name, ok := b.cache.Pop(); ok { + comps = append(comps, b.resource, name) + + return &PostDelDiscardRequester{ + builder: b, + name: name, + operation: "DELETE", + DiscardRequester: DiscardRequester{ + BaseRequester: BaseRequester{ + method: "DELETE", + req: cli.Delete().AbsPath(comps...). + MaxRetries(b.maxRetries), + }, + }, + } + } + // If cache is empty, fall through to POST + } + + // POST logic - create resource and add to cache if successful + comps = append(comps, b.resource) + + // Use builder's atomic counter for synchronized unique ID generation + counter := atomic.AddInt64(&b.resourceCounter, 1) + timestamp := time.Now().UnixNano() + name := fmt.Sprintf("%d-%d", timestamp, counter) + + body, err := utils.RenderTemplate(b.resource, map[string]interface{}{ + "namePattern": name, + "namespace": b.namespace, + "payload": RandomPayload(b.payloadSize), + "nodeSelector": b.nodeSelector, + "labels": b.labels, + "serviceAccountName": b.serviceAccountName, + "schedulerName": b.schedulerName, + }) + if err != nil { + panic(fmt.Errorf("failed to render %s template: %w", b.resource, err)) + } + + return &PostDelDiscardRequester{ + builder: b, + name: name, + operation: "POST", + DiscardRequester: DiscardRequester{ + BaseRequester: BaseRequester{ + method: "POST", + req: cli.Post().AbsPath(comps...).Body(body).MaxRetries(b.maxRetries), + }, + }, + } +} + +// PostDelDiscardRequester handles both POST and DELETE requests with cache management +type PostDelDiscardRequester struct { + builder *requestPostDelBuilder + name string + operation string // "POST" or "DELETE" + DiscardRequester +} + +func (reqr *PostDelDiscardRequester) Do(ctx context.Context) (bytes int64, err error) { + // Use DiscardRequester's Do method to discard response body + bytes, err = reqr.DiscardRequester.Do(ctx) + + switch reqr.operation { + case "POST": + // Only add to cache if POST request was successful + if err == nil { + reqr.builder.cache.Push(reqr.name) + } + case "DELETE": + // If DELETE request failed, restore the item back to cache + // since the resource still exists in Kubernetes + if err != nil { + reqr.builder.cache.Push(reqr.name) + } + } + + return bytes, err +} + +func toPtr[T any](v T) *T { + return &v +} + +// requestPutBuilder builds PUT requests that overwrite an existing ConfigMap +// named `-` with a body whose `data.payload` field +// carries `payloadSize` random bytes. The target ConfigMap must already exist +// (e.g. created via `runkperf data cm add`); otherwise the apiserver returns +// 404 and the request is counted as a failure. Only `configmaps` is supported +// today — see RequestPut.Validate. +type requestPutBuilder struct { + version schema.GroupVersion + resource string + namespace string + name string + keySpaceSize int + payloadSize int + maxRetries int +} + +func newRequestPutBuilder(src *types.RequestPut, maxRetries int) *requestPutBuilder { + return &requestPutBuilder{ + version: schema.GroupVersion{ + Group: src.Group, + Version: src.Version, + }, + resource: src.Resource, + namespace: src.Namespace, + name: src.Name, + keySpaceSize: src.KeySpaceSize, + payloadSize: src.PayloadSize, + maxRetries: maxRetries, + } +} + +// Build implements RESTRequestBuilder.Build. +func (b *requestPutBuilder) Build(cli rest.Interface) Requester { + comps := []string{"api", b.version.Version, "namespaces", b.namespace, b.resource} + + randomInt, _ := rand.Int(rand.Reader, big.NewInt(int64(b.keySpaceSize))) + finalName := fmt.Sprintf("%s-%d", b.name, randomInt.Int64()) + comps = append(comps, finalName) + + cm := &corev1.ConfigMap{ + TypeMeta: metav1.TypeMeta{APIVersion: "v1", Kind: "ConfigMap"}, + ObjectMeta: metav1.ObjectMeta{ + Name: finalName, + Namespace: b.namespace, + }, + Data: map[string]string{"payload": RandomPayload(b.payloadSize)}, + } + body, _ := json.Marshal(cm) + + return &DiscardRequester{ + BaseRequester: BaseRequester{ + method: "PUT", + req: cli.Put().AbsPath(comps...).Body(body).MaxRetries(b.maxRetries), + }, + } +} + // 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. // It returns "" when n <= 0 and panics if reading random bytes fails. diff --git a/request/random_test.go b/request/random_test.go index 7430fc81..f5fbe7bd 100644 --- a/request/random_test.go +++ b/request/random_test.go @@ -4,12 +4,46 @@ package request import ( + "context" + "encoding/json" + "io" + "net/http" "strconv" + "strings" + "sync" "testing" + "github.com/Azure/kperf/api/types" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/client-go/kubernetes/scheme" + "k8s.io/client-go/rest/fake" ) +func TestRenderPatchBody(t *testing.T) { + const tmpl = `{"metadata":{"labels":{"bench-tick":"{{.Random}}"}}}` + + // Placeholder is replaced with a random string, so the result no longer + // contains the placeholder and differs from the raw template. + rendered := string(renderPatchBody(tmpl)) + assert.NotContains(t, rendered, patchRandomPlaceholder) + assert.NotEqual(t, tmpl, rendered) + + // Each render produces a distinct body. + assert.NotEqual(t, string(renderPatchBody(tmpl)), string(renderPatchBody(tmpl))) + + // Every occurrence is replaced. + multi := string(renderPatchBody(`{"a":"{{.Random}}","b":"{{.Random}}"}`)) + assert.NotContains(t, multi, patchRandomPlaceholder) + + // A body without the placeholder is returned unchanged (backward compatible). + const static = `{"metadata":{"labels":{"bench-tick":"1"}}}` + assert.Equal(t, static, string(renderPatchBody(static))) +} + func TestRandomPayload(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) { @@ -44,3 +78,116 @@ func TestRandBytes(t *testing.T) { }) } } + +func TestRequestPatchBuilderRendersUniqueBodies(t *testing.T) { + b := newRequestPatchBuilder(&types.RequestPatch{ + KubeGroupVersionResource: types.KubeGroupVersionResource{ + Version: "v1", + Resource: "configmaps", + }, + Namespace: "bench", + Name: "cm", + KeySpaceSize: 100, + PatchType: "merge", + Body: `{"metadata":{"labels":{"bench-tick":"{{.Random}}"}}}`, + }, "", 0) + + // Each call renders a fresh random body, producing a distinct result. + first := string(renderPatchBody(b.body)) + second := string(renderPatchBody(b.body)) + assert.NotEqual(t, first, second) + assert.NotContains(t, first, patchRandomPlaceholder) + assert.NotContains(t, second, patchRandomPlaceholder) +} + +func TestRequestPatchBuilderRendersConcurrencySafe(t *testing.T) { + b := newRequestPatchBuilder(&types.RequestPatch{ + KubeGroupVersionResource: types.KubeGroupVersionResource{ + Version: "v1", + Resource: "configmaps", + }, + Name: "cm", + PatchType: "merge", + Body: `{"n":"{{.Random}}"}`, + }, "", 0) + + const goroutines = 16 + const perGoroutine = 100 + + var wg sync.WaitGroup + var mu sync.Mutex + seen := make(map[string]struct{}, goroutines*perGoroutine) + + for i := 0; i < goroutines; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for j := 0; j < perGoroutine; j++ { + body := string(renderPatchBody(b.body)) + assert.NotContains(t, body, patchRandomPlaceholder) + mu.Lock() + seen[body] = struct{}{} + mu.Unlock() + } + }() + } + wg.Wait() + + // Random bodies are effectively unique across all requests. + assert.Equal(t, goroutines*perGoroutine, len(seen)) +} + +func TestPodOptions(t *testing.T) { + for _, tt := range []struct { + name string + options *types.PostDelPodOptions + labels map[string]string + }{ + {name: "defaults", labels: map[string]string{"app": "fake-pod"}}, + {name: "empty app", options: &types.PostDelPodOptions{ + Labels: map[string]string{"app": ""}, + }, labels: map[string]string{"app": ""}}, + {name: "other labels", options: &types.PostDelPodOptions{ + Labels: map[string]string{"team": "benchmark"}, + }, labels: map[string]string{"app": "fake-pod", "team": "benchmark"}}, + {name: "string labels", options: &types.PostDelPodOptions{ + Labels: map[string]string{"true": "false", "123": "456", "null": "null"}, + }, labels: map[string]string{"app": "fake-pod", "true": "false", "123": "456", "null": "null"}}, + {name: "custom", options: &types.PostDelPodOptions{ + Labels: map[string]string{"app": "custom-app", "team": "benchmark"}, NodeSelector: map[string]string{"agentpool": "benchmark"}, + ServiceAccountName: "custom-account", SchedulerName: "custom-scheduler", + }, labels: map[string]string{"app": "custom-app", "team": "benchmark"}}, + } { + t.Run(tt.name, func(t *testing.T) { + builder := newRequestPostDelBuilder(&types.RequestPostDel{ + KubeGroupVersionResource: types.KubeGroupVersionResource{Version: "v1", Resource: "pods"}, + Namespace: "default", + Pod: tt.options, + }, "", 0) + var pod corev1.Pod + cli := &fake.RESTClient{ + GroupVersion: schema.GroupVersion{Version: "v1"}, + NegotiatedSerializer: scheme.Codecs.WithoutConversion(), + Client: fake.CreateHTTPClient(func(r *http.Request) (*http.Response, error) { + require.Equal(t, "POST", r.Method) + require.NoError(t, json.NewDecoder(r.Body).Decode(&pod)) + return &http.Response{ + StatusCode: http.StatusCreated, Header: http.Header{"Content-Type": {"application/json"}}, + Body: io.NopCloser(strings.NewReader("{}")), + }, nil + }), + } + _, err := builder.Build(cli).Do(context.Background()) + require.NoError(t, err) + require.Equal(t, "default", pod.Namespace) + options := tt.options + if options == nil { + options = &types.PostDelPodOptions{} + } + require.Equal(t, tt.labels, pod.Labels) + require.Equal(t, options.NodeSelector, pod.Spec.NodeSelector) + require.Equal(t, options.ServiceAccountName, pod.Spec.ServiceAccountName) + require.Equal(t, options.SchedulerName, pod.Spec.SchedulerName) + }) + } +}