feat(scheduler): account for NvFractions GPU memory

.changes/unreleased/added-20260810-110236.yaml

@@ -0,0 +1,3 @@
+kind: Added
+body: |-
+  Schedule NvFractions GPU memory requests

.changes/unreleased/fixed-20260810-125620.yaml

@@ -0,0 +1,3 @@
+kind: Fixed
+body: |-
+  Preserve reservation pods during BindRequest cache lag

pkg/binder/binding/resourcereservation/resource_reservation.go

@@ -226,26 +226,66 @@ func (rsc *service) syncForPods(ctx context.Context, pods []*v1.Pod, gpuGroupToS
 	return nil
 }
 
-// hasActiveBindRequestsForGpuGroup checks if any non-terminal BindRequests reference
-// the given GPU group. This prevents premature reservation pod deletion when the
-// informer cache has not yet propagated GPU group labels on recently-bound fraction pods.
+// hasActiveBindRequestsForGpuGroup checks if BindRequests still protect the GPU group.
+// A succeeded BindRequest also protects the reservation while its pod is still alive:
+// the pod label can lag behind the BindRequest status in the controller cache.
 func (rsc *service) hasActiveBindRequestsForGpuGroup(ctx context.Context, gpuGroup string) (bool, error) {
 	bindRequestList := &schedulingv1alpha2.BindRequestList{}
 	if err := rsc.kubeClient.List(ctx, bindRequestList); err != nil {
 		return false, fmt.Errorf("failed to list BindRequests: %w", err)
 	}
 
 	for _, br := range bindRequestList.Items {
-		if br.Status.Phase == schedulingv1alpha2.BindRequestPhaseSucceeded ||
-			br.Status.Phase == schedulingv1alpha2.BindRequestPhaseFailed {
+		if !slices.Contains(br.Spec.SelectedGPUGroups, gpuGroup) {
 			continue
 		}
-		if slices.Contains(br.Spec.SelectedGPUGroups, gpuGroup) {
-			return true, nil
+
+		if br.Status.Phase == schedulingv1alpha2.BindRequestPhaseFailed {
+			continue
 		}
+
+		if br.Status.Phase == schedulingv1alpha2.BindRequestPhaseSucceeded {
+			hasLivePod, err := rsc.hasLivePodForBindRequest(ctx, &br)
+			if err != nil {
+				return false, err
+			}
+			if hasLivePod {
+				return true, nil
+			}
+			continue
+		}
+
+		return true, nil
 	}
 	return false, nil
 }
+
+func (rsc *service) hasLivePodForBindRequest(ctx context.Context, bindRequest *schedulingv1alpha2.BindRequest) (bool, error) {
+	pod := &v1.Pod{}
+	err := rsc.kubeClient.Get(ctx, client.ObjectKey{
+		Namespace: bindRequest.Namespace,
+		Name:      bindRequest.Spec.PodName,
+	}, pod)
+	if apierrors.IsNotFound(err) {
+		return false, nil
+	}
+	if err != nil {
+		return false, fmt.Errorf("failed to get pod for BindRequest <%s/%s>: %w",
+			bindRequest.Namespace, bindRequest.Name, err)
+	}
+
+	if slices.Contains([]v1.PodPhase{v1.PodSucceeded, v1.PodFailed}, pod.Status.Phase) {
+		return false, nil
+	}
+
+	for _, gpuGroup := range bindRequest.Spec.SelectedGPUGroups {
+		if slices.Contains(resources.GetGpuGroups(pod), gpuGroup) {
+			return true, nil
+		}
+	}
+
+	return true, nil
+}
 func (rsc *service) ReserveGpuDevice(ctx context.Context, pod *v1.Pod, nodeName string, gpuGroup string) (string, error) {
 	logger := log.FromContext(ctx)
 

pkg/binder/binding/resourcereservation/resource_reservation_test.go

@@ -1437,6 +1437,47 @@ var _ = Describe("Race condition: reservation pod deleted during concurrent bind
 				"Reservation pod should be deleted when only terminal BindRequests exist")
 		})
 
+		It("should preserve reservation pod when succeeded BindRequest still has a live pod", func() {
+			livePod := &v1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "fraction-pod-1",
+					Namespace: "team-a",
+				},
+				Status: v1.PodStatus{
+					Phase: v1.PodPending,
+				},
+			}
+			succeededBindRequest := &schedulingv1alpha2.BindRequest{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "bind-request-done",
+					Namespace: "team-a",
+				},
+				Spec: schedulingv1alpha2.BindRequestSpec{
+					PodName:           livePod.Name,
+					SelectedNode:      nodeName,
+					SelectedGPUGroups: []string{gpuGroup},
+				},
+				Status: schedulingv1alpha2.BindRequestStatus{
+					Phase: schedulingv1alpha2.BindRequestPhaseSucceeded,
+				},
+			}
+
+			clientWithObjs := fake.NewClientBuilder().WithScheme(testScheme).
+				WithRuntimeObjects(reservationPod.DeepCopy(), livePod, succeededBindRequest).
+				WithIndex(&v1.Pod{}, "spec.nodeName", nodeNameIndexer).Build()
+			rsc := initializeTestService(clientWithObjs)
+
+			err := rsc.SyncForGpuGroup(context.TODO(), gpuGroup)
+			Expect(err).To(Succeed())
+
+			pods := &v1.PodList{}
+			err = clientWithObjs.List(context.Background(), pods,
+				runtimeClient.InNamespace(resourceReservationNameSpace))
+			Expect(err).To(Succeed())
+			Expect(len(pods.Items)).To(Equal(1),
+				"Reservation pod should be preserved while the bound pod may still be missing the gpu-group label")
+		})
+
 		It("should delete reservation pod when only failed BindRequests exist", func() {
 			failedBindRequest := &schedulingv1alpha2.BindRequest{
 				ObjectMeta: metav1.ObjectMeta{

pkg/nodescaleadjuster/scale_adjuster/calculator.go

@@ -77,11 +77,11 @@ func (c *calculator) getGPUFraction(pod *v1.Pod) (float64, error) {
 		}
 		return gpuFraction, nil
 	}
-	if pod.Annotations[constants.GpuMemory] != "" {
-		_, err := resources.GetGPUMemory(pod)
-		if err != nil {
-			return 0, err
-		}
+	gpuMemory, err := resources.GetGPUMemory(pod)
+	if err != nil {
+		return 0, err
+	}
+	if gpuMemory > 0 {
 		return c.gpuMemoryToFractionRatio, nil
 	}
 	return 0, fmt.Errorf("pod %v/%v does not have GPU fraction or memory annotation", pod.Namespace, pod.Name)

pkg/nodescaleadjuster/scale_adjuster/scale_adjuster.go

@@ -14,7 +14,6 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
-	"github.com/kai-scheduler/KAI-scheduler/pkg/common/constants"
 	"github.com/kai-scheduler/KAI-scheduler/pkg/common/resources"
 	"github.com/kai-scheduler/KAI-scheduler/pkg/nodescaleadjuster/scaler"
 )
@@ -160,7 +159,7 @@ func (sa *ScaleAdjuster) getUnschedulablePods() ([]*corev1.Pod, error) {
 		if pod.Spec.SchedulerName != sa.schedulerName {
 			continue
 		}
-		if !requestFractionalGPU(&pod) {
+		if !resources.RequestsGPUFraction(&pod) {
 			continue
 		}
 		if !isPodAlive(&pod) {
@@ -175,10 +174,6 @@ func (sa *ScaleAdjuster) getUnschedulablePods() ([]*corev1.Pod, error) {
 	return pods, nil
 }
 
-func requestFractionalGPU(pod *corev1.Pod) bool {
-	return pod.Annotations[constants.GpuFraction] != "" || pod.Annotations[constants.GpuMemory] != ""
-}
-
 func isPodAlive(pod *corev1.Pod) bool {
 	return !slices.Contains([]corev1.PodPhase{corev1.PodSucceeded, corev1.PodFailed}, pod.Status.Phase)
 }

pkg/podgroupcontroller/controllers/resources/fraction.go

@@ -16,6 +16,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/kai-scheduler/KAI-scheduler/pkg/common/constants"
+	"github.com/kai-scheduler/KAI-scheduler/pkg/common/resources"
 )
 
 const (
@@ -31,28 +32,23 @@ var (
 func calculateAllocatedFraction(
 	ctx context.Context, pod *v1.Pod, kubeClient client.Client,
 ) (resource.Quantity, error) {
-	gpuFractionStr, hasFractionAnnotation := pod.Annotations[constants.GpuFraction]
-	if hasFractionAnnotation {
-		return resource.MustParse(gpuFractionStr), nil
+	req, err := resources.ParsePodGPUFractionRequest(pod)
+	if err != nil {
+		return resource.Quantity{}, fmt.Errorf("failed to parse GPU fraction for pod %s/%s: %s", pod.Namespace, pod.Name, err)
 	}
-
-	gpuMemoryStr, hasMemoryAnnotation := pod.Annotations[constants.GpuMemory]
-	if !hasMemoryAnnotation {
-		return resource.Quantity{}, fmt.Errorf(
-			"cannot calculate fraction because the pod doesn't a fraction or memory annotation")
+	if req == nil {
+		return resource.Quantity{}, fmt.Errorf("cannot calculate fraction because the pod doesn't have a fraction or memory annotation")
 	}
 
-	return getFractionFromMemoryRequest(ctx, gpuMemoryStr, pod.Spec.NodeName, kubeClient)
+	if req.Portion > 0 {
+		return resource.MustParse(fmt.Sprintf("%g", req.Portion)), nil
+	}
+	return getFractionFromMemoryRequest(ctx, req.Memory.Value()/resources.BytesInMiB, pod.Spec.NodeName, kubeClient)
 }
 
 func getFractionFromMemoryRequest(
-	ctx context.Context, gpuMemoryStr string, nodeName string, kubeClient client.Client,
+	ctx context.Context, gpuMemory int64, nodeName string, kubeClient client.Client,
 ) (resource.Quantity, error) {
-	gpuMemory, err := strconv.ParseInt(gpuMemoryStr, 10, 64)
-	if err != nil {
-		return resource.Quantity{}, fmt.Errorf("failed to parse %s annotation to int: %w", constants.GpuMemory, err)
-	}
-
 	nodeGpuMemory, err := getNodeSingleGpuMemory(ctx, nodeName, kubeClient)
 	if err != nil {
 		return resource.Quantity{}, fmt.Errorf("failed extract node gpu memory for node %s : %w",

pkg/podgroupcontroller/controllers/resources/fraction_test.go

@@ -15,6 +15,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 
 	"github.com/kai-scheduler/KAI-scheduler/pkg/common/constants"
+	"github.com/kai-scheduler/KAI-scheduler/pkg/common/resources"
 )
 
 func Test_getReceivedFraction(t *testing.T) {
@@ -53,6 +54,23 @@ func Test_getReceivedFraction(t *testing.T) {
 			resource.MustParse("0.5"),
 			false,
 		},
+		{
+			"NvFractions request + Nvidia node",
+			&v1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{resources.CalcGpuFractionAnnotationForContainer(""): "2000Mi"},
+				},
+				Spec: v1.PodSpec{NodeName: "n1"},
+			},
+			&v1.Node{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:   "n1",
+					Labels: map[string]string{constants.NvidiaGpuMemory: "4000"},
+				},
+			},
+			resource.MustParse("0.5"),
+			false,
+		},
 		{
 			"Memory request + Amd node",
 			&v1.Pod{
@@ -143,8 +161,8 @@ func Test_getReceivedFraction(t *testing.T) {
 
 func Test_getFractionFromMemoryRequest(t *testing.T) {
 	type args struct {
-		gpuMemoryStr string
-		nodeName     string
+		gpuMemory int64
+		nodeName  string
 	}
 	tests := []struct {
 		name    string
@@ -156,8 +174,8 @@ func Test_getFractionFromMemoryRequest(t *testing.T) {
 		{
 			"Node with Nvidia memory label",
 			args{
-				gpuMemoryStr: "2000",
-				nodeName:     "n1",
+				gpuMemory: 2000,
+				nodeName:  "n1",
 			},
 			&v1.Node{
 				ObjectMeta: metav1.ObjectMeta{
@@ -171,8 +189,8 @@ func Test_getFractionFromMemoryRequest(t *testing.T) {
 		{
 			"Node with Amd memory label",
 			args{
-				gpuMemoryStr: "4000",
-				nodeName:     "n1",
+				gpuMemory: 4000,
+				nodeName:  "n1",
 			},
 			&v1.Node{
 				ObjectMeta: metav1.ObjectMeta{
@@ -183,21 +201,6 @@ func Test_getFractionFromMemoryRequest(t *testing.T) {
 			resource.MustParse("0.25"),
 			false,
 		},
-		{
-			"invalid gpu memory value",
-			args{
-				gpuMemoryStr: "abc",
-				nodeName:     "n1",
-			},
-			&v1.Node{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:   "n1",
-					Labels: map[string]string{constants.NvidiaGpuMemory: "4000"},
-				},
-			},
-			resource.Quantity{},
-			true,
-		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -208,7 +211,7 @@ func Test_getFractionFromMemoryRequest(t *testing.T) {
 			}
 			kubeClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(tt.node).Build()
 
-			got, err := getFractionFromMemoryRequest(context.TODO(), tt.args.gpuMemoryStr, tt.args.nodeName, kubeClient)
+			got, err := getFractionFromMemoryRequest(context.TODO(), tt.args.gpuMemory, tt.args.nodeName, kubeClient)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("getFractionFromMemoryRequest() error = %v, wantErr %v", err, tt.wantErr)
 				return