Merge pull request #2036 from kai-scheduler/claude/upstream-resource-helpers-prototype

refactor(scheduler): delegate effective-request accounting to k8s.io/component-helpers

Author: gshaibi

pkg/admission/webhook/v1alpha2/podhooks/pod_resize_validator.go

@@ -10,8 +10,8 @@ import (
 
 	corev1 "k8s.io/api/core/v1"
 	schedulingv1 "k8s.io/api/scheduling/v1"
-	"k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/apimachinery/pkg/runtime"
+	resourcehelpers "k8s.io/component-helpers/resource"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
@@ -24,15 +24,6 @@ import (
 
 var resizeLog = logf.Log.WithName("pod-resize-validator")
 
-func isPodResizeInfeasible(pod *corev1.Pod) bool {
-	for _, c := range pod.Status.Conditions {
-		if c.Type == corev1.PodResizePending {
-			return c.Status == corev1.ConditionTrue && c.Reason == corev1.PodReasonInfeasible
-		}
-	}
-	return false
-}
-
 // memoryLimitBytesPerUnit converts a Queue Memory.Limit (in megabytes) to bytes.
 const memoryLimitBytesPerUnit = 1_000_000
 
@@ -135,135 +126,40 @@ func (v *PodResizeValidator) validateResize(ctx context.Context, oldPod, newPod
 // podResizeDelta computes the net per-resource increase this resize introduces
 // relative to what the queue already accounts for.
 //
-// The queue's Status.Allocated is a pod-level sum of per-container effective
-// requests, so the delta is computed at the same granularity:
+// All three aggregates come from the same upstream helper the scheduler uses in
+// getPodResourceRequest, so the delta baseline is guaranteed to match queue
+// accounting by construction rather than by parallel hand-written logic:
 //
-//  1. For each regular container and each restartable init container (sidecar),
-//     accumulate three pod-level sums: newSpecSum, oldSpecSum, effectiveOldSum.
-//  2. Skip a resource if its pod-level spec is unchanged (newSpecSum[r] ==
-//     oldSpecSum[r]) — that resource was not part of this resize request and
-//     must not generate spurious delta even when an earlier infeasible attempt
-//     left an unresolved stale spec.
-//  3. For changed resources, delta = max(0, newSpecSum[r] - effectiveOldSum[r]).
-//     Pod-level aggregation naturally handles CPU/memory moved between containers
-//     (redistribution produces zero net delta at the pod level).
+//   - newSpec      : the resize target (spec only)
+//   - oldSpec      : the pre-resize target (spec only)
+//   - effectiveOld : what the queue currently charges, i.e. the KEP-1287
+//     effective request max(spec, enacted, allocated), or max(enacted, allocated)
+//     when the kubelet marked the resize Infeasible
 //
-// Effective old baseline per container:
-//   - If old pod has a current Infeasible condition: max(enacted, allocated).
-//     The infeasible spec was never committed by the kubelet, so the queue only
-//     reflects enacted/allocated.
-//   - Otherwise (normal / Deferred / InProgress): max(spec, enacted, allocated),
-//     matching what the scheduler charges for those states.
-//   - Falls back to old spec when no ContainerStatus is available.
+// A resource whose spec is unchanged is skipped: it is not part of this resize,
+// and an unresolved Infeasible spec on it must not produce phantom delta.
 func podResizeDelta(oldPod, newPod *corev1.Pod) corev1.ResourceList {
-	oldInfeasible := isPodResizeInfeasible(oldPod)
-
-	statusByName := make(map[string]*corev1.ContainerStatus, len(oldPod.Status.ContainerStatuses))
-	for i := range oldPod.Status.ContainerStatuses {
-		statusByName[oldPod.Status.ContainerStatuses[i].Name] = &oldPod.Status.ContainerStatuses[i]
-	}
-	oldByName := make(map[string]*corev1.Container, len(oldPod.Spec.Containers))
-	for i := range oldPod.Spec.Containers {
-		oldByName[oldPod.Spec.Containers[i].Name] = &oldPod.Spec.Containers[i]
-	}
-
-	initStatusByName := make(map[string]*corev1.ContainerStatus, len(oldPod.Status.InitContainerStatuses))
-	for i := range oldPod.Status.InitContainerStatuses {
-		initStatusByName[oldPod.Status.InitContainerStatuses[i].Name] = &oldPod.Status.InitContainerStatuses[i]
-	}
-	oldInitByName := make(map[string]*corev1.Container, len(oldPod.Spec.InitContainers))
-	for i := range oldPod.Spec.InitContainers {
-		oldInitByName[oldPod.Spec.InitContainers[i].Name] = &oldPod.Spec.InitContainers[i]
-	}
+	specOnly := resourcehelpers.PodResourcesOptions{}
+	withStatus := resourcehelpers.PodResourcesOptions{UseStatusResources: true}
 
-	newSpecSum := corev1.ResourceList{}
-	oldSpecSum := corev1.ResourceList{}
-	effectiveOldSum := corev1.ResourceList{}
+	newSpec := resourcehelpers.AggregateContainerRequests(newPod, specOnly)
+	oldSpec := resourcehelpers.AggregateContainerRequests(oldPod, specOnly)
+	effectiveOld := resourcehelpers.AggregateContainerRequests(oldPod, withStatus)
 
-	for i := range newPod.Spec.Containers {
-		accumulateDeltaSums(&newPod.Spec.Containers[i], oldByName, statusByName, oldInfeasible, newSpecSum, oldSpecSum, effectiveOldSum)
-	}
-	for i := range newPod.Spec.InitContainers {
-		c := &newPod.Spec.InitContainers[i]
-		if c.RestartPolicy == nil || *c.RestartPolicy != corev1.ContainerRestartPolicyAlways {
-			continue
-		}
-		accumulateDeltaSums(c, oldInitByName, initStatusByName, oldInfeasible, newSpecSum, oldSpecSum, effectiveOldSum)
-	}
-
-	zero := resource.MustParse("0")
 	delta := corev1.ResourceList{}
-	for r, newQty := range newSpecSum {
-		if oldQty, ok := oldSpecSum[r]; ok && newQty.Cmp(oldQty) == 0 {
-			continue // resource not changed by this resize at the pod level
+	for resName, newQty := range newSpec {
+		if oldQty, ok := oldSpec[resName]; ok && newQty.Cmp(oldQty) == 0 {
+			continue // resource not changed by this resize
 		}
-		effectiveOld := effectiveOldSum[r]
 		diff := newQty.DeepCopy()
-		diff.Sub(effectiveOld)
-		if diff.Cmp(zero) > 0 {
-			delta[r] = diff
+		diff.Sub(effectiveOld[resName])
+		if diff.Sign() > 0 {
+			delta[resName] = diff
 		}
 	}
 	return delta
 }
 
-// accumulateDeltaSums accumulates one container's contribution into three pod-level
-// sums. The effective old baseline follows the queue's own accounting:
-//   - infeasible old pod: max(enacted, allocated)
-//   - otherwise:          max(old_spec, enacted, allocated)
-func accumulateDeltaSums(
-	newC *corev1.Container,
-	oldByName map[string]*corev1.Container,
-	statusByName map[string]*corev1.ContainerStatus,
-	oldInfeasible bool,
-	newSpecSum, oldSpecSum, effectiveOldSum corev1.ResourceList,
-) {
-	oldC := oldByName[newC.Name]
-	cs := statusByName[newC.Name]
-
-	for resName, newQty := range newC.Resources.Requests {
-		cur := newSpecSum[resName]
-		cur.Add(newQty)
-		newSpecSum[resName] = cur
-
-		var oldSpecQty resource.Quantity
-		if oldC != nil {
-			oldSpecQty = oldC.Resources.Requests[resName]
-		}
-		cur = oldSpecSum[resName]
-		cur.Add(oldSpecQty)
-		oldSpecSum[resName] = cur
-
-		effectiveOld := oldSpecQty
-		if cs != nil {
-			candidates := []resource.Quantity{}
-			if !oldInfeasible {
-				candidates = append(candidates, oldSpecQty)
-			}
-			if cs.Resources != nil {
-				if enacted, ok := cs.Resources.Requests[resName]; ok {
-					candidates = append(candidates, enacted)
-				}
-			}
-			if alloc, ok := cs.AllocatedResources[resName]; ok {
-				candidates = append(candidates, alloc)
-			}
-			if len(candidates) > 0 {
-				best := candidates[0]
-				for _, q := range candidates[1:] {
-					if q.Cmp(best) > 0 {
-						best = q
-					}
-				}
-				effectiveOld = best
-			}
-		}
-		cur = effectiveOldSum[resName]
-		cur.Add(effectiveOld)
-		effectiveOldSum[resName] = cur
-	}
-}
-
 func checkQueueCapacity(
 	queue *v2.Queue,
 	delta corev1.ResourceList,

pkg/admission/webhook/v1alpha2/podhooks/pod_resize_validator_test.go

@@ -519,3 +519,37 @@ func TestPodResizeValidator_Redistribution_AllowedAtLimit(t *testing.T) {
 	resp := v.Handle(context.Background(), makeRequest(t, oldPod, newPod))
 	assert.True(t, resp.Allowed, "CPU redistribution with unchanged pod total should be allowed even at limit")
 }
+
+// TestPodResizeValidator_InitPeakDominates_NoDelta covers a pod whose queue charge is
+// dominated by the init-phase peak rather than the steady-state sum. Resizing a regular
+// container below that peak does not change what the queue charges, so the delta must be
+// zero even though the container itself grew.
+func TestPodResizeValidator_InitPeakDominates_NoDelta(t *testing.T) {
+	scheme := buildScheme()
+	// Queue is exactly at its limit: 10 CPU limit, 10 CPU allocated (the init peak).
+	queue := newQueue("q", 10000, -1, 0, "10", "0")
+	pg := newPodGroup("pg", "ns", "q")
+	c := fake.NewClientBuilder().WithScheme(scheme).WithObjects(queue, pg).Build()
+	v := NewPodResizeValidator(c, scheme, testSchedulerName, true, false)
+
+	makePod := func(mainCPU string) *corev1.Pod {
+		p := podWithRequests("ns", "p", "pg", testSchedulerName, mainCPU, "0")
+		p.Spec.InitContainers = []corev1.Container{
+			{
+				Name: "heavy-init",
+				Resources: corev1.ResourceRequirements{
+					Requests: corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("10")},
+				},
+			},
+		}
+		return p
+	}
+	// Pod charge = max(steady, initPeak) = max(1,10) = 10 before, max(2,10) = 10 after.
+	oldPod := makePod("1")
+	newPod := makePod("2")
+
+	assert.Empty(t, podResizeDelta(oldPod, newPod), "init-peak-dominated resize should produce no delta")
+
+	resp := v.Handle(context.Background(), makeRequest(t, oldPod, newPod))
+	assert.True(t, resp.Allowed, "queue charge is unchanged, so the resize must be admitted at the limit")
+}