Implement podGroup Status

Signed-off-by: helayoty <heelayot@microsoft.com>

pkg/scheduler/schedule_one_podgroup.go

@@ -24,14 +24,18 @@ import (
 	"time"
 
 	v1 "k8s.io/api/core/v1"
+	apimeta "k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	corev1helpers "k8s.io/component-helpers/scheduling/corev1"
 	"k8s.io/klog/v2"
 	fwk "k8s.io/kube-scheduler/framework"
+	schedulingapi "k8s.io/kubernetes/pkg/apis/scheduling"
 	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/scheduler/framework"
 	"k8s.io/kubernetes/pkg/scheduler/metrics"
+	"k8s.io/kubernetes/pkg/scheduler/util"
 	"k8s.io/utils/ptr"
 )
 
@@ -487,12 +491,27 @@ func (sched *Scheduler) submitPodGroupAlgorithmResult(ctx context.Context, sched
 		}
 	}
 	logger := klog.FromContext(ctx)
+	var condition *metav1.Condition
 	switch {
 	case podGroupResult.status.IsSuccess():
+		condition = &metav1.Condition{
+			Type:    schedulingapi.PodGroupScheduled,
+			Status:  metav1.ConditionTrue,
+			Reason:  "Scheduled",
+			Message: podGroupResult.status.Message(),
+		}
 		logger.V(2).Info("Successfully scheduled a pod group", "podGroup", klog.KObj(podGroupInfo), "scheduledPods", scheduledPods, "unschedulablePods", unschedulablePods)
 		metrics.PodGroupScheduled(schedFwk.ProfileName(), metrics.SinceInSeconds(start))
 	case podGroupResult.status.IsRejected():
+		condition = &metav1.Condition{
+			Type:    schedulingapi.PodGroupScheduled,
+			Status:  metav1.ConditionFalse,
+			Reason:  schedulingapi.PodGroupReasonUnschedulable,
+			Message: podGroupResult.status.Message(),
+		}
 		if podGroupResult.waitingOnPreemption {
+			condition.Message = "waiting for preemption to complete"
+
 			logger.V(2).Info("Pod group is waiting for preemption", "podGroup", klog.KObj(podGroupInfo), "unschedulablePods", unschedulablePods)
 			metrics.PodGroupWaitingOnPreemption(schedFwk.ProfileName(), metrics.SinceInSeconds(start))
 		} else {
@@ -500,9 +519,40 @@ func (sched *Scheduler) submitPodGroupAlgorithmResult(ctx context.Context, sched
 			metrics.PodGroupUnschedulable(schedFwk.ProfileName(), metrics.SinceInSeconds(start))
 		}
 	default:
+		condition = &metav1.Condition{
+			Type:    schedulingapi.PodGroupScheduled,
+			Status:  metav1.ConditionFalse,
+			Reason:  schedulingapi.PodGroupReasonSchedulerError,
+			Message: podGroupResult.status.AsError().Error(),
+		}
 		utilruntime.HandleErrorWithContext(ctx, podGroupResult.status.AsError(), "Error scheduling pod group", "podGroup", klog.KObj(podGroupInfo), "errorPods", len(podGroupInfo.QueuedPodInfos))
 		metrics.PodGroupScheduleError(schedFwk.ProfileName(), metrics.SinceInSeconds(start))
 	}
+	if condition != nil {
+		sched.updatePodGroupScheduledCondition(ctx, podGroupInfo, condition)
+	}
+}
+
+// updatePodGroupScheduledCondition patches the given condition on a PodGroup.
+func (sched *Scheduler) updatePodGroupScheduledCondition(ctx context.Context,
+	podGroupInfo *framework.QueuedPodGroupInfo, condition *metav1.Condition) {
+	logger := klog.FromContext(ctx)
+
+	pg, err := sched.podGroupLister.PodGroups(podGroupInfo.Namespace).Get(podGroupInfo.Name)
+	if err != nil {
+		utilruntime.HandleErrorWithLogger(logger, err, "Failed to get PodGroup for status update", "podGroup", klog.KObj(podGroupInfo))
+		return
+	}
+
+	condition.ObservedGeneration = pg.Generation
+	newStatus := pg.Status.DeepCopy()
+	if !apimeta.SetStatusCondition(&newStatus.Conditions, *condition) {
+		return
+	}
+
+	if err := util.PatchPodGroupStatus(ctx, sched.client, podGroupInfo.Name, podGroupInfo.Namespace, &pg.Status, newStatus); err != nil {
+		utilruntime.HandleErrorWithLogger(logger, err, "Failed to update PodGroup status", "podGroup", klog.KObj(podGroupInfo))
+	}
 }
 
 // placementResult associates pod group algorithm result with the placement.

pkg/scheduler/schedule_one_podgroup_test.go

