From 8d2599992b9672b60aaeea8bd05eaa43b7cd7a17 Mon Sep 17 00:00:00 2001 From: Patrik Flykt Date: Fri, 3 Jul 2026 08:51:40 +0300 Subject: [PATCH 1/5] Add checks against missing xpumd mounts Signed-off-by: Patrik Flykt --- internal/controller/deviceplugin_controller.go | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/internal/controller/deviceplugin_controller.go b/internal/controller/deviceplugin_controller.go index 21588f6..59ded00 100644 --- a/internal/controller/deviceplugin_controller.go +++ b/internal/controller/deviceplugin_controller.go @@ -71,6 +71,11 @@ func hexArgStr(s []string) string { } func addXpumdMounts(spec *core.PodSpec) { + if len(spec.Containers) == 0 { + klog.Warning("Cannot add xpumd mounts: no containers in pod spec") + return + } + for _, v := range spec.Volumes { if v.Name == xpumdVolumeName { return @@ -98,6 +103,11 @@ func addXpumdMounts(spec *core.PodSpec) { } func removeXpumdMounts(spec *core.PodSpec) { + if len(spec.Containers) == 0 { + klog.Warning("Cannot remove xpumd mounts: no containers in pod spec") + return + } + for i, v := range spec.Volumes { if v.Name == xpumdVolumeName { spec.Volumes = slices.Delete(spec.Volumes, i, i+1) @@ -145,6 +155,11 @@ func dpArgs(spec *v1alpha.ClusterPolicy) []string { } func (r *DevicePluginReconciler) updateDaemonSetObject(ds *apps.DaemonSet, spec *v1alpha.ClusterPolicy) { + if len(ds.Spec.Template.Spec.Containers) == 0 { + klog.Error(nil, "Cannot update DaemonSet: no containers in pod spec", "DaemonSet", ds.Name) + return + } + name := fmt.Sprintf("%s-device-plugin", spec.Name) ds.Name = name From 75a05462103e6aabbb5bcba6f0597853499d460f Mon Sep 17 00:00:00 2001 From: Patrik Flykt Date: Fri, 3 Jul 2026 09:58:07 +0300 Subject: [PATCH 2/5] Update Kueue to v0.18.2 with kustomization Signed-off-by: Patrik Flykt --- config/kueue/kustomization.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/config/kueue/kustomization.yaml b/config/kueue/kustomization.yaml index e9f532b..183cebe 100644 --- a/config/kueue/kustomization.yaml +++ b/config/kueue/kustomization.yaml @@ -1,4 +1,4 @@ apiVersion: kustomize.config.k8s.io/v1beta1 kind: Kustomization resources: -- "https://github.com/kubernetes-sigs/kueue/releases/download/v0.17.1/manifests.yaml" +- "https://github.com/kubernetes-sigs/kueue/releases/download/v0.18.2/manifests.yaml" From 7ead3d47c211a1bb0e79da6ea71c903a921a018c Mon Sep 17 00:00:00 2001 From: Patrik Flykt Date: Fri, 26 Jun 2026 13:00:50 +0300 Subject: [PATCH 3/5] Set controller reference for Kueue objects Signed-off-by: Patrik Flykt --- internal/controller/misc_controller.go | 37 ++++++++++++++++----- internal/controller/misc_controller_test.go | 2 ++ 2 files changed, 30 insertions(+), 9 deletions(-) diff --git a/internal/controller/misc_controller.go b/internal/controller/misc_controller.go index 5ebef36..24cabdf 100644 --- a/internal/controller/misc_controller.go +++ b/internal/controller/misc_controller.go @@ -461,7 +461,7 @@ func (r *MiscReconciler) createResourceFlavor() *kueuev1beta2.ResourceFlavor { } } -func (r *MiscReconciler) modifyResourceFlavor(ctx context.Context) error { +func (r *MiscReconciler) modifyResourceFlavor(ctx context.Context, cp *v1alpha.ClusterPolicy) error { currentUnstructured := &unstructured.Unstructured{} currentUnstructured.SetKind(kueueResourceFlavorKind) currentUnstructured.SetAPIVersion(kueueAPIVersion) @@ -476,7 +476,14 @@ func (r *MiscReconciler) modifyResourceFlavor(ctx context.Context) error { } if !found { - err := r.Create(ctx, r.createResourceFlavor()) + resourceFlavor := r.createResourceFlavor() + + if err := ctrl.SetControllerReference(cp, resourceFlavor, r.Scheme); err != nil { + klog.Error(err, "unable to set Kueue ResourceFlavor controller reference") + return err + } + + err := r.Create(ctx, resourceFlavor) if client.IgnoreAlreadyExists(err) != nil { klog.Error(err, "unable to create Kueue ResourceFlavor '%s'", kueueFlavorName) return err @@ -590,7 +597,7 @@ func (r *MiscReconciler) createLocalQueue(clusterQueueName string, localQueueSpe return &localQueue } -func (r *MiscReconciler) modifyLocalQueues(ctx context.Context, clusterQueue *v1alpha.ClusterQueueSpec) error { +func (r *MiscReconciler) modifyLocalQueues(ctx context.Context, clusterQueue *v1alpha.ClusterQueueSpec, cp *v1alpha.ClusterPolicy) error { if len(clusterQueue.LocalQueues) == 0 { klog.V(3).Infof("No LocalQueues defined for ClusterQueue '%s'", clusterQueue.Name) return nil @@ -613,6 +620,11 @@ func (r *MiscReconciler) modifyLocalQueues(ctx context.Context, clusterQueue *v1 newLocalQueue := r.createLocalQueue(clusterQueue.Name, &localQueue) if !found { + if err := ctrl.SetControllerReference(cp, newLocalQueue, r.Scheme); err != nil { + klog.Error(err, "unable to set Kueue LocalQueue controller reference") + return err + } + err := r.Create(ctx, newLocalQueue) if client.IgnoreAlreadyExists(err) != nil { klog.Error(err, "unable to create Kueue LocalQueue '%s/%s'", localQueue.Namespace, localQueue.Name) @@ -692,7 +704,9 @@ func (r *MiscReconciler) createClusterQueue(resources clusterResourceMap, cluste return clusterQueue } -func (r *MiscReconciler) modifyClusterQueue(ctx context.Context, resources clusterResourceMap, kueueSpec *v1alpha.KueueQueueSpec) error { +func (r *MiscReconciler) modifyClusterQueue(ctx context.Context, resources clusterResourceMap, cp *v1alpha.ClusterPolicy) error { + kueueSpec := cp.Spec.Kueue + if len(kueueSpec.EqualResources) == 0 { klog.V(3).Infof("At least one EqualResources queue should be configured") return nil @@ -720,6 +734,11 @@ func (r *MiscReconciler) modifyClusterQueue(ctx context.Context, resources clust newClusterQueue := r.createClusterQueue(clusterResources[n], &clusterQueue) if !found { + if err := ctrl.SetControllerReference(cp, newClusterQueue, r.Scheme); err != nil { + klog.Error(err, "unable to set Kueue ClusterQueue controller reference") + return err + } + err = r.Create(ctx, newClusterQueue) if client.IgnoreAlreadyExists(err) != nil { klog.Error(err, "unable to create Kueue ClusterQueue '%s'", clusterQueue.Name) @@ -752,7 +771,7 @@ func (r *MiscReconciler) modifyClusterQueue(ctx context.Context, resources clust } } - if err := r.modifyLocalQueues(ctx, &clusterQueue); err != nil { + if err := r.modifyLocalQueues(ctx, &clusterQueue, cp); err != nil { return err } } @@ -760,17 +779,17 @@ func (r *MiscReconciler) modifyClusterQueue(ctx context.Context, resources clust return nil } -func (r *MiscReconciler) createKueueQueues(ctx context.Context, resources clusterResourceMap, kueueSpec *v1alpha.KueueQueueSpec) error { +func (r *MiscReconciler) createKueueQueues(ctx context.Context, resources clusterResourceMap, cp *v1alpha.ClusterPolicy) error { if resources == nil { klog.V(3).Infof("no resources defined when attempting to create Kueue queues") return nil } - if err := r.modifyResourceFlavor(ctx); err != nil { + if err := r.modifyResourceFlavor(ctx, cp); err != nil { return err } - if err := r.modifyClusterQueue(ctx, resources, kueueSpec); err != nil { + if err := r.modifyClusterQueue(ctx, resources, cp); err != nil { return err } @@ -881,7 +900,7 @@ func (r *MiscReconciler) reconcileKueueObjects(ctx context.Context, cp *v1alpha. return nil } - return r.createKueueQueues(ctx, resources, cp.Spec.Kueue) + return r.createKueueQueues(ctx, resources, cp) } func (r *MiscReconciler) Reconcile(ctx context.Context, cp *v1alpha.ClusterPolicy) (ctrl.Result, error) { diff --git a/internal/controller/misc_controller_test.go b/internal/controller/misc_controller_test.go index 2508506..af85137 100644 --- a/internal/controller/misc_controller_test.go +++ b/internal/controller/misc_controller_test.go @@ -414,6 +414,7 @@ var _ = Describe("Misc Kueue", func() { Expect(core.AddToScheme(s)).To(Succeed()) Expect(kueuev1beta2.AddToScheme(s)).To(Succeed()) Expect(apiextensionsv1.AddToScheme(s)).To(Succeed()) + Expect(v1alpha.AddToScheme(s)).To(Succeed()) cp := &v1alpha.ClusterPolicy{ ObjectMeta: metav1.ObjectMeta{ @@ -444,6 +445,7 @@ var _ = Describe("Misc Kueue", func() { r := MiscReconciler{} r.Client = fake.NewClientBuilder().WithScheme(s).WithObjects(crd1, crd2, crd3, node).Build() r.Opts = ControllerOpts{ReqName: "test-cluster-policy"} + r.Scheme = s err := r.reconcileKueueObjects(ctx, cp) Expect(err).NotTo(HaveOccurred()) From 1f4575c7d06e5e501b887ea20a5e4909a92e3e01 Mon Sep 17 00:00:00 2001 From: Patrik Flykt Date: Fri, 3 Jul 2026 15:13:39 +0300 Subject: [PATCH 4/5] Prune old queues in case cluster policy renamed them Go through the existing cluster and local queues and remove those that are not specified in the cluster policy received for this reconciliation round. Signed-off-by: Patrik Flykt --- internal/controller/misc_controller.go | 94 ++++++++++++- internal/controller/misc_controller_test.go | 147 ++++++++++++++++++++ 2 files changed, 240 insertions(+), 1 deletion(-) diff --git a/internal/controller/misc_controller.go b/internal/controller/misc_controller.go index 24cabdf..b6b6dd4 100644 --- a/internal/controller/misc_controller.go +++ b/internal/controller/misc_controller.go @@ -597,6 +597,90 @@ func (r *MiscReconciler) createLocalQueue(clusterQueueName string, localQueueSpe return &localQueue } +func (r *MiscReconciler) keepLocalQueue(lq *kueuev1beta2.LocalQueue, keepLQs []v1alpha.LocalQueueSpec) bool { + for _, keep := range keepLQs { + if lq.Name == keep.Name && lq.Namespace == keep.Namespace { + return true + } + } + return false +} + +func (r *MiscReconciler) pruneLocalQueues(ctx context.Context, lqList *kueuev1beta2.LocalQueueList, keepLQs []v1alpha.LocalQueueSpec, cqName string, cp *v1alpha.ClusterPolicy) { + for _, lq := range lqList.Items { + klog.V(3).Infof("Checking local queue '%s' attached to cluster queue '%s'", lq.Name, cqName) + if lq.Spec.ClusterQueue != kueuev1beta2.ClusterQueueReference(cqName) { + continue + } + + if r.keepLocalQueue(&lq, keepLQs) { + klog.V(3).Infof("Keeping LocalQueue %s (in ClusterQueue %s)", lq.Name, cqName) + continue + } + + // delete this local queue + if err := r.Delete(ctx, &lq); client.IgnoreNotFound(err) != nil { + klog.Warningf("Error when attempting to delete LocalQueue %s: %v", lq.Name, err) + } else { + klog.V(3).Infof("Deleted LocalQueue %s namespace %s (in ClusterQueue %s)", lq.Name, lq.Namespace, cqName) + } + } +} + +func (r *MiscReconciler) clusterQueueExists(name string, cp *v1alpha.ClusterPolicy) []v1alpha.LocalQueueSpec { + kueue := cp.Spec.Kueue + if kueue == nil { + return nil + } + + for _, eq := range kueue.EqualResources { + if name == eq.Name { + return eq.LocalQueues + } + } + + return nil +} + +func (r *MiscReconciler) pruneQueues(ctx context.Context, cp *v1alpha.ClusterPolicy) error { + matchLabels := map[string]string{ + "app": kueueAppLabel, + "owner": r.Opts.ReqName, + } + + cqList := &kueuev1beta2.ClusterQueueList{} + if err := r.List(ctx, cqList, client.MatchingLabels(matchLabels)); client.IgnoreNotFound(err) != nil { + return fmt.Errorf("Could not fetch ClusterQueue list: %v", err) + } + + lqList := &kueuev1beta2.LocalQueueList{} + if err := r.List(ctx, lqList, client.InNamespace(metav1.NamespaceAll), client.MatchingLabels(matchLabels)); err != nil { + if client.IgnoreNotFound(err) != nil { + return fmt.Errorf("Could not fetch LocalQueue list: %v", err) + } else { + klog.Warningf("No LocalQueues found") + } + } + + for _, cq := range cqList.Items { + lq := r.clusterQueueExists(cq.Name, cp) + if lq == nil { + if err := r.Delete(ctx, &cq); client.IgnoreNotFound(err) != nil { + klog.Warningf("Error when attempting to delete ClusterQueue %s: %v", cq.Name, err) + } else { + klog.V(3).Infof("Deleted ClusterQueue %s", cq.Name) + } + } else { + klog.V(3).Infof("Keep ClusterQueue %s", cq.Name) + } + // delete lqs, lq = keep, lq = nil -> delete all + r.pruneLocalQueues(ctx, lqList, lq, cq.Name, cp) + + } + + return nil +} + func (r *MiscReconciler) modifyLocalQueues(ctx context.Context, clusterQueue *v1alpha.ClusterQueueSpec, cp *v1alpha.ClusterPolicy) error { if len(clusterQueue.LocalQueues) == 0 { klog.V(3).Infof("No LocalQueues defined for ClusterQueue '%s'", clusterQueue.Name) @@ -717,6 +801,10 @@ func (r *MiscReconciler) modifyClusterQueue(ctx context.Context, resources clust return err } + if err := r.pruneQueues(ctx, cp); err != nil { + klog.Warningf("Pruning Queues returned error but continuing anyway: %v", err) + } + for n, clusterQueue := range kueueSpec.EqualResources { currentUnstructured := &unstructured.Unstructured{} currentUnstructured.SetKind(kueueClusterQueueKind) @@ -837,7 +925,11 @@ func (r *MiscReconciler) removeKueueObjects(ctx context.Context, crName string) localQueues := &kueuev1beta2.LocalQueueList{} if err := r.List(ctx, localQueues, client.InNamespace(metav1.NamespaceAll), client.MatchingLabels(matchLabels)); err != nil { - klog.Warningf("Error when deleting Kueue LocalQueues: %v", err) + if client.IgnoreNotFound(err) == nil { + klog.V(3).Infof("No Kueue LocalQueues to delete") + } else { + klog.Warningf("Error when deleting Kueue LocalQueues: %v", err) + } } else { for i := range localQueues.Items { diff --git a/internal/controller/misc_controller_test.go b/internal/controller/misc_controller_test.go index af85137..0737202 100644 --- a/internal/controller/misc_controller_test.go +++ b/internal/controller/misc_controller_test.go @@ -474,6 +474,153 @@ var _ = Describe("Misc Kueue", func() { err = r.Get(ctx, types.NamespacedName{Name: kueueFlavorName}, flavor) Expect(err).To(HaveOccurred()) }) + + It("update Kueue queues", func() { + s := runtime.NewScheme() + Expect(core.AddToScheme(s)).To(Succeed()) + Expect(kueuev1beta2.AddToScheme(s)).To(Succeed()) + Expect(apiextensionsv1.AddToScheme(s)).To(Succeed()) + Expect(v1alpha.AddToScheme(s)).To(Succeed()) + + cp := &v1alpha.ClusterPolicy{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-cluster-policy", + }, + Spec: v1alpha.ClusterPolicySpec{ + ResourceRegistration: "dp", + EnableKueue: true, + + Kueue: &v1alpha.KueueQueueSpec{ + EqualResources: []v1alpha.ClusterQueueSpec{ + { + Name: "keep-clusterqueue", + LocalQueues: []v1alpha.LocalQueueSpec{ + {Name: "keepy-queue-1", Namespace: "default"}, + {Name: "keep-queue-2", Namespace: "foo"}, + }, + }, + { + Name: "update-local-queues", + LocalQueues: []v1alpha.LocalQueueSpec{ + {Name: "my-queue-1", Namespace: "foo"}, + {Name: "my-queue-2", Namespace: "default"}, + }, + }, + { + Name: "remove-these-queues", + LocalQueues: []v1alpha.LocalQueueSpec{ + {Name: "remove-queue-A", Namespace: "bar"}, + {Name: "remove-queue-B", Namespace: "foo"}, + }, + }, + }, + }, + }, + } + cpUpdated := &v1alpha.ClusterPolicy{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-cluster-policy", + }, + Spec: v1alpha.ClusterPolicySpec{ + ResourceRegistration: "dp", + EnableKueue: true, + + Kueue: &v1alpha.KueueQueueSpec{ + EqualResources: []v1alpha.ClusterQueueSpec{ + { + Name: "update-local-queues", + LocalQueues: []v1alpha.LocalQueueSpec{ + {Name: "my-queue-1", Namespace: "bar"}, + {Name: "my-queue-B", Namespace: "default"}, + }, + }, + { + Name: "keep-clusterqueue", + LocalQueues: []v1alpha.LocalQueueSpec{ + {Name: "keepy-queue-1", Namespace: "default"}, + {Name: "keep-queue-2", Namespace: "default"}, + }, + }, + { + Name: "test-spec-name", + LocalQueues: []v1alpha.LocalQueueSpec{ + {Name: "my-queue-A", Namespace: "default"}, + {Name: "my-queue-B", Namespace: "default"}, + }, + }, + }, + }, + }, + } + // specify which cluster and local queues are left after + // pruning them against 'cpUpdated' cluster policy + queuesAfterPrune := map[string]map[string]v1alpha.LocalQueueSpec{ + "keep-clusterqueue": map[string]v1alpha.LocalQueueSpec{ + "keepy-queue-1": v1alpha.LocalQueueSpec{ + Name: "keepy-queue-1", + Namespace: "default", + }, + }, + "update-local-queues": nil, + } + + ctx := context.Background() + + crd1, crd2, crd3 := kueueCRDObjects() + node := gpuNode() + + r := MiscReconciler{} + r.Client = fake.NewClientBuilder().WithScheme(s).WithObjects(crd1, crd2, crd3, node).Build() + r.Opts = ControllerOpts{ReqName: "test-cluster-policy"} + r.Scheme = s + + // add CR queues + _, err := r.Reconcile(ctx, cp) + Expect(err).NotTo(HaveOccurred()) + + // prune queues + err = r.pruneQueues(ctx, cpUpdated) + Expect(err).NotTo(HaveOccurred()) + + lqList := kueuev1beta2.LocalQueueList{} + err = r.List(ctx, &lqList) + Expect(err).NotTo(HaveOccurred()) + + for _, lq := range lqList.Items { + clusterQueueName := string(lq.Spec.ClusterQueue) + + localQueues, found := queuesAfterPrune[clusterQueueName] + Expect(found).To(BeTrue()) + + localQ, found := localQueues[lq.Name] + Expect(found).To(BeTrue()) + Expect(localQ.Name).To(Equal(lq.Name)) + Expect(localQ.Namespace).To(Equal(lq.Namespace)) + + // local queue was found + delete(localQueues, lq.Name) + } + + cqList := kueuev1beta2.ClusterQueueList{} + err = r.List(ctx, &cqList) + Expect(err).NotTo(HaveOccurred()) + + for _, cq := range cqList.Items { + localQueues, found := queuesAfterPrune[cq.Name] + + // verify each cluster queues is found + Expect(found).To(BeTrue()) + + // verify each local queue has been found above + Expect(len(localQueues)).To(Equal(0)) + + // cluster queue was found + delete(queuesAfterPrune, cq.Name) + } + + // all cluster queues have been found + Expect(len(queuesAfterPrune)).To(Equal(0)) + }) }) }) From 73e611e97a466cd802528e86c7f030523c244c34 Mon Sep 17 00:00:00 2001 From: Patrik Flykt Date: Fri, 3 Jul 2026 10:16:09 +0300 Subject: [PATCH 5/5] More test cases Signed-off-by: Patrik Flykt --- internal/controller/misc_controller_test.go | 378 ++++++++++++++++++++ 1 file changed, 378 insertions(+) diff --git a/internal/controller/misc_controller_test.go b/internal/controller/misc_controller_test.go index 0737202..8bc4947 100644 --- a/internal/controller/misc_controller_test.go +++ b/internal/controller/misc_controller_test.go @@ -621,6 +621,384 @@ var _ = Describe("Misc Kueue", func() { // all cluster queues have been found Expect(len(queuesAfterPrune)).To(Equal(0)) }) + + It("modify LocalQueue specifications", func() { + s := runtime.NewScheme() + Expect(core.AddToScheme(s)).To(Succeed()) + Expect(kueuev1beta2.AddToScheme(s)).To(Succeed()) + Expect(apiextensionsv1.AddToScheme(s)).To(Succeed()) + Expect(v1alpha.AddToScheme(s)).To(Succeed()) + + cp := &v1alpha.ClusterPolicy{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-cluster-policy", + }, + Spec: v1alpha.ClusterPolicySpec{ + ResourceRegistration: "dp", + EnableKueue: true, + Kueue: &v1alpha.KueueQueueSpec{ + EqualResources: []v1alpha.ClusterQueueSpec{ + { + Name: "test-queue", + LocalQueues: []v1alpha.LocalQueueSpec{ + {Name: "lq-1", Namespace: "default"}, + {Name: "lq-2", Namespace: "test-ns"}, + }, + }, + }, + }, + }, + } + + ctx := context.Background() + crd1, crd2, crd3 := kueueCRDObjects() + node := gpuNode() + + r := MiscReconciler{} + r.Client = fake.NewClientBuilder().WithScheme(s).WithObjects(crd1, crd2, crd3, node).Build() + r.Opts = ControllerOpts{ReqName: "test-cluster-policy"} + r.Scheme = s + + // Create initial queues + _, err := r.Reconcile(ctx, cp) + Expect(err).NotTo(HaveOccurred()) + + // Verify initial LocalQueues were created + lqList := kueuev1beta2.LocalQueueList{} + err = r.List(ctx, &lqList) + Expect(err).NotTo(HaveOccurred()) + Expect(len(lqList.Items)).To(Equal(2)) + + // Update ClusterPolicy with modified LocalQueue (different namespace) + cpUpdated := cp.DeepCopy() + cpUpdated.Spec.Kueue.EqualResources[0].LocalQueues[1].Namespace = "modified-ns" + + // Reconcile with updated policy + _, err = r.Reconcile(ctx, cpUpdated) + Expect(err).NotTo(HaveOccurred()) + + // Verify LocalQueue was updated + lqModified := &kueuev1beta2.LocalQueue{} + err = r.Get(ctx, types.NamespacedName{Name: "lq-2", Namespace: "modified-ns"}, lqModified) + Expect(err).NotTo(HaveOccurred()) + Expect(lqModified.Spec.ClusterQueue).To(Equal(kueuev1beta2.ClusterQueueReference("test-queue"))) + + // Old LocalQueue should no longer exist + lqOld := &kueuev1beta2.LocalQueue{} + err = r.Get(ctx, types.NamespacedName{Name: "lq-2", Namespace: "test-ns"}, lqOld) + Expect(err).To(HaveOccurred()) + }) + + It("add new LocalQueues to existing ClusterQueue", func() { + s := runtime.NewScheme() + Expect(core.AddToScheme(s)).To(Succeed()) + Expect(kueuev1beta2.AddToScheme(s)).To(Succeed()) + Expect(apiextensionsv1.AddToScheme(s)).To(Succeed()) + Expect(v1alpha.AddToScheme(s)).To(Succeed()) + + cp := &v1alpha.ClusterPolicy{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-cluster-policy", + }, + Spec: v1alpha.ClusterPolicySpec{ + ResourceRegistration: "dp", + EnableKueue: true, + Kueue: &v1alpha.KueueQueueSpec{ + EqualResources: []v1alpha.ClusterQueueSpec{ + { + Name: "test-queue", + LocalQueues: []v1alpha.LocalQueueSpec{ + {Name: "lq-1", Namespace: "default"}, + }, + }, + }, + }, + }, + } + + ctx := context.Background() + crd1, crd2, crd3 := kueueCRDObjects() + node := gpuNode() + + r := MiscReconciler{} + r.Client = fake.NewClientBuilder().WithScheme(s).WithObjects(crd1, crd2, crd3, node).Build() + r.Opts = ControllerOpts{ReqName: "test-cluster-policy"} + r.Scheme = s + + // Create initial queue with one LocalQueue + _, err := r.Reconcile(ctx, cp) + Expect(err).NotTo(HaveOccurred()) + + // Add more LocalQueues + cpUpdated := cp.DeepCopy() + cpUpdated.Spec.Kueue.EqualResources[0].LocalQueues = append( + cpUpdated.Spec.Kueue.EqualResources[0].LocalQueues, + v1alpha.LocalQueueSpec{Name: "lq-2", Namespace: "ns-2"}, + v1alpha.LocalQueueSpec{Name: "lq-3", Namespace: "ns-3"}, + ) + + // Reconcile with updated policy + _, err = r.Reconcile(ctx, cpUpdated) + Expect(err).NotTo(HaveOccurred()) + + // Verify all LocalQueues exist + lqList := kueuev1beta2.LocalQueueList{} + err = r.List(ctx, &lqList) + Expect(err).NotTo(HaveOccurred()) + Expect(len(lqList.Items)).To(Equal(3)) + + for _, lq := range lqList.Items { + Expect(lq.Spec.ClusterQueue).To(Equal(kueuev1beta2.ClusterQueueReference("test-queue"))) + } + }) + + It("remove LocalQueues from existing ClusterQueue", func() { + s := runtime.NewScheme() + Expect(core.AddToScheme(s)).To(Succeed()) + Expect(kueuev1beta2.AddToScheme(s)).To(Succeed()) + Expect(apiextensionsv1.AddToScheme(s)).To(Succeed()) + Expect(v1alpha.AddToScheme(s)).To(Succeed()) + + cp := &v1alpha.ClusterPolicy{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-cluster-policy", + }, + Spec: v1alpha.ClusterPolicySpec{ + ResourceRegistration: "dp", + EnableKueue: true, + Kueue: &v1alpha.KueueQueueSpec{ + EqualResources: []v1alpha.ClusterQueueSpec{ + { + Name: "test-queue", + LocalQueues: []v1alpha.LocalQueueSpec{ + {Name: "lq-1", Namespace: "default"}, + {Name: "lq-2", Namespace: "ns-2"}, + {Name: "lq-3", Namespace: "ns-3"}, + }, + }, + }, + }, + }, + } + + ctx := context.Background() + crd1, crd2, crd3 := kueueCRDObjects() + node := gpuNode() + + r := MiscReconciler{} + r.Client = fake.NewClientBuilder().WithScheme(s).WithObjects(crd1, crd2, crd3, node).Build() + r.Opts = ControllerOpts{ReqName: "test-cluster-policy"} + r.Scheme = s + + // Create initial queues + _, err := r.Reconcile(ctx, cp) + Expect(err).NotTo(HaveOccurred()) + + // Remove middle LocalQueue + cpUpdated := cp.DeepCopy() + cpUpdated.Spec.Kueue.EqualResources[0].LocalQueues = []v1alpha.LocalQueueSpec{ + {Name: "lq-1", Namespace: "default"}, + {Name: "lq-3", Namespace: "ns-3"}, + } + + // Reconcile with updated policy + _, err = r.Reconcile(ctx, cpUpdated) + Expect(err).NotTo(HaveOccurred()) + + // Prune old queues + err = r.pruneQueues(ctx, cpUpdated) + Expect(err).NotTo(HaveOccurred()) + + // Verify lq-2 was deleted + lqDeleted := &kueuev1beta2.LocalQueue{} + err = r.Get(ctx, types.NamespacedName{Name: "lq-2", Namespace: "ns-2"}, lqDeleted) + Expect(err).To(HaveOccurred()) + + // Verify remaining LocalQueues still exist + lqList := kueuev1beta2.LocalQueueList{} + err = r.List(ctx, &lqList) + Expect(err).NotTo(HaveOccurred()) + Expect(len(lqList.Items)).To(Equal(2)) + }) + + It("handle ClusterQueue with empty LocalQueues", func() { + s := runtime.NewScheme() + Expect(core.AddToScheme(s)).To(Succeed()) + Expect(kueuev1beta2.AddToScheme(s)).To(Succeed()) + Expect(apiextensionsv1.AddToScheme(s)).To(Succeed()) + Expect(v1alpha.AddToScheme(s)).To(Succeed()) + + cp := &v1alpha.ClusterPolicy{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-cluster-policy", + }, + Spec: v1alpha.ClusterPolicySpec{ + ResourceRegistration: "dp", + EnableKueue: true, + Kueue: &v1alpha.KueueQueueSpec{ + EqualResources: []v1alpha.ClusterQueueSpec{ + { + Name: "test-queue-empty", + LocalQueues: []v1alpha.LocalQueueSpec{}, + }, + }, + }, + }, + } + + ctx := context.Background() + crd1, crd2, crd3 := kueueCRDObjects() + node := gpuNode() + + r := MiscReconciler{} + r.Client = fake.NewClientBuilder().WithScheme(s).WithObjects(crd1, crd2, crd3, node).Build() + r.Opts = ControllerOpts{ReqName: "test-cluster-policy"} + r.Scheme = s + + // Create queue with empty LocalQueues + _, err := r.Reconcile(ctx, cp) + Expect(err).NotTo(HaveOccurred()) + + // Verify ClusterQueue was created even with empty LocalQueues + cqList := kueuev1beta2.ClusterQueueList{} + err = r.List(ctx, &cqList) + Expect(err).NotTo(HaveOccurred()) + Expect(len(cqList.Items)).To(Equal(1)) + Expect(cqList.Items[0].Name).To(Equal("test-queue-empty")) + }) + + It("prune queues when none exist", func() { + s := runtime.NewScheme() + Expect(core.AddToScheme(s)).To(Succeed()) + Expect(kueuev1beta2.AddToScheme(s)).To(Succeed()) + Expect(apiextensionsv1.AddToScheme(s)).To(Succeed()) + Expect(v1alpha.AddToScheme(s)).To(Succeed()) + + cp := &v1alpha.ClusterPolicy{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-cluster-policy", + }, + Spec: v1alpha.ClusterPolicySpec{ + ResourceRegistration: "dp", + EnableKueue: true, + Kueue: &v1alpha.KueueQueueSpec{ + EqualResources: []v1alpha.ClusterQueueSpec{}, + }, + }, + } + + ctx := context.Background() + crd1, crd2, crd3 := kueueCRDObjects() + node := gpuNode() + + r := MiscReconciler{} + r.Client = fake.NewClientBuilder().WithScheme(s).WithObjects(crd1, crd2, crd3, node).Build() + r.Opts = ControllerOpts{ReqName: "test-cluster-policy"} + r.Scheme = s + + // Prune when no queues exist should not error + err := r.pruneQueues(ctx, cp) + Expect(err).NotTo(HaveOccurred()) + }) + + It("replace all ClusterQueues and LocalQueues", func() { + s := runtime.NewScheme() + Expect(core.AddToScheme(s)).To(Succeed()) + Expect(kueuev1beta2.AddToScheme(s)).To(Succeed()) + Expect(apiextensionsv1.AddToScheme(s)).To(Succeed()) + Expect(v1alpha.AddToScheme(s)).To(Succeed()) + + cpInitial := &v1alpha.ClusterPolicy{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-cluster-policy", + }, + Spec: v1alpha.ClusterPolicySpec{ + ResourceRegistration: "dp", + EnableKueue: true, + Kueue: &v1alpha.KueueQueueSpec{ + EqualResources: []v1alpha.ClusterQueueSpec{ + { + Name: "old-queue", + LocalQueues: []v1alpha.LocalQueueSpec{ + {Name: "old-lq", Namespace: "old-ns"}, + }, + }, + }, + }, + }, + } + + cpNew := &v1alpha.ClusterPolicy{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-cluster-policy", + }, + Spec: v1alpha.ClusterPolicySpec{ + ResourceRegistration: "dp", + EnableKueue: true, + Kueue: &v1alpha.KueueQueueSpec{ + EqualResources: []v1alpha.ClusterQueueSpec{ + { + Name: "new-queue-1", + LocalQueues: []v1alpha.LocalQueueSpec{ + {Name: "new-lq-1", Namespace: "new-ns-1"}, + }, + }, + { + Name: "new-queue-2", + LocalQueues: []v1alpha.LocalQueueSpec{ + {Name: "new-lq-2", Namespace: "new-ns-2"}, + {Name: "new-lq-3", Namespace: "new-ns-3"}, + }, + }, + }, + }, + }, + } + + ctx := context.Background() + crd1, crd2, crd3 := kueueCRDObjects() + node := gpuNode() + + r := MiscReconciler{} + r.Client = fake.NewClientBuilder().WithScheme(s).WithObjects(crd1, crd2, crd3, node).Build() + r.Opts = ControllerOpts{ReqName: "test-cluster-policy"} + r.Scheme = s + + // Create initial queues + _, err := r.Reconcile(ctx, cpInitial) + Expect(err).NotTo(HaveOccurred()) + + // Verify initial state + lqListInitial := kueuev1beta2.LocalQueueList{} + err = r.List(ctx, &lqListInitial) + Expect(err).NotTo(HaveOccurred()) + Expect(len(lqListInitial.Items)).To(Equal(1)) + + // Create new queues + _, err = r.Reconcile(ctx, cpNew) + Expect(err).NotTo(HaveOccurred()) + + // Prune old queues + err = r.pruneQueues(ctx, cpNew) + Expect(err).NotTo(HaveOccurred()) + + // Verify old queue is gone + cqOld := &kueuev1beta2.ClusterQueue{} + err = r.Get(ctx, types.NamespacedName{Name: "old-queue"}, cqOld) + Expect(err).To(HaveOccurred()) + + // Verify new queues exist + cqList := kueuev1beta2.ClusterQueueList{} + err = r.List(ctx, &cqList) + Expect(err).NotTo(HaveOccurred()) + Expect(len(cqList.Items)).To(Equal(2)) + + // Verify new LocalQueues exist + lqListNew := kueuev1beta2.LocalQueueList{} + err = r.List(ctx, &lqListNew) + Expect(err).NotTo(HaveOccurred()) + Expect(len(lqListNew.Items)).To(Equal(3)) + }) }) })