github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2026-07-21 00:33:15 +00:00
Code Issues Projects Releases Wiki Activity

scheduler_perf: block after creating ResourceSlices

Browse Source 
After creating ResourceSlices, the workload was allowed to proceed even while
the scheduler was still busy receiving those new ResourceSlices. This blurred
the line between "setup" and "measurement" phase of DRA workloads. It's not
immediately clear how much that affected results, but it is cleaner to block.

This is done by returning the scheduler instance to the main scheduler_perf
loop and then pass the SharedDRAManager into the driver setup operation. There
it can be used to poll until that manager has processed all ResourceSlices.
This commit is contained in:
Patrick Ohly
2025-09-08 14:14:39 +02:00
parent 597a684bb0
commit 8ff5cec261
3 changed files with 29 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
test/integration/scheduler_perf/dra.go
View File

@@ -23,6 +23,7 @@ import (
"reflect"
"sync"
"github.com/onsi/gomega"
"github.com/stretchr/testify/require"
v1 "k8s.io/api/core/v1"
@@ -38,6 +39,7 @@ import (
"k8s.io/dynamic-resource-allocation/cel"
resourceslicetracker "k8s.io/dynamic-resource-allocation/resourceslice/tracker"
"k8s.io/dynamic-resource-allocation/structured"
"k8s.io/kube-scheduler/framework"
"k8s.io/kubernetes/pkg/features"
"k8s.io/kubernetes/pkg/scheduler/framework/plugins/dynamicresources"
"k8s.io/kubernetes/pkg/scheduler/util/assumecache"
@@ -141,7 +143,6 @@ type createResourceDriverOp struct {
}
var _ realOp = &createResourceDriverOp{}
var _ runnableOp = &createResourceDriverOp{}
func (op *createResourceDriverOp) isValid(allowParameterization bool) error {
if !isValidCount(allowParameterization, op.MaxClaimsPerNode, op.MaxClaimsPerNodeParam) {
@@ -170,9 +171,7 @@ func (op *createResourceDriverOp) patchParams(w *workload) (realOp, error) {
return op, op.isValid(false)
}
func (op *createResourceDriverOp) requiredNamespaces() []string { return nil }
func (op *createResourceDriverOp) run(tCtx ktesting.TContext) {
func (op *createResourceDriverOp) run(tCtx ktesting.TContext, draManager framework.SharedDRAManager) {
tCtx.Logf("creating resource driver %q for nodes matching %q", op.DriverName, op.Nodes)
var driverNodes []string
@@ -191,11 +190,20 @@ func (op *createResourceDriverOp) run(tCtx ktesting.TContext) {
}
}
numSlices := 0
for _, nodeName := range driverNodes {
slice := resourceSlice(op.DriverName, nodeName, op.MaxClaimsPerNode)
_, err := tCtx.Client().ResourceV1().ResourceSlices().Create(tCtx, slice, metav1.CreateOptions{})
tCtx.ExpectNoError(err, "create node resource slice")
numSlices++
}
ktesting.Eventually(tCtx, func(tCtx ktesting.TContext) int {
slices, err := draManager.ResourceSlices().ListWithDeviceTaintRules()
tCtx.ExpectNoError(err, "list ResourceSlices")
return len(slices)
}).Should(gomega.Equal(numSlices), "informer has received all ResourceSlices")
tCtx.CleanupCtx(func(tCtx ktesting.TContext) {
err := tCtx.Client().ResourceV1().ResourceSlices().DeleteCollection(tCtx,
metav1.DeleteOptions{},

20
test/integration/scheduler_perf/scheduler_perf.go
View File

@@ -61,6 +61,7 @@ import (
"k8s.io/component-base/metrics/testutil"
"k8s.io/klog/v2"
"k8s.io/kubernetes/pkg/features"
"k8s.io/kubernetes/pkg/scheduler"
"k8s.io/kubernetes/pkg/scheduler/apis/config"
"k8s.io/kubernetes/pkg/scheduler/apis/config/scheme"
"k8s.io/kubernetes/pkg/scheduler/apis/config/validation"
@@ -1020,7 +1021,7 @@ func initTestOutput(tb testing.TB) io.Writer {
var specialFilenameChars = regexp.MustCompile(`[^a-zA-Z0-9-_]`)
func setupTestCase(t testing.TB, tc *testCase, featureGates map[featuregate.Feature]bool, outOfTreePluginRegistry frameworkruntime.Registry) (informers.SharedInformerFactory, ktesting.TContext) {
func setupTestCase(t testing.TB, tc *testCase, featureGates map[featuregate.Feature]bool, outOfTreePluginRegistry frameworkruntime.Registry) (*scheduler.Scheduler, informers.SharedInformerFactory, ktesting.TContext) {
tCtx := ktesting.Init(t, initoption.PerTestOutput(UseTestingLog))
artifacts, doArtifacts := os.LookupEnv("ARTIFACTS")
if !UseTestingLog && doArtifacts {
@@ -1205,7 +1206,7 @@ func RunBenchmarkPerfScheduling(b *testing.B, configFile string, topicName strin
fixJSONOutput(b)
featureGates := featureGatesMerge(tc.FeatureGates, w.FeatureGates)
informerFactory, tCtx := setupTestCase(b, tc, featureGates, outOfTreePluginRegistry)
scheduler, informerFactory, tCtx := setupTestCase(b, tc, featureGates, outOfTreePluginRegistry)
err := w.isValid(tc.MetricsCollectorConfig)
if err != nil {
@@ -1219,7 +1220,7 @@ func RunBenchmarkPerfScheduling(b *testing.B, configFile string, topicName strin
}
}
results, err := runWorkload(tCtx, tc, w, informerFactory)
results, err := runWorkload(tCtx, tc, w, scheduler, informerFactory)
if err != nil {
tCtx.Fatalf("Error running workload %s: %s", w.Name, err)
}
@@ -1314,13 +1315,13 @@ func RunIntegrationPerfScheduling(t *testing.T, configFile string) {
t.Skipf("disabled by label filter %q", TestSchedulingLabelFilter)
}
featureGates := featureGatesMerge(tc.FeatureGates, w.FeatureGates)
informerFactory, tCtx := setupTestCase(t, tc, featureGates, nil)
scheduler, informerFactory, tCtx := setupTestCase(t, tc, featureGates, nil)
err := w.isValid(tc.MetricsCollectorConfig)
if err != nil {
t.Fatalf("workload %s is not valid: %v", w.Name, err)
}
_, err = runWorkload(tCtx, tc, w, informerFactory)
_, err = runWorkload(tCtx, tc, w, scheduler, informerFactory)
if err != nil {
tCtx.Fatalf("Error running workload %s: %s", w.Name, err)
}
@@ -1380,7 +1381,7 @@ func unrollWorkloadTemplate(tb ktesting.TB, wt []op, w *workload) []op {
return unrolled
}
func setupClusterForWorkload(tCtx ktesting.TContext, configPath string, featureGates map[featuregate.Feature]bool, outOfTreePluginRegistry frameworkruntime.Registry) (informers.SharedInformerFactory, ktesting.TContext) {
func setupClusterForWorkload(tCtx ktesting.TContext, configPath string, featureGates map[featuregate.Feature]bool, outOfTreePluginRegistry frameworkruntime.Registry) (*scheduler.Scheduler, informers.SharedInformerFactory, ktesting.TContext) {
var cfg *config.KubeSchedulerConfiguration
var err error
if configPath != "" {
@@ -1486,6 +1487,7 @@ func stopCollectingMetrics(tCtx ktesting.TContext, collectorCtx ktesting.TContex
type WorkloadExecutor struct {
tCtx ktesting.TContext
scheduler *scheduler.Scheduler
wg sync.WaitGroup
collectorCtx ktesting.TContext
collectorWG sync.WaitGroup
@@ -1499,7 +1501,7 @@ type WorkloadExecutor struct {
nextNodeIndex int
}
func runWorkload(tCtx ktesting.TContext, tc *testCase, w *workload, informerFactory informers.SharedInformerFactory) ([]DataItem, error) {
func runWorkload(tCtx ktesting.TContext, tc *testCase, w *workload, scheduler *scheduler.Scheduler, informerFactory informers.SharedInformerFactory) ([]DataItem, error) {
b, benchmarking := tCtx.TB().(*testing.B)
if benchmarking {
start := time.Now()
@@ -1533,6 +1535,7 @@ func runWorkload(tCtx ktesting.TContext, tc *testCase, w *workload, informerFact
executor := WorkloadExecutor{
tCtx: tCtx,
scheduler: scheduler,
numPodsScheduledPerNamespace: make(map[string]int),
podInformer: podInformer,
throughputErrorMargin: throughputErrorMargin,
@@ -1593,6 +1596,9 @@ func (e *WorkloadExecutor) runOp(op realOp, opIndex int) error {
return e.runStartCollectingMetricsOp(opIndex, concreteOp)
case *stopCollectingMetricsOp:
return e.runStopCollectingMetrics(opIndex)
case *createResourceDriverOp:
concreteOp.run(e.tCtx, e.scheduler.Profiles["default-scheduler"].SharedDRAManager())
return nil
default:
return e.runDefaultOp(opIndex, concreteOp)
}

7
test/integration/scheduler_perf/util.go
View File

@@ -51,6 +51,7 @@ import (
kubeschedulerconfigv1 "k8s.io/kube-scheduler/config/v1"
apiservertesting "k8s.io/kubernetes/cmd/kube-apiserver/app/testing"
"k8s.io/kubernetes/pkg/features"
"k8s.io/kubernetes/pkg/scheduler"
"k8s.io/kubernetes/pkg/scheduler/apis/config"
kubeschedulerscheme "k8s.io/kubernetes/pkg/scheduler/apis/config/scheme"
frameworkruntime "k8s.io/kubernetes/pkg/scheduler/framework/runtime"
@@ -91,7 +92,7 @@ func newDefaultComponentConfig() (*config.KubeSchedulerConfiguration, error) {
// remove resources after finished.
// Notes on rate limiter:
// - client rate limit is set to 5000.
func mustSetupCluster(tCtx ktesting.TContext, config *config.KubeSchedulerConfiguration, enabledFeatures map[featuregate.Feature]bool, outOfTreePluginRegistry frameworkruntime.Registry) (informers.SharedInformerFactory, ktesting.TContext) {
func mustSetupCluster(tCtx ktesting.TContext, config *config.KubeSchedulerConfiguration, enabledFeatures map[featuregate.Feature]bool, outOfTreePluginRegistry frameworkruntime.Registry) (*scheduler.Scheduler, informers.SharedInformerFactory, ktesting.TContext) {
var runtimeConfig []string
if enabledFeatures[features.DynamicResourceAllocation] {
runtimeConfig = append(runtimeConfig, fmt.Sprintf("%s=true", resourceapi.SchemeGroupVersion))
@@ -135,7 +136,7 @@ func mustSetupCluster(tCtx ktesting.TContext, config *config.KubeSchedulerConfig
// Not all config options will be effective but only those mostly related with scheduler performance will
// be applied to start a scheduler, most of them are defined in `scheduler.schedulerOptions`.
_, informerFactory := util.StartScheduler(tCtx, config, outOfTreePluginRegistry)
scheduler, informerFactory := util.StartScheduler(tCtx, config, outOfTreePluginRegistry)
util.StartFakePVController(tCtx, tCtx.Client(), informerFactory)
runGC := util.CreateGCController(tCtx, tCtx, *cfg, informerFactory)
runNS := util.CreateNamespaceController(tCtx, tCtx, *cfg, informerFactory)
@@ -153,7 +154,7 @@ func mustSetupCluster(tCtx ktesting.TContext, config *config.KubeSchedulerConfig
go runNS()
go runResourceClaimController()
return informerFactory, tCtx
return scheduler, informerFactory, tCtx
}
func isAttempted(pod *v1.Pod) bool {