koord-scheduler: coscheduling supports sensing PodGroup changes (#1299)

Signed-off-by: wangjianyu.wjy <wangjianyu.wjy@alibaba-inc.com>
Co-authored-by: wangjianyu.wjy <wangjianyu.wjy@alibaba-inc.com>

Author: ZiMengSheng

pkg/scheduler/plugins/coscheduling/core/core.go

@@ -77,9 +77,6 @@ type Manager interface {
 type PodGroupManager struct {
 	// pgClient is a podGroup client
 	pgClient pgclientset.Interface
-	// scheduleTimeout is the default timeout for podgroup scheduling.
-	// If podgroup's scheduleTimeoutSeconds is set, it will be used.
-	scheduleTimeout *time.Duration
 	// pgLister is podgroup lister
 	pgLister pglister.PodGroupLister
 	// podLister is pod lister
@@ -110,6 +107,7 @@ func NewPodGroupManager(
 
 	podGroupEventHandler := cache.ResourceEventHandlerFuncs{
 		AddFunc:    gangCache.onPodGroupAdd,
+		UpdateFunc: gangCache.onPodGroupUpdate,
 		DeleteFunc: gangCache.onPodGroupDelete,
 	}
 	frameworkexthelper.ForceSyncFromInformer(context.TODO().Done(), pgSharedInformerFactory, pgInformer.Informer(), podGroupEventHandler)

pkg/scheduler/plugins/coscheduling/core/gang.go

@@ -170,9 +170,6 @@ func (gang *Gang) tryInitByPodConfig(pod *v1.Pod, args *config.CoschedulingArgs)
 func (gang *Gang) tryInitByPodGroup(pg *v1alpha1.PodGroup, args *config.CoschedulingArgs) {
 	gang.lock.Lock()
 	defer gang.lock.Unlock()
-	if gang.HasGangInit {
-		return
-	}
 	minRequiredNumber := pg.Spec.MinMember
 	gang.MinRequiredNumber = int(minRequiredNumber)
 

pkg/scheduler/plugins/coscheduling/core/gang_cache.go

@@ -109,9 +109,6 @@ func (gangCache *GangCache) onPodAdd(obj interface{}) {
 	}
 }
 
-func (gangCache *GangCache) onPodUpdate(oldObj interface{}, newObj interface{}) {
-}
-
 func (gangCache *GangCache) onPodDelete(obj interface{}) {
 	pod, ok := obj.(*v1.Pod)
 	if !ok {
@@ -144,16 +141,25 @@ func (gangCache *GangCache) onPodGroupAdd(obj interface{}) {
 	gangName := pg.Name
 
 	gangId := util.GetId(gangNamespace, gangName)
-	gang := gangCache.getGangFromCacheByGangId(gangId, false)
-	if gang == nil {
-		gang = gangCache.getGangFromCacheByGangId(gangId, true)
-		klog.Infof("Create gang by podGroup on add, gangName: %v", gangId)
-	}
-
+	gang := gangCache.getGangFromCacheByGangId(gangId, true)
 	gang.tryInitByPodGroup(pg, gangCache.pluginArgs)
 }
 
 func (gangCache *GangCache) onPodGroupUpdate(oldObj interface{}, newObj interface{}) {
+	pg, ok := newObj.(*v1alpha1.PodGroup)
+	if !ok {
+		return
+	}
+	gangNamespace := pg.Namespace
+	gangName := pg.Name
+
+	gangId := util.GetId(gangNamespace, gangName)
+	gang := gangCache.getGangFromCacheByGangId(gangId, false)
+	if gang == nil {
+		klog.Errorf("Gang object isn't exist when got Update Event")
+		return
+	}
+	gang.tryInitByPodGroup(pg, gangCache.pluginArgs)
 }
 
 func (gangCache *GangCache) onPodGroupDelete(obj interface{}) {

pkg/scheduler/plugins/coscheduling/core/gang_cache_test.go

@@ -926,5 +926,40 @@ func TestGangCache_OnGangDelete(t *testing.T) {
 	cacheGang := cache.getGangFromCacheByGangId("default/gangb", false)
 	wantedGang.GangGroupId = util.GetGangGroupId(wantedGang.GangGroup)
 	assert.Equal(t, wantedGang, cacheGang)
+}
+
+func TestGangCache_onPodGroupUpdate(t *testing.T) {
+	pgClient := fakepgclientset.NewSimpleClientset()
+	preTimeNowFn := timeNowFn
+	defer func() {
+		timeNowFn = preTimeNowFn
+	}()
+	timeNowFn = fakeTimeNowFn
+
+	pgInformerFactory := pgformers.NewSharedInformerFactory(pgClient, 0)
+	pgInformer := pgInformerFactory.Scheduling().V1alpha1().PodGroups()
+	pglister := pgInformer.Lister()
+	cache := NewGangCache(&config.CoschedulingArgs{DefaultTimeout: &metav1.Duration{Duration: time.Second}}, nil, pglister, pgClient)
+
+	// init gang
+	podGroup := &v1alpha1.PodGroup{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "default",
+			Name:      "ganga",
+		},
+		Spec: v1alpha1.PodGroupSpec{
+			MinMember: 2,
+		},
+	}
+	gangId := util.GetId("default", "ganga")
+	cache.onPodGroupAdd(podGroup)
+	gang := cache.getGangFromCacheByGangId(gangId, false)
+	assert.Equal(t, gang.MinRequiredNumber, int(podGroup.Spec.MinMember))
 
+	// update gang
+	newPodGroup := podGroup.DeepCopy()
+	newPodGroup.Spec.MinMember = 3
+	cache.onPodGroupUpdate(podGroup, newPodGroup)
+	gang = cache.getGangFromCacheByGangId(gangId, false)
+	assert.Equal(t, gang.MinRequiredNumber, int(newPodGroup.Spec.MinMember))
 }