@@ -24,7 +24,11 @@ import (
 	"time"
 
 	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
 	v1 "k8s.io/api/core/v1"
+	schedulingv1alpha2 "k8s.io/api/scheduling/v1alpha2"
+	apimeta "k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
@@ -32,12 +36,15 @@ import (
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/client-go/informers"
 	clientsetfake "k8s.io/client-go/kubernetes/fake"
+	schedulinglisters "k8s.io/client-go/listers/scheduling/v1alpha2"
 	clienttesting "k8s.io/client-go/testing"
+	toolscache "k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/events"
 	featuregatetesting "k8s.io/component-base/featuregate/testing"
 	"k8s.io/klog/v2"
 	"k8s.io/klog/v2/ktesting"
 	fwk "k8s.io/kube-scheduler/framework"
+	schedulingapi "k8s.io/kubernetes/pkg/apis/scheduling"
 	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/scheduler/apis/config"
 	internalcache "k8s.io/kubernetes/pkg/scheduler/backend/cache"
@@ -372,6 +379,7 @@ func TestPodGroupCycle_UpdateSnapshotError(t *testing.T) {
 		Profiles:        profile.Map{"test-scheduler": schedFwk},
 		SchedulingQueue: internalqueue.NewTestQueue(ctx, nil),
 		Cache:           cache,
+		client:          clientsetfake.NewClientset(),
 		FailureHandler: func(ctx context.Context, fwk framework.Framework, p *framework.QueuedPodInfo, status *fwk.Status, ni *fwk.NominatingInfo, start time.Time) {
 			failureHandlerCalled = true
 			if cmp.Equal(updateSnapshotErr.Error(), status.AsError()) {
@@ -897,9 +905,15 @@ func TestSubmitPodGroupAlgorithmResult(t *testing.T) {
 	qInfo2 := &framework.QueuedPodInfo{PodInfo: &framework.PodInfo{Pod: p2}}
 	qInfo3 := &framework.QueuedPodInfo{PodInfo: &framework.PodInfo{Pod: p3}}
 
+	testPG := &schedulingv1alpha2.PodGroup{
+		ObjectMeta: metav1.ObjectMeta{Name: "pg", Namespace: "default"},
+	}
+
 	pgInfo := &framework.QueuedPodGroupInfo{
 		QueuedPodInfos: []*framework.QueuedPodInfo{qInfo1, qInfo2, qInfo3},
 		PodGroupInfo: &framework.PodGroupInfo{
+			Name:            "pg",
+			Namespace:       "default",
 			UnscheduledPods: []*v1.Pod{p1, p2, p3},
 		},
 	}
@@ -910,6 +924,7 @@ func TestSubmitPodGroupAlgorithmResult(t *testing.T) {
 		expectBound      sets.Set[string]
 		expectPreempting sets.Set[string]
 		expectFailed     sets.Set[string]
+		expectCondition  *metav1.Condition
 	}{
 		{
 			name: "All pods feasible",
@@ -930,11 +945,16 @@ func TestSubmitPodGroupAlgorithmResult(t *testing.T) {
 				}},
 			},
 			expectBound: sets.New("p1", "p2", "p3"),
+			expectCondition: &metav1.Condition{
+				Type:   schedulingapi.PodGroupScheduled,
+				Status: metav1.ConditionTrue,
+				Reason: "Scheduled",
+			},
 		},
 		{
 			name: "All pods feasible, but all waiting",
 			algorithmResult: podGroupAlgorithmResult{
-				status: fwk.NewStatus(fwk.Unschedulable),
+				status: fwk.NewStatus(fwk.Unschedulable, "not enough capacity for the gang"),
 				podResults: []algorithmResult{{
 					scheduleResult: ScheduleResult{SuggestedHost: "node1"},
 					status:         nil,
@@ -950,6 +970,12 @@ func TestSubmitPodGroupAlgorithmResult(t *testing.T) {
 				}},
 			},
 			expectFailed: sets.New("p1", "p2", "p3"),
+			expectCondition: &metav1.Condition{
+				Type:    schedulingapi.PodGroupScheduled,
+				Status:  metav1.ConditionFalse,
+				Reason:  schedulingapi.PodGroupReasonUnschedulable,
+				Message: "not enough capacity for the gang",
+			},
 		},
 		{
 			name: "All pods feasible, but last waiting",
@@ -970,6 +996,11 @@ func TestSubmitPodGroupAlgorithmResult(t *testing.T) {
 				}},
 			},
 			expectBound: sets.New("p1", "p2", "p3"),
+			expectCondition: &metav1.Condition{
+				Type:   schedulingapi.PodGroupScheduled,
+				Status: metav1.ConditionTrue,
+				Reason: "Scheduled",
+			},
 		},
 		{
 			name: "All pods feasible, one waiting, one unschedulable",
@@ -990,6 +1021,11 @@ func TestSubmitPodGroupAlgorithmResult(t *testing.T) {
 			},
 			expectBound:  sets.New("p1", "p3"),
 			expectFailed: sets.New("p2"),
+			expectCondition: &metav1.Condition{
+				Type:   schedulingapi.PodGroupScheduled,
+				Status: metav1.ConditionTrue,
+				Reason: "Scheduled",
+			},
 		},
 		{
 			name: "All pods require preemption",
@@ -1023,11 +1059,17 @@ func TestSubmitPodGroupAlgorithmResult(t *testing.T) {
 				}},
 			},
 			expectPreempting: sets.New("p1", "p2", "p3"),
+			expectCondition: &metav1.Condition{
+				Type:    schedulingapi.PodGroupScheduled,
+				Status:  metav1.ConditionFalse,
+				Reason:  schedulingapi.PodGroupReasonUnschedulable,
+				Message: "waiting for preemption to complete",
+			},
 		},
 		{
 			name: "All pods require preemption, but waiting",
 			algorithmResult: podGroupAlgorithmResult{
-				status: fwk.NewStatus(fwk.Unschedulable),
+				status: fwk.NewStatus(fwk.Unschedulable, "preemption required but not feasible"),
 				podResults: []algorithmResult{{
 					scheduleResult: ScheduleResult{
 						SuggestedHost:  "node1",
@@ -1055,6 +1097,12 @@ func TestSubmitPodGroupAlgorithmResult(t *testing.T) {
 				}},
 			},
 			expectFailed: sets.New("p1", "p2", "p3"),
+			expectCondition: &metav1.Condition{
+				Type:    schedulingapi.PodGroupScheduled,
+				Status:  metav1.ConditionFalse,
+				Reason:  schedulingapi.PodGroupReasonUnschedulable,
+				Message: "preemption required but not feasible",
+			},
 		},
 		{
 			name: "One pod requires preemption, but waiting, two are feasible",
@@ -1080,6 +1128,12 @@ func TestSubmitPodGroupAlgorithmResult(t *testing.T) {
 				}},
 			},
 			expectPreempting: sets.New("p1", "p2", "p3"),
+			expectCondition: &metav1.Condition{
+				Type:    schedulingapi.PodGroupScheduled,
+				Status:  metav1.ConditionFalse,
+				Reason:  schedulingapi.PodGroupReasonUnschedulable,
+				Message: "waiting for preemption to complete",
+			},
 		},
 		{
 			name: "One pod unschedulable, one requires preemption, one feasible",
@@ -1104,11 +1158,17 @@ func TestSubmitPodGroupAlgorithmResult(t *testing.T) {
 			},
 			expectPreempting: sets.New("p1", "p3"),
 			expectFailed:     sets.New("p2"),
+			expectCondition: &metav1.Condition{
+				Type:    schedulingapi.PodGroupScheduled,
+				Status:  metav1.ConditionFalse,
+				Reason:  schedulingapi.PodGroupReasonUnschedulable,
+				Message: "waiting for preemption to complete",
+			},
 		},
 		{
 			name: "All pods unschedulable",
 			algorithmResult: podGroupAlgorithmResult{
-				status: fwk.NewStatus(fwk.Unschedulable),
+				status: fwk.NewStatus(fwk.Unschedulable, "0/3 nodes are available: insufficient cpu"),
 				podResults: []algorithmResult{{
 					scheduleResult: ScheduleResult{SuggestedHost: "", nominatingInfo: clearNominatedNode},
 					status:         fwk.NewStatus(fwk.Unschedulable),
@@ -1122,11 +1182,17 @@ func TestSubmitPodGroupAlgorithmResult(t *testing.T) {
 			},
 			expectBound:  sets.New[string](),
 			expectFailed: sets.New("p1", "p2", "p3"),
+			expectCondition: &metav1.Condition{
+				Type:    schedulingapi.PodGroupScheduled,
+				Status:  metav1.ConditionFalse,
+				Reason:  schedulingapi.PodGroupReasonUnschedulable,
+				Message: "0/3 nodes are available: insufficient cpu",
+			},
 		},
 		{
 			name: "All pods unschedulable with nil nominatingInfo",
 			algorithmResult: podGroupAlgorithmResult{
-				status: fwk.NewStatus(fwk.Unschedulable),
+				status: fwk.NewStatus(fwk.Unschedulable, "0/3 nodes are available: insufficient cpu"),
 				podResults: []algorithmResult{{
 					scheduleResult: ScheduleResult{SuggestedHost: "", nominatingInfo: nil},
 					status:         fwk.NewStatus(fwk.Unschedulable),
@@ -1140,15 +1206,27 @@ func TestSubmitPodGroupAlgorithmResult(t *testing.T) {
 			},
 			expectBound:  sets.New[string](),
 			expectFailed: sets.New("p1", "p2", "p3"),
+			expectCondition: &metav1.Condition{
+				Type:    schedulingapi.PodGroupScheduled,
+				Status:  metav1.ConditionFalse,
+				Reason:  schedulingapi.PodGroupReasonUnschedulable,
+				Message: "0/3 nodes are available: insufficient cpu",
+			},
 		},
 		{
 			name: "Unschedulable for the entire pod group",
 			algorithmResult: podGroupAlgorithmResult{
-				status:     fwk.NewStatus(fwk.Unschedulable),
+				status:     fwk.NewStatus(fwk.Unschedulable, "node affinity mismatch"),
 				podResults: []algorithmResult{},
 			},
 			expectBound:  sets.New[string](),
 			expectFailed: sets.New("p1", "p2", "p3"),
+			expectCondition: &metav1.Condition{
+				Type:    schedulingapi.PodGroupScheduled,
+				Status:  metav1.ConditionFalse,
+				Reason:  schedulingapi.PodGroupReasonUnschedulable,
+				Message: "node affinity mismatch",
+			},
 		},
 		{
 			name: "Error for one pod",
@@ -1197,7 +1275,7 @@ func TestSubmitPodGroupAlgorithmResult(t *testing.T) {
 			preemptingPods := sets.New[string]()
 			failedPods := sets.New[string]()
 
-			client := clientsetfake.NewClientset(testNode)
+			client := clientsetfake.NewClientset(testNode, testPG)
 			client.PrependReactor("create", "pods", func(action clienttesting.Action) (bool, runtime.Object, error) {
 				if action.GetSubresource() != "binding" {
 					return false, nil, nil
@@ -1239,7 +1317,12 @@ func TestSubmitPodGroupAlgorithmResult(t *testing.T) {
 			cache := internalcache.New(ctx, nil, true)
 			cache.AddNode(klog.FromContext(ctx), testNode)
 
+			pgIndexer := toolscache.NewIndexer(toolscache.MetaNamespaceKeyFunc, toolscache.Indexers{toolscache.NamespaceIndex: toolscache.MetaNamespaceIndexFunc})
+			pgIndexer.Add(testPG)
+
 			sched := &Scheduler{
+				client:          client,
+				podGroupLister:  schedulinglisters.NewPodGroupLister(pgIndexer),
 				Cache:           cache,
 				Profiles:        profile.Map{"test-scheduler": schedFwk},
 				SchedulingQueue: internalqueue.NewTestQueue(ctx, nil),
@@ -1279,6 +1362,163 @@ func TestSubmitPodGroupAlgorithmResult(t *testing.T) {
 			if !tt.expectFailed.Equal(failedPods) {
 				t.Errorf("Expected failed pods: %v, but got: %v", tt.expectFailed, failedPods)
 			}
+
+			updatedPG, err := client.SchedulingV1alpha2().PodGroups("default").Get(ctx, "pg", metav1.GetOptions{})
+			if err != nil {
+				t.Fatalf("Failed to get PodGroup: %v", err)
+			}
+			cond := apimeta.FindStatusCondition(updatedPG.Status.Conditions, schedulingapi.PodGroupScheduled)
+			if tt.expectCondition != nil {
+				if cond == nil {
+					t.Fatalf("Expected PodGroupScheduled condition to be set, but it was not found")
+				}
+				if diff := cmp.Diff(*tt.expectCondition, *cond, cmpopts.IgnoreFields(metav1.Condition{}, "LastTransitionTime")); diff != "" {
+					t.Errorf("Unexpected PodGroupScheduled condition (-want +got):\n%s", diff)
+				}
+			}
+		})
+	}
+}
+
+func TestUpdatePodGroupScheduledCondition(t *testing.T) {
+	tests := []struct {
+		name            string
+		existingPG      *schedulingv1alpha2.PodGroup
+		namespace       string
+		pgName          string
+		condition       *metav1.Condition
+		expectCondition bool
+	}{
+		{
+			name: "set Scheduled condition to True on empty status",
+			existingPG: &schedulingv1alpha2.PodGroup{
+				ObjectMeta: metav1.ObjectMeta{Name: "pg1", Namespace: "ns1"},
+			},
+			namespace: "ns1",
+			pgName:    "pg1",
+			condition: &metav1.Condition{
+				Type:    schedulingapi.PodGroupScheduled,
+				Status:  metav1.ConditionTrue,
+				Reason:  "SomeReason",
+				Message: "All required pods have been successfully scheduled",
+			},
+			expectCondition: true,
+		},
+		{
+			name: "set Scheduled condition to False with Unschedulable reason",
+			existingPG: &schedulingv1alpha2.PodGroup{
+				ObjectMeta: metav1.ObjectMeta{Name: "pg2", Namespace: "ns1"},
+			},
+			namespace: "ns1",
+			pgName:    "pg2",
+			condition: &metav1.Condition{
+				Type:    schedulingapi.PodGroupScheduled,
+				Status:  metav1.ConditionFalse,
+				Reason:  schedulingapi.PodGroupReasonUnschedulable,
+				Message: "0/3 nodes are available: insufficient cpu",
+			},
+			expectCondition: true,
+		},
+		{
+			name: "set Scheduled condition to False with SchedulerError reason",
+			existingPG: &schedulingv1alpha2.PodGroup{
+				ObjectMeta: metav1.ObjectMeta{Name: "pg3", Namespace: "ns1"},
+			},
+			namespace: "ns1",
+			pgName:    "pg3",
+			condition: &metav1.Condition{
+				Type:    schedulingapi.PodGroupScheduled,
+				Status:  metav1.ConditionFalse,
+				Reason:  schedulingapi.PodGroupReasonSchedulerError,
+				Message: "Internal scheduling error",
+			},
+			expectCondition: true,
+		},
+		{
+			name: "transition from Unschedulable to Scheduled",
+			existingPG: &schedulingv1alpha2.PodGroup{
+				ObjectMeta: metav1.ObjectMeta{Name: "pg4", Namespace: "ns1"},
+				Status: schedulingv1alpha2.PodGroupStatus{
+					Conditions: []metav1.Condition{
+						{
+							Type:               schedulingapi.PodGroupScheduled,
+							Status:             metav1.ConditionFalse,
+							Reason:             schedulingapi.PodGroupReasonUnschedulable,
+							Message:            "previously unschedulable",
+							LastTransitionTime: metav1.Now(),
+						},
+					},
+				},
+			},
+			namespace: "ns1",
+			pgName:    "pg4",
+			condition: &metav1.Condition{
+				Type:    schedulingapi.PodGroupScheduled,
+				Status:  metav1.ConditionTrue,
+				Reason:  "SomeReason",
+				Message: "All required pods have been successfully scheduled",
+			},
+			expectCondition: true,
+		},
+		{
+			name:      "PodGroup not found does not panic",
+			namespace: "ns1",
+			pgName:    "nonexistent",
+			condition: &metav1.Condition{
+				Type:    schedulingapi.PodGroupScheduled,
+				Status:  metav1.ConditionTrue,
+				Reason:  "SomeReason",
+				Message: "test",
+			},
+			expectCondition: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, ctx := ktesting.NewTestContext(t)
+
+			var client *clientsetfake.Clientset
+			pgIndexer := toolscache.NewIndexer(toolscache.MetaNamespaceKeyFunc, toolscache.Indexers{toolscache.NamespaceIndex: toolscache.MetaNamespaceIndexFunc})
+			if tt.existingPG != nil {
+				client = clientsetfake.NewClientset(tt.existingPG)
+				pgIndexer.Add(tt.existingPG)
+			} else {
+				client = clientsetfake.NewClientset()
+			}
+			sched := &Scheduler{client: client, podGroupLister: schedulinglisters.NewPodGroupLister(pgIndexer)}
+
+			pgInfo := &framework.QueuedPodGroupInfo{
+				PodGroupInfo: &framework.PodGroupInfo{
+					Namespace: tt.namespace,
+					Name:      tt.pgName,
+				},
+			}
+			sched.updatePodGroupScheduledCondition(ctx, pgInfo, tt.condition)
+
+			if !tt.expectCondition {
+				return
+			}
+
+			updatedPG, err := client.SchedulingV1alpha2().PodGroups(tt.namespace).Get(ctx, tt.pgName, metav1.GetOptions{})
+			if err != nil {
+				t.Fatalf("Failed to get PodGroup: %v", err)
+			}
+
+			cond := apimeta.FindStatusCondition(updatedPG.Status.Conditions, tt.condition.Type)
+			if cond == nil {
+				t.Fatalf("Expected %s condition to be set, but it was not found", tt.condition.Type)
+			}
+
+			if cond.Status != tt.condition.Status {
+				t.Errorf("Condition status: got %v, want %v", cond.Status, tt.condition.Status)
+			}
+			if cond.Reason != tt.condition.Reason {
+				t.Errorf("Condition reason: got %v, want %v", cond.Reason, tt.condition.Reason)
+			}
+			if cond.Message != tt.condition.Message {
+				t.Errorf("Condition message: got %v, want %v", cond.Message, tt.condition.Message)
+			}
 		})
 	}
 }

pkg/scheduler/scheduler.go

@@ -31,6 +31,7 @@ import (
 	"k8s.io/client-go/informers"
 	coreinformers "k8s.io/client-go/informers/core/v1"
 	clientset "k8s.io/client-go/kubernetes"
+	schedulinglisters "k8s.io/client-go/listers/scheduling/v1alpha2"
 	restclient "k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
 	resourceslicetracker "k8s.io/dynamic-resource-allocation/resourceslice/tracker"
@@ -113,6 +114,8 @@ type Scheduler struct {
 	// panic.
 	logger klog.Logger
 
+	podGroupLister schedulinglisters.PodGroupLister
+
 	// registeredHandlers contains the registrations of all handlers. It's used to check if all handlers have finished syncing before the scheduling cycles start.
 	registeredHandlers []cache.ResourceEventHandlerRegistration
 
@@ -309,6 +312,7 @@ func New(ctx context.Context,
 
 	podLister := informerFactory.Core().V1().Pods().Lister()
 	nodeLister := informerFactory.Core().V1().Nodes().Lister()
+	podGroupLister := informerFactory.Scheduling().V1alpha2().PodGroups().Lister()
 
 	snapshot := internalcache.NewEmptySnapshot()
 	metricsRecorder := metrics.NewMetricsAsyncRecorder(1000, time.Second, stopEverything)
@@ -445,6 +449,8 @@ func New(ctx context.Context,
 		logger:                                 logger,
 		APIDispatcher:                          apiDispatcher,
 		nominatedNodeNameForExpectationEnabled: feature.DefaultFeatureGate.Enabled(features.NominatedNodeNameForExpectation),
+		PodGroupManager:                        podGroupManager,
+		podGroupLister:                         podGroupLister,
 		genericWorkloadEnabled:                 feature.DefaultFeatureGate.Enabled(features.GenericWorkload),
 	}
 	sched.NextPod = podQueue.Pop

pkg/scheduler/util/utils.go

@@ -25,6 +25,7 @@ import (
 
 	v1 "k8s.io/api/core/v1"
 	resourceapi "k8s.io/api/resource/v1"
+	schedulingv1alpha2 "k8s.io/api/scheduling/v1alpha2"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -133,6 +134,41 @@ func PatchPodStatus(ctx context.Context, cs kubernetes.Interface, name string, n
 	return retry.OnError(retry.DefaultBackoff, Retriable, patchFn)
 }
 
+// PatchPodGroupStatus calculates the delta bytes change from <old.Status> to <newStatus>,
+// and then submits a request to API server to patch the PodGroup status changes.
+func PatchPodGroupStatus(ctx context.Context, cs kubernetes.Interface, name string,
+	namespace string, oldStatus *schedulingv1alpha2.PodGroupStatus,
+	newStatus *schedulingv1alpha2.PodGroupStatus) error {
+	if newStatus == nil {
+		return nil
+	}
+
+	oldData, err := json.Marshal(schedulingv1alpha2.PodGroup{Status: *oldStatus})
+	if err != nil {
+		return err
+	}
+
+	newData, err := json.Marshal(schedulingv1alpha2.PodGroup{Status: *newStatus})
+	if err != nil {
+		return err
+	}
+	patchBytes, err := strategicpatch.CreateTwoWayMergePatch(oldData, newData, &schedulingv1alpha2.PodGroup{})
+	if err != nil {
+		return fmt.Errorf("failed to create merge patch for podgroup %q/%q: %w", namespace, name, err)
+	}
+
+	if "{}" == string(patchBytes) {
+		return nil
+	}
+
+	patchFn := func() error {
+		_, err := cs.SchedulingV1alpha2().PodGroups(namespace).Patch(ctx, name, types.StrategicMergePatchType, patchBytes, metav1.PatchOptions{}, "status")
+		return err
+	}
+
+	return retry.OnError(retry.DefaultBackoff, Retriable, patchFn)
+}
+
 // DeletePod deletes the given <pod> from API server
 func DeletePod(ctx context.Context, cs kubernetes.Interface, pod *v1.Pod) error {
 	return cs.CoreV1().Pods(pod.Namespace).Delete(ctx, pod.Name, metav1.DeleteOptions{})

pkg/scheduler/util/utils_test.go

@@ -27,6 +27,7 @@ import (
 	"github.com/google/go-cmp/cmp"
 
 	v1 "k8s.io/api/core/v1"
+	schedulingv1alpha2 "k8s.io/api/scheduling/v1alpha2"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/net"
@@ -333,6 +334,195 @@ func TestPatchPodStatus(t *testing.T) {
 	}
 }
 
+func TestPatchPodGroupStatus(t *testing.T) {
+	now := metav1.NewTime(time.Now().Truncate(time.Second))
+
+	tests := []struct {
+		name   string
+		pg     schedulingv1alpha2.PodGroup
+		client *clientsetfake.Clientset
+		// validateErr checks if error returned from PatchPodGroupStatus is expected one or not.
+		// (true means error is expected one.)
+		validateErr    func(goterr error) bool
+		statusToUpdate *schedulingv1alpha2.PodGroupStatus
+	}{
+		{
+			name:   "Should update podgroup conditions successfully",
+			client: clientsetfake.NewClientset(),
+			pg: schedulingv1alpha2.PodGroup{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "ns",
+					Name:      "pg1",
+				},
+			},
+			statusToUpdate: &schedulingv1alpha2.PodGroupStatus{
+				Conditions: []metav1.Condition{
+					{
+						Type:               "PodGroupScheduled",
+						Status:             metav1.ConditionFalse,
+						Reason:             "Unschedulable",
+						Message:            "not enough capacity for the gang",
+						LastTransitionTime: now,
+					},
+				},
+			},
+		},
+		{
+			name:   "no-op when status is unchanged",
+			client: clientsetfake.NewClientset(),
+			pg: schedulingv1alpha2.PodGroup{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "ns",
+					Name:      "pg1",
+				},
+				Status: schedulingv1alpha2.PodGroupStatus{
+					Conditions: []metav1.Condition{
+						{
+							Type:               "PodGroupScheduled",
+							Status:             metav1.ConditionFalse,
+							Reason:             "Unschedulable",
+							Message:            "not enough capacity",
+							LastTransitionTime: now,
+						},
+					},
+				},
+			},
+			statusToUpdate: &schedulingv1alpha2.PodGroupStatus{
+				Conditions: []metav1.Condition{
+					{
+						Type:               "PodGroupScheduled",
+						Status:             metav1.ConditionFalse,
+						Reason:             "Unschedulable",
+						Message:            "not enough capacity",
+						LastTransitionTime: now,
+					},
+				},
+			},
+		},
+		{
+			name:   "nil newStatus returns nil",
+			client: clientsetfake.NewClientset(),
+			pg: schedulingv1alpha2.PodGroup{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "ns",
+					Name:      "pg1",
+				},
+			},
+			statusToUpdate: nil,
+		},
+		{
+			name: "retry patch request when a 'connection refused' error is returned",
+			client: func() *clientsetfake.Clientset {
+				client := clientsetfake.NewClientset()
+
+				reqcount := 0
+				client.PrependReactor("patch", "podgroups", func(action clienttesting.Action) (bool, runtime.Object, error) {
+					defer func() { reqcount++ }()
+					if reqcount == 0 {
+						return true, &schedulingv1alpha2.PodGroup{}, fmt.Errorf("connection refused: %w", syscall.ECONNREFUSED)
+					}
+					if reqcount == 1 {
+						return false, &schedulingv1alpha2.PodGroup{}, nil
+					}
+					return true, nil, errors.New("requests comes in more than three times.")
+				})
+
+				return client
+			}(),
+			pg: schedulingv1alpha2.PodGroup{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "ns",
+					Name:      "pg1",
+				},
+			},
+			statusToUpdate: &schedulingv1alpha2.PodGroupStatus{
+				Conditions: []metav1.Condition{
+					{
+						Type:               "PodGroupScheduled",
+						Status:             metav1.ConditionFalse,
+						Reason:             "Unschedulable",
+						Message:            "not enough capacity for the gang",
+						LastTransitionTime: now,
+					},
+				},
+			},
+		},
+		{
+			name: "only 4 retries at most",
+			client: func() *clientsetfake.Clientset {
+				client := clientsetfake.NewClientset()
+
+				reqcount := 0
+				client.PrependReactor("patch", "podgroups", func(action clienttesting.Action) (bool, runtime.Object, error) {
+					defer func() { reqcount++ }()
+					if reqcount >= 4 {
+						return true, nil, errors.New("requests comes in more than four times.")
+					}
+					return true, &schedulingv1alpha2.PodGroup{}, fmt.Errorf("connection refused: %w", syscall.ECONNREFUSED)
+				})
+
+				return client
+			}(),
+			pg: schedulingv1alpha2.PodGroup{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "ns",
+					Name:      "pg1",
+				},
+			},
+			validateErr: net.IsConnectionRefused,
+			statusToUpdate: &schedulingv1alpha2.PodGroupStatus{
+				Conditions: []metav1.Condition{
+					{
+						Type:               "PodGroupScheduled",
+						Status:             metav1.ConditionFalse,
+						Reason:             "Unschedulable",
+						Message:            "not enough capacity for the gang",
+						LastTransitionTime: now,
+					},
+				},
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			client := tc.client
+			_, err := client.SchedulingV1alpha2().PodGroups(tc.pg.Namespace).Create(context.TODO(), &tc.pg, metav1.CreateOptions{})
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			_, ctx := ktesting.NewTestContext(t)
+			ctx, cancel := context.WithCancel(ctx)
+			defer cancel()
+
+			err = PatchPodGroupStatus(ctx, client, tc.pg.Name, tc.pg.Namespace, &tc.pg.Status, tc.statusToUpdate)
+			if err != nil && tc.validateErr == nil {
+				t.Fatal(err)
+			}
+			if tc.validateErr != nil {
+				if !tc.validateErr(err) {
+					t.Fatalf("Returned unexpected error: %v", err)
+				}
+				return
+			}
+
+			retrievedPG, err := client.SchedulingV1alpha2().PodGroups(tc.pg.Namespace).Get(ctx, tc.pg.Name, metav1.GetOptions{})
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			wantStatus := tc.pg.Status
+			if tc.statusToUpdate != nil {
+				wantStatus = *tc.statusToUpdate
+			}
+			if diff := cmp.Diff(wantStatus, retrievedPG.Status); diff != "" {
+				t.Errorf("unexpected podgroup status (-want,+got):\n%s", diff)
+			}
+		})
+	}
+}
+
 // Test_As tests the As function with Pod.
 func Test_As_Pod(t *testing.T) {
 	tests := []struct {

test/integration/scheduler/podgroup/podgroup_test.go

@@ -24,10 +24,12 @@ import (
 	v1 "k8s.io/api/core/v1"
 	schedulingapi "k8s.io/api/scheduling/v1alpha2"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	apimeta "k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	featuregatetesting "k8s.io/component-base/featuregate/testing"
+	scheduling "k8s.io/kubernetes/pkg/apis/scheduling"
 	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/scheduler"
 	"k8s.io/kubernetes/pkg/scheduler/backend/queue"
@@ -85,6 +87,13 @@ func TestPodGroupScheduling(t *testing.T) {
 		numUnschedulable int
 	}
 
+	type waitForPodGroupCondition struct {
+		podGroupName    string
+		conditionType   string
+		conditionStatus metav1.ConditionStatus
+		reason          string
+	}
+
 	// step represents a single step in a test scenario.
 	type step struct {
 		name                         string
@@ -95,6 +104,7 @@ func TestPodGroupScheduling(t *testing.T) {
 		waitForPodsUnschedulable     []string
 		waitForPodsScheduled         []string
 		waitForAnyPodsScheduled      *waitForAnyPodsScheduled
+		waitForPodGroupCondition     *waitForPodGroupCondition
 	}
 
 	tests := []struct {
@@ -383,6 +393,83 @@ func TestPodGroupScheduling(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "gang PodGroup status set to Scheduled when all pods scheduled",
+			steps: []step{
+				{
+					name:           "Create the PodGroup object",
+					createPodGroup: gangPodGroup,
+				},
+				{
+					name:       "Create all pods belonging to the gang",
+					createPods: []*v1.Pod{p1, p2, p3},
+				},
+				{
+					name:                 "Verify all gang pods are scheduled",
+					waitForPodsScheduled: []string{"p1", "p2", "p3"},
+				},
+				{
+					name: "Verify PodGroup condition is Scheduled",
+					waitForPodGroupCondition: &waitForPodGroupCondition{
+						podGroupName:    "pg1",
+						conditionType:   scheduling.PodGroupScheduled,
+						conditionStatus: metav1.ConditionTrue,
+						reason:          scheduling.PodGroupReasonScheduled,
+					},
+				},
+			},
+		},
+		{
+			name: "gang PodGroup status transitions from Unschedulable to Scheduled",
+			steps: []step{
+				{
+					name:       "Create the resource-blocking pod",
+					createPods: []*v1.Pod{blockerPod},
+				},
+				{
+					name:                 "Schedule the resource-blocking pod",
+					waitForPodsScheduled: []string{"blocker"},
+				},
+				{
+					name:           "Create the PodGroup object",
+					createPodGroup: gangPodGroup,
+				},
+				{
+					name:       "Create gang pods",
+					createPods: []*v1.Pod{p1, p2, p3},
+				},
+				{
+					name:                     "Verify pods become unschedulable",
+					waitForPodsUnschedulable: []string{"p1", "p2", "p3"},
+				},
+				{
+					name: "Verify PodGroup condition is Unschedulable",
+					waitForPodGroupCondition: &waitForPodGroupCondition{
+						podGroupName:    "pg1",
+						conditionType:   scheduling.PodGroupScheduled,
+						conditionStatus: metav1.ConditionFalse,
+						reason:          scheduling.PodGroupReasonUnschedulable,
+					},
+				},
+				{
+					name:       "Delete the resource-blocking pod",
+					deletePods: []string{"blocker"},
+				},
+				{
+					name:                 "Verify the entire gang is now scheduled",
+					waitForPodsScheduled: []string{"p1", "p2", "p3"},
+				},
+				{
+					name: "Verify PodGroup condition transitions to Scheduled",
+					waitForPodGroupCondition: &waitForPodGroupCondition{
+						podGroupName:    "pg1",
+						conditionType:   scheduling.PodGroupScheduled,
+						conditionStatus: metav1.ConditionTrue,
+						reason:          scheduling.PodGroupReasonScheduled,
+					},
+				},
+			},
+		},
 	}
 
 	for _, tt := range tests {
@@ -496,6 +583,25 @@ func TestPodGroupScheduling(t *testing.T) {
 					if err != nil {
 						t.Fatalf("Step %d: Failed to wait for pods to be scheduled or unschedulable: %v", i, err)
 					}
+				case step.waitForPodGroupCondition != nil:
+					wc := step.waitForPodGroupCondition
+					err := wait.PollUntilContextTimeout(testCtx.Ctx, 200*time.Millisecond, wait.ForeverTestTimeout, false,
+						func(ctx context.Context) (bool, error) {
+							pg, err := cs.SchedulingV1alpha2().PodGroups(ns).Get(ctx, wc.podGroupName, metav1.GetOptions{})
+							if err != nil {
+								return false, err
+							}
+							cond := apimeta.FindStatusCondition(pg.Status.Conditions, wc.conditionType)
+							if cond == nil {
+								return false, nil
+							}
+							return cond.Status == wc.conditionStatus && (wc.reason == "" || cond.Reason == wc.reason), nil
+						},
+					)
+					if err != nil {
+						t.Fatalf("Step %d: Timed out waiting for PodGroup %s condition %s=%s reason=%s: %v",
+							i, wc.podGroupName, wc.conditionType, wc.conditionStatus, wc.reason, err)
+					}
 				}
 			}
 		})