mirror of
https://github.com/k3s-io/kubernetes.git
synced 2026-07-18 04:54:54 +00:00
Merge pull request #136980 from Argh4k/scheduler-perf-refactor
Add new template functions to scheduler_perf
This commit is contained in:
Kubernetes Prow Robot
GitHub
@@ -1,189 +0,0 @@
|
||||
/*
|
||||
Copyright 2019 The Kubernetes Authors.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
package benchmark
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"html/template"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
"k8s.io/apimachinery/pkg/api/meta"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/client-go/discovery/cached/memory"
|
||||
"k8s.io/client-go/restmapper"
|
||||
"k8s.io/klog/v2"
|
||||
"k8s.io/kubernetes/test/utils/ktesting"
|
||||
"k8s.io/utils/ptr"
|
||||
"sigs.k8s.io/yaml"
|
||||
)
|
||||
|
||||
// createAny defines an op where some object gets created from a YAML file.
|
||||
// The nameset can be specified.
|
||||
type createAny struct {
|
||||
// Must match createAnyOpcode.
|
||||
Opcode operationCode
|
||||
// Namespace the object should be created in. Must be empty for cluster-scoped objects.
|
||||
Namespace string
|
||||
// Path to spec file describing the object to create.
|
||||
// This will be processed with text/template.
|
||||
// .Index will be in the range [0, Count-1] when creating
|
||||
// more than one object. .Count is the total number of objects.
|
||||
TemplatePath string
|
||||
// Count determines how many objects get created. Defaults to 1 if unset.
|
||||
Count *int
|
||||
CountParam string
|
||||
}
|
||||
|
||||
var _ runnableOp = &createAny{}
|
||||
|
||||
func (c *createAny) isValid(allowParameterization bool) error {
|
||||
if c.TemplatePath == "" {
|
||||
return fmt.Errorf("TemplatePath must be set")
|
||||
}
|
||||
// The namespace can only be checked during later because we don't know yet
|
||||
// whether the object is namespaced or cluster-scoped.
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *createAny) collectsMetrics() bool {
|
||||
return false
|
||||
}
|
||||
|
||||
func (c createAny) patchParams(w *workload) (realOp, error) {
|
||||
if c.CountParam != "" {
|
||||
count, err := w.Params.get(c.CountParam[1:])
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
c.Count = ptr.To(count)
|
||||
}
|
||||
return &c, c.isValid(false)
|
||||
}
|
||||
|
||||
func (c *createAny) requiredNamespaces() []string {
|
||||
if c.Namespace == "" {
|
||||
return nil
|
||||
}
|
||||
return []string{c.Namespace}
|
||||
}
|
||||
|
||||
func (c *createAny) run(tCtx ktesting.TContext) {
|
||||
count := 1
|
||||
if c.Count != nil {
|
||||
count = *c.Count
|
||||
}
|
||||
for index := 0; index < count; index++ {
|
||||
c.create(tCtx, map[string]any{"Index": index, "Count": count})
|
||||
}
|
||||
}
|
||||
|
||||
func (c *createAny) create(tCtx ktesting.TContext, env map[string]any) {
|
||||
var obj *unstructured.Unstructured
|
||||
if err := getSpecFromTextTemplateFile(c.TemplatePath, env, &obj); err != nil {
|
||||
tCtx.Fatalf("%s: parsing failed: %v", c.TemplatePath, err)
|
||||
}
|
||||
|
||||
// Not caching the discovery result isn't very efficient, but good enough when
|
||||
// createAny isn't done often.
|
||||
mapping, err := restMappingFromUnstructuredObj(tCtx, obj)
|
||||
if err != nil {
|
||||
tCtx.Fatalf("%s: %v", c.TemplatePath, err)
|
||||
}
|
||||
resourceClient := tCtx.Dynamic().Resource(mapping.Resource)
|
||||
|
||||
create := func() error {
|
||||
options := metav1.CreateOptions{
|
||||
// If the YAML input is invalid, then we want the
|
||||
// apiserver to tell us via an error. This can
|
||||
// happen because decoding into an unstructured object
|
||||
// doesn't validate.
|
||||
FieldValidation: "Strict",
|
||||
}
|
||||
if c.Namespace != "" {
|
||||
if mapping.Scope.Name() != meta.RESTScopeNameNamespace {
|
||||
return fmt.Errorf("namespace %q set for %q, but %q has scope %q", c.Namespace, c.TemplatePath, mapping.GroupVersionKind, mapping.Scope.Name())
|
||||
}
|
||||
_, err = resourceClient.Namespace(c.Namespace).Create(tCtx, obj, options)
|
||||
} else {
|
||||
if mapping.Scope.Name() != meta.RESTScopeNameRoot {
|
||||
return fmt.Errorf("namespace not set for %q, but %q has scope %q", c.TemplatePath, mapping.GroupVersionKind, mapping.Scope.Name())
|
||||
}
|
||||
_, err = resourceClient.Create(tCtx, obj, options)
|
||||
}
|
||||
return err
|
||||
}
|
||||
// Retry, some errors (like CRD just created and type not ready for use yet) are temporary.
|
||||
ctx, cancel := context.WithTimeout(tCtx, 20*time.Second)
|
||||
defer cancel()
|
||||
for {
|
||||
err := create()
|
||||
if err == nil {
|
||||
return
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
tCtx.Fatalf("%s: timed out (%q) while creating %q, last error was: %v", c.TemplatePath, context.Cause(ctx), klog.KObj(obj), err)
|
||||
case <-time.After(time.Second):
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func getSpecFromTextTemplateFile(path string, env map[string]any, spec interface{}) error {
|
||||
content, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
fm := template.FuncMap{"div": func(a, b int) int {
|
||||
return a / b
|
||||
}}
|
||||
modFn := template.FuncMap{"mod": func(a, b int) int {
|
||||
return a % b
|
||||
}}
|
||||
tmpl, err := template.New("object").Funcs(fm).Funcs(modFn).Parse(string(content))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var buffer bytes.Buffer
|
||||
if err := tmpl.Execute(&buffer, env); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return yaml.UnmarshalStrict(buffer.Bytes(), spec)
|
||||
}
|
||||
|
||||
func restMappingFromUnstructuredObj(tCtx ktesting.TContext, obj *unstructured.Unstructured) (*meta.RESTMapping, error) {
|
||||
discoveryCache := memory.NewMemCacheClient(tCtx.Client().Discovery())
|
||||
restMapper := restmapper.NewDeferredDiscoveryRESTMapper(discoveryCache)
|
||||
gv, err := schema.ParseGroupVersion(obj.GetAPIVersion())
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("extract group+version from object %q: %w", klog.KObj(obj), err)
|
||||
}
|
||||
gk := schema.GroupKind{Group: gv.Group, Kind: obj.GetKind()}
|
||||
|
||||
mapping, err := restMapper.RESTMapping(gk, gv.Version)
|
||||
if err != nil {
|
||||
// Cached mapping might be stale, refresh on next try.
|
||||
restMapper.Reset()
|
||||
return nil, fmt.Errorf("failed mapping %q to resource: %w", gk, err)
|
||||
}
|
||||
return mapping, nil
|
||||
}
|
||||
@@ -12,4 +12,4 @@ spec:
|
||||
resourceClaims:
|
||||
- name: resource
|
||||
# Five pods share access to the same claim.
|
||||
resourceClaimName: test-claim-{{div .Index 5}}
|
||||
resourceClaimName: test-claim-{{DivideInt .Index 5}}
|
||||
|
||||
@@ -10,7 +10,7 @@ spec:
|
||||
name: pause
|
||||
resources:
|
||||
requests:
|
||||
example.com/gpu-{{mod .Index 25}}: 1
|
||||
example.com/gpu-{{Mod .Index 25}}: 1
|
||||
limits:
|
||||
example.com/gpu-{{mod .Index 25}}: 1
|
||||
example.com/gpu-{{Mod .Index 25}}: 1
|
||||
|
||||
|
||||
@@ -10,7 +10,7 @@ spec:
|
||||
name: pause
|
||||
resources:
|
||||
requests:
|
||||
example.com/gpu-{{mod .Index 50}}: 1
|
||||
example.com/gpu-{{Mod .Index 50}}: 1
|
||||
limits:
|
||||
example.com/gpu-{{mod .Index 50}}: 1
|
||||
example.com/gpu-{{Mod .Index 50}}: 1
|
||||
|
||||
|
||||
@@ -11,6 +11,6 @@ spec:
|
||||
name: pause
|
||||
resources:
|
||||
requests:
|
||||
deviceclass.resource.kubernetes.io/test-class-{{mod .Index 25}}: 1
|
||||
deviceclass.resource.kubernetes.io/test-class-{{Mod .Index 25}}: 1
|
||||
limits:
|
||||
deviceclass.resource.kubernetes.io/test-class-{{mod .Index 25}}: 1
|
||||
deviceclass.resource.kubernetes.io/test-class-{{Mod .Index 25}}: 1
|
||||
|
||||
@@ -4,7 +4,7 @@ metadata:
|
||||
generateName: pod-unsched-
|
||||
spec:
|
||||
tolerations:
|
||||
- key: toleration-{{ div .Index 10 }}
|
||||
- key: toleration-{{ DivideInt .Index 10 }}
|
||||
operator: Exists
|
||||
effect: NoSchedule
|
||||
containers:
|
||||
|
||||
@@ -18,8 +18,8 @@ spec:
|
||||
- image: registry.k8s.io/pause:3.10.1
|
||||
name: pause
|
||||
ports:
|
||||
- hostPort: 8{{ mod .Index 12 }}
|
||||
containerPort: 8{{ mod .Index 12 }}
|
||||
- hostPort: 8{{ Mod .Index 12 }}
|
||||
containerPort: 8{{ Mod .Index 12 }}
|
||||
resources:
|
||||
requests:
|
||||
cpu: 0.35
|
||||
|
||||
@@ -9,5 +9,5 @@ spec:
|
||||
- image: registry.k8s.io/pause:3.10.1
|
||||
name: pause
|
||||
ports:
|
||||
- hostPort: 8{{ mod .Index 12 }}
|
||||
containerPort: 8{{ mod .Index 12 }}
|
||||
- hostPort: 8{{ Mod .Index 12 }}
|
||||
containerPort: 8{{ Mod .Index 12 }}
|
||||
|
||||
@@ -9,5 +9,5 @@ spec:
|
||||
resources:
|
||||
requests:
|
||||
cpu: 0.0001
|
||||
memory: {{ div 30000 .Count }}Mi
|
||||
memory: {{ DivideInt 30000 .Count }}Mi
|
||||
nodeName: scheduler-perf-node
|
||||
|
||||
@@ -12,4 +12,4 @@ spec:
|
||||
resources:
|
||||
requests:
|
||||
cpu: 0.0001
|
||||
memory: {{ div 30000 .Count }}Mi
|
||||
memory: {{ DivideInt 30000 .Count }}Mi
|
||||
|
||||
@@ -11,4 +11,4 @@ spec:
|
||||
resources:
|
||||
requests:
|
||||
cpu: 0.0001
|
||||
memory: {{ div 30000 .Count }}Mi
|
||||
memory: {{ DivideInt 30000 .Count }}Mi
|
||||
|
||||
997
test/integration/scheduler_perf/executor.go
Normal file
997
test/integration/scheduler_perf/executor.go
Normal file
@@ -0,0 +1,997 @@
|
||||
/*
|
||||
Copyright The Kubernetes Authors.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
package benchmark
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"math"
|
||||
"os"
|
||||
"runtime"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
v1 "k8s.io/api/core/v1"
|
||||
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
||||
"k8s.io/apimachinery/pkg/api/meta"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
|
||||
"k8s.io/apimachinery/pkg/labels"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/apimachinery/pkg/util/wait"
|
||||
cacheddiscovery "k8s.io/client-go/discovery/cached/memory"
|
||||
"k8s.io/client-go/dynamic"
|
||||
coreinformers "k8s.io/client-go/informers/core/v1"
|
||||
clientset "k8s.io/client-go/kubernetes"
|
||||
"k8s.io/client-go/restmapper"
|
||||
"k8s.io/client-go/tools/cache"
|
||||
"k8s.io/klog/v2"
|
||||
"k8s.io/kubernetes/pkg/scheduler"
|
||||
schedutil "k8s.io/kubernetes/pkg/scheduler/util"
|
||||
testutils "k8s.io/kubernetes/test/utils"
|
||||
"k8s.io/kubernetes/test/utils/ktesting"
|
||||
"k8s.io/utils/ptr"
|
||||
"sigs.k8s.io/yaml"
|
||||
)
|
||||
|
||||
type WorkloadExecutor struct {
|
||||
scheduler *scheduler.Scheduler
|
||||
wg sync.WaitGroup
|
||||
collectorCancel func(string)
|
||||
collectorWG sync.WaitGroup
|
||||
collectors []testDataCollector
|
||||
dataItems []DataItem
|
||||
numPodsScheduledPerNamespace map[string]int
|
||||
podInformer coreinformers.PodInformer
|
||||
throughputErrorMargin float64
|
||||
testCase *testCase
|
||||
workload *workload
|
||||
topicName string
|
||||
nextNodeIndex int
|
||||
}
|
||||
|
||||
func (e *WorkloadExecutor) wait() {
|
||||
e.collectorWG.Wait()
|
||||
e.wg.Wait()
|
||||
}
|
||||
|
||||
func (e *WorkloadExecutor) runOp(tCtx ktesting.TContext, op realOp, opIndex int) error {
|
||||
switch concreteOp := op.(type) {
|
||||
case *createNodesOp:
|
||||
return e.runCreateNodesOp(tCtx, opIndex, concreteOp)
|
||||
case *createNamespacesOp:
|
||||
return e.runCreateNamespaceOp(tCtx, opIndex, concreteOp)
|
||||
case *createPodsOp:
|
||||
return e.runCreatePodsOp(tCtx, opIndex, concreteOp)
|
||||
case *deletePodsOp:
|
||||
return e.runDeletePodsOp(tCtx, opIndex, concreteOp)
|
||||
case *churnOp:
|
||||
return e.runChurnOp(tCtx, opIndex, concreteOp)
|
||||
case *barrierOp:
|
||||
return e.runBarrierOp(tCtx, opIndex, concreteOp)
|
||||
case *sleepOp:
|
||||
return e.runSleepOp(tCtx, concreteOp)
|
||||
case *startCollectingMetricsOp:
|
||||
return e.runStartCollectingMetricsOp(tCtx, opIndex, concreteOp)
|
||||
case *stopCollectingMetricsOp:
|
||||
return e.runStopCollectingMetrics(tCtx, opIndex)
|
||||
case *createResourceDriverOp:
|
||||
concreteOp.run(tCtx, e.scheduler.Profiles["default-scheduler"].SharedDRAManager())
|
||||
return nil
|
||||
default:
|
||||
return e.runDefaultOp(tCtx, opIndex, concreteOp)
|
||||
}
|
||||
}
|
||||
|
||||
func (e *WorkloadExecutor) runCreateNodesOp(tCtx ktesting.TContext, opIndex int, op *createNodesOp) error {
|
||||
nodePreparer, err := getNodePreparer(fmt.Sprintf("node-%d-", opIndex), op, tCtx.Client())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := nodePreparer.PrepareNodes(tCtx, e.nextNodeIndex); err != nil {
|
||||
return err
|
||||
}
|
||||
e.nextNodeIndex += op.Count
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *WorkloadExecutor) runCreateNamespaceOp(tCtx ktesting.TContext, opIndex int, op *createNamespacesOp) error {
|
||||
nsPreparer, err := newNamespacePreparer(tCtx, op)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := nsPreparer.prepare(tCtx); err != nil {
|
||||
err2 := nsPreparer.cleanup(tCtx)
|
||||
if err2 != nil {
|
||||
err = fmt.Errorf("prepare: %w; cleanup: %w", err, err2)
|
||||
}
|
||||
return err
|
||||
}
|
||||
for _, n := range nsPreparer.namespaces() {
|
||||
if _, ok := e.numPodsScheduledPerNamespace[n]; ok {
|
||||
// this namespace has been already created.
|
||||
continue
|
||||
}
|
||||
e.numPodsScheduledPerNamespace[n] = 0
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *WorkloadExecutor) runBarrierOp(tCtx ktesting.TContext, opIndex int, op *barrierOp) error {
|
||||
for _, namespace := range op.Namespaces {
|
||||
if _, ok := e.numPodsScheduledPerNamespace[namespace]; !ok {
|
||||
return fmt.Errorf("unknown namespace %s", namespace)
|
||||
}
|
||||
}
|
||||
switch op.StageRequirement {
|
||||
case Attempted:
|
||||
if err := waitUntilPodsAttempted(tCtx, e.podInformer, op.LabelSelector, op.Namespaces, e.numPodsScheduledPerNamespace); err != nil {
|
||||
return err
|
||||
}
|
||||
case Scheduled:
|
||||
// Default should be treated like "Scheduled", so handling both in the same way.
|
||||
fallthrough
|
||||
default:
|
||||
if err := waitUntilPodsScheduled(tCtx, e.podInformer, op.LabelSelector, op.Namespaces, e.numPodsScheduledPerNamespace); err != nil {
|
||||
return err
|
||||
}
|
||||
// At the end of the barrier, we can be sure that there are no pods
|
||||
// pending scheduling in the namespaces that we just blocked on.
|
||||
if len(op.Namespaces) == 0 {
|
||||
e.numPodsScheduledPerNamespace = make(map[string]int)
|
||||
} else {
|
||||
for _, namespace := range op.Namespaces {
|
||||
delete(e.numPodsScheduledPerNamespace, namespace)
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *WorkloadExecutor) runSleepOp(tCtx ktesting.TContext, op *sleepOp) error {
|
||||
select {
|
||||
case <-tCtx.Done():
|
||||
case <-time.After(op.Duration.Duration):
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *WorkloadExecutor) runStopCollectingMetrics(tCtx ktesting.TContext, opIndex int) error {
|
||||
items, err := stopCollectingMetrics(tCtx, e.collectorCancel, &e.collectorWG, e.workload.Threshold.Get(e.topicName), *e.workload.ThresholdMetricSelector, opIndex, e.collectors)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
e.dataItems = append(e.dataItems, items...)
|
||||
e.collectorCancel = nil
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *WorkloadExecutor) runCreatePodsOp(tCtx ktesting.TContext, opIndex int, op *createPodsOp) error {
|
||||
// define Pod's namespace automatically, and create that namespace.
|
||||
namespace := fmt.Sprintf("namespace-%d", opIndex)
|
||||
if op.Namespace != nil {
|
||||
namespace = *op.Namespace
|
||||
}
|
||||
err := createNamespaceIfNotPresent(tCtx, namespace, &e.numPodsScheduledPerNamespace)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if op.PodTemplatePath == nil {
|
||||
op.PodTemplatePath = e.testCase.DefaultPodTemplatePath
|
||||
}
|
||||
|
||||
if op.CollectMetrics {
|
||||
if e.collectorCancel != nil {
|
||||
return fmt.Errorf("metrics collection is overlapping. Probably second collector was started before stopping a previous one")
|
||||
}
|
||||
var err error
|
||||
e.collectors, e.collectorCancel, err = startCollectingMetrics(tCtx, &e.collectorWG, e.podInformer, e.testCase.MetricsCollectorConfig, e.throughputErrorMargin, opIndex, namespace, []string{namespace}, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if err := createPodsRapidly(tCtx, namespace, op); err != nil {
|
||||
return err
|
||||
}
|
||||
switch {
|
||||
case op.SkipWaitToCompletion:
|
||||
// Only record those namespaces that may potentially require barriers
|
||||
// in the future.
|
||||
e.numPodsScheduledPerNamespace[namespace] += op.Count
|
||||
case op.SteadyState:
|
||||
if err := createPodsSteadily(tCtx, namespace, e.podInformer, op); err != nil {
|
||||
return err
|
||||
}
|
||||
default:
|
||||
if err := waitUntilPodsScheduledInNamespace(tCtx, e.podInformer, nil, namespace, op.Count); err != nil {
|
||||
return fmt.Errorf("error in waiting for pods to get scheduled: %w", err)
|
||||
}
|
||||
}
|
||||
if op.CollectMetrics {
|
||||
// CollectMetrics and SkipWaitToCompletion can never be true at the
|
||||
// same time, so if we're here, it means that all pods have been
|
||||
// scheduled.
|
||||
items, err := stopCollectingMetrics(tCtx, e.collectorCancel, &e.collectorWG, e.workload.Threshold.Get(e.topicName), *e.workload.ThresholdMetricSelector, opIndex, e.collectors)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
e.dataItems = append(e.dataItems, items...)
|
||||
e.collectorCancel = nil
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *WorkloadExecutor) runDeletePodsOp(tCtx ktesting.TContext, opIndex int, op *deletePodsOp) error {
|
||||
labelSelector := labels.ValidatedSetSelector(op.LabelSelector)
|
||||
|
||||
podsToDelete, err := e.podInformer.Lister().Pods(op.Namespace).List(labelSelector)
|
||||
if err != nil {
|
||||
return fmt.Errorf("error in listing pods in the namespace %s: %w", op.Namespace, err)
|
||||
}
|
||||
|
||||
deletePods := func(opIndex int) {
|
||||
if op.DeletePodsPerSecond > 0 {
|
||||
ticker := time.NewTicker(time.Second / time.Duration(op.DeletePodsPerSecond))
|
||||
defer ticker.Stop()
|
||||
|
||||
for i := range podsToDelete {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
if err := tCtx.Client().CoreV1().Pods(op.Namespace).Delete(tCtx, podsToDelete[i].Name, metav1.DeleteOptions{}); err != nil {
|
||||
if errors.Is(err, context.Canceled) {
|
||||
return
|
||||
}
|
||||
tCtx.Errorf("op %d: unable to delete pod %v: %v", opIndex, podsToDelete[i].Name, err)
|
||||
}
|
||||
case <-tCtx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
listOpts := metav1.ListOptions{
|
||||
LabelSelector: labelSelector.String(),
|
||||
}
|
||||
if err := tCtx.Client().CoreV1().Pods(op.Namespace).DeleteCollection(tCtx, metav1.DeleteOptions{}, listOpts); err != nil {
|
||||
if errors.Is(err, context.Canceled) {
|
||||
return
|
||||
}
|
||||
tCtx.Errorf("op %d: unable to delete pods in namespace %v: %v", opIndex, op.Namespace, err)
|
||||
}
|
||||
}
|
||||
|
||||
if op.SkipWaitToCompletion {
|
||||
e.wg.Add(1)
|
||||
go func(opIndex int) {
|
||||
defer e.wg.Done()
|
||||
deletePods(opIndex)
|
||||
}(opIndex)
|
||||
} else {
|
||||
deletePods(opIndex)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *WorkloadExecutor) runChurnOp(tCtx ktesting.TContext, opIndex int, op *churnOp) error {
|
||||
var namespace string
|
||||
if op.Namespace != nil {
|
||||
namespace = *op.Namespace
|
||||
} else {
|
||||
namespace = fmt.Sprintf("namespace-%d", opIndex)
|
||||
}
|
||||
restMapper := restmapper.NewDeferredDiscoveryRESTMapper(cacheddiscovery.NewMemCacheClient(tCtx.Client().Discovery()))
|
||||
// Ensure the namespace exists.
|
||||
nsObj := &v1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: namespace}}
|
||||
if _, err := tCtx.Client().CoreV1().Namespaces().Create(tCtx, nsObj, metav1.CreateOptions{}); err != nil && !apierrors.IsAlreadyExists(err) {
|
||||
return fmt.Errorf("unable to create namespace %v: %w", namespace, err)
|
||||
}
|
||||
|
||||
var churnFns []func(name string) string
|
||||
|
||||
for i, path := range op.TemplatePaths {
|
||||
unstructuredObj, gvk, err := getUnstructuredFromFile(path)
|
||||
if err != nil {
|
||||
return fmt.Errorf("unable to parse the %v-th template path: %w", i, err)
|
||||
}
|
||||
// Obtain GVR.
|
||||
mapping, err := restMapper.RESTMapping(gvk.GroupKind(), gvk.Version)
|
||||
if err != nil {
|
||||
return fmt.Errorf("unable to find GVR for %v: %w", gvk, err)
|
||||
}
|
||||
gvr := mapping.Resource
|
||||
// Distinguish cluster-scoped with namespaced API objects.
|
||||
var dynRes dynamic.ResourceInterface
|
||||
if mapping.Scope.Name() == meta.RESTScopeNameNamespace {
|
||||
dynRes = tCtx.Dynamic().Resource(gvr).Namespace(namespace)
|
||||
} else {
|
||||
dynRes = tCtx.Dynamic().Resource(gvr)
|
||||
}
|
||||
|
||||
churnFns = append(churnFns, func(name string) string {
|
||||
if name != "" {
|
||||
if err := dynRes.Delete(tCtx, name, metav1.DeleteOptions{}); err != nil && !errors.Is(err, context.Canceled) {
|
||||
tCtx.Errorf("op %d: unable to delete %v: %v", opIndex, name, err)
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
live, err := dynRes.Create(tCtx, unstructuredObj, metav1.CreateOptions{})
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
return live.GetName()
|
||||
})
|
||||
}
|
||||
|
||||
var interval int64 = 500
|
||||
if op.IntervalMilliseconds != 0 {
|
||||
interval = op.IntervalMilliseconds
|
||||
}
|
||||
ticker := time.NewTicker(time.Duration(interval) * time.Millisecond)
|
||||
|
||||
switch op.Mode {
|
||||
case Create:
|
||||
e.wg.Add(1)
|
||||
go func() {
|
||||
defer e.wg.Done()
|
||||
defer ticker.Stop()
|
||||
count, threshold := 0, op.Number
|
||||
if threshold == 0 {
|
||||
threshold = math.MaxInt32
|
||||
}
|
||||
for count < threshold {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
for i := range churnFns {
|
||||
churnFns[i]("")
|
||||
}
|
||||
count++
|
||||
case <-tCtx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
case Recreate:
|
||||
e.wg.Add(1)
|
||||
go func() {
|
||||
defer e.wg.Done()
|
||||
defer ticker.Stop()
|
||||
retVals := make([][]string, len(churnFns))
|
||||
// For each churn function, instantiate a slice of strings with length "op.Number".
|
||||
for i := range retVals {
|
||||
retVals[i] = make([]string, op.Number)
|
||||
}
|
||||
|
||||
count := 0
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
for i := range churnFns {
|
||||
retVals[i][count%op.Number] = churnFns[i](retVals[i][count%op.Number])
|
||||
}
|
||||
count++
|
||||
case <-tCtx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *WorkloadExecutor) runDefaultOp(tCtx ktesting.TContext, opIndex int, op realOp) error {
|
||||
runnable, ok := op.(runnableOp)
|
||||
if !ok {
|
||||
return fmt.Errorf("invalid op %v", op)
|
||||
}
|
||||
for _, namespace := range runnable.requiredNamespaces() {
|
||||
err := createNamespaceIfNotPresent(tCtx, namespace, &e.numPodsScheduledPerNamespace)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
runnable.run(tCtx)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *WorkloadExecutor) runStartCollectingMetricsOp(tCtx ktesting.TContext, opIndex int, op *startCollectingMetricsOp) error {
|
||||
if e.collectorCancel != nil {
|
||||
return fmt.Errorf("metrics collection is overlapping. Probably second collector was started before stopping a previous one")
|
||||
}
|
||||
var err error
|
||||
e.collectors, e.collectorCancel, err = startCollectingMetrics(tCtx, &e.collectorWG, e.podInformer, e.testCase.MetricsCollectorConfig, e.throughputErrorMargin, opIndex, op.Name, op.Namespaces, op.LabelSelector)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func startCollectingMetrics(tCtx ktesting.TContext, collectorWG *sync.WaitGroup, podInformer coreinformers.PodInformer, mcc *metricsCollectorConfig, throughputErrorMargin float64, opIndex int, name string, namespaces []string, labelSelector map[string]string) ([]testDataCollector, func(string), error) {
|
||||
collectorCtx := tCtx.WithCancel()
|
||||
workloadName := tCtx.Name()
|
||||
|
||||
// Clean up memory usage from the initial setup phase.
|
||||
runtime.GC()
|
||||
|
||||
// The first part is the same for each workload, therefore we can strip it.
|
||||
workloadName = workloadName[strings.Index(name, "/")+1:]
|
||||
collectorsList := getTestDataCollectors(podInformer, fmt.Sprintf("%s/%s", workloadName, name), namespaces, labelSelector, mcc, throughputErrorMargin)
|
||||
for _, collector := range collectorsList {
|
||||
// Need loop-local variable for function below.
|
||||
err := collector.init()
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("failed to initialize data collector: %w", err)
|
||||
}
|
||||
tCtx.TB().Cleanup(func() {
|
||||
collectorCtx.Cancel("cleaning up")
|
||||
})
|
||||
collectorWG.Add(1)
|
||||
go func() {
|
||||
defer collectorWG.Done()
|
||||
collector.run(collectorCtx)
|
||||
}()
|
||||
}
|
||||
if b, ok := tCtx.TB().(*testing.B); ok {
|
||||
b.ResetTimer()
|
||||
}
|
||||
tCtx.Log("Started metrics collection")
|
||||
return collectorsList, collectorCtx.Cancel, nil
|
||||
}
|
||||
|
||||
func stopCollectingMetrics(tCtx ktesting.TContext, collectorCancel func(string), collectorWG *sync.WaitGroup, threshold float64, tms thresholdMetricSelector, opIndex int, collectors []testDataCollector) ([]DataItem, error) {
|
||||
if b, ok := tCtx.TB().(*testing.B); ok {
|
||||
b.StopTimer()
|
||||
}
|
||||
if collectorCancel == nil {
|
||||
return nil, fmt.Errorf("missing startCollectingMetrics operation before stopping")
|
||||
}
|
||||
collectorCancel("collecting metrics, collector must stop first")
|
||||
collectorWG.Wait()
|
||||
var dataItems []DataItem
|
||||
for _, collector := range collectors {
|
||||
items := collector.collect()
|
||||
dataItems = append(dataItems, items...)
|
||||
err := applyThreshold(items, threshold, tms)
|
||||
if err != nil {
|
||||
tCtx.Errorf("op %d: %s", opIndex, err)
|
||||
}
|
||||
}
|
||||
tCtx.Log("Stopped metrics collection")
|
||||
return dataItems, nil
|
||||
}
|
||||
|
||||
type testDataCollector interface {
|
||||
init() error
|
||||
run(tCtx ktesting.TContext)
|
||||
collect() []DataItem
|
||||
}
|
||||
|
||||
// var for mocking in tests.
|
||||
var getTestDataCollectors = func(podInformer coreinformers.PodInformer, name string, namespaces []string, labelSelector map[string]string, mcc *metricsCollectorConfig, throughputErrorMargin float64) []testDataCollector {
|
||||
if mcc == nil {
|
||||
mcc = &defaultMetricsCollectorConfig
|
||||
}
|
||||
return []testDataCollector{
|
||||
newThroughputCollector(podInformer, map[string]string{"Name": name}, labelSelector, namespaces, throughputErrorMargin),
|
||||
newMetricsCollector(mcc, map[string]string{"Name": name}),
|
||||
newMemoryCollector(map[string]string{"Name": name}, 500*time.Millisecond),
|
||||
newSchedulingDurationCollector(map[string]string{"Name": name}),
|
||||
}
|
||||
}
|
||||
|
||||
func createNamespaceIfNotPresent(tCtx ktesting.TContext, namespace string, podsPerNamespace *map[string]int) error {
|
||||
if _, ok := (*podsPerNamespace)[namespace]; !ok {
|
||||
// The namespace has not created yet.
|
||||
// So, create that and register it.
|
||||
_, err := tCtx.Client().CoreV1().Namespaces().Create(tCtx, &v1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: namespace}}, metav1.CreateOptions{})
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to create namespace for Pod: %v", namespace)
|
||||
}
|
||||
(*podsPerNamespace)[namespace] = 0
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// createPodsRapidly implements the "create pods rapidly" mode of [createPodsOp].
|
||||
// It's a nop when cpo.SteadyState is true.
|
||||
func createPodsRapidly(tCtx ktesting.TContext, namespace string, cpo *createPodsOp) error {
|
||||
if cpo.SteadyState {
|
||||
return nil
|
||||
}
|
||||
strategy, err := getPodStrategy(cpo)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
tCtx.Logf("creating %d pods in namespace %q", cpo.Count, namespace)
|
||||
config := testutils.NewTestPodCreatorConfig()
|
||||
config.AddStrategy(namespace, cpo.Count, strategy)
|
||||
podCreator := testutils.NewTestPodCreator(tCtx.Client(), config)
|
||||
return podCreator.CreatePods(tCtx)
|
||||
}
|
||||
|
||||
// createPodsSteadily implements the "create pods and delete pods" mode of [createPodsOp].
|
||||
// It's a nop when cpo.SteadyState is false.
|
||||
func createPodsSteadily(tCtx ktesting.TContext, namespace string, podInformer coreinformers.PodInformer, cpo *createPodsOp) error {
|
||||
if !cpo.SteadyState {
|
||||
return nil
|
||||
}
|
||||
strategy, err := getPodStrategy(cpo)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
tCtx.Logf("creating pods in namespace %q for %s", namespace, cpo.Duration)
|
||||
tCtx = tCtx.WithTimeout(cpo.Duration.Duration, fmt.Sprintf("the operation ran for the configured %s", cpo.Duration.Duration))
|
||||
|
||||
// Start watching pods in the namespace. Any pod which is seen as being scheduled
|
||||
// gets deleted.
|
||||
scheduledPods := make(chan *v1.Pod, cpo.Count)
|
||||
scheduledPodsClosed := false
|
||||
var mutex sync.Mutex
|
||||
defer func() {
|
||||
mutex.Lock()
|
||||
defer mutex.Unlock()
|
||||
close(scheduledPods)
|
||||
scheduledPodsClosed = true
|
||||
}()
|
||||
|
||||
existingPods := 0
|
||||
runningPods := 0
|
||||
onPodChange := func(oldObj, newObj any) {
|
||||
oldPod, newPod, err := schedutil.As[*v1.Pod](oldObj, newObj)
|
||||
if err != nil {
|
||||
tCtx.Errorf("unexpected pod events: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
mutex.Lock()
|
||||
defer mutex.Unlock()
|
||||
if oldPod == nil {
|
||||
existingPods++
|
||||
}
|
||||
if (oldPod == nil || oldPod.Spec.NodeName == "") && newPod.Spec.NodeName != "" {
|
||||
// Got scheduled.
|
||||
runningPods++
|
||||
|
||||
// Only ask for deletion in our namespace.
|
||||
if newPod.Namespace != namespace {
|
||||
return
|
||||
}
|
||||
if !scheduledPodsClosed {
|
||||
select {
|
||||
case <-tCtx.Done():
|
||||
case scheduledPods <- newPod:
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
handle, err := podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
|
||||
AddFunc: func(obj any) {
|
||||
onPodChange(nil, obj)
|
||||
},
|
||||
UpdateFunc: func(oldObj, newObj any) {
|
||||
onPodChange(oldObj, newObj)
|
||||
},
|
||||
DeleteFunc: func(obj any) {
|
||||
pod, _, err := schedutil.As[*v1.Pod](obj, nil)
|
||||
if err != nil {
|
||||
tCtx.Errorf("unexpected pod events: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
mutex.Lock()
|
||||
defer mutex.Unlock()
|
||||
existingPods--
|
||||
if pod.Spec.NodeName != "" {
|
||||
runningPods--
|
||||
}
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("register event handler: %w", err)
|
||||
}
|
||||
defer func() {
|
||||
tCtx.ExpectNoError(podInformer.Informer().RemoveEventHandler(handle), "remove event handler")
|
||||
}()
|
||||
|
||||
// Seed the namespace with the initial number of pods.
|
||||
if err := strategy(tCtx, tCtx.Client(), namespace, cpo.Count); err != nil {
|
||||
return fmt.Errorf("create initial %d pods: %w", cpo.Count, err)
|
||||
}
|
||||
|
||||
// Now loop until we are done. Report periodically how many pods were scheduled.
|
||||
countScheduledPods := 0
|
||||
lastCountScheduledPods := 0
|
||||
logPeriod := time.Second
|
||||
ticker := time.NewTicker(logPeriod)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-tCtx.Done():
|
||||
tCtx.Logf("Completed after seeing %d scheduled pod: %v", countScheduledPods, context.Cause(tCtx))
|
||||
|
||||
// Sanity check: at least one pod should have been scheduled,
|
||||
// giving us a non-zero average.
|
||||
//
|
||||
// This is important because otherwise "no pods scheduled because of constant
|
||||
// failure" would not get detected in unit tests. For benchmarks
|
||||
// it's less important, but indicates that the time period
|
||||
// might have been too small.
|
||||
//
|
||||
// The collector logs "Failed to measure SchedulingThroughput ... Increase pods and/or nodes to make scheduling take longer"
|
||||
// but that is hard to spot.
|
||||
//
|
||||
// The non-steady case blocks until all pods have been scheduled
|
||||
// and doesn't need this check.
|
||||
if countScheduledPods == 0 {
|
||||
return errors.New("no pod at all got scheduled, either because of a problem or because the test interval was too small")
|
||||
}
|
||||
return nil
|
||||
case <-scheduledPods:
|
||||
countScheduledPods++
|
||||
if countScheduledPods%cpo.Count == 0 {
|
||||
// All scheduled. Start over with a new batch.
|
||||
err := tCtx.Client().CoreV1().Pods(namespace).DeleteCollection(tCtx, metav1.DeleteOptions{
|
||||
GracePeriodSeconds: ptr.To(int64(0)),
|
||||
PropagationPolicy: ptr.To(metav1.DeletePropagationBackground), // Foreground will block.
|
||||
}, metav1.ListOptions{})
|
||||
// Ignore errors when the time is up. errors.Is(context.Canceled) would
|
||||
// be more precise, but doesn't work because client-go doesn't reliably
|
||||
// propagate it.
|
||||
if tCtx.Err() != nil {
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
// Worse, sometimes rate limiting gives up *before* the context deadline is reached.
|
||||
// Then we get here with this error:
|
||||
// client rate limiter Wait returned an error: rate: Wait(n=1) would exceed context deadline
|
||||
//
|
||||
// This also can be ignored. We'll retry if the test is not done yet.
|
||||
if strings.Contains(err.Error(), "would exceed context deadline") {
|
||||
continue
|
||||
}
|
||||
return fmt.Errorf("delete scheduled pods: %w", err)
|
||||
}
|
||||
err = strategy(tCtx, tCtx.Client(), namespace, cpo.Count)
|
||||
if tCtx.Err() != nil {
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
return fmt.Errorf("create next batch of pods: %w", err)
|
||||
}
|
||||
}
|
||||
case <-ticker.C:
|
||||
delta := countScheduledPods - lastCountScheduledPods
|
||||
lastCountScheduledPods = countScheduledPods
|
||||
func() {
|
||||
mutex.Lock()
|
||||
defer mutex.Unlock()
|
||||
|
||||
tCtx.Logf("%d pods got scheduled in total in namespace %q, overall %d out of %d pods scheduled: %f pods/s in last interval",
|
||||
countScheduledPods, namespace,
|
||||
runningPods, existingPods,
|
||||
float64(delta)/logPeriod.Seconds(),
|
||||
)
|
||||
}()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// waitUntilPodsScheduled blocks until the all pods in the given namespaces are
|
||||
// scheduled.
|
||||
func waitUntilPodsScheduled(tCtx ktesting.TContext, podInformer coreinformers.PodInformer, labelSelector map[string]string, namespaces []string, numPodsScheduledPerNamespace map[string]int) error {
|
||||
// If unspecified, default to all known namespaces.
|
||||
if len(namespaces) == 0 {
|
||||
for namespace := range numPodsScheduledPerNamespace {
|
||||
namespaces = append(namespaces, namespace)
|
||||
}
|
||||
}
|
||||
for _, namespace := range namespaces {
|
||||
select {
|
||||
case <-tCtx.Done():
|
||||
return context.Cause(tCtx)
|
||||
default:
|
||||
}
|
||||
wantCount, ok := numPodsScheduledPerNamespace[namespace]
|
||||
if !ok {
|
||||
return fmt.Errorf("unknown namespace %s", namespace)
|
||||
}
|
||||
if err := waitUntilPodsScheduledInNamespace(tCtx, podInformer, labelSelector, namespace, wantCount); err != nil {
|
||||
return fmt.Errorf("error waiting for pods in namespace %q: %w", namespace, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// waitUntilPodsAttempted blocks until the all pods in the given namespaces are
|
||||
// attempted (at least once went through a scheduling cycle).
|
||||
func waitUntilPodsAttempted(tCtx ktesting.TContext, podInformer coreinformers.PodInformer, labelSelector map[string]string, namespaces []string, numPodsScheduledPerNamespace map[string]int) error {
|
||||
// If unspecified, default to all known namespaces.
|
||||
if len(namespaces) == 0 {
|
||||
for namespace := range numPodsScheduledPerNamespace {
|
||||
namespaces = append(namespaces, namespace)
|
||||
}
|
||||
}
|
||||
for _, namespace := range namespaces {
|
||||
select {
|
||||
case <-tCtx.Done():
|
||||
return context.Cause(tCtx)
|
||||
default:
|
||||
}
|
||||
wantCount, ok := numPodsScheduledPerNamespace[namespace]
|
||||
if !ok {
|
||||
return fmt.Errorf("unknown namespace %s", namespace)
|
||||
}
|
||||
if err := waitUntilPodsAttemptedInNamespace(tCtx, podInformer, labelSelector, namespace, wantCount); err != nil {
|
||||
return fmt.Errorf("error waiting for pods in namespace %q: %w", namespace, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// waitUntilPodsAttemptedInNamespace blocks until all pods in the given
|
||||
// namespace at least once went through a scheduling cycle.
|
||||
// Times out after 10 minutes similarly to waitUntilPodsScheduledInNamespace.
|
||||
func waitUntilPodsAttemptedInNamespace(tCtx ktesting.TContext, podInformer coreinformers.PodInformer, labelSelector map[string]string, namespace string, wantCount int) error {
|
||||
var pendingPod *v1.Pod
|
||||
|
||||
err := wait.PollUntilContextTimeout(tCtx, 1*time.Second, 10*time.Minute, true, func(ctx context.Context) (bool, error) {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return true, ctx.Err()
|
||||
default:
|
||||
}
|
||||
scheduled, attempted, unattempted, err := getScheduledPods(podInformer, labelSelector, namespace)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
if len(scheduled)+len(attempted) >= wantCount {
|
||||
tCtx.Logf("all pods attempted to be scheduled")
|
||||
return true, nil
|
||||
}
|
||||
tCtx.Logf("namespace: %s, attempted pods: want %d, got %d", namespace, wantCount, len(scheduled)+len(attempted))
|
||||
if len(unattempted) > 0 {
|
||||
pendingPod = unattempted[0]
|
||||
} else {
|
||||
pendingPod = nil
|
||||
}
|
||||
return false, nil
|
||||
})
|
||||
|
||||
if err != nil && pendingPod != nil {
|
||||
err = fmt.Errorf("at least pod %s is not attempted: %w", klog.KObj(pendingPod), err)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func getNodePreparer(prefix string, cno *createNodesOp, clientset clientset.Interface) (testutils.TestNodePreparer, error) {
|
||||
var nodeStrategy testutils.PrepareNodeStrategy = &testutils.TrivialNodePrepareStrategy{}
|
||||
if cno.NodeAllocatableStrategy != nil {
|
||||
nodeStrategy = cno.NodeAllocatableStrategy
|
||||
} else if cno.LabelNodePrepareStrategy != nil {
|
||||
nodeStrategy = cno.LabelNodePrepareStrategy
|
||||
} else if cno.UniqueNodeLabelStrategy != nil {
|
||||
nodeStrategy = cno.UniqueNodeLabelStrategy
|
||||
}
|
||||
|
||||
nodeTemplate := StaticNodeTemplate(makeBaseNode(prefix))
|
||||
if cno.NodeTemplatePath != nil {
|
||||
nodeTemplate = nodeTemplateFromFile(*cno.NodeTemplatePath)
|
||||
}
|
||||
|
||||
return NewIntegrationTestNodePreparer(
|
||||
clientset,
|
||||
[]testutils.CountToStrategy{{Count: cno.Count, Strategy: nodeStrategy}},
|
||||
nodeTemplate,
|
||||
), nil
|
||||
}
|
||||
|
||||
// waitUntilPodsScheduledInNamespace blocks until all pods in the given
|
||||
// namespace are scheduled. Times out after 10 minutes because even at the
|
||||
// lowest observed QPS of ~10 pods/sec, a 5000-node test should complete.
|
||||
func waitUntilPodsScheduledInNamespace(tCtx ktesting.TContext, podInformer coreinformers.PodInformer, labelSelector map[string]string, namespace string, wantCount int) error {
|
||||
var pendingPod *v1.Pod
|
||||
|
||||
err := wait.PollUntilContextTimeout(tCtx, 1*time.Second, 10*time.Minute, true, func(ctx context.Context) (bool, error) {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return true, ctx.Err()
|
||||
default:
|
||||
}
|
||||
scheduled, attempted, unattempted, err := getScheduledPods(podInformer, labelSelector, namespace)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
if len(scheduled) >= wantCount {
|
||||
tCtx.Logf("scheduling succeed")
|
||||
return true, nil
|
||||
}
|
||||
tCtx.Logf("namespace: %s, pods: want %d, got %d", namespace, wantCount, len(scheduled))
|
||||
if len(attempted) > 0 {
|
||||
pendingPod = attempted[0]
|
||||
} else if len(unattempted) > 0 {
|
||||
pendingPod = unattempted[0]
|
||||
} else {
|
||||
pendingPod = nil
|
||||
}
|
||||
return false, nil
|
||||
})
|
||||
|
||||
if err != nil && pendingPod != nil {
|
||||
err = fmt.Errorf("at least pod %s is not scheduled: %w", klog.KObj(pendingPod), err)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func getPodStrategy(cpo *createPodsOp) (testutils.TestPodCreateStrategy, error) {
|
||||
podTemplate := testutils.StaticPodTemplate(makeBasePod())
|
||||
if cpo.PodTemplatePath != nil {
|
||||
podTemplate = podTemplateFromFile(*cpo.PodTemplatePath)
|
||||
}
|
||||
if cpo.PersistentVolumeClaimTemplatePath == nil {
|
||||
return testutils.NewCustomCreatePodStrategy(podTemplate), nil
|
||||
}
|
||||
|
||||
pvTemplate, err := getPersistentVolumeSpecFromFile(cpo.PersistentVolumeTemplatePath)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
pvcTemplate, err := getPersistentVolumeClaimSpecFromFile(cpo.PersistentVolumeClaimTemplatePath)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return testutils.NewCreatePodWithPersistentVolumeStrategy(pvcTemplate, getCustomVolumeFactory(pvTemplate), podTemplate), nil
|
||||
}
|
||||
|
||||
type nodeTemplateFromFile string
|
||||
|
||||
func (f nodeTemplateFromFile) GetNodeTemplate(index, count int) (*v1.Node, error) {
|
||||
nodeSpec := &v1.Node{}
|
||||
if err := getSpecFromTextTemplateFile(string(f), map[string]any{"Index": index, "Count": count}, nodeSpec); err != nil {
|
||||
return nil, fmt.Errorf("parsing Node: %w", err)
|
||||
}
|
||||
return nodeSpec, nil
|
||||
}
|
||||
|
||||
type podTemplateFromFile string
|
||||
|
||||
func (f podTemplateFromFile) GetPodTemplate(index, count int) (*v1.Pod, error) {
|
||||
podSpec := &v1.Pod{}
|
||||
if err := getSpecFromTextTemplateFile(string(f), map[string]any{"Index": index, "Count": count}, podSpec); err != nil {
|
||||
return nil, fmt.Errorf("parsing Pod: %w", err)
|
||||
}
|
||||
return podSpec, nil
|
||||
}
|
||||
|
||||
func getPersistentVolumeSpecFromFile(path *string) (*v1.PersistentVolume, error) {
|
||||
persistentVolumeSpec := &v1.PersistentVolume{}
|
||||
if err := getSpecFromFile(path, persistentVolumeSpec); err != nil {
|
||||
return nil, fmt.Errorf("parsing PersistentVolume: %w", err)
|
||||
}
|
||||
return persistentVolumeSpec, nil
|
||||
}
|
||||
|
||||
func getPersistentVolumeClaimSpecFromFile(path *string) (*v1.PersistentVolumeClaim, error) {
|
||||
persistentVolumeClaimSpec := &v1.PersistentVolumeClaim{}
|
||||
if err := getSpecFromFile(path, persistentVolumeClaimSpec); err != nil {
|
||||
return nil, fmt.Errorf("parsing PersistentVolumeClaim: %w", err)
|
||||
}
|
||||
return persistentVolumeClaimSpec, nil
|
||||
}
|
||||
|
||||
func getCustomVolumeFactory(pvTemplate *v1.PersistentVolume) func(id int) *v1.PersistentVolume {
|
||||
return func(id int) *v1.PersistentVolume {
|
||||
pv := pvTemplate.DeepCopy()
|
||||
volumeID := fmt.Sprintf("vol-%d", id)
|
||||
pv.ObjectMeta.Name = volumeID
|
||||
pvs := pv.Spec.PersistentVolumeSource
|
||||
if pvs.CSI != nil {
|
||||
pvs.CSI.VolumeHandle = volumeID
|
||||
} else if pvs.AWSElasticBlockStore != nil {
|
||||
pvs.AWSElasticBlockStore.VolumeID = volumeID
|
||||
}
|
||||
return pv
|
||||
}
|
||||
}
|
||||
|
||||
// namespacePreparer holds configuration information for the test namespace preparer.
|
||||
type namespacePreparer struct {
|
||||
count int
|
||||
prefix string
|
||||
spec *v1.Namespace
|
||||
}
|
||||
|
||||
func newNamespacePreparer(tCtx ktesting.TContext, cno *createNamespacesOp) (*namespacePreparer, error) {
|
||||
ns := &v1.Namespace{}
|
||||
if cno.NamespaceTemplatePath != nil {
|
||||
if err := getSpecFromFile(cno.NamespaceTemplatePath, ns); err != nil {
|
||||
return nil, fmt.Errorf("parsing NamespaceTemplate: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
return &namespacePreparer{
|
||||
count: cno.Count,
|
||||
prefix: cno.Prefix,
|
||||
spec: ns,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// namespaces returns namespace names have been (or will be) created by this namespacePreparer
|
||||
func (p *namespacePreparer) namespaces() []string {
|
||||
namespaces := make([]string, p.count)
|
||||
for i := 0; i < p.count; i++ {
|
||||
namespaces[i] = fmt.Sprintf("%s-%d", p.prefix, i)
|
||||
}
|
||||
return namespaces
|
||||
}
|
||||
|
||||
// prepare creates the namespaces.
|
||||
func (p *namespacePreparer) prepare(tCtx ktesting.TContext) error {
|
||||
base := &v1.Namespace{}
|
||||
if p.spec != nil {
|
||||
base = p.spec
|
||||
}
|
||||
tCtx.Logf("Making %d namespaces with prefix %q and template %v", p.count, p.prefix, *base)
|
||||
for i := 0; i < p.count; i++ {
|
||||
n := base.DeepCopy()
|
||||
n.Name = fmt.Sprintf("%s-%d", p.prefix, i)
|
||||
if err := testutils.RetryWithExponentialBackOff(func() (bool, error) {
|
||||
_, err := tCtx.Client().CoreV1().Namespaces().Create(tCtx, n, metav1.CreateOptions{})
|
||||
return err == nil || apierrors.IsAlreadyExists(err), nil
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// cleanup deletes existing test namespaces.
|
||||
func (p *namespacePreparer) cleanup(tCtx ktesting.TContext) error {
|
||||
var errRet error
|
||||
for i := 0; i < p.count; i++ {
|
||||
n := fmt.Sprintf("%s-%d", p.prefix, i)
|
||||
if err := tCtx.Client().CoreV1().Namespaces().Delete(tCtx, n, metav1.DeleteOptions{}); err != nil {
|
||||
tCtx.Errorf("Deleting Namespace: %v", err)
|
||||
errRet = err
|
||||
}
|
||||
}
|
||||
return errRet
|
||||
}
|
||||
|
||||
func getUnstructuredFromFile(path string) (*unstructured.Unstructured, *schema.GroupVersionKind, error) {
|
||||
bytes, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
bytes, err = yaml.YAMLToJSONStrict(bytes)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("cannot covert YAML to JSON: %w", err)
|
||||
}
|
||||
|
||||
obj, gvk, err := unstructured.UnstructuredJSONScheme.Decode(bytes, nil, nil)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
unstructuredObj, ok := obj.(*unstructured.Unstructured)
|
||||
if !ok {
|
||||
return nil, nil, fmt.Errorf("cannot convert spec file in %v to an unstructured obj", path)
|
||||
}
|
||||
return unstructuredObj, gvk, nil
|
||||
}
|
||||
674
test/integration/scheduler_perf/executor_test.go
Normal file
674
test/integration/scheduler_perf/executor_test.go
Normal file
@@ -0,0 +1,674 @@
|
||||
/*
|
||||
Copyright The Kubernetes Authors.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
package benchmark
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"github.com/google/go-cmp/cmp"
|
||||
"github.com/google/go-cmp/cmp/cmpopts"
|
||||
v1 "k8s.io/api/core/v1"
|
||||
"k8s.io/apimachinery/pkg/api/resource"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/util/sets"
|
||||
coreinformers "k8s.io/client-go/informers/core/v1"
|
||||
"k8s.io/client-go/kubernetes/fake"
|
||||
testutils "k8s.io/kubernetes/test/utils"
|
||||
"k8s.io/kubernetes/test/utils/ktesting"
|
||||
"k8s.io/utils/ptr"
|
||||
)
|
||||
|
||||
type verifyFunc func(t *testing.T, tCtx ktesting.TContext, op realOp) error
|
||||
|
||||
func TestRunOp(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
op realOp
|
||||
expectedFailure bool
|
||||
verifyFuncs []verifyFunc
|
||||
}{
|
||||
{
|
||||
name: "Create Single Node",
|
||||
op: &createNodesOp{
|
||||
Opcode: createNodesOpcode,
|
||||
Count: 1,
|
||||
},
|
||||
verifyFuncs: []verifyFunc{
|
||||
verifyCount(1),
|
||||
verifyObj(
|
||||
&v1.Node{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
GenerateName: "node-0-",
|
||||
},
|
||||
Status: v1.NodeStatus{
|
||||
Capacity: v1.ResourceList{
|
||||
v1.ResourceCPU: resource.MustParse("4"),
|
||||
v1.ResourceMemory: resource.MustParse("32Gi"),
|
||||
v1.ResourcePods: resource.MustParse("110"),
|
||||
},
|
||||
Allocatable: v1.ResourceList{
|
||||
v1.ResourceCPU: resource.MustParse("4"),
|
||||
v1.ResourceMemory: resource.MustParse("32Gi"),
|
||||
v1.ResourcePods: resource.MustParse("110"),
|
||||
},
|
||||
},
|
||||
}),
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "Create Multiple Nodes",
|
||||
op: &createNodesOp{
|
||||
Opcode: createNodesOpcode,
|
||||
Count: 5,
|
||||
},
|
||||
verifyFuncs: []verifyFunc{
|
||||
verifyCount(5),
|
||||
verifyObj(
|
||||
&v1.Node{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
GenerateName: "node-0-",
|
||||
},
|
||||
Status: v1.NodeStatus{
|
||||
Capacity: v1.ResourceList{
|
||||
v1.ResourceCPU: resource.MustParse("4"),
|
||||
v1.ResourceMemory: resource.MustParse("32Gi"),
|
||||
v1.ResourcePods: resource.MustParse("110"),
|
||||
},
|
||||
Allocatable: v1.ResourceList{
|
||||
v1.ResourceCPU: resource.MustParse("4"),
|
||||
v1.ResourceMemory: resource.MustParse("32Gi"),
|
||||
v1.ResourcePods: resource.MustParse("110"),
|
||||
},
|
||||
},
|
||||
}),
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "Create Nodes with Label Strategy",
|
||||
op: &createNodesOp{
|
||||
Opcode: createNodesOpcode,
|
||||
Count: 3,
|
||||
LabelNodePrepareStrategy: testutils.NewLabelNodePrepareStrategy("test-label", "value1", "value2", "value3"),
|
||||
},
|
||||
verifyFuncs: []verifyFunc{
|
||||
verifyCount(3),
|
||||
verifyLabelValuesAllowed("test-label", sets.New("value1", "value2", "value3")),
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "Create Nodes with Unique Label Strategy",
|
||||
op: &createNodesOp{
|
||||
Opcode: createNodesOpcode,
|
||||
Count: 2,
|
||||
UniqueNodeLabelStrategy: testutils.NewUniqueNodeLabelStrategy("unique-test-label"),
|
||||
},
|
||||
verifyFuncs: []verifyFunc{
|
||||
verifyCount(2),
|
||||
verifyUniqueLabelValues("unique-test-label"),
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "Create Nodes with Node Allocatable Strategy",
|
||||
op: &createNodesOp{
|
||||
Opcode: createNodesOpcode,
|
||||
Count: 2,
|
||||
NodeAllocatableStrategy: testutils.NewNodeAllocatableStrategy(
|
||||
map[v1.ResourceName]string{
|
||||
v1.ResourceCPU: "2",
|
||||
v1.ResourceMemory: "4Gi",
|
||||
},
|
||||
nil, // no CSI node allocatable
|
||||
nil, // no migrated plugins
|
||||
),
|
||||
},
|
||||
verifyFuncs: []verifyFunc{
|
||||
verifyCount(2),
|
||||
verifyObj(
|
||||
&v1.Node{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
GenerateName: "node-0-",
|
||||
},
|
||||
Status: v1.NodeStatus{
|
||||
Capacity: v1.ResourceList{
|
||||
v1.ResourceCPU: resource.MustParse("4"),
|
||||
v1.ResourceMemory: resource.MustParse("32Gi"),
|
||||
v1.ResourcePods: resource.MustParse("110"),
|
||||
},
|
||||
Allocatable: v1.ResourceList{
|
||||
v1.ResourceCPU: resource.MustParse("2"),
|
||||
v1.ResourceMemory: resource.MustParse("4Gi"),
|
||||
v1.ResourcePods: resource.MustParse("110"),
|
||||
},
|
||||
},
|
||||
}),
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "Create Nodes with Custom Template",
|
||||
op: &createNodesOp{
|
||||
Opcode: createNodesOpcode,
|
||||
Count: 2,
|
||||
NodeTemplatePath: createObjTemplateFile(t,
|
||||
&v1.Node{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
GenerateName: "custom-node-",
|
||||
},
|
||||
Status: v1.NodeStatus{
|
||||
Capacity: v1.ResourceList{
|
||||
v1.ResourcePods: resource.MustParse("100"),
|
||||
v1.ResourceCPU: resource.MustParse("4"),
|
||||
v1.ResourceMemory: resource.MustParse("8Gi"),
|
||||
},
|
||||
},
|
||||
},
|
||||
),
|
||||
},
|
||||
verifyFuncs: []verifyFunc{
|
||||
verifyCount(2),
|
||||
verifyObj(
|
||||
&v1.Node{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
GenerateName: "custom-node-",
|
||||
},
|
||||
Status: v1.NodeStatus{
|
||||
Capacity: v1.ResourceList{
|
||||
v1.ResourcePods: resource.MustParse("100"),
|
||||
v1.ResourceCPU: resource.MustParse("4"),
|
||||
v1.ResourceMemory: resource.MustParse("8Gi"),
|
||||
},
|
||||
},
|
||||
},
|
||||
),
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "Invalid Node Template Path",
|
||||
op: &createNodesOp{
|
||||
Opcode: createNodesOpcode,
|
||||
Count: 1,
|
||||
NodeTemplatePath: ptr.To("non-existent-file.json"),
|
||||
},
|
||||
expectedFailure: true,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
_, tCtx := ktesting.NewTestContext(t)
|
||||
client := fake.NewSimpleClientset()
|
||||
tCtx = tCtx.WithClients(nil, nil, client, nil, nil)
|
||||
|
||||
exec := &WorkloadExecutor{
|
||||
numPodsScheduledPerNamespace: make(map[string]int),
|
||||
nextNodeIndex: 0,
|
||||
}
|
||||
|
||||
err := exec.runOp(tCtx, tt.op, 0)
|
||||
|
||||
if tt.expectedFailure {
|
||||
if err == nil {
|
||||
t.Fatalf("Expected error but got none")
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to run operation: %v", err)
|
||||
}
|
||||
|
||||
if tt.verifyFuncs != nil {
|
||||
for i, vf := range tt.verifyFuncs {
|
||||
if err := vf(t, tCtx, tt.op); err != nil {
|
||||
t.Fatalf("Verification function %d failed for test %q: %v", i, tt.name, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// verifyCount returns a verification function that checks if the number of existing objects
|
||||
// matches the expected count based on the operation type.
|
||||
func verifyCount(expectedCount int) verifyFunc {
|
||||
return func(t *testing.T, tCtx ktesting.TContext, op realOp) error {
|
||||
switch op.(type) {
|
||||
case *createNodesOp:
|
||||
nodes, err := tCtx.Client().CoreV1().Nodes().List(context.Background(), metav1.ListOptions{})
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to list nodes: %w", err)
|
||||
}
|
||||
if got := len(nodes.Items); got != expectedCount {
|
||||
return fmt.Errorf("unexpected node count: got %d, want %d", got, expectedCount)
|
||||
}
|
||||
default:
|
||||
return fmt.Errorf("verifyCount doesn't support this operation type: %T", op)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// verifyLabelValuesAllowed returns a verification function that checks if the label values for a given key.
|
||||
func verifyLabelValuesAllowed(key string, allowValues sets.Set[string]) verifyFunc {
|
||||
return func(t *testing.T, tCtx ktesting.TContext, op realOp) error {
|
||||
labelValues, _, err := objLabelValues(t, tCtx, op, key)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to get label values: %w", err)
|
||||
}
|
||||
|
||||
for labelValue := range labelValues {
|
||||
if !allowValues.Has(labelValue) {
|
||||
return fmt.Errorf("Node has unexpected label value %s for key %s", labelValue, key)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// verifyUniqueLabelValues returns a verification function that checks if the label values for a given key are unique.
|
||||
func verifyUniqueLabelValues(key string) verifyFunc {
|
||||
return func(t *testing.T, tCtx ktesting.TContext, op realOp) error {
|
||||
_, duplicatedValues, err := objLabelValues(t, tCtx, op, key)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to get label values: %w", err)
|
||||
}
|
||||
|
||||
if duplicatedValues.Len() > 0 {
|
||||
return fmt.Errorf("Node has duplicate label values %v for key %s", duplicatedValues.UnsortedList(), key)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// objLabelValues is a helper function to extract label values from the listed objects.
|
||||
// The listed objects are dependent on the operation type.
|
||||
// It returns two sets: one with deduplicated labelValues and second with duplicated labels.
|
||||
func objLabelValues(t *testing.T, tCtx ktesting.TContext, op realOp, key string) (sets.Set[string], sets.Set[string], error) {
|
||||
t.Helper()
|
||||
|
||||
labelValues := sets.New[string]()
|
||||
duplicatedValues := sets.New[string]()
|
||||
|
||||
switch op.(type) {
|
||||
case *createNodesOp:
|
||||
nodes, err := tCtx.Client().CoreV1().Nodes().List(context.Background(), metav1.ListOptions{})
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("failed to list nodes for label verification: %w", err)
|
||||
}
|
||||
|
||||
for _, node := range nodes.Items {
|
||||
if labelValue, exists := node.Labels[key]; exists {
|
||||
if labelValues.Has(labelValue) {
|
||||
duplicatedValues.Insert(labelValue)
|
||||
}
|
||||
labelValues.Insert(labelValue)
|
||||
} else {
|
||||
return nil, nil, fmt.Errorf("Node %s is missing expected label %s", node.Name, key)
|
||||
}
|
||||
}
|
||||
default:
|
||||
return nil, nil, fmt.Errorf("verifyLabel doesn't support this operation type: %T", op)
|
||||
}
|
||||
|
||||
return labelValues, duplicatedValues, nil
|
||||
}
|
||||
|
||||
// verifyObj checks if listed objects match the expected template object using cmp.Diff.
|
||||
func verifyObj(expectedObj any) verifyFunc {
|
||||
return func(t *testing.T, tCtx ktesting.TContext, op realOp) error {
|
||||
var got, want any
|
||||
var cmpOpts []cmp.Option
|
||||
switch opDetails := op.(type) {
|
||||
case *createNodesOp:
|
||||
nodesList, listErr := tCtx.Client().CoreV1().Nodes().List(context.Background(), metav1.ListOptions{})
|
||||
if listErr != nil {
|
||||
return fmt.Errorf("failed to list nodes: %w", listErr)
|
||||
}
|
||||
gotNodes := nodesList.Items
|
||||
|
||||
expectedNodeTemplate, ok := expectedObj.(*v1.Node)
|
||||
if !ok {
|
||||
return fmt.Errorf("expectedObj must be *v1.Node when op is *createNodesOp, got %T", expectedObj)
|
||||
}
|
||||
|
||||
wantNodes := make([]v1.Node, len(gotNodes))
|
||||
// we don't need to verify len(), we just need to check if all of them are the same as the expected one.
|
||||
for i := range gotNodes {
|
||||
wantNodes[i] = *expectedNodeTemplate
|
||||
}
|
||||
|
||||
cmpOpts = []cmp.Option{
|
||||
cmpopts.EquateEmpty(),
|
||||
cmpopts.IgnoreFields(metav1.ObjectMeta{},
|
||||
"UID", "ResourceVersion", "Generation", "CreationTimestamp", "ManagedFields", "SelfLink", "Name",
|
||||
"Labels", // verifyObj doesn't care about labels.
|
||||
),
|
||||
cmpopts.IgnoreFields(v1.NodeStatus{}, // This test isn't interested in these fields.
|
||||
"Conditions",
|
||||
"Phase",
|
||||
),
|
||||
}
|
||||
got = gotNodes
|
||||
want = wantNodes
|
||||
default:
|
||||
return fmt.Errorf("verifyObj doesn't support this operation type for cmp.Diff: %T", opDetails)
|
||||
}
|
||||
|
||||
if diff := cmp.Diff(want, got, cmpOpts...); diff != "" {
|
||||
return fmt.Errorf("unexpected difference (-want +got):\\n%s", diff)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// createObjTemplateFile creates a temporary file with the given object serialized as JSON.
|
||||
func createObjTemplateFile(t *testing.T, obj any) *string {
|
||||
t.Helper()
|
||||
|
||||
dir, err := os.MkdirTemp("", "scheduler-perf-test")
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to create temp dir for the template: %v", err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
if err := os.RemoveAll(dir); err != nil {
|
||||
t.Errorf("Failed to remove temp dir: %v", err)
|
||||
}
|
||||
})
|
||||
|
||||
templateFile := filepath.Join(dir, "template.json")
|
||||
f, err := os.Create(templateFile)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to create the template file %s: %v", templateFile, err)
|
||||
}
|
||||
defer func() {
|
||||
if err := f.Close(); err != nil {
|
||||
t.Errorf("Failed to close file: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
switch obj := obj.(type) {
|
||||
case *v1.Node:
|
||||
if err := json.NewEncoder(f).Encode(obj); err != nil {
|
||||
t.Fatalf("Failed to encode the template to %s: %v", templateFile, err)
|
||||
}
|
||||
default:
|
||||
t.Fatalf("Unsupported object type for template file: %T", obj)
|
||||
}
|
||||
return &templateFile
|
||||
}
|
||||
|
||||
// mockDataCollector always returns the same data items, to be used for mocking data collector in unit tests.
|
||||
type mockDataCollector struct {
|
||||
dataItems []DataItem
|
||||
}
|
||||
|
||||
// init does nothing.
|
||||
func (mc *mockDataCollector) init() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// run does nothing.
|
||||
func (mc *mockDataCollector) run(_ ktesting.TContext) {}
|
||||
|
||||
// collect always returns DataItems defined in the collector.
|
||||
func (mc *mockDataCollector) collect() []DataItem {
|
||||
return mc.dataItems
|
||||
}
|
||||
|
||||
func TestMetricThreshold(t *testing.T) {
|
||||
testCases := []struct {
|
||||
name string
|
||||
thresholdValue float64
|
||||
dataItems []DataItem
|
||||
thresholdMetricSelector *thresholdMetricSelector
|
||||
expectCollectionFailure bool
|
||||
expectedDataItemsWithThresholdIndices []int
|
||||
expectedThresholdName string
|
||||
}{
|
||||
{
|
||||
name: "value is above threshold, no error",
|
||||
thresholdValue: 100,
|
||||
dataItems: []DataItem{
|
||||
{
|
||||
Data: map[string]float64{
|
||||
"Average": 150,
|
||||
},
|
||||
Labels: map[string]string{
|
||||
"Metric": "throughput",
|
||||
},
|
||||
},
|
||||
},
|
||||
thresholdMetricSelector: &thresholdMetricSelector{
|
||||
Name: "throughput",
|
||||
DataBucket: "Average",
|
||||
},
|
||||
expectedDataItemsWithThresholdIndices: []int{0},
|
||||
expectedThresholdName: "AverageThreshold",
|
||||
},
|
||||
{
|
||||
name: "value is below threshold, expect error",
|
||||
thresholdValue: 100,
|
||||
dataItems: []DataItem{
|
||||
{
|
||||
Data: map[string]float64{
|
||||
"Average": 70,
|
||||
"Max": 90,
|
||||
},
|
||||
Labels: map[string]string{
|
||||
"Metric": "throughput",
|
||||
},
|
||||
},
|
||||
},
|
||||
thresholdMetricSelector: &thresholdMetricSelector{
|
||||
Name: "throughput",
|
||||
DataBucket: "Max",
|
||||
},
|
||||
expectCollectionFailure: true,
|
||||
expectedDataItemsWithThresholdIndices: []int{0},
|
||||
expectedThresholdName: "MaxThreshold",
|
||||
},
|
||||
{
|
||||
name: "no error if the labels do not match",
|
||||
thresholdValue: 100,
|
||||
dataItems: []DataItem{
|
||||
{
|
||||
Data: map[string]float64{
|
||||
"Average": 70,
|
||||
},
|
||||
Labels: map[string]string{
|
||||
"Metric": "throughput",
|
||||
"label": "value",
|
||||
},
|
||||
},
|
||||
},
|
||||
thresholdMetricSelector: &thresholdMetricSelector{
|
||||
Name: "throughput",
|
||||
DataBucket: "Average",
|
||||
Labels: map[string]string{
|
||||
"label": "value2",
|
||||
},
|
||||
},
|
||||
expectedDataItemsWithThresholdIndices: []int{},
|
||||
expectedThresholdName: "AverageThreshold",
|
||||
},
|
||||
{
|
||||
name: "out of multiple data items only matching are selected",
|
||||
thresholdValue: 100,
|
||||
dataItems: []DataItem{
|
||||
{
|
||||
Data: map[string]float64{
|
||||
"Average": 70,
|
||||
},
|
||||
Labels: map[string]string{
|
||||
"Metric": "throughput",
|
||||
"label": "value",
|
||||
},
|
||||
},
|
||||
{
|
||||
Data: map[string]float64{
|
||||
"Average": 150,
|
||||
},
|
||||
Labels: map[string]string{
|
||||
"Metric": "throughput",
|
||||
"label": "value2",
|
||||
},
|
||||
},
|
||||
},
|
||||
thresholdMetricSelector: &thresholdMetricSelector{
|
||||
Name: "throughput",
|
||||
DataBucket: "Average",
|
||||
Labels: map[string]string{
|
||||
"label": "value2",
|
||||
},
|
||||
},
|
||||
expectedDataItemsWithThresholdIndices: []int{1},
|
||||
expectedThresholdName: "AverageThreshold",
|
||||
},
|
||||
{
|
||||
name: "threshold value is added for all matching entries",
|
||||
thresholdValue: 100,
|
||||
dataItems: []DataItem{
|
||||
{
|
||||
Data: map[string]float64{
|
||||
"Average": 130,
|
||||
},
|
||||
Labels: map[string]string{
|
||||
"Metric": "throughput",
|
||||
"label": "value",
|
||||
},
|
||||
},
|
||||
{
|
||||
Data: map[string]float64{
|
||||
"Average": 150,
|
||||
},
|
||||
Labels: map[string]string{
|
||||
"Metric": "throughput",
|
||||
"label": "value2",
|
||||
},
|
||||
},
|
||||
},
|
||||
thresholdMetricSelector: &thresholdMetricSelector{
|
||||
Name: "throughput",
|
||||
DataBucket: "Average",
|
||||
},
|
||||
expectedDataItemsWithThresholdIndices: []int{0, 1},
|
||||
expectedThresholdName: "AverageThreshold",
|
||||
},
|
||||
{
|
||||
name: "threshold value is added for all matching entries even with error",
|
||||
thresholdValue: 100,
|
||||
dataItems: []DataItem{
|
||||
{
|
||||
Data: map[string]float64{
|
||||
"Average": 70,
|
||||
},
|
||||
Labels: map[string]string{
|
||||
"Metric": "throughput",
|
||||
"label": "value",
|
||||
},
|
||||
},
|
||||
{
|
||||
Data: map[string]float64{
|
||||
"Average": 80,
|
||||
},
|
||||
Labels: map[string]string{
|
||||
"Metric": "throughput",
|
||||
"label": "value2",
|
||||
},
|
||||
},
|
||||
{
|
||||
Data: map[string]float64{
|
||||
"Average": 130,
|
||||
},
|
||||
Labels: map[string]string{
|
||||
"Metric": "throughput",
|
||||
"label": "value3",
|
||||
},
|
||||
},
|
||||
},
|
||||
thresholdMetricSelector: &thresholdMetricSelector{
|
||||
Name: "throughput",
|
||||
DataBucket: "Average",
|
||||
},
|
||||
expectCollectionFailure: true,
|
||||
expectedDataItemsWithThresholdIndices: []int{0, 1, 2},
|
||||
expectedThresholdName: "AverageThreshold",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range testCases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
_, tCtx := ktesting.NewTestContext(t)
|
||||
var capturedErr error
|
||||
capturingCtx, finalize := tCtx.WithError(&capturedErr)
|
||||
defer finalize()
|
||||
|
||||
originalGetTestDataCollectors := getTestDataCollectors
|
||||
defer func() { getTestDataCollectors = originalGetTestDataCollectors }()
|
||||
getTestDataCollectors = func(_ coreinformers.PodInformer, _ string, _ []string, _ map[string]string, _ *metricsCollectorConfig, _ float64) []testDataCollector {
|
||||
return []testDataCollector{&mockDataCollector{dataItems: tc.dataItems}}
|
||||
}
|
||||
|
||||
workload := &workload{
|
||||
Name: "some/workload",
|
||||
Threshold: thresholds{
|
||||
valuesByTopic: map[string]float64{"example": tc.thresholdValue},
|
||||
},
|
||||
ThresholdMetricSelector: tc.thresholdMetricSelector,
|
||||
}
|
||||
exec := &WorkloadExecutor{
|
||||
topicName: "example",
|
||||
testCase: &testCase{},
|
||||
numPodsScheduledPerNamespace: make(map[string]int),
|
||||
workload: workload,
|
||||
}
|
||||
|
||||
start := &startCollectingMetricsOp{
|
||||
Opcode: startCollectingMetricsOpcode,
|
||||
Name: "test-collection",
|
||||
Namespaces: []string{"test-namespaces"},
|
||||
}
|
||||
err := exec.runOp(capturingCtx, start, 0)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to start metric collection")
|
||||
}
|
||||
stop := &stopCollectingMetricsOp{Opcode: stopCollectingMetricsOpcode}
|
||||
err = exec.runOp(capturingCtx, stop, 0)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to stop metric collection")
|
||||
}
|
||||
|
||||
if tc.expectCollectionFailure != capturingCtx.Failed() {
|
||||
t.Fatalf("expectCollectionFailure=%v but got %v", tc.expectCollectionFailure, capturingCtx.Failed())
|
||||
}
|
||||
for _, idx := range tc.expectedDataItemsWithThresholdIndices {
|
||||
if idx >= len(exec.dataItems) {
|
||||
t.Fatalf("expectedDataItemsWithThresholdIndex out of data items range")
|
||||
}
|
||||
if _, ok := exec.dataItems[idx].Data[tc.expectedThresholdName]; !ok {
|
||||
t.Fatalf("expected data item at index=%d to have %s field", idx, tc.expectedThresholdName)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
712
test/integration/scheduler_perf/operations.go
Normal file
712
test/integration/scheduler_perf/operations.go
Normal file
@@ -0,0 +1,712 @@
|
||||
/*
|
||||
Copyright 2019 The Kubernetes Authors.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
package benchmark
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"html/template"
|
||||
"os"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"k8s.io/apimachinery/pkg/api/meta"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/client-go/discovery/cached/memory"
|
||||
"k8s.io/client-go/restmapper"
|
||||
"k8s.io/klog/v2"
|
||||
testutils "k8s.io/kubernetes/test/utils"
|
||||
"k8s.io/kubernetes/test/utils/ktesting"
|
||||
"k8s.io/utils/ptr"
|
||||
"sigs.k8s.io/yaml"
|
||||
)
|
||||
|
||||
// createAny defines an op where some object gets created from a YAML file.
|
||||
// The nameset can be specified.
|
||||
type createAny struct {
|
||||
// Must match createAnyOpcode.
|
||||
Opcode operationCode
|
||||
// Namespace the object should be created in. Must be empty for cluster-scoped objects.
|
||||
Namespace string
|
||||
// Path to spec file describing the object to create.
|
||||
// This will be processed with text/template.
|
||||
// .Index will be in the range [0, Count-1] when creating
|
||||
// more than one object. .Count is the total number of objects.
|
||||
TemplatePath string
|
||||
// Count determines how many objects get created. Defaults to 1 if unset.
|
||||
Count *int
|
||||
CountParam string
|
||||
}
|
||||
|
||||
var _ runnableOp = &createAny{}
|
||||
|
||||
func (c *createAny) isValid(allowParameterization bool) error {
|
||||
if c.TemplatePath == "" {
|
||||
return fmt.Errorf("TemplatePath must be set")
|
||||
}
|
||||
// The namespace can only be checked during later because we don't know yet
|
||||
// whether the object is namespaced or cluster-scoped.
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *createAny) collectsMetrics() bool {
|
||||
return false
|
||||
}
|
||||
|
||||
func (c createAny) patchParams(w *workload) (realOp, error) {
|
||||
if c.CountParam != "" {
|
||||
count, err := w.Params.get(c.CountParam[1:])
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
c.Count = ptr.To(count)
|
||||
}
|
||||
return &c, c.isValid(false)
|
||||
}
|
||||
|
||||
func (c *createAny) requiredNamespaces() []string {
|
||||
if c.Namespace == "" {
|
||||
return nil
|
||||
}
|
||||
return []string{c.Namespace}
|
||||
}
|
||||
|
||||
func (c *createAny) run(tCtx ktesting.TContext) {
|
||||
count := 1
|
||||
if c.Count != nil {
|
||||
count = *c.Count
|
||||
}
|
||||
for index := 0; index < count; index++ {
|
||||
c.create(tCtx, map[string]any{"Index": index, "Count": count})
|
||||
}
|
||||
}
|
||||
|
||||
func (c *createAny) create(tCtx ktesting.TContext, env map[string]any) {
|
||||
var obj *unstructured.Unstructured
|
||||
if err := getSpecFromTextTemplateFile(c.TemplatePath, env, &obj); err != nil {
|
||||
tCtx.Fatalf("%s: parsing failed: %v", c.TemplatePath, err)
|
||||
}
|
||||
|
||||
// Not caching the discovery result isn't very efficient, but good enough when
|
||||
// createAny isn't done often.
|
||||
mapping, err := restMappingFromUnstructuredObj(tCtx, obj)
|
||||
if err != nil {
|
||||
tCtx.Fatalf("%s: %v", c.TemplatePath, err)
|
||||
}
|
||||
resourceClient := tCtx.Dynamic().Resource(mapping.Resource)
|
||||
|
||||
create := func() error {
|
||||
options := metav1.CreateOptions{
|
||||
// If the YAML input is invalid, then we want the
|
||||
// apiserver to tell us via an error. This can
|
||||
// happen because decoding into an unstructured object
|
||||
// doesn't validate.
|
||||
FieldValidation: "Strict",
|
||||
}
|
||||
if c.Namespace != "" {
|
||||
if mapping.Scope.Name() != meta.RESTScopeNameNamespace {
|
||||
return fmt.Errorf("namespace %q set for %q, but %q has scope %q", c.Namespace, c.TemplatePath, mapping.GroupVersionKind, mapping.Scope.Name())
|
||||
}
|
||||
_, err = resourceClient.Namespace(c.Namespace).Create(tCtx, obj, options)
|
||||
} else {
|
||||
if mapping.Scope.Name() != meta.RESTScopeNameRoot {
|
||||
return fmt.Errorf("namespace not set for %q, but %q has scope %q", c.TemplatePath, mapping.GroupVersionKind, mapping.Scope.Name())
|
||||
}
|
||||
_, err = resourceClient.Create(tCtx, obj, options)
|
||||
}
|
||||
return err
|
||||
}
|
||||
// Retry, some errors (like CRD just created and type not ready for use yet) are temporary.
|
||||
ctx, cancel := context.WithTimeout(tCtx, 20*time.Second)
|
||||
defer cancel()
|
||||
for {
|
||||
err := create()
|
||||
if err == nil {
|
||||
return
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
tCtx.Fatalf("%s: timed out (%q) while creating %q, last error was: %v", c.TemplatePath, context.Cause(ctx), klog.KObj(obj), err)
|
||||
case <-time.After(time.Second):
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func getSpecFromTextTemplateFile(path string, env map[string]any, spec interface{}) error {
|
||||
content, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
tmpl, err := template.New("object").Funcs(getTemplateFuncs()).Parse(string(content))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var buffer bytes.Buffer
|
||||
if err := tmpl.Execute(&buffer, env); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return yaml.UnmarshalStrict(buffer.Bytes(), spec)
|
||||
}
|
||||
|
||||
func restMappingFromUnstructuredObj(tCtx ktesting.TContext, obj *unstructured.Unstructured) (*meta.RESTMapping, error) {
|
||||
discoveryCache := memory.NewMemCacheClient(tCtx.Client().Discovery())
|
||||
restMapper := restmapper.NewDeferredDiscoveryRESTMapper(discoveryCache)
|
||||
gv, err := schema.ParseGroupVersion(obj.GetAPIVersion())
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("extract group+version from object %q: %w", klog.KObj(obj), err)
|
||||
}
|
||||
gk := schema.GroupKind{Group: gv.Group, Kind: obj.GetKind()}
|
||||
|
||||
mapping, err := restMapper.RESTMapping(gk, gv.Version)
|
||||
if err != nil {
|
||||
// Cached mapping might be stale, refresh on next try.
|
||||
restMapper.Reset()
|
||||
return nil, fmt.Errorf("failed mapping %q to resource: %w", gk, err)
|
||||
}
|
||||
return mapping, nil
|
||||
}
|
||||
|
||||
// createNodesOp defines an op where nodes are created as a part of a workload.
|
||||
type createNodesOp struct {
|
||||
// Must be "createNodes".
|
||||
Opcode operationCode
|
||||
// Number of nodes to create. Parameterizable through CountParam.
|
||||
Count int
|
||||
// Template parameter for Count.
|
||||
CountParam string
|
||||
// Path to spec file describing the nodes to create.
|
||||
// Optional
|
||||
NodeTemplatePath *string
|
||||
// At most one of the following strategies can be defined. Defaults
|
||||
// to TrivialNodePrepareStrategy if unspecified.
|
||||
// Optional
|
||||
NodeAllocatableStrategy *testutils.NodeAllocatableStrategy
|
||||
LabelNodePrepareStrategy *testutils.LabelNodePrepareStrategy
|
||||
UniqueNodeLabelStrategy *testutils.UniqueNodeLabelStrategy
|
||||
}
|
||||
|
||||
func (cno *createNodesOp) isValid(allowParameterization bool) error {
|
||||
if !isValidCount(allowParameterization, cno.Count, cno.CountParam) {
|
||||
return fmt.Errorf("invalid Count=%d / CountParam=%q", cno.Count, cno.CountParam)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (*createNodesOp) collectsMetrics() bool {
|
||||
return false
|
||||
}
|
||||
|
||||
func (cno createNodesOp) patchParams(w *workload) (realOp, error) {
|
||||
if cno.CountParam != "" {
|
||||
var err error
|
||||
cno.Count, err = w.Params.get(cno.CountParam[1:])
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return &cno, (&cno).isValid(false)
|
||||
}
|
||||
|
||||
// createNamespacesOp defines an op for creating namespaces
|
||||
type createNamespacesOp struct {
|
||||
// Must be "createNamespaces".
|
||||
Opcode operationCode
|
||||
// Name prefix of the Namespace. The format is "<prefix>-<number>", where number is
|
||||
// between 0 and count-1.
|
||||
Prefix string
|
||||
// Number of namespaces to create. Parameterizable through CountParam.
|
||||
Count int
|
||||
// Template parameter for Count. Takes precedence over Count if both set.
|
||||
CountParam string
|
||||
// Path to spec file describing the Namespaces to create.
|
||||
// Optional
|
||||
NamespaceTemplatePath *string
|
||||
}
|
||||
|
||||
func (cmo *createNamespacesOp) isValid(allowParameterization bool) error {
|
||||
if !isValidCount(allowParameterization, cmo.Count, cmo.CountParam) {
|
||||
return fmt.Errorf("invalid Count=%d / CountParam=%q", cmo.Count, cmo.CountParam)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (*createNamespacesOp) collectsMetrics() bool {
|
||||
return false
|
||||
}
|
||||
|
||||
func (cmo createNamespacesOp) patchParams(w *workload) (realOp, error) {
|
||||
if cmo.CountParam != "" {
|
||||
var err error
|
||||
cmo.Count, err = w.Params.get(cmo.CountParam[1:])
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return &cmo, (&cmo).isValid(false)
|
||||
}
|
||||
|
||||
// createPodsOp defines an op where pods are scheduled as a part of a workload.
|
||||
// The test can block on the completion of this op before moving forward or
|
||||
// continue asynchronously.
|
||||
type createPodsOp struct {
|
||||
// Must be "createPods".
|
||||
Opcode operationCode
|
||||
// Number of pods to schedule. Parameterizable through CountParam.
|
||||
Count int
|
||||
// Template parameter for Count.
|
||||
CountParam string
|
||||
// If false, Count pods get created rapidly. This can be used to
|
||||
// measure how quickly the scheduler can fill up a cluster.
|
||||
//
|
||||
// If true, Count pods get created, the operation waits for
|
||||
// a pod to get scheduled, deletes it and then creates another.
|
||||
// This continues until the configured Duration is over.
|
||||
// Metrics collection, if enabled, runs in parallel.
|
||||
//
|
||||
// This mode can be used to measure how the scheduler behaves
|
||||
// in a steady state where the cluster is always at roughly the
|
||||
// same level of utilization. Pods can be created in a separate,
|
||||
// earlier operation to simulate non-empty clusters.
|
||||
//
|
||||
// Note that the operation will delete any scheduled pod in
|
||||
// the namespace, so use different namespaces for pods that
|
||||
// are supposed to be kept running.
|
||||
SteadyState bool
|
||||
// Template parameter for SteadyState.
|
||||
SteadyStateParam string
|
||||
// How long to keep the cluster in a steady state.
|
||||
Duration metav1.Duration
|
||||
// Template parameter for Duration.
|
||||
DurationParam string
|
||||
// Whether to enable metrics collection for this createPodsOp.
|
||||
// Optional. Both CollectMetrics and SkipWaitToCompletion cannot be true at
|
||||
// the same time for a particular createPodsOp.
|
||||
CollectMetrics bool
|
||||
// Namespace the pods should be created in. Defaults to a unique
|
||||
// namespace of the format "namespace-<number>".
|
||||
// Optional
|
||||
Namespace *string
|
||||
// Path to spec file describing the pods to schedule.
|
||||
// If nil, DefaultPodTemplatePath will be used.
|
||||
// Optional
|
||||
PodTemplatePath *string
|
||||
// Whether to wait for all pods in this op to get scheduled.
|
||||
// Defaults to false if not specified.
|
||||
// Optional
|
||||
SkipWaitToCompletion bool
|
||||
// Persistent volume settings for the pods to be scheduled.
|
||||
// Optional
|
||||
PersistentVolumeTemplatePath *string
|
||||
PersistentVolumeClaimTemplatePath *string
|
||||
}
|
||||
|
||||
func (cpo *createPodsOp) isValid(allowParameterization bool) error {
|
||||
if !isValidCount(allowParameterization, cpo.Count, cpo.CountParam) {
|
||||
return fmt.Errorf("invalid Count=%d / CountParam=%q", cpo.Count, cpo.CountParam)
|
||||
}
|
||||
if cpo.CollectMetrics && cpo.SkipWaitToCompletion {
|
||||
// While it's technically possible to achieve this, the additional
|
||||
// complexity is not worth it, especially given that we don't have any
|
||||
// use-cases right now.
|
||||
return fmt.Errorf("collectMetrics and skipWaitToCompletion cannot be true at the same time")
|
||||
}
|
||||
if cpo.SkipWaitToCompletion && cpo.SteadyState {
|
||||
return errors.New("skipWaitToCompletion and steadyState cannot be true at the same time")
|
||||
}
|
||||
if cpo.SteadyState && !allowParameterization && cpo.Duration.Duration <= 0 {
|
||||
return errors.New("when creating pods in a steady state, the test duration must be > 0")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (cpo *createPodsOp) collectsMetrics() bool {
|
||||
return cpo.CollectMetrics
|
||||
}
|
||||
|
||||
func (cpo createPodsOp) patchParams(w *workload) (realOp, error) {
|
||||
if cpo.CountParam != "" {
|
||||
var err error
|
||||
cpo.Count, err = w.Params.get(cpo.CountParam[1:])
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
if cpo.DurationParam != "" {
|
||||
durationStr, err := getParam[string](w.Params, cpo.DurationParam[1:])
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if cpo.Duration.Duration, err = time.ParseDuration(durationStr); err != nil {
|
||||
return nil, fmt.Errorf("parsing duration parameter %s: %w", cpo.DurationParam, err)
|
||||
}
|
||||
}
|
||||
if cpo.SteadyStateParam != "" {
|
||||
var err error
|
||||
cpo.SteadyState, err = getParam[bool](w.Params, cpo.SteadyStateParam[1:])
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return &cpo, (&cpo).isValid(false)
|
||||
}
|
||||
|
||||
// createPodSetsOp defines an op where a set of createPodsOps is created in each unique namespace.
|
||||
type createPodSetsOp struct {
|
||||
// Must be "createPodSets".
|
||||
Opcode operationCode
|
||||
// Number of sets to create.
|
||||
Count int
|
||||
// Template parameter for Count.
|
||||
CountParam string
|
||||
// Each set of pods will be created in a namespace of the form namespacePrefix-<number>,
|
||||
// where number is from 0 to count-1
|
||||
NamespacePrefix string
|
||||
// The template of a createPodsOp.
|
||||
CreatePodsOp createPodsOp
|
||||
}
|
||||
|
||||
func (cpso *createPodSetsOp) isValid(allowParameterization bool) error {
|
||||
if !isValidCount(allowParameterization, cpso.Count, cpso.CountParam) {
|
||||
return fmt.Errorf("invalid Count=%d / CountParam=%q", cpso.Count, cpso.CountParam)
|
||||
}
|
||||
return cpso.CreatePodsOp.isValid(allowParameterization)
|
||||
}
|
||||
|
||||
func (cpso *createPodSetsOp) collectsMetrics() bool {
|
||||
return cpso.CreatePodsOp.CollectMetrics
|
||||
}
|
||||
|
||||
func (cpso createPodSetsOp) patchParams(w *workload) (realOp, error) {
|
||||
if cpso.CountParam != "" {
|
||||
var err error
|
||||
cpso.Count, err = w.Params.get(cpso.CountParam[1:])
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return &cpso, (&cpso).isValid(true)
|
||||
}
|
||||
|
||||
// deletePodsOp defines an op where previously created pods are deleted.
|
||||
// The test can block on the completion of this op before moving forward or
|
||||
// continue asynchronously.
|
||||
type deletePodsOp struct {
|
||||
// Must be "deletePods".
|
||||
Opcode operationCode
|
||||
// Namespace the pods should be deleted from.
|
||||
Namespace string
|
||||
// Labels used to filter the pods to delete.
|
||||
// If empty, it will delete all Pods in the namespace.
|
||||
// Optional.
|
||||
LabelSelector map[string]string
|
||||
// Whether to wait for all pods in this op to be deleted.
|
||||
// Defaults to false if not specified.
|
||||
// Optional
|
||||
SkipWaitToCompletion bool
|
||||
// Number of pods to be deleted per second.
|
||||
// If zero, all pods are deleted at once.
|
||||
// Optional
|
||||
DeletePodsPerSecond int
|
||||
}
|
||||
|
||||
func (dpo *deletePodsOp) isValid(allowParameterization bool) error {
|
||||
if dpo.Opcode != deletePodsOpcode {
|
||||
return fmt.Errorf("invalid opcode %q; expected %q", dpo.Opcode, deletePodsOpcode)
|
||||
}
|
||||
if dpo.DeletePodsPerSecond < 0 {
|
||||
return fmt.Errorf("invalid DeletePodsPerSecond=%d; should be non-negative", dpo.DeletePodsPerSecond)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (dpo *deletePodsOp) collectsMetrics() bool {
|
||||
return false
|
||||
}
|
||||
|
||||
func (dpo deletePodsOp) patchParams(w *workload) (realOp, error) {
|
||||
return &dpo, nil
|
||||
}
|
||||
|
||||
// churnOp defines an op where services are created as a part of a workload.
|
||||
type churnOp struct {
|
||||
// Must be "churnOp".
|
||||
Opcode operationCode
|
||||
// Value must be one of the followings:
|
||||
// - recreate. In this mode, API objects will be created for N cycles, and then
|
||||
// deleted in the next N cycles. N is specified by the "Number" field.
|
||||
// - create. In this mode, API objects will be created (without deletion) until
|
||||
// reaching a threshold - which is specified by the "Number" field.
|
||||
Mode string
|
||||
// Maximum number of API objects to be created.
|
||||
// Defaults to 0, which means unlimited.
|
||||
Number int
|
||||
// Intervals of churning. Defaults to 500 millisecond.
|
||||
IntervalMilliseconds int64
|
||||
// Namespace the churning objects should be created in. Defaults to a unique
|
||||
// namespace of the format "namespace-<number>".
|
||||
// Optional
|
||||
Namespace *string
|
||||
// Path of API spec files.
|
||||
TemplatePaths []string
|
||||
}
|
||||
|
||||
func (co *churnOp) isValid(_ bool) error {
|
||||
if co.Mode != Recreate && co.Mode != Create {
|
||||
return fmt.Errorf("invalid mode: %v. must be one of %v", co.Mode, []string{Recreate, Create})
|
||||
}
|
||||
if co.Number < 0 {
|
||||
return fmt.Errorf("number (%v) cannot be negative", co.Number)
|
||||
}
|
||||
if co.Mode == Recreate && co.Number == 0 {
|
||||
return fmt.Errorf("number cannot be 0 when mode is %v", Recreate)
|
||||
}
|
||||
if len(co.TemplatePaths) == 0 {
|
||||
return fmt.Errorf("at least one template spec file needs to be specified")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (*churnOp) collectsMetrics() bool {
|
||||
return false
|
||||
}
|
||||
|
||||
func (co churnOp) patchParams(w *workload) (realOp, error) {
|
||||
return &co, nil
|
||||
}
|
||||
|
||||
type SchedulingStage string
|
||||
|
||||
const (
|
||||
Scheduled SchedulingStage = "Scheduled"
|
||||
Attempted SchedulingStage = "Attempted"
|
||||
)
|
||||
|
||||
// barrierOp defines an op that can be used to wait until all scheduled pods of
|
||||
// one or many namespaces have been bound to nodes. This is useful when pods
|
||||
// were scheduled with SkipWaitToCompletion set to true.
|
||||
type barrierOp struct {
|
||||
// Must be "barrier".
|
||||
Opcode operationCode
|
||||
// Namespaces to block on. Empty array or not specifying this field signifies
|
||||
// that the barrier should block on all namespaces.
|
||||
Namespaces []string
|
||||
// Labels used to filter the pods to block on.
|
||||
// If empty, it won't filter the labels.
|
||||
// Optional.
|
||||
LabelSelector map[string]string
|
||||
// Determines what stage of pods scheduling the barrier should wait for.
|
||||
// If empty, it is interpreted as "Scheduled".
|
||||
// Optional
|
||||
StageRequirement SchedulingStage
|
||||
}
|
||||
|
||||
func (bo *barrierOp) isValid(allowParameterization bool) error {
|
||||
if bo.StageRequirement != "" && bo.StageRequirement != Scheduled && bo.StageRequirement != Attempted {
|
||||
return fmt.Errorf("invalid StageRequirement %s", bo.StageRequirement)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (*barrierOp) collectsMetrics() bool {
|
||||
return false
|
||||
}
|
||||
|
||||
func (bo barrierOp) patchParams(w *workload) (realOp, error) {
|
||||
if bo.StageRequirement == "" {
|
||||
bo.StageRequirement = Scheduled
|
||||
}
|
||||
return &bo, nil
|
||||
}
|
||||
|
||||
// sleepOp defines an op that can be used to sleep for a specified amount of time.
|
||||
// This is useful in simulating workloads that require some sort of time-based synchronisation.
|
||||
type sleepOp struct {
|
||||
// Must be "sleep".
|
||||
Opcode operationCode
|
||||
// Duration of sleep.
|
||||
Duration metav1.Duration
|
||||
// Template parameter for Duration.
|
||||
DurationParam string
|
||||
}
|
||||
|
||||
func (so *sleepOp) isValid(_ bool) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (so *sleepOp) collectsMetrics() bool {
|
||||
return false
|
||||
}
|
||||
|
||||
func (so sleepOp) patchParams(w *workload) (realOp, error) {
|
||||
if so.DurationParam != "" {
|
||||
durationStr, err := getParam[string](w.Params, so.DurationParam[1:])
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if so.Duration.Duration, err = time.ParseDuration(durationStr); err != nil {
|
||||
return nil, fmt.Errorf("invalid duration parameter %s: %w", so.DurationParam, err)
|
||||
}
|
||||
}
|
||||
return &so, nil
|
||||
}
|
||||
|
||||
// startCollectingMetricsOp defines an op that starts metrics collectors.
|
||||
// stopCollectingMetricsOp has to be used after this op to finish collecting.
|
||||
type startCollectingMetricsOp struct {
|
||||
// Must be "startCollectingMetrics".
|
||||
Opcode operationCode
|
||||
// Name appended to workload's name in results.
|
||||
Name string
|
||||
// Namespaces for which the scheduling throughput metric is calculated.
|
||||
Namespaces []string
|
||||
// Labels used to filter the pods for which the scheduling throughput metric is collected.
|
||||
// If empty, it will collect the metric for all pods in the selected namespaces.
|
||||
// Optional.
|
||||
LabelSelector map[string]string
|
||||
}
|
||||
|
||||
func (scm *startCollectingMetricsOp) isValid(_ bool) error {
|
||||
if len(scm.Namespaces) == 0 {
|
||||
return fmt.Errorf("namespaces cannot be empty")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (*startCollectingMetricsOp) collectsMetrics() bool {
|
||||
return false
|
||||
}
|
||||
|
||||
func (scm startCollectingMetricsOp) patchParams(_ *workload) (realOp, error) {
|
||||
return &scm, nil
|
||||
}
|
||||
|
||||
// stopCollectingMetricsOp defines an op that stops collecting the metrics
|
||||
// and writes them into the result slice.
|
||||
// startCollectingMetricsOp has be used before this op to begin collecting.
|
||||
type stopCollectingMetricsOp struct {
|
||||
// Must be "stopCollectingMetrics".
|
||||
Opcode operationCode
|
||||
}
|
||||
|
||||
func (scm *stopCollectingMetricsOp) isValid(_ bool) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (*stopCollectingMetricsOp) collectsMetrics() bool {
|
||||
return true
|
||||
}
|
||||
|
||||
func (scm stopCollectingMetricsOp) patchParams(_ *workload) (realOp, error) {
|
||||
return &scm, nil
|
||||
}
|
||||
|
||||
func getTemplateFuncs() template.FuncMap {
|
||||
return template.FuncMap{
|
||||
"AddFloat": addFloat,
|
||||
"AddInt": addInt,
|
||||
"DivideFloat": divideFloat,
|
||||
"DivideInt": divideInt,
|
||||
"Mod": mod,
|
||||
"MultiplyFloat": multiplyFloat,
|
||||
"MultiplyInt": multiplyInt,
|
||||
"SubtractFloat": subtractFloat,
|
||||
"SubtractInt": subtractInt,
|
||||
}
|
||||
}
|
||||
|
||||
func toFloat64(val any) float64 {
|
||||
switch i := val.(type) {
|
||||
case float64:
|
||||
return i
|
||||
case float32:
|
||||
return float64(i)
|
||||
case int64:
|
||||
return float64(i)
|
||||
case int32:
|
||||
return float64(i)
|
||||
case int:
|
||||
return float64(i)
|
||||
case uint64:
|
||||
return float64(i)
|
||||
case uint32:
|
||||
return float64(i)
|
||||
case uint:
|
||||
return float64(i)
|
||||
case string:
|
||||
f, err := strconv.ParseFloat(i, 64)
|
||||
if err == nil {
|
||||
return f
|
||||
}
|
||||
}
|
||||
panic(fmt.Sprintf("cannot cast %v to float64", val))
|
||||
}
|
||||
|
||||
func addInt(numbers ...any) int {
|
||||
return int(addFloat(numbers...))
|
||||
}
|
||||
|
||||
func subtractInt(i, j any) int {
|
||||
return int(subtractFloat(i, j))
|
||||
}
|
||||
|
||||
func multiplyInt(numbers ...any) int {
|
||||
return int(multiplyFloat(numbers...))
|
||||
}
|
||||
|
||||
func divideInt(i, j any) int {
|
||||
return int(divideFloat(i, j))
|
||||
}
|
||||
|
||||
func addFloat(numbers ...any) float64 {
|
||||
sum := 0.0
|
||||
for _, number := range numbers {
|
||||
sum += toFloat64(number)
|
||||
}
|
||||
return sum
|
||||
}
|
||||
|
||||
func subtractFloat(i, j any) float64 {
|
||||
typedI := toFloat64(i)
|
||||
typedJ := toFloat64(j)
|
||||
return typedI - typedJ
|
||||
}
|
||||
|
||||
func multiplyFloat(numbers ...any) float64 {
|
||||
product := 1.0
|
||||
for _, number := range numbers {
|
||||
product *= toFloat64(number)
|
||||
}
|
||||
return product
|
||||
}
|
||||
|
||||
func divideFloat(i, j any) float64 {
|
||||
typedI := toFloat64(i)
|
||||
typedJ := toFloat64(j)
|
||||
if typedJ == 0 {
|
||||
panic("division by zero")
|
||||
}
|
||||
return typedI / typedJ
|
||||
}
|
||||
|
||||
func mod(a, b any) int {
|
||||
return int(toFloat64(a)) % int(toFloat64(b))
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
@@ -17,408 +17,12 @@ limitations under the License.
|
||||
package benchmark
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"github.com/google/go-cmp/cmp"
|
||||
"github.com/google/go-cmp/cmp/cmpopts"
|
||||
v1 "k8s.io/api/core/v1"
|
||||
"k8s.io/apimachinery/pkg/api/resource"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/util/sets"
|
||||
coreinformers "k8s.io/client-go/informers/core/v1"
|
||||
"k8s.io/client-go/kubernetes/fake"
|
||||
"k8s.io/component-base/featuregate"
|
||||
testutils "k8s.io/kubernetes/test/utils"
|
||||
ktesting "k8s.io/kubernetes/test/utils/ktesting"
|
||||
"k8s.io/utils/ptr"
|
||||
)
|
||||
|
||||
type verifyFunc func(t *testing.T, tCtx ktesting.TContext, op realOp) error
|
||||
|
||||
func TestRunOp(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
op realOp
|
||||
expectedFailure bool
|
||||
verifyFuncs []verifyFunc
|
||||
}{
|
||||
{
|
||||
name: "Create Single Node",
|
||||
op: &createNodesOp{
|
||||
Opcode: createNodesOpcode,
|
||||
Count: 1,
|
||||
},
|
||||
verifyFuncs: []verifyFunc{
|
||||
verifyCount(1),
|
||||
verifyObj(
|
||||
&v1.Node{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
GenerateName: "node-0-",
|
||||
},
|
||||
Status: v1.NodeStatus{
|
||||
Capacity: v1.ResourceList{
|
||||
v1.ResourceCPU: resource.MustParse("4"),
|
||||
v1.ResourceMemory: resource.MustParse("32Gi"),
|
||||
v1.ResourcePods: resource.MustParse("110"),
|
||||
},
|
||||
Allocatable: v1.ResourceList{
|
||||
v1.ResourceCPU: resource.MustParse("4"),
|
||||
v1.ResourceMemory: resource.MustParse("32Gi"),
|
||||
v1.ResourcePods: resource.MustParse("110"),
|
||||
},
|
||||
},
|
||||
}),
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "Create Multiple Nodes",
|
||||
op: &createNodesOp{
|
||||
Opcode: createNodesOpcode,
|
||||
Count: 5,
|
||||
},
|
||||
verifyFuncs: []verifyFunc{
|
||||
verifyCount(5),
|
||||
verifyObj(
|
||||
&v1.Node{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
GenerateName: "node-0-",
|
||||
},
|
||||
Status: v1.NodeStatus{
|
||||
Capacity: v1.ResourceList{
|
||||
v1.ResourceCPU: resource.MustParse("4"),
|
||||
v1.ResourceMemory: resource.MustParse("32Gi"),
|
||||
v1.ResourcePods: resource.MustParse("110"),
|
||||
},
|
||||
Allocatable: v1.ResourceList{
|
||||
v1.ResourceCPU: resource.MustParse("4"),
|
||||
v1.ResourceMemory: resource.MustParse("32Gi"),
|
||||
v1.ResourcePods: resource.MustParse("110"),
|
||||
},
|
||||
},
|
||||
}),
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "Create Nodes with Label Strategy",
|
||||
op: &createNodesOp{
|
||||
Opcode: createNodesOpcode,
|
||||
Count: 3,
|
||||
LabelNodePrepareStrategy: testutils.NewLabelNodePrepareStrategy("test-label", "value1", "value2", "value3"),
|
||||
},
|
||||
verifyFuncs: []verifyFunc{
|
||||
verifyCount(3),
|
||||
verifyLabelValuesAllowed("test-label", sets.New("value1", "value2", "value3")),
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "Create Nodes with Unique Label Strategy",
|
||||
op: &createNodesOp{
|
||||
Opcode: createNodesOpcode,
|
||||
Count: 2,
|
||||
UniqueNodeLabelStrategy: testutils.NewUniqueNodeLabelStrategy("unique-test-label"),
|
||||
},
|
||||
verifyFuncs: []verifyFunc{
|
||||
verifyCount(2),
|
||||
verifyUniqueLabelValues("unique-test-label"),
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "Create Nodes with Node Allocatable Strategy",
|
||||
op: &createNodesOp{
|
||||
Opcode: createNodesOpcode,
|
||||
Count: 2,
|
||||
NodeAllocatableStrategy: testutils.NewNodeAllocatableStrategy(
|
||||
map[v1.ResourceName]string{
|
||||
v1.ResourceCPU: "2",
|
||||
v1.ResourceMemory: "4Gi",
|
||||
},
|
||||
nil, // no CSI node allocatable
|
||||
nil, // no migrated plugins
|
||||
),
|
||||
},
|
||||
verifyFuncs: []verifyFunc{
|
||||
verifyCount(2),
|
||||
verifyObj(
|
||||
&v1.Node{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
GenerateName: "node-0-",
|
||||
},
|
||||
Status: v1.NodeStatus{
|
||||
Capacity: v1.ResourceList{
|
||||
v1.ResourceCPU: resource.MustParse("4"),
|
||||
v1.ResourceMemory: resource.MustParse("32Gi"),
|
||||
v1.ResourcePods: resource.MustParse("110"),
|
||||
},
|
||||
Allocatable: v1.ResourceList{
|
||||
v1.ResourceCPU: resource.MustParse("2"),
|
||||
v1.ResourceMemory: resource.MustParse("4Gi"),
|
||||
v1.ResourcePods: resource.MustParse("110"),
|
||||
},
|
||||
},
|
||||
}),
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "Create Nodes with Custom Template",
|
||||
op: &createNodesOp{
|
||||
Opcode: createNodesOpcode,
|
||||
Count: 2,
|
||||
NodeTemplatePath: createObjTemplateFile(t,
|
||||
&v1.Node{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
GenerateName: "custom-node-",
|
||||
},
|
||||
Status: v1.NodeStatus{
|
||||
Capacity: v1.ResourceList{
|
||||
v1.ResourcePods: resource.MustParse("100"),
|
||||
v1.ResourceCPU: resource.MustParse("4"),
|
||||
v1.ResourceMemory: resource.MustParse("8Gi"),
|
||||
},
|
||||
},
|
||||
},
|
||||
),
|
||||
},
|
||||
verifyFuncs: []verifyFunc{
|
||||
verifyCount(2),
|
||||
verifyObj(
|
||||
&v1.Node{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
GenerateName: "custom-node-",
|
||||
},
|
||||
Status: v1.NodeStatus{
|
||||
Capacity: v1.ResourceList{
|
||||
v1.ResourcePods: resource.MustParse("100"),
|
||||
v1.ResourceCPU: resource.MustParse("4"),
|
||||
v1.ResourceMemory: resource.MustParse("8Gi"),
|
||||
},
|
||||
},
|
||||
},
|
||||
),
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "Invalid Node Template Path",
|
||||
op: &createNodesOp{
|
||||
Opcode: createNodesOpcode,
|
||||
Count: 1,
|
||||
NodeTemplatePath: ptr.To("non-existent-file.json"),
|
||||
},
|
||||
expectedFailure: true,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
_, tCtx := ktesting.NewTestContext(t)
|
||||
client := fake.NewSimpleClientset()
|
||||
tCtx = tCtx.WithClients(nil, nil, client, nil, nil)
|
||||
|
||||
exec := &WorkloadExecutor{
|
||||
tCtx: tCtx,
|
||||
numPodsScheduledPerNamespace: make(map[string]int),
|
||||
nextNodeIndex: 0,
|
||||
}
|
||||
|
||||
err := exec.runOp(tt.op, 0)
|
||||
|
||||
if tt.expectedFailure {
|
||||
if err == nil {
|
||||
t.Fatalf("Expected error but got none")
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to run operation: %v", err)
|
||||
}
|
||||
|
||||
if tt.verifyFuncs != nil {
|
||||
for i, vf := range tt.verifyFuncs {
|
||||
if err := vf(t, tCtx, tt.op); err != nil {
|
||||
t.Fatalf("Verification function %d failed for test %q: %v", i, tt.name, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// verifyCount returns a verification function that checks if the number of existing objects
|
||||
// matches the expected count based on the operation type.
|
||||
func verifyCount(expectedCount int) verifyFunc {
|
||||
return func(t *testing.T, tCtx ktesting.TContext, op realOp) error {
|
||||
switch op.(type) {
|
||||
case *createNodesOp:
|
||||
nodes, err := tCtx.Client().CoreV1().Nodes().List(context.Background(), metav1.ListOptions{})
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to list nodes: %w", err)
|
||||
}
|
||||
if got := len(nodes.Items); got != expectedCount {
|
||||
return fmt.Errorf("unexpected node count: got %d, want %d", got, expectedCount)
|
||||
}
|
||||
default:
|
||||
return fmt.Errorf("verifyCount doesn't support this operation type: %T", op)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// verifyLabelValuesAllowed returns a verification function that checks if the label values for a given key.
|
||||
func verifyLabelValuesAllowed(key string, allowValues sets.Set[string]) verifyFunc {
|
||||
return func(t *testing.T, tCtx ktesting.TContext, op realOp) error {
|
||||
labelValues, _, err := objLabelValues(t, tCtx, op, key)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to get label values: %w", err)
|
||||
}
|
||||
|
||||
for labelValue := range labelValues {
|
||||
if !allowValues.Has(labelValue) {
|
||||
return fmt.Errorf("Node has unexpected label value %s for key %s", labelValue, key)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// verifyUniqueLabelValues returns a verification function that checks if the label values for a given key are unique.
|
||||
func verifyUniqueLabelValues(key string) verifyFunc {
|
||||
return func(t *testing.T, tCtx ktesting.TContext, op realOp) error {
|
||||
_, duplicatedValues, err := objLabelValues(t, tCtx, op, key)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to get label values: %w", err)
|
||||
}
|
||||
|
||||
if duplicatedValues.Len() > 0 {
|
||||
return fmt.Errorf("Node has duplicate label values %v for key %s", duplicatedValues.UnsortedList(), key)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// objLabelValues is a helper function to extract label values from the listed objects.
|
||||
// The listed objects are dependent on the operation type.
|
||||
// It returns two sets: one with deduplicated labelValues and second with duplicated labels.
|
||||
func objLabelValues(t *testing.T, tCtx ktesting.TContext, op realOp, key string) (sets.Set[string], sets.Set[string], error) {
|
||||
t.Helper()
|
||||
|
||||
labelValues := sets.New[string]()
|
||||
duplicatedValues := sets.New[string]()
|
||||
|
||||
switch op.(type) {
|
||||
case *createNodesOp:
|
||||
nodes, err := tCtx.Client().CoreV1().Nodes().List(context.Background(), metav1.ListOptions{})
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("failed to list nodes for label verification: %w", err)
|
||||
}
|
||||
|
||||
for _, node := range nodes.Items {
|
||||
if labelValue, exists := node.Labels[key]; exists {
|
||||
if labelValues.Has(labelValue) {
|
||||
duplicatedValues.Insert(labelValue)
|
||||
}
|
||||
labelValues.Insert(labelValue)
|
||||
} else {
|
||||
return nil, nil, fmt.Errorf("Node %s is missing expected label %s", node.Name, key)
|
||||
}
|
||||
}
|
||||
default:
|
||||
return nil, nil, fmt.Errorf("verifyLabel doesn't support this operation type: %T", op)
|
||||
}
|
||||
|
||||
return labelValues, duplicatedValues, nil
|
||||
}
|
||||
|
||||
// verifyObj checks if listed objects match the expected template object using cmp.Diff.
|
||||
func verifyObj(expectedObj any) verifyFunc {
|
||||
return func(t *testing.T, tCtx ktesting.TContext, op realOp) error {
|
||||
var got, want any
|
||||
var cmpOpts []cmp.Option
|
||||
switch opDetails := op.(type) {
|
||||
case *createNodesOp:
|
||||
nodesList, listErr := tCtx.Client().CoreV1().Nodes().List(context.Background(), metav1.ListOptions{})
|
||||
if listErr != nil {
|
||||
return fmt.Errorf("failed to list nodes: %w", listErr)
|
||||
}
|
||||
gotNodes := nodesList.Items
|
||||
|
||||
expectedNodeTemplate, ok := expectedObj.(*v1.Node)
|
||||
if !ok {
|
||||
return fmt.Errorf("expectedObj must be *v1.Node when op is *createNodesOp, got %T", expectedObj)
|
||||
}
|
||||
|
||||
wantNodes := make([]v1.Node, len(gotNodes))
|
||||
// we don't need to verify len(), we just need to check if all of them are the same as the expected one.
|
||||
for i := range gotNodes {
|
||||
wantNodes[i] = *expectedNodeTemplate
|
||||
}
|
||||
|
||||
cmpOpts = []cmp.Option{
|
||||
cmpopts.EquateEmpty(),
|
||||
cmpopts.IgnoreFields(metav1.ObjectMeta{},
|
||||
"UID", "ResourceVersion", "Generation", "CreationTimestamp", "ManagedFields", "SelfLink", "Name",
|
||||
"Labels", // verifyObj doesn't care about labels.
|
||||
),
|
||||
cmpopts.IgnoreFields(v1.NodeStatus{}, // This test isn't interested in these fields.
|
||||
"Conditions",
|
||||
"Phase",
|
||||
),
|
||||
}
|
||||
got = gotNodes
|
||||
want = wantNodes
|
||||
default:
|
||||
return fmt.Errorf("verifyObj doesn't support this operation type for cmp.Diff: %T", opDetails)
|
||||
}
|
||||
|
||||
if diff := cmp.Diff(want, got, cmpOpts...); diff != "" {
|
||||
return fmt.Errorf("unexpected difference (-want +got):\\n%s", diff)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// createObjTemplateFile creates a temporary file with the given object serialized as JSON.
|
||||
func createObjTemplateFile(t *testing.T, obj any) *string {
|
||||
t.Helper()
|
||||
|
||||
dir, err := os.MkdirTemp("", "scheduler-perf-test")
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to create temp dir for the template: %v", err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
if err := os.RemoveAll(dir); err != nil {
|
||||
t.Errorf("Failed to remove temp dir: %v", err)
|
||||
}
|
||||
})
|
||||
|
||||
templateFile := filepath.Join(dir, "template.json")
|
||||
f, err := os.Create(templateFile)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to create the template file %s: %v", templateFile, err)
|
||||
}
|
||||
defer func() {
|
||||
if err := f.Close(); err != nil {
|
||||
t.Errorf("Failed to close file: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
switch obj := obj.(type) {
|
||||
case *v1.Node:
|
||||
if err := json.NewEncoder(f).Encode(obj); err != nil {
|
||||
t.Fatalf("Failed to encode the template to %s: %v", templateFile, err)
|
||||
}
|
||||
default:
|
||||
t.Fatalf("Unsupported object type for template file: %T", obj)
|
||||
}
|
||||
return &templateFile
|
||||
}
|
||||
|
||||
func TestFeatureGatesMerge(t *testing.T) {
|
||||
const (
|
||||
FeatureA featuregate.Feature = "FeatureA"
|
||||
@@ -741,260 +345,3 @@ func TestApplyThreshold(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// mockDataCollector always returns the same data items, to be used for mocking data collector in unit tests.
|
||||
type mockDataCollector struct {
|
||||
dataItems []DataItem
|
||||
}
|
||||
|
||||
// init does nothing.
|
||||
func (mc *mockDataCollector) init() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// run does nothing.
|
||||
func (mc *mockDataCollector) run(_ ktesting.TContext) {}
|
||||
|
||||
// collect always returns DataItems defined in the collector.
|
||||
func (mc *mockDataCollector) collect() []DataItem {
|
||||
return mc.dataItems
|
||||
}
|
||||
|
||||
func TestMetricThreshold(t *testing.T) {
|
||||
testCases := []struct {
|
||||
name string
|
||||
thresholdValue float64
|
||||
dataItems []DataItem
|
||||
thresholdMetricSelector *thresholdMetricSelector
|
||||
expectCollectionFailure bool
|
||||
expectedDataItemsWithThresholdIndices []int
|
||||
expectedThresholdName string
|
||||
}{
|
||||
{
|
||||
name: "value is above threshold, no error",
|
||||
thresholdValue: 100,
|
||||
dataItems: []DataItem{
|
||||
{
|
||||
Data: map[string]float64{
|
||||
"Average": 150,
|
||||
},
|
||||
Labels: map[string]string{
|
||||
"Metric": "throughput",
|
||||
},
|
||||
},
|
||||
},
|
||||
thresholdMetricSelector: &thresholdMetricSelector{
|
||||
Name: "throughput",
|
||||
DataBucket: "Average",
|
||||
},
|
||||
expectedDataItemsWithThresholdIndices: []int{0},
|
||||
expectedThresholdName: "AverageThreshold",
|
||||
},
|
||||
{
|
||||
name: "value is below threshold, expect error",
|
||||
thresholdValue: 100,
|
||||
dataItems: []DataItem{
|
||||
{
|
||||
Data: map[string]float64{
|
||||
"Average": 70,
|
||||
"Max": 90,
|
||||
},
|
||||
Labels: map[string]string{
|
||||
"Metric": "throughput",
|
||||
},
|
||||
},
|
||||
},
|
||||
thresholdMetricSelector: &thresholdMetricSelector{
|
||||
Name: "throughput",
|
||||
DataBucket: "Max",
|
||||
},
|
||||
expectCollectionFailure: true,
|
||||
expectedDataItemsWithThresholdIndices: []int{0},
|
||||
expectedThresholdName: "MaxThreshold",
|
||||
},
|
||||
{
|
||||
name: "no error if the labels do not match",
|
||||
thresholdValue: 100,
|
||||
dataItems: []DataItem{
|
||||
{
|
||||
Data: map[string]float64{
|
||||
"Average": 70,
|
||||
},
|
||||
Labels: map[string]string{
|
||||
"Metric": "throughput",
|
||||
"label": "value",
|
||||
},
|
||||
},
|
||||
},
|
||||
thresholdMetricSelector: &thresholdMetricSelector{
|
||||
Name: "throughput",
|
||||
DataBucket: "Average",
|
||||
Labels: map[string]string{
|
||||
"label": "value2",
|
||||
},
|
||||
},
|
||||
expectedDataItemsWithThresholdIndices: []int{},
|
||||
expectedThresholdName: "AverageThreshold",
|
||||
},
|
||||
{
|
||||
name: "out of multiple data items only matching are selected",
|
||||
thresholdValue: 100,
|
||||
dataItems: []DataItem{
|
||||
{
|
||||
Data: map[string]float64{
|
||||
"Average": 70,
|
||||
},
|
||||
Labels: map[string]string{
|
||||
"Metric": "throughput",
|
||||
"label": "value",
|
||||
},
|
||||
},
|
||||
{
|
||||
Data: map[string]float64{
|
||||
"Average": 150,
|
||||
},
|
||||
Labels: map[string]string{
|
||||
"Metric": "throughput",
|
||||
"label": "value2",
|
||||
},
|
||||
},
|
||||
},
|
||||
thresholdMetricSelector: &thresholdMetricSelector{
|
||||
Name: "throughput",
|
||||
DataBucket: "Average",
|
||||
Labels: map[string]string{
|
||||
"label": "value2",
|
||||
},
|
||||
},
|
||||
expectedDataItemsWithThresholdIndices: []int{1},
|
||||
expectedThresholdName: "AverageThreshold",
|
||||
},
|
||||
{
|
||||
name: "threshold value is added for all matching entries",
|
||||
thresholdValue: 100,
|
||||
dataItems: []DataItem{
|
||||
{
|
||||
Data: map[string]float64{
|
||||
"Average": 130,
|
||||
},
|
||||
Labels: map[string]string{
|
||||
"Metric": "throughput",
|
||||
"label": "value",
|
||||
},
|
||||
},
|
||||
{
|
||||
Data: map[string]float64{
|
||||
"Average": 150,
|
||||
},
|
||||
Labels: map[string]string{
|
||||
"Metric": "throughput",
|
||||
"label": "value2",
|
||||
},
|
||||
},
|
||||
},
|
||||
thresholdMetricSelector: &thresholdMetricSelector{
|
||||
Name: "throughput",
|
||||
DataBucket: "Average",
|
||||
},
|
||||
expectedDataItemsWithThresholdIndices: []int{0, 1},
|
||||
expectedThresholdName: "AverageThreshold",
|
||||
},
|
||||
{
|
||||
name: "threshold value is added for all matching entries even with error",
|
||||
thresholdValue: 100,
|
||||
dataItems: []DataItem{
|
||||
{
|
||||
Data: map[string]float64{
|
||||
"Average": 70,
|
||||
},
|
||||
Labels: map[string]string{
|
||||
"Metric": "throughput",
|
||||
"label": "value",
|
||||
},
|
||||
},
|
||||
{
|
||||
Data: map[string]float64{
|
||||
"Average": 80,
|
||||
},
|
||||
Labels: map[string]string{
|
||||
"Metric": "throughput",
|
||||
"label": "value2",
|
||||
},
|
||||
},
|
||||
{
|
||||
Data: map[string]float64{
|
||||
"Average": 130,
|
||||
},
|
||||
Labels: map[string]string{
|
||||
"Metric": "throughput",
|
||||
"label": "value3",
|
||||
},
|
||||
},
|
||||
},
|
||||
thresholdMetricSelector: &thresholdMetricSelector{
|
||||
Name: "throughput",
|
||||
DataBucket: "Average",
|
||||
},
|
||||
expectCollectionFailure: true,
|
||||
expectedDataItemsWithThresholdIndices: []int{0, 1, 2},
|
||||
expectedThresholdName: "AverageThreshold",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range testCases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
_, tCtx := ktesting.NewTestContext(t)
|
||||
var capturedErr error
|
||||
capturingCtx, finalize := tCtx.WithError(&capturedErr)
|
||||
defer finalize()
|
||||
|
||||
originalGetTestDataCollectors := getTestDataCollectors
|
||||
defer func() { getTestDataCollectors = originalGetTestDataCollectors }()
|
||||
getTestDataCollectors = func(_ coreinformers.PodInformer, _ string, _ []string, _ map[string]string, _ *metricsCollectorConfig, _ float64) []testDataCollector {
|
||||
return []testDataCollector{&mockDataCollector{dataItems: tc.dataItems}}
|
||||
}
|
||||
|
||||
workload := &workload{
|
||||
Name: "some/workload",
|
||||
Threshold: thresholds{
|
||||
valuesByTopic: map[string]float64{"example": tc.thresholdValue},
|
||||
},
|
||||
ThresholdMetricSelector: tc.thresholdMetricSelector,
|
||||
}
|
||||
exec := &WorkloadExecutor{
|
||||
topicName: "example",
|
||||
testCase: &testCase{},
|
||||
tCtx: capturingCtx,
|
||||
numPodsScheduledPerNamespace: make(map[string]int),
|
||||
workload: workload,
|
||||
}
|
||||
|
||||
start := &startCollectingMetricsOp{
|
||||
Opcode: startCollectingMetricsOpcode,
|
||||
Name: "test-collection",
|
||||
Namespaces: []string{"test-namespaces"},
|
||||
}
|
||||
err := exec.runOp(start, 0)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to start metric collection")
|
||||
}
|
||||
stop := &stopCollectingMetricsOp{Opcode: stopCollectingMetricsOpcode}
|
||||
err = exec.runOp(stop, 0)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to stop metric collection")
|
||||
}
|
||||
|
||||
if tc.expectCollectionFailure != capturingCtx.Failed() {
|
||||
t.Fatalf("expectCollectionFailure=%v but got %v", tc.expectCollectionFailure, capturingCtx.Failed())
|
||||
}
|
||||
for _, idx := range tc.expectedDataItemsWithThresholdIndices {
|
||||
if idx >= len(exec.dataItems) {
|
||||
t.Fatalf("expectedDataItemsWithThresholdIndex out of data items range")
|
||||
}
|
||||
if _, ok := exec.dataItems[idx].Data[tc.expectedThresholdName]; !ok {
|
||||
t.Fatalf("expected data item at index=%d to have %s field", idx, tc.expectedThresholdName)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user