From 60c36432f2a58e85c4a78da9fc585f72ee53c065 Mon Sep 17 00:00:00 2001 From: Patrick Ohly Date: Tue, 20 May 2025 17:19:23 +0200 Subject: [PATCH 1/3] DRA integration: set up nodes for scheduling This enables proper scheduling tests. Most of them are probably better done in scheduler_perf where the same test then can also be used for benchmarking and creating objects is a bit better supported (from YAML, for example), but some special cases (in particular, anything involving error injection) are better done here. --- test/integration/dra/dra_test.go | 87 +++++++++++++++++++++++++++++--- 1 file changed, 79 insertions(+), 8 deletions(-) diff --git a/test/integration/dra/dra_test.go b/test/integration/dra/dra_test.go index 624012ff828..1054fa871e4 100644 --- a/test/integration/dra/dra_test.go +++ b/test/integration/dra/dra_test.go @@ -54,6 +54,7 @@ import ( st "k8s.io/kubernetes/pkg/scheduler/testing" "k8s.io/kubernetes/test/integration/framework" "k8s.io/kubernetes/test/integration/util" + "k8s.io/kubernetes/test/utils/format" "k8s.io/kubernetes/test/utils/ktesting" "k8s.io/utils/ptr" ) @@ -85,6 +86,8 @@ var ( Namespace(namespace). RequestWithPrioritizedList(className). Obj() + + numNodes = 2 ) // createTestNamespace creates a namespace with a name that is derived from the @@ -115,6 +118,11 @@ func TestDRA(t *testing.T) { // change the global DefaultFeatureGate. For each configuration, // multiple tests can run in parallel as long as they are careful // about what they create. + // + // Each configuration starts with two Nodes (ready, sufficient RAM and CPU for multiple pods) + // and no ResourceSlices. To test scheduling, a sub-test must create ResourceSlices. + // createTestNamespace can be used to create a unique per-test namespace. The name of that + // namespace then can be used to create cluster-scoped objects without conflicts between tests. for name, tc := range map[string]struct { apis map[schema.GroupVersion]bool features map[featuregate.Feature]bool @@ -216,11 +224,63 @@ func TestDRA(t *testing.T) { tCtx.Cleanup(server.TearDownFn) tCtx = ktesting.WithRESTConfig(tCtx, server.ClientConfig) + createNodes(tCtx) + tc.f(tCtx) }) } } +func createNodes(tCtx ktesting.TContext) { + for i := 0; i < numNodes; i++ { + // Create node. + node := &v1.Node{ + ObjectMeta: metav1.ObjectMeta{ + Name: fmt.Sprintf("worker-%d", i), + }, + } + node, err := tCtx.Client().CoreV1().Nodes().Create(tCtx, node, metav1.CreateOptions{}) + tCtx.ExpectNoError(err, fmt.Sprintf("creating node #%d", i)) + + // Make the node ready. + node.Status = v1.NodeStatus{ + Capacity: v1.ResourceList{ + v1.ResourceCPU: resource.MustParse("100"), + v1.ResourceMemory: resource.MustParse("1000"), + v1.ResourcePods: resource.MustParse("100"), + }, + Phase: v1.NodeRunning, + Conditions: []v1.NodeCondition{ + { + Type: v1.NodeReady, + Status: v1.ConditionTrue, + }, + }, + } + node, err = tCtx.Client().CoreV1().Nodes().UpdateStatus(tCtx, node, metav1.UpdateOptions{}) + tCtx.ExpectNoError(err, fmt.Sprintf("setting status of node #%d", i)) + + // Remove taint added by TaintNodesByCondition admission check. + node.Spec.Taints = nil + _, err = tCtx.Client().CoreV1().Nodes().Update(tCtx, node, metav1.UpdateOptions{}) + tCtx.ExpectNoError(err, fmt.Sprintf("removing node taint from #%d", i)) + } + + tCtx.CleanupCtx(func(tCtx ktesting.TContext) { + if !tCtx.Failed() { + return + } + + // Dump information about the cluster. + nodes, err := tCtx.Client().CoreV1().Nodes().List(tCtx, metav1.ListOptions{}) + if err != nil { + tCtx.Logf("Retrieving nodes failed: %v", err) + } else { + tCtx.Logf("Nodes:\n%s", format.Object(nodes.Items, 1)) + } + }) +} + func startScheduler(tCtx ktesting.TContext) { // Run scheduler with default configuration. tCtx.Log("Scheduler starting...") @@ -323,11 +383,23 @@ func testAdminAccess(tCtx ktesting.TContext, adminAccessEnabled bool) { func testPrioritizedList(tCtx ktesting.TContext, enabled bool) { tCtx.Parallel() + namespace := createTestNamespace(tCtx, nil) + class := class.DeepCopy() + class.Name = namespace + class.Spec.Selectors = []resourceapi.DeviceSelector{{ + CEL: &resourceapi.CELDeviceSelector{ + Expression: fmt.Sprintf("device.driver == %q", namespace), + }, + }} _, err := tCtx.Client().ResourceV1beta1().DeviceClasses().Create(tCtx, class, metav1.CreateOptions{}) tCtx.ExpectNoError(err, "create class") - namespace := createTestNamespace(tCtx, nil) + tCtx.CleanupCtx(func(tCtx ktesting.TContext) { + err := tCtx.Client().ResourceV1beta1().DeviceClasses().Delete(tCtx, class.Name, metav1.DeleteOptions{}) + tCtx.ExpectNoError(err, "delete class") + }) claim := claimPrioritizedList.DeepCopy() claim.Namespace = namespace + claim.Spec.Devices.Requests[0].FirstAvailable[0].DeviceClassName = class.Name claim, err = tCtx.Client().ResourceV1beta1().ResourceClaims(namespace).Create(tCtx, claim, metav1.CreateOptions{}) if !enabled { @@ -339,10 +411,9 @@ func testPrioritizedList(tCtx ktesting.TContext, enabled bool) { tCtx.Run("scheduler", func(tCtx ktesting.TContext) { startScheduler(tCtx) - // The fake cluster configuration is not complete enough to actually schedule pods. - // That is covered over in test/integration/scheduler_perf. - // Here we only test that we get to the point where it notices that, without failing - // during PreFilter because of FirstAvailable. + // We could create ResourceSlices for some node with the right driver. + // But failing during Filter is sufficient to determine that it did + // not fail during PreFilter because of FirstAvailable. pod := podWithClaimName.DeepCopy() pod.Namespace = namespace _, err := tCtx.Client().CoreV1().Pods(namespace).Create(tCtx, pod, metav1.CreateOptions{}) @@ -352,14 +423,14 @@ func testPrioritizedList(tCtx ktesting.TContext, enabled bool) { "Type": gomega.Equal(v1.PodScheduled), "Status": gomega.Equal(v1.ConditionFalse), "Reason": gomega.Equal("Unschedulable"), - "Message": gomega.Equal("no nodes available to schedule pods"), + "Message": gomega.Equal("0/2 nodes are available: 2 cannot allocate all claims. still not schedulable, preemption: 0/2 nodes are available: 2 Preemption is not helpful for scheduling."), }), )) ktesting.Eventually(tCtx, func(tCtx ktesting.TContext) *v1.Pod { pod, err := tCtx.Client().CoreV1().Pods(namespace).Get(tCtx, pod.Name, metav1.GetOptions{}) tCtx.ExpectNoError(err, "get pod") return pod - }).WithTimeout(time.Minute).WithPolling(time.Second).Should(schedulingAttempted) + }).WithTimeout(10 * time.Second).WithPolling(time.Second).Should(schedulingAttempted) }) } @@ -481,7 +552,7 @@ func testPublishResourceSlices(tCtx ktesting.TContext, disabledFeatures ...featu ktesting.Eventually(tCtx, getStats).WithTimeout(10*time.Second).Should(gomega.HaveField("NumDeletes", gomega.BeNumerically(">=", int64(1))), "Slice should have been removed.") ktesting.Eventually(tCtx, func(tCtx ktesting.TContext) bool { return gotValidationError.Load() - }).WithTimeout(10 * time.Second).Should(gomega.BeTrueBecause("Should have gotten another error because the slice is invalid.")) + }).WithTimeout(time.Minute).Should(gomega.BeTrueBecause("Should have gotten another error because the slice is invalid.")) } From 50f152440be59fce98ee6c149c3b6866ca78b647 Mon Sep 17 00:00:00 2001 From: Patrick Ohly Date: Wed, 21 May 2025 14:40:41 +0200 Subject: [PATCH 2/3] DRA integration: start scheduler on demand As soon as we have more than one test using the scheduler, we need some coordination between tests. This is handled by a singleton which starts the scheduler for the first user and stops it after the last one is gone. To avoid having to pass around an additional parameter, the context is used to access the singleton under the hood. --- test/integration/dra/dra_test.go | 101 ++++++++++++++++++++++++++----- 1 file changed, 86 insertions(+), 15 deletions(-) diff --git a/test/integration/dra/dra_test.go b/test/integration/dra/dra_test.go index 1054fa871e4..13e1e0e0ef4 100644 --- a/test/integration/dra/dra_test.go +++ b/test/integration/dra/dra_test.go @@ -23,6 +23,7 @@ import ( "regexp" "sort" "strings" + "sync" "sync/atomic" "testing" "time" @@ -42,6 +43,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime/schema" utilfeature "k8s.io/apiserver/pkg/util/feature" + "k8s.io/client-go/informers" "k8s.io/component-base/featuregate" featuregatetesting "k8s.io/component-base/featuregate/testing" "k8s.io/dynamic-resource-allocation/resourceslice" @@ -221,10 +223,14 @@ func TestDRA(t *testing.T) { } apiServerFlags = append(apiServerFlags, "--runtime-config="+strings.Join(runtimeConfigs, ",")) server := kubeapiservertesting.StartTestServerOrDie(t, apiServerOptions, apiServerFlags, etcdOptions) - tCtx.Cleanup(server.TearDownFn) + tCtx.CleanupCtx(func(tCtx ktesting.TContext) { + tCtx.Log("Stopping the apiserver...") + server.TearDownFn() + }) tCtx = ktesting.WithRESTConfig(tCtx, server.ClientConfig) createNodes(tCtx) + tCtx = prepareScheduler(tCtx) tc.f(tCtx) }) @@ -281,21 +287,84 @@ func createNodes(tCtx ktesting.TContext) { }) } -func startScheduler(tCtx ktesting.TContext) { - // Run scheduler with default configuration. - tCtx.Log("Scheduler starting...") - schedulerCtx := klog.NewContext(tCtx, klog.LoggerWithName(tCtx.Logger(), "scheduler")) - schedulerCtx, cancel := context.WithCancelCause(schedulerCtx) - _, informerFactory := util.StartScheduler(schedulerCtx, tCtx.Client(), tCtx.RESTConfig(), newDefaultSchedulerComponentConfig(tCtx), nil) - // Stop clients of the apiserver before stopping the apiserver itself, - // otherwise it delays its shutdown. - tCtx.Cleanup(informerFactory.Shutdown) - tCtx.Cleanup(func() { - tCtx.Log("Stoping scheduler...") - cancel(errors.New("test is done")) - }) +// prepareScheduler returns a TContext which can be passed to startScheduler +// to actually start the scheduler when there is a demand for it. +// +// Under the hood, schedulerSingleton ensures that at most one scheduler +// instance is created and tears it down when the last test using it is +// done. This could lead to starting and stopping it multiple times, but in +// practice tests start together in parallel and share a single instance. +func prepareScheduler(tCtx ktesting.TContext) ktesting.TContext { + scheduler := &schedulerSingleton{ + rootCtx: tCtx, + } + + return ktesting.WithValue(tCtx, schedulerKey, scheduler) } +func startScheduler(tCtx ktesting.TContext) { + value := tCtx.Value(schedulerKey) + if value == nil { + tCtx.Fatal("internal error: startScheduler without a prior prepareScheduler call") + } + scheduler := value.(*schedulerSingleton) + scheduler.start(tCtx) +} + +type schedulerSingleton struct { + rootCtx ktesting.TContext + + mutex sync.Mutex + usageCount int + informerFactory informers.SharedInformerFactory + cancel func(err error) +} + +func (scheduler *schedulerSingleton) start(tCtx ktesting.TContext) { + scheduler.mutex.Lock() + defer scheduler.mutex.Unlock() + + scheduler.usageCount++ + tCtx.CleanupCtx(scheduler.stop) + if scheduler.usageCount > 1 { + // Already started earlier. + return + } + + // Run scheduler with default configuration. This must use the root context because + // the per-test tCtx passed to start will get canceled once the test which triggered + // starting the scheduler is done. + schedulerCtx := scheduler.rootCtx + schedulerCtx.Logf("Starting the scheduler for test %s...", tCtx.Name()) + ctx := klog.NewContext(schedulerCtx, klog.LoggerWithName(schedulerCtx.Logger(), "scheduler")) + ctx, scheduler.cancel = context.WithCancelCause(ctx) + _, scheduler.informerFactory = util.StartScheduler(ctx, schedulerCtx.Client(), schedulerCtx.RESTConfig(), newDefaultSchedulerComponentConfig(schedulerCtx), nil) + schedulerCtx.Logf("Started the scheduler for test %s.", tCtx.Name()) +} + +func (scheduler *schedulerSingleton) stop(tCtx ktesting.TContext) { + scheduler.mutex.Lock() + defer scheduler.mutex.Unlock() + + scheduler.usageCount-- + if scheduler.usageCount > 0 { + // Still in use by some other test. + return + } + + scheduler.rootCtx.Logf("Stopping the scheduler after test %s...", tCtx.Name()) + if scheduler.cancel != nil { + scheduler.cancel(errors.New("test is done")) + } + if scheduler.informerFactory != nil { + scheduler.informerFactory.Shutdown() + } +} + +type schedulerKeyType int + +var schedulerKey schedulerKeyType + func newDefaultSchedulerComponentConfig(tCtx ktesting.TContext) *config.KubeSchedulerConfiguration { gvk := kubeschedulerconfigv1.SchemeGroupVersion.WithKind("KubeSchedulerConfiguration") cfg := config.KubeSchedulerConfiguration{} @@ -394,6 +463,7 @@ func testPrioritizedList(tCtx ktesting.TContext, enabled bool) { _, err := tCtx.Client().ResourceV1beta1().DeviceClasses().Create(tCtx, class, metav1.CreateOptions{}) tCtx.ExpectNoError(err, "create class") tCtx.CleanupCtx(func(tCtx ktesting.TContext) { + tCtx.Log("Cleaning up DeviceClass...") err := tCtx.Client().ResourceV1beta1().DeviceClasses().Delete(tCtx, class.Name, metav1.DeleteOptions{}) tCtx.ExpectNoError(err, "delete class") }) @@ -438,7 +508,8 @@ func testPublishResourceSlices(tCtx ktesting.TContext, disabledFeatures ...featu tCtx.Parallel() tCtx = ktesting.WithTimeout(tCtx, 30*time.Second, "test timed out") - driverName := "dra.example.com" + namespace := createTestNamespace(tCtx, nil) + driverName := namespace + ".example.com" poolName := "global" resources := &resourceslice.DriverResources{ Pools: map[string]resourceslice.Pool{ From e63019a8701e11df0792ca4dd746c6b21bfe9573 Mon Sep 17 00:00:00 2001 From: Patrick Ohly Date: Wed, 21 May 2025 21:39:18 +0200 Subject: [PATCH 3/3] DRA integration: refactor code to support other tests Creating class, claim and pod is expected to be fairly common. --- test/integration/dra/dra_test.go | 123 ++++++++++++++++++++----------- 1 file changed, 82 insertions(+), 41 deletions(-) diff --git a/test/integration/dra/dra_test.go b/test/integration/dra/dra_test.go index 13e1e0e0ef4..8a0bc352162 100644 --- a/test/integration/dra/dra_test.go +++ b/test/integration/dra/dra_test.go @@ -114,6 +114,62 @@ func createTestNamespace(tCtx ktesting.TContext, labels map[string]string) strin return ns.Name } +// createTestClass creates a DeviceClass with a driver name derived from the test namespace +func createTestClass(tCtx ktesting.TContext, namespace string) (*resourceapi.DeviceClass, string) { + driverName := namespace + ".driver" + class := class.DeepCopy() + class.Name = namespace + ".class" + class.Spec.Selectors = []resourceapi.DeviceSelector{{ + CEL: &resourceapi.CELDeviceSelector{ + Expression: fmt.Sprintf("device.driver == %q", driverName), + }, + }} + _, err := tCtx.Client().ResourceV1beta1().DeviceClasses().Create(tCtx, class, metav1.CreateOptions{}) + tCtx.ExpectNoError(err, "create class") + tCtx.CleanupCtx(func(tCtx ktesting.TContext) { + tCtx.Log("Cleaning up DeviceClass...") + err := tCtx.Client().ResourceV1beta1().DeviceClasses().Delete(tCtx, class.Name, metav1.DeleteOptions{}) + tCtx.ExpectNoError(err, "delete class") + }) + + return class, driverName +} + +// createClaim creates a claim and in the namespace. +// The class must already exist and is used for all requests. +func createClaim(tCtx ktesting.TContext, namespace string, suffix string, class *resourceapi.DeviceClass, claim *resourceapi.ResourceClaim) *resourceapi.ResourceClaim { + claim = claim.DeepCopy() + claim.Namespace = namespace + claim.Name += suffix + claimName := claim.Name + for i := range claim.Spec.Devices.Requests { + request := &claim.Spec.Devices.Requests[i] + if request.DeviceClassName != "" { + request.DeviceClassName = class.Name + continue + } + for e := range request.FirstAvailable { + subRequest := &request.FirstAvailable[e] + subRequest.DeviceClassName = class.Name + } + } + claim, err := tCtx.Client().ResourceV1beta1().ResourceClaims(namespace).Create(tCtx, claim, metav1.CreateOptions{}) + tCtx.ExpectNoError(err, "create claim "+claimName) + return claim +} + +// createPod create a pod in the namespace, referencing the given claim. +func createPod(tCtx ktesting.TContext, namespace string, suffix string, claim *resourceapi.ResourceClaim, pod *v1.Pod) *v1.Pod { + pod = pod.DeepCopy() + pod.Name += suffix + podName := pod.Name + pod.Namespace = namespace + pod.Spec.ResourceClaims[0].ResourceClaimName = &claim.Name + pod, err := tCtx.Client().CoreV1().Pods(namespace).Create(tCtx, pod, metav1.CreateOptions{}) + tCtx.ExpectNoError(err, "create pod "+podName) + return pod +} + func TestDRA(t *testing.T) { // Each sub-test brings up the API server in a certain // configuration. These sub-tests must run sequentially because they @@ -453,24 +509,14 @@ func testAdminAccess(tCtx ktesting.TContext, adminAccessEnabled bool) { func testPrioritizedList(tCtx ktesting.TContext, enabled bool) { tCtx.Parallel() namespace := createTestNamespace(tCtx, nil) - class := class.DeepCopy() - class.Name = namespace - class.Spec.Selectors = []resourceapi.DeviceSelector{{ - CEL: &resourceapi.CELDeviceSelector{ - Expression: fmt.Sprintf("device.driver == %q", namespace), - }, - }} - _, err := tCtx.Client().ResourceV1beta1().DeviceClasses().Create(tCtx, class, metav1.CreateOptions{}) - tCtx.ExpectNoError(err, "create class") - tCtx.CleanupCtx(func(tCtx ktesting.TContext) { - tCtx.Log("Cleaning up DeviceClass...") - err := tCtx.Client().ResourceV1beta1().DeviceClasses().Delete(tCtx, class.Name, metav1.DeleteOptions{}) - tCtx.ExpectNoError(err, "delete class") - }) - claim := claimPrioritizedList.DeepCopy() - claim.Namespace = namespace - claim.Spec.Devices.Requests[0].FirstAvailable[0].DeviceClassName = class.Name - claim, err = tCtx.Client().ResourceV1beta1().ResourceClaims(namespace).Create(tCtx, claim, metav1.CreateOptions{}) + class, _ := createTestClass(tCtx, namespace) + // This is allowed to fail if the feature is disabled. + // createClaim normally doesn't return errors because this is unusual, but we can get it indirectly. + claim, err := func() (claim *resourceapi.ResourceClaim, finalError error) { + tCtx, finalize := ktesting.WithError(tCtx, &finalError) + defer finalize() + return createClaim(tCtx, namespace, "", class, claimPrioritizedList), nil + }() if !enabled { require.Error(tCtx, err, "claim should have become invalid after dropping FirstAvailable") @@ -478,30 +524,25 @@ func testPrioritizedList(tCtx ktesting.TContext, enabled bool) { } require.NotEmpty(tCtx, claim.Spec.Devices.Requests[0].FirstAvailable, "should store FirstAvailable") - tCtx.Run("scheduler", func(tCtx ktesting.TContext) { - startScheduler(tCtx) + startScheduler(tCtx) - // We could create ResourceSlices for some node with the right driver. - // But failing during Filter is sufficient to determine that it did - // not fail during PreFilter because of FirstAvailable. - pod := podWithClaimName.DeepCopy() - pod.Namespace = namespace - _, err := tCtx.Client().CoreV1().Pods(namespace).Create(tCtx, pod, metav1.CreateOptions{}) - tCtx.ExpectNoError(err, "create pod") - schedulingAttempted := gomega.HaveField("Status.Conditions", gomega.ContainElement( - gstruct.MatchFields(gstruct.IgnoreExtras, gstruct.Fields{ - "Type": gomega.Equal(v1.PodScheduled), - "Status": gomega.Equal(v1.ConditionFalse), - "Reason": gomega.Equal("Unschedulable"), - "Message": gomega.Equal("0/2 nodes are available: 2 cannot allocate all claims. still not schedulable, preemption: 0/2 nodes are available: 2 Preemption is not helpful for scheduling."), - }), - )) - ktesting.Eventually(tCtx, func(tCtx ktesting.TContext) *v1.Pod { - pod, err := tCtx.Client().CoreV1().Pods(namespace).Get(tCtx, pod.Name, metav1.GetOptions{}) - tCtx.ExpectNoError(err, "get pod") - return pod - }).WithTimeout(10 * time.Second).WithPolling(time.Second).Should(schedulingAttempted) - }) + // We could create ResourceSlices for some node with the right driver. + // But failing during Filter is sufficient to determine that it did + // not fail during PreFilter because of FirstAvailable. + pod := createPod(tCtx, namespace, "", claim, podWithClaimName) + schedulingAttempted := gomega.HaveField("Status.Conditions", gomega.ContainElement( + gstruct.MatchFields(gstruct.IgnoreExtras, gstruct.Fields{ + "Type": gomega.Equal(v1.PodScheduled), + "Status": gomega.Equal(v1.ConditionFalse), + "Reason": gomega.Equal("Unschedulable"), + "Message": gomega.Equal("0/2 nodes are available: 2 cannot allocate all claims. still not schedulable, preemption: 0/2 nodes are available: 2 Preemption is not helpful for scheduling."), + }), + )) + ktesting.Eventually(tCtx, func(tCtx ktesting.TContext) *v1.Pod { + pod, err := tCtx.Client().CoreV1().Pods(namespace).Get(tCtx, pod.Name, metav1.GetOptions{}) + tCtx.ExpectNoError(err, "get pod") + return pod + }).WithTimeout(10 * time.Second).WithPolling(time.Second).Should(schedulingAttempted) } func testPublishResourceSlices(tCtx ktesting.TContext, disabledFeatures ...featuregate.Feature) {