From 8ff5cec26175ce5dacf67a4746af9fed648b4554 Mon Sep 17 00:00:00 2001 From: Patrick Ohly Date: Mon, 8 Sep 2025 14:14:39 +0200 Subject: [PATCH] scheduler_perf: block after creating ResourceSlices 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. --- test/integration/scheduler_perf/dra.go | 16 +++++++++++---- .../scheduler_perf/scheduler_perf.go | 20 ++++++++++++------- test/integration/scheduler_perf/util.go | 7 ++++--- 3 files changed, 29 insertions(+), 14 deletions(-) diff --git a/test/integration/scheduler_perf/dra.go b/test/integration/scheduler_perf/dra.go index 75f11e58e3f..b4d7a91c230 100644 --- a/test/integration/scheduler_perf/dra.go +++ b/test/integration/scheduler_perf/dra.go @@ -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{}, diff --git a/test/integration/scheduler_perf/scheduler_perf.go b/test/integration/scheduler_perf/scheduler_perf.go index 4c4c2c30bfe..9763ccb6fef 100644 --- a/test/integration/scheduler_perf/scheduler_perf.go +++ b/test/integration/scheduler_perf/scheduler_perf.go @@ -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) } diff --git a/test/integration/scheduler_perf/util.go b/test/integration/scheduler_perf/util.go index 74c79455607..85c99d84d7f 100644 --- a/test/integration/scheduler_perf/util.go +++ b/test/integration/scheduler_perf/util.go @@ -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 {