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" 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 diff --git a/internal/controller/misc_controller.go b/internal/controller/misc_controller.go index 5ebef36..b6b6dd4 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,91 @@ func (r *MiscReconciler) createLocalQueue(clusterQueueName string, localQueueSpe return &localQueue } -func (r *MiscReconciler) modifyLocalQueues(ctx context.Context, clusterQueue *v1alpha.ClusterQueueSpec) error { +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) return nil @@ -613,6 +704,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 +788,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 @@ -703,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) @@ -720,6 +822,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 +859,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 +867,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 } @@ -818,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 { @@ -881,7 +992,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..8bc4947 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()) @@ -472,6 +474,531 @@ 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)) + }) + + 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)) + }) }) })