fix volcano podgroup update issue

kubeflow · May 2, 2024 · f3c56ef · f3c56ef
1 parent f8f7363
commit f3c56ef
Show file tree

Hide file tree

Showing 2 changed files with 26 additions and 16 deletions.
diff --git a/pkg/controller.v1/common/job.go b/pkg/controller.v1/common/job.go
@@ -279,31 +279,33 @@ func (jc *JobController) ReconcileJobs(
 			var pgSpecFill FillPodGroupSpecFunc
 			switch jc.Config.GangScheduling {
 			case GangSchedulerVolcano:
-				pgSpecFill = func(pg metav1.Object) error {
+				pgSpecFill = func(pg metav1.Object) (metav1.Object, error) {
 					volcanoPodGroup, match := pg.(*volcanov1beta1.PodGroup)
+					volcanoPodGroup = volcanoPodGroup.DeepCopy()
 					if !match {
-						return fmt.Errorf("unable to recognize PodGroup: %v", klog.KObj(pg))
+						return nil, fmt.Errorf("unable to recognize PodGroup: %v", klog.KObj(pg))
 					}
 					volcanoPodGroup.Spec = volcanov1beta1.PodGroupSpec{
 						MinMember:         minMember,
 						Queue:             queue,
 						PriorityClassName: priorityClass,
 						MinResources:      minResources,
 					}
-					return nil
+					return volcanoPodGroup, nil
 				}
 			default:
-				pgSpecFill = func(pg metav1.Object) error {
+				pgSpecFill = func(pg metav1.Object) (metav1.Object, error) {
 					schedulerPluginsPodGroup, match := pg.(*schedulerpluginsv1alpha1.PodGroup)
+					schedulerPluginsPodGroup = schedulerPluginsPodGroup.DeepCopy()
 					if !match {
-						return fmt.Errorf("unable to recognize PodGroup: %v", klog.KObj(pg))
+						return nil, fmt.Errorf("unable to recognize PodGroup: %v", klog.KObj(pg))
 					}
 					schedulerPluginsPodGroup.Spec = schedulerpluginsv1alpha1.PodGroupSpec{
 						MinMember:              minMember,
 						MinResources:           *minResources,
 						ScheduleTimeoutSeconds: schedulerTimeout,
 					}
-					return nil
+					return schedulerPluginsPodGroup, nil
 				}
 			}
 

diff --git a/pkg/controller.v1/common/scheduling.go b/pkg/controller.v1/common/scheduling.go
@@ -19,15 +19,16 @@ package common
 import (
 	"fmt"
 
-	"github.com/google/go-cmp/cmp"
+	volcanov1beta1 "volcano.sh/apis/pkg/apis/scheduling/v1beta1"
+
 	log "github.com/sirupsen/logrus"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/klog/v2"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
-type FillPodGroupSpecFunc func(object metav1.Object) error
+type FillPodGroupSpecFunc func(object metav1.Object) (metav1.Object, error)
 
 func (jc *JobController) SyncPodGroup(job metav1.Object, specFunc FillPodGroupSpecFunc) (metav1.Object, error) {
 	pgctl := jc.PodGroupControl
@@ -36,14 +37,20 @@ func (jc *JobController) SyncPodGroup(job metav1.Object, specFunc FillPodGroupSp
 	podGroup, err := pgctl.GetPodGroup(job.GetNamespace(), job.GetName())
 	if err == nil {
 		// update podGroup for gang scheduling
-		oldPodGroup := &podGroup
-		if err = specFunc(podGroup); err != nil {
+		updatedSpecPodGroup, err := specFunc(podGroup)
+		if err != nil {
 			return nil, fmt.Errorf("unable to fill the spec of PodGroup, '%v': %v", klog.KObj(podGroup), err)
 		}
-		if diff := cmp.Diff(oldPodGroup, podGroup); len(diff) != 0 {
-			return podGroup, pgctl.UpdatePodGroup(podGroup.(client.Object))
+
+		existVolcanoPodGroup := podGroup.(*volcanov1beta1.PodGroup)
+		updatedSpecVolcanoPodGroup := updatedSpecPodGroup.(*volcanov1beta1.PodGroup)
+		// The hpa-controller may update the num of replicas
+		// https://github.com/kubeflow/common/pull/207
+		if existVolcanoPodGroup.Spec.MinMember != updatedSpecVolcanoPodGroup.Spec.MinMember {
+			// The queue name should not be changed after the pg is created
+			updatedSpecVolcanoPodGroup.Spec.Queue = existVolcanoPodGroup.Spec.Queue
+			return updatedSpecPodGroup, pgctl.UpdatePodGroup(updatedSpecPodGroup.(client.Object))
 		}
-		return podGroup, nil
 	} else if client.IgnoreNotFound(err) != nil {
 		return nil, fmt.Errorf("unable to get a PodGroup: %v", err)
 	} else {
@@ -53,13 +60,14 @@ func (jc *JobController) SyncPodGroup(job metav1.Object, specFunc FillPodGroupSp
 		newPodGroup.SetNamespace(job.GetNamespace())
 		newPodGroup.SetAnnotations(job.GetAnnotations())
 		newPodGroup.SetOwnerReferences([]metav1.OwnerReference{*jc.GenOwnerReference(job)})
-		if err = specFunc(newPodGroup); err != nil {
+		updatedSpecPodGroup, err := specFunc(newPodGroup)
+		if err != nil {
 			return nil, fmt.Errorf("unable to fill the spec of PodGroup, '%v': %v", klog.KObj(newPodGroup), err)
 		}
 
-		err = pgctl.CreatePodGroup(newPodGroup)
+		err = pgctl.CreatePodGroup(updatedSpecPodGroup.(client.Object))
 		if err != nil {
-			return podGroup, fmt.Errorf("unable to create PodGroup: %v", err)
+			return updatedSpecPodGroup, fmt.Errorf("unable to create PodGroup: %v", err)
 		}
 		createdPodGroupsCount.Inc()
 	}