add podGroup backoff time for coscheduling

apis/config/scheme/scheme_test.go

@@ -29,7 +29,7 @@ import (
 	"k8s.io/kubernetes/pkg/scheduler/apis/config/testing/defaults"
 
 	"sigs.k8s.io/scheduler-plugins/apis/config"
-	"sigs.k8s.io/scheduler-plugins/apis/config/v1"
+	v1 "sigs.k8s.io/scheduler-plugins/apis/config/v1"
 	"sigs.k8s.io/scheduler-plugins/apis/config/v1beta2"
 	"sigs.k8s.io/scheduler-plugins/apis/config/v1beta3"
 	"sigs.k8s.io/scheduler-plugins/pkg/coscheduling"
@@ -66,6 +66,7 @@ profiles:
   - name: Coscheduling
     args:
       permitWaitingTimeSeconds: 10
+      podGroupBackoffSeconds: 0
       deniedPGExpirationTimeSeconds: 3
   - name: NodeResourcesAllocatable
     args:
@@ -472,6 +473,7 @@ kind: KubeSchedulerConfiguration
 profiles:
 - schedulerName: scheduler-plugins
   pluginConfig:
+  - name: Coscheduling # Test argument defaulting logic
   - name: TopologicalSort
     args:
       namespaces:
@@ -488,6 +490,12 @@ profiles:
 					SchedulerName: "scheduler-plugins",
 					Plugins:       defaults.PluginsV1,
 					PluginConfig: []schedconfig.PluginConfig{
+						{
+							Name: coscheduling.Name,
+							Args: &config.CoschedulingArgs{
+								PermitWaitingTimeSeconds: 60,
+							},
+						},
 						{
 							Name: topologicalsort.Name,
 							Args: &config.TopologicalSortArgs{
@@ -728,6 +736,7 @@ profiles:
       apiVersion: kubescheduler.config.k8s.io/v1beta2
       kind: CoschedulingArgs
       permitWaitingTimeSeconds: 10
+      podGroupBackoffSeconds: 0
     name: Coscheduling
   - args:
       apiVersion: kubescheduler.config.k8s.io/v1beta2
@@ -781,6 +790,7 @@ profiles:
 								Name: coscheduling.Name,
 								Args: &config.CoschedulingArgs{
 									PermitWaitingTimeSeconds: 10,
+									PodGroupBackoffSeconds:   20,
 								},
 							},
 							{
@@ -868,6 +878,7 @@ profiles:
       apiVersion: kubescheduler.config.k8s.io/v1beta3
       kind: CoschedulingArgs
       permitWaitingTimeSeconds: 10
+      podGroupBackoffSeconds: 20
     name: Coscheduling
   - args:
       apiVersion: kubescheduler.config.k8s.io/v1beta3
@@ -1022,6 +1033,7 @@ profiles:
       apiVersion: kubescheduler.config.k8s.io/v1
       kind: CoschedulingArgs
       permitWaitingTimeSeconds: 10
+      podGroupBackoffSeconds: 0
     name: Coscheduling
   - args:
       apiVersion: kubescheduler.config.k8s.io/v1

apis/config/types.go

@@ -30,6 +30,8 @@ type CoschedulingArgs struct {
 
 	// PermitWaitingTimeSeconds is the waiting timeout in seconds.
 	PermitWaitingTimeSeconds int64
+	// PodGroupBackoffSeconds is the backoff time in seconds before a pod group can be scheduled again.
+	PodGroupBackoffSeconds int64
 }
 
 // ModeType is a "string" type.

apis/config/v1/defaults.go

@@ -28,6 +28,7 @@ import (
 
 var (
 	defaultPermitWaitingTimeSeconds int64 = 60
+	defaultPodGroupBackoffSeconds   int64 = 0
 
 	defaultNodeResourcesAllocatableMode = Least
 
@@ -83,6 +84,9 @@ func SetDefaults_CoschedulingArgs(obj *CoschedulingArgs) {
 	if obj.PermitWaitingTimeSeconds == nil {
 		obj.PermitWaitingTimeSeconds = &defaultPermitWaitingTimeSeconds
 	}
+	if obj.PodGroupBackoffSeconds == nil {
+		obj.PodGroupBackoffSeconds = &defaultPodGroupBackoffSeconds
+	}
 }
 
 // SetDefaults_NodeResourcesAllocatableArgs sets the defaults parameters for NodeResourceAllocatable.

apis/config/v1/defaults_test.go

@@ -40,15 +40,18 @@ func TestSchedulingDefaults(t *testing.T) {
 			config: &CoschedulingArgs{},
 			expect: &CoschedulingArgs{
 				PermitWaitingTimeSeconds: pointer.Int64Ptr(60),
+				PodGroupBackoffSeconds:   pointer.Int64Ptr(0),
 			},
 		},
 		{
 			name: "set non default CoschedulingArgs",
 			config: &CoschedulingArgs{
 				PermitWaitingTimeSeconds: pointer.Int64Ptr(60),
+				PodGroupBackoffSeconds:   pointer.Int64Ptr(20),
 			},
 			expect: &CoschedulingArgs{
 				PermitWaitingTimeSeconds: pointer.Int64Ptr(60),
+				PodGroupBackoffSeconds:   pointer.Int64Ptr(20),
 			},
 		},
 		{

apis/config/v1/types.go

@@ -30,6 +30,8 @@ type CoschedulingArgs struct {
 
 	// PermitWaitingTimeSeconds is the waiting timeout in seconds.
 	PermitWaitingTimeSeconds *int64 `json:"permitWaitingTimeSeconds,omitempty"`
+	// PodGroupBackoffSeconds is the backoff time in seconds before a pod group can be scheduled again.
+	PodGroupBackoffSeconds *int64 `json:"podGroupBackoffSeconds,omitempty"`
 }
 
 // ModeType is a type "string".

apis/config/v1beta2/defaults.go

@@ -27,6 +27,7 @@ import (
 
 var (
 	defaultPermitWaitingTimeSeconds      int64 = 60
+	defaultPodGroupBackoffSeconds        int64 = 0
 	defaultDeniedPGExpirationTimeSeconds int64 = 20
 
 	defaultNodeResourcesAllocatableMode = Least
@@ -81,6 +82,9 @@ func SetDefaults_CoschedulingArgs(obj *CoschedulingArgs) {
 	if obj.DeniedPGExpirationTimeSeconds == nil {
 		obj.DeniedPGExpirationTimeSeconds = &defaultDeniedPGExpirationTimeSeconds
 	}
+	if obj.PodGroupBackoffSeconds == nil {
+		obj.PodGroupBackoffSeconds = &defaultPodGroupBackoffSeconds
+	}
 }
 
 // SetDefaults_NodeResourcesAllocatableArgs sets the defaults parameters for NodeResourceAllocatable.

apis/config/v1beta2/defaults_test.go

@@ -41,17 +41,20 @@ func TestSchedulingDefaults(t *testing.T) {
 			expect: &CoschedulingArgs{
 				PermitWaitingTimeSeconds:      pointer.Int64Ptr(60),
 				DeniedPGExpirationTimeSeconds: pointer.Int64Ptr(20),
+				PodGroupBackoffSeconds:        pointer.Int64Ptr(0),
 			},
 		},
 		{
 			name: "set non default CoschedulingArgs",
 			config: &CoschedulingArgs{
 				PermitWaitingTimeSeconds:      pointer.Int64Ptr(60),
 				DeniedPGExpirationTimeSeconds: pointer.Int64Ptr(10),
+				PodGroupBackoffSeconds:        pointer.Int64Ptr(20),
 			},
 			expect: &CoschedulingArgs{
 				PermitWaitingTimeSeconds:      pointer.Int64Ptr(60),
 				DeniedPGExpirationTimeSeconds: pointer.Int64Ptr(10),
+				PodGroupBackoffSeconds:        pointer.Int64Ptr(20),
 			},
 		},
 		{

apis/config/v1beta2/types.go

@@ -30,6 +30,8 @@ type CoschedulingArgs struct {
 
 	// PermitWaitingTimeSeconds is the waiting timeout in seconds.
 	PermitWaitingTimeSeconds *int64 `json:"permitWaitingTimeSeconds,omitempty"`
+	// PodGroupBackoffSeconds is the backoff time in seconds before a pod group can be scheduled again.
+	PodGroupBackoffSeconds *int64 `json:"podGroupBackoffSeconds,omitempty"`
 	// DeniedPGExpirationTimeSeconds is the expiration time of the denied podgroup store.
 	DeniedPGExpirationTimeSeconds *int64 `json:"deniedPGExpirationTimeSeconds,omitempty"`
 }

apis/config/v1beta3/defaults.go

@@ -17,19 +17,20 @@ limitations under the License.
 package v1beta3
 
 import (
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"strconv"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	schedulerconfigv1beta3 "k8s.io/kube-scheduler/config/v1beta3"
 	k8sschedulerconfigv1beta3 "k8s.io/kubernetes/pkg/scheduler/apis/config/v1beta3"
 )
 
 var (
-	defaultPermitWaitingTimeSeconds int64 = 60
-
-	defaultNodeResourcesAllocatableMode = Least
+	defaultPermitWaitingTimeSeconds     int64 = 60
+	defaultPodGroupBackoffSeconds       int64 = 0
+	defaultNodeResourcesAllocatableMode       = Least
 
 	// defaultResourcesToWeightMap is used to set the default resourceToWeight map for CPU and memory
 	// used by the NodeResourcesAllocatable scoring plugin.
@@ -83,6 +84,9 @@ func SetDefaults_CoschedulingArgs(obj *CoschedulingArgs) {
 	if obj.PermitWaitingTimeSeconds == nil {
 		obj.PermitWaitingTimeSeconds = &defaultPermitWaitingTimeSeconds
 	}
+	if obj.PodGroupBackoffSeconds == nil {
+		obj.PodGroupBackoffSeconds = &defaultPodGroupBackoffSeconds
+	}
 }
 
 // SetDefaults_NodeResourcesAllocatableArgs sets the defaults parameters for NodeResourceAllocatable.

apis/config/v1beta3/defaults_test.go

@@ -40,15 +40,18 @@ func TestSchedulingDefaults(t *testing.T) {
 			config: &CoschedulingArgs{},
 			expect: &CoschedulingArgs{
 				PermitWaitingTimeSeconds: pointer.Int64Ptr(60),
+				PodGroupBackoffSeconds:   pointer.Int64Ptr(0),
 			},
 		},
 		{
 			name: "set non default CoschedulingArgs",
 			config: &CoschedulingArgs{
 				PermitWaitingTimeSeconds: pointer.Int64Ptr(60),
+				PodGroupBackoffSeconds:   pointer.Int64Ptr(20),
 			},
 			expect: &CoschedulingArgs{
 				PermitWaitingTimeSeconds: pointer.Int64Ptr(60),
+				PodGroupBackoffSeconds:   pointer.Int64Ptr(20),
 			},
 		},
 		{

apis/config/v1beta3/types.go

@@ -30,6 +30,8 @@ type CoschedulingArgs struct {
 
 	// PermitWaitingTimeSeconds is the waiting timeout in seconds.
 	PermitWaitingTimeSeconds *int64 `json:"permitWaitingTimeSeconds,omitempty"`
+	// PodGroupBackoffSeconds is the backoff time in seconds before a pod group can be scheduled again.
+	PodGroupBackoffSeconds *int64 `json:"podGroupBackoffSeconds,omitempty"`
 }
 
 // ModeType is a type "string".

pkg/coscheduling/core/core.go

@@ -61,6 +61,7 @@ type Manager interface {
 	DeletePermittedPodGroup(string)
 	CalculateAssignedPods(string, string) int
 	ActivateSiblings(pod *corev1.Pod, state *framework.CycleState)
+	BackoffPodGroup(string, time.Duration)
 }
 
 // PodGroupManager defines the scheduling operation called
@@ -74,6 +75,8 @@ type PodGroupManager struct {
 	scheduleTimeout *time.Duration
 	// permittedPG stores the podgroup name which has passed the pre resource check.
 	permittedPG *gochache.Cache
+	// backedOffPG stores the podgorup name which failed scheudling recently.
+	backedOffPG *gochache.Cache
 	// pgLister is podgroup lister
 	pgLister pglister.PodGroupLister
 	// podLister is pod lister
@@ -91,10 +94,18 @@ func NewPodGroupManager(pgClient pgclientset.Interface, snapshotSharedLister fra
 		pgLister:             pgInformer.Lister(),
 		podLister:            podInformer.Lister(),
 		permittedPG:          gochache.New(3*time.Second, 3*time.Second),
+		backedOffPG:          gochache.New(10*time.Second, 10*time.Second),
 	}
 	return pgMgr
 }
 
+func (pgMgr *PodGroupManager) BackoffPodGroup(pgName string, backoff time.Duration) {
+	if backoff == time.Duration(0) {
+		return
+	}
+	pgMgr.backedOffPG.Add(pgName, nil, backoff)
+}
+
 // ActivateSiblings stashes the pods belonging to the same PodGroup of the given pod
 // in the given state, with a reserved key "kubernetes.io/pods-to-activate".
 func (pgMgr *PodGroupManager) ActivateSiblings(pod *corev1.Pod, state *framework.CycleState) {
@@ -143,6 +154,10 @@ func (pgMgr *PodGroupManager) PreFilter(ctx context.Context, pod *corev1.Pod) er
 		return nil
 	}
 
+	if _, exist := pgMgr.backedOffPG.Get(pgFullName); exist {
+		return fmt.Errorf("podGroup %v failed recently", pgFullName)
+	}
+
 	pods, err := pgMgr.podLister.Pods(pod.Namespace).List(
 		labels.SelectorFromSet(labels.Set{v1alpha1.PodGroupLabel: util.GetPodGroupLabel(pod)}),
 	)

pkg/coscheduling/core/core_test.go

@@ -139,7 +139,7 @@ func TestPreFilter(t *testing.T) {
 			existingPods, allNodes := testutil.MakeNodesAndPods(map[string]string{"test": "a"}, 60, 30)
 			snapshot := testutil.NewFakeSharedLister(existingPods, allNodes)
 			pgMgr := &PodGroupManager{pgLister: pgLister, permittedPG: newCache(),
-				snapshotSharedLister: snapshot, podLister: podInformer.Lister(), scheduleTimeout: &scheduleTimeout}
+				snapshotSharedLister: snapshot, podLister: podInformer.Lister(), scheduleTimeout: &scheduleTimeout, backedOffPG: gochache.New(10*time.Second, 10*time.Second)}
 			informerFactory.Start(ctx.Done())
 			if !clicache.WaitForCacheSync(ctx.Done(), podInformer.Informer().HasSynced) {
 				t.Fatal("WaitForCacheSync failed")

pkg/coscheduling/coscheduling.go

@@ -22,6 +22,7 @@ import (
 	"time"
 
 	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/tools/cache"
 	corev1helpers "k8s.io/component-helpers/scheduling/corev1"
@@ -30,6 +31,7 @@ import (
 
 	"sigs.k8s.io/scheduler-plugins/apis/config"
 	"sigs.k8s.io/scheduler-plugins/apis/scheduling"
+	"sigs.k8s.io/scheduler-plugins/apis/scheduling/v1alpha1"
 	"sigs.k8s.io/scheduler-plugins/pkg/coscheduling/core"
 	pgclientset "sigs.k8s.io/scheduler-plugins/pkg/generated/clientset/versioned"
 	pgformers "sigs.k8s.io/scheduler-plugins/pkg/generated/informers/externalversions"
@@ -41,6 +43,7 @@ type Coscheduling struct {
 	frameworkHandler framework.Handle
 	pgMgr            core.Manager
 	scheduleTimeout  *time.Duration
+	pgBackoff        *time.Duration
 }
 
 var _ framework.QueueSortPlugin = &Coscheduling{}
@@ -78,6 +81,14 @@ func New(obj runtime.Object, handle framework.Handle) (framework.Plugin, error)
 		pgMgr:            pgMgr,
 		scheduleTimeout:  &scheduleTimeDuration,
 	}
+	if args.PodGroupBackoffSeconds < 0 {
+		err := fmt.Errorf("Parse Arguments Failed")
+		klog.ErrorS(err, "PodGroupBackoffSeconds cannot be negative")
+		return nil, err
+	} else if args.PodGroupBackoffSeconds > 0 {
+		pgBackoff := time.Duration(args.PodGroupBackoffSeconds) * time.Second
+		plugin.pgBackoff = &pgBackoff
+	}
 	pgInformerFactory.Start(ctx.Done())
 	if !cache.WaitForCacheSync(ctx.Done(), pgInformer.Informer().HasSynced) {
 		err := fmt.Errorf("WaitForCacheSync failed")
@@ -166,6 +177,16 @@ func (cs *Coscheduling) PostFilter(ctx context.Context, state *framework.CycleSt
 			waitingPod.Reject(cs.Name(), "optimistic rejection in PostFilter")
 		}
 	})
+
+	if cs.pgBackoff != nil {
+		pods, err := cs.frameworkHandler.SharedInformerFactory().Core().V1().Pods().Lister().Pods(pod.Namespace).List(
+			labels.SelectorFromSet(labels.Set{v1alpha1.PodGroupLabel: util.GetPodGroupLabel(pod)}),
+		)
+		if err == nil && len(pods) >= int(pg.Spec.MinMember) {
+			cs.pgMgr.BackoffPodGroup(pgName, *cs.pgBackoff)
+		}
+	}
+
 	cs.pgMgr.DeletePermittedPodGroup(pgName)
 	return &framework.PostFilterResult{}, framework.NewStatus(framework.Unschedulable,
 		fmt.Sprintf("PodGroup %v gets rejected due to Pod %v is unschedulable even after PostFilter", pgName, pod.Name))