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 {