From 57ba60526f6acbe4d16378ebe074ce2187c59588 Mon Sep 17 00:00:00 2001 From: helayoty Date: Thu, 5 Mar 2026 23:10:52 +0000 Subject: [PATCH] Implement podGroup Status Signed-off-by: helayoty --- pkg/scheduler/schedule_one_podgroup.go | 50 ++++ pkg/scheduler/schedule_one_podgroup_test.go | 252 +++++++++++++++++- pkg/scheduler/scheduler.go | 6 + pkg/scheduler/util/utils.go | 36 +++ pkg/scheduler/util/utils_test.go | 190 +++++++++++++ .../scheduler/podgroup/podgroup_test.go | 106 ++++++++ 6 files changed, 634 insertions(+), 6 deletions(-) diff --git a/pkg/scheduler/schedule_one_podgroup.go b/pkg/scheduler/schedule_one_podgroup.go index 1f295873143..b93884825d6 100644 --- a/pkg/scheduler/schedule_one_podgroup.go +++ b/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. diff --git a/pkg/scheduler/schedule_one_podgroup_test.go b/pkg/scheduler/schedule_one_podgroup_test.go index d4531680c15..1e70ffa05af 100644 --- a/pkg/scheduler/schedule_one_podgroup_test.go +++ b/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) + } }) } } diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go index 226e9ce8a85..cb9b92e2cd7 100644 --- a/pkg/scheduler/scheduler.go +++ b/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 diff --git a/pkg/scheduler/util/utils.go b/pkg/scheduler/util/utils.go index 06ba395030d..5bf1a3b4e70 100644 --- a/pkg/scheduler/util/utils.go +++ b/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 to , +// 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 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{}) diff --git a/pkg/scheduler/util/utils_test.go b/pkg/scheduler/util/utils_test.go index 59c0f3d1a19..c53e22aa57d 100644 --- a/pkg/scheduler/util/utils_test.go +++ b/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 { diff --git a/test/integration/scheduler/podgroup/podgroup_test.go b/test/integration/scheduler/podgroup/podgroup_test.go index ef6a58d7d48..77b408a48ff 100644 --- a/test/integration/scheduler/podgroup/podgroup_test.go +++ b/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) + } } } })