Fix proactive scale up injecting fake pods for scheduling gated pods

Author: abdelrahman882

cluster-autoscaler/processors/podinjection/pod_group.go

@@ -20,6 +20,7 @@ import (
 	apiv1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	kube_util "k8s.io/autoscaler/cluster-autoscaler/utils/kubernetes"
 )
 
 type podGroup struct {
@@ -45,6 +46,30 @@ func groupPods(pods []*apiv1.Pod, controllers []controller) map[types.UID]podGro
 	return podGroups
 }
 
+func filterOutSchedulingGatedPods(groups map[types.UID]podGroup, allPods []*apiv1.Pod) map[types.UID]podGroup {
+	if groups != nil {
+		podsWithSchedulingGates := kube_util.SchedulingGatedPods(allPods)
+		for _, pod := range podsWithSchedulingGates {
+			if pod != nil {
+				groups = removeSchedulingGatedPodFromGroups(groups, pod)
+			}
+		}
+	}
+	return groups
+}
+
+func removeSchedulingGatedPodFromGroups(groups map[types.UID]podGroup, pod *apiv1.Pod) map[types.UID]podGroup {
+	for _, podOwnerRef := range pod.OwnerReferences {
+		// SchedulingGated pods can't be unschedualable nor unprocessed nor scheduled so it is not expected
+		// to have them as group sample nor in pod count, so decreasing desiredReplicas by one is enough
+		if grp, found := groups[podOwnerRef.UID]; found {
+			grp.desiredReplicas -= 1
+			groups[podOwnerRef.UID] = grp
+		}
+	}
+	return groups
+}
+
 // updatePodGroups updates the pod group if ownerRef is the controller of the pod
 func updatePodGroups(pod *apiv1.Pod, ownerRef metav1.OwnerReference, podGroups map[types.UID]podGroup) map[types.UID]podGroup {
 	if ownerRef.Controller == nil {

cluster-autoscaler/processors/podinjection/pod_injection_processor.go

@@ -60,10 +60,16 @@ func (p *PodInjectionPodListProcessor) Process(ctx *context.AutoscalingContext,
 		return unschedulablePods, fmt.Errorf("failed to list nodeInfos from cluster snapshot: %v", err)
 	}
 	scheduledPods := podsFromNodeInfos(nodeInfos)
-
 	groupedPods := groupPods(append(scheduledPods, unschedulablePods...), controllers)
-	var podsToInject []*apiv1.Pod
 
+	allPods, err := ctx.AllPodLister().List()
+	if err == nil {
+		groupedPods = filterOutSchedulingGatedPods(groupedPods, allPods)
+	} else {
+		klog.Warningf("Pod injection processor failed to list pods to filter scheduling gated pods with error %v", err.Error())
+	}
+
+	var podsToInject []*apiv1.Pod
 	for _, groupedPod := range groupedPods {
 		var fakePodCount = groupedPod.fakePodCount()
 		fakePods := makeFakePods(groupedPod.ownerUid, groupedPod.sample, fakePodCount)

cluster-autoscaler/processors/podinjection/pod_injection_processor_test.go

@@ -38,13 +38,22 @@ import (
 	. "k8s.io/autoscaler/cluster-autoscaler/utils/test"
 )
 
+func setStatusConditionGated(pod *apiv1.Pod) {
+	pod.Status.Conditions = []apiv1.PodCondition{{
+		Type:   apiv1.PodScheduled,
+		Status: apiv1.ConditionFalse,
+		Reason: apiv1.PodReasonSchedulingGated,
+	}}
+}
 func TestTargetCountInjectionPodListProcessor(t *testing.T) {
 	node := BuildTestNode("node1", 100, 0)
 
 	replicaSet1 := createTestReplicaSet("rep-set-1", "default", 5)
 	scheduledPodRep1Copy1 := buildTestPod("default", "-scheduled-pod-rep1-1", WithControllerOwnerRef(replicaSet1.Name, "ReplicaSet", replicaSet1.UID), WithNodeName(node.Name))
 	podRep1Copy1 := buildTestPod("default", "pod-rep1-1", WithControllerOwnerRef(replicaSet1.Name, "ReplicaSet", replicaSet1.UID))
 	podRep1Copy2 := buildTestPod("default", "pod-rep1-2", WithControllerOwnerRef(replicaSet1.Name, "ReplicaSet", replicaSet1.UID))
+	podRep1Copy3Gated := buildTestPod("default", "pod-rep1-3", WithControllerOwnerRef(replicaSet1.Name, "ReplicaSet", replicaSet1.UID))
+	setStatusConditionGated(podRep1Copy3Gated)
 
 	job1 := createTestJob("job-1", "default", 10, 10, 0)
 	scheduledPodJob1Copy1 := buildTestPod("default", "scheduled-pod-job1-1", WithControllerOwnerRef(job1.Name, "Job", job1.UID), WithNodeName(node.Name))
@@ -55,6 +64,8 @@ func TestTargetCountInjectionPodListProcessor(t *testing.T) {
 	scheduledParallelStatefulsetPod := buildTestPod("default", "parallel-scheduled-pod-statefulset-1", WithControllerOwnerRef(parallelStatefulset.Name, "StatefulSet", parallelStatefulset.UID), WithNodeName(node.Name))
 	parallelStatefulsetPodCopy1 := buildTestPod("default", "parallel-pod-statefulset1-1", WithControllerOwnerRef(parallelStatefulset.Name, "StatefulSet", parallelStatefulset.UID))
 	parallelStatefulsetPodCopy2 := buildTestPod("default", "parallel-pod-statefulset1-2", WithControllerOwnerRef(parallelStatefulset.Name, "StatefulSet", parallelStatefulset.UID))
+	parallelStatefulsetPodCopy3Gated := buildTestPod("default", "parallel-pod-statefulset1-3", WithControllerOwnerRef(parallelStatefulset.Name, "StatefulSet", parallelStatefulset.UID))
+	setStatusConditionGated(parallelStatefulsetPodCopy3Gated)
 
 	sequentialStatefulset := createTestStatefulset("sequential-statefulset-1", "default", appsv1.OrderedReadyPodManagement, 10)
 	scheduledSequentialStatefulsetPod := buildTestPod("default", "sequential-scheduled-pod-statefulset-1", WithControllerOwnerRef(sequentialStatefulset.Name, "StatefulSet", sequentialStatefulset.UID), WithNodeName(node.Name))
@@ -72,6 +83,7 @@ func TestTargetCountInjectionPodListProcessor(t *testing.T) {
 		name              string
 		scheduledPods     []*apiv1.Pod
 		unschedulablePods []*apiv1.Pod
+		otherPods         []*apiv1.Pod
 		wantPods          []*apiv1.Pod
 	}{
 		{
@@ -111,6 +123,20 @@ func TestTargetCountInjectionPodListProcessor(t *testing.T) {
 				makeFakePods(parallelStatefulset.UID, scheduledParallelStatefulsetPod, 7)...,
 			),
 		},
+		{
+			name:              "Mix of controllers with scheduling gated pods",
+			scheduledPods:     []*apiv1.Pod{scheduledPodRep1Copy1, scheduledPodJob1Copy1, scheduledParallelStatefulsetPod},
+			unschedulablePods: []*apiv1.Pod{podRep1Copy1, podRep1Copy2, podJob1Copy1, podJob1Copy2, parallelStatefulsetPodCopy1, parallelStatefulsetPodCopy2},
+			otherPods:         []*apiv1.Pod{parallelStatefulsetPodCopy3Gated, podRep1Copy3Gated},
+			wantPods: append(
+				append(
+					append(
+						[]*apiv1.Pod{podRep1Copy1, podRep1Copy2, podJob1Copy1, podJob1Copy2, parallelStatefulsetPodCopy1, parallelStatefulsetPodCopy2},
+						makeFakePods(replicaSet1.UID, scheduledPodRep1Copy1, 1)...),
+					makeFakePods(job1.UID, scheduledPodJob1Copy1, 7)...),
+				makeFakePods(parallelStatefulset.UID, scheduledParallelStatefulsetPod, 6)...,
+			),
+		},
 	}
 
 	for _, tc := range testCases {
@@ -119,15 +145,16 @@ func TestTargetCountInjectionPodListProcessor(t *testing.T) {
 			clusterSnapshot := testsnapshot.NewCustomTestSnapshotOrDie(t, store.NewDeltaSnapshotStore(16))
 			err := clusterSnapshot.AddNodeInfo(framework.NewTestNodeInfo(node, tc.scheduledPods...))
 			assert.NoError(t, err)
+			allPodsLister := fakeAllPodsLister{podsToList: append(append(tc.scheduledPods, tc.unschedulablePods...), tc.otherPods...)}
 			ctx := context.AutoscalingContext{
 				AutoscalingKubeClients: context.AutoscalingKubeClients{
-					ListerRegistry: kubernetes.NewListerRegistry(nil, nil, nil, nil, nil, nil, jobLister, replicaSetLister, statefulsetLister),
+					ListerRegistry: kubernetes.NewListerRegistry(nil, nil, &allPodsLister, nil, nil, nil, jobLister, replicaSetLister, statefulsetLister),
 				},
 				ClusterSnapshot: clusterSnapshot,
 			}
 			pods, err := p.Process(&ctx, tc.unschedulablePods)
 			assert.NoError(t, err)
-			assert.ElementsMatch(t, tc.wantPods, pods)
+			assert.Equal(t, tc.wantPods, pods)
 		})
 	}
 }
@@ -433,3 +460,11 @@ func buildTestPod(namespace, name string, opts ...podOption) *apiv1.Pod {
 }
 
 type podOption func(*apiv1.Pod)
+
+type fakeAllPodsLister struct {
+	podsToList []*apiv1.Pod
+}
+
+func (l *fakeAllPodsLister) List() ([]*apiv1.Pod, error) {
+	return l.podsToList, nil
+}

cluster-autoscaler/utils/kubernetes/listers.go

@@ -204,6 +204,30 @@ func SchedulerUnprocessedPods(allPods []*apiv1.Pod, bypassedSchedulers map[strin
 	return unprocessedPods
 }
 
+// SchedulingGatedPods is a helper method that returns all pods which has scheduling gate
+// SchedulingGated pods are not scheduled nor deleted by the implementation and are not unschedulable nor unprocessed by definition
+func SchedulingGatedPods(allPods []*apiv1.Pod) []*apiv1.Pod {
+	var schedulingGatedPods []*apiv1.Pod
+	for _, pod := range allPods {
+		if isSchedulingGated(pod) {
+			schedulingGatedPods = append(schedulingGatedPods, pod)
+		}
+	}
+	return schedulingGatedPods
+}
+
+// isSchedulingGated returns true in case PodScheduled is false with reason PodReasonSchedulingGated
+func isSchedulingGated(pod *apiv1.Pod) bool {
+	if isScheduled(pod) || isDeleted(pod) {
+		return false
+	}
+	_, condition := podv1.GetPodCondition(&pod.Status, apiv1.PodScheduled)
+	if condition != nil && condition.Status == apiv1.ConditionFalse && condition.Reason == apiv1.PodReasonSchedulingGated {
+		return true
+	}
+	return false
+}
+
 // UnschedulablePods is a helper method that returns all unschedulable pods from given pod list.
 func UnschedulablePods(allPods []*apiv1.Pod) []*apiv1.Pod {
 	var unschedulablePods []*apiv1.Pod