github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2026-07-19 23:51:08 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #133045 from natasha41575/retry_pending_resizes

Browse Source 
[FG:InPlacePodVerticalScaling] Retry pending resizes only if aggregated requests are decreasing
This commit is contained in:
Kubernetes Prow Robot
2025-07-18 20:28:31 -07:00
committed by GitHub
parent 5e83b9c2c2 ae5247afc1
commit 7e6f41f9d7
5 changed files with 191 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
pkg/kubelet/allocation/allocation_manager.go
View File

@@ -116,6 +116,9 @@ type Manager interface {
// PushPendingResize queues a pod with a pending resize request for later reevaluation.
PushPendingResize(uid types.UID)
// HasPendingResizes returns whether there are currently any pending resizes.
HasPendingResizes() bool
// RetryPendingResizes retries all pending resizes.
RetryPendingResizes(trigger string)
@@ -449,6 +452,13 @@ func (m *manager) isResizeIncreasingRequests(pod *v1.Pod) bool {
newRequest.Cpu().Cmp(*oldRequest.Cpu()) > 0
}
func (m *manager) HasPendingResizes() bool {
m.allocationMutex.Lock()
defer m.allocationMutex.Unlock()
return len(m.podsWithPendingResizes) > 0
}
// GetContainerResourceAllocation returns the last checkpointed AllocatedResources values
// If checkpoint manager has not been initialized, it returns nil, false
func (m *manager) GetContainerResourceAllocation(podUID types.UID, containerName string) (v1.ResourceRequirements, bool) {
@@ -737,18 +747,6 @@ func (m *manager) canResizePod(allocatedPods []*v1.Pod, pod *v1.Pod) (bool, stri
return false, v1.PodReasonInfeasible, msg
}
for i := range allocatedPods {
for j, c := range allocatedPods[i].Status.ContainerStatuses {
actuatedResources, exists := m.GetActuatedResources(allocatedPods[i].UID, c.Name)
if exists {
// Overwrite the actual resources in the status with the actuated resources.
// This lets us reuse the existing scheduler libraries without having to wait
// for the actual resources in the status to be updated.
allocatedPods[i].Status.ContainerStatuses[j].Resources = &actuatedResources
}
}
}
if ok, failReason, failMessage := m.canAdmitPod(allocatedPods, pod); !ok {
// Log reason and return.
klog.V(3).InfoS("Resize cannot be accommodated", "pod", klog.KObj(pod), "reason", failReason, "message", failMessage)
@@ -770,8 +768,6 @@ func (m *manager) CheckPodResizeInProgress(allocatedPod *v1.Pod, podStatus *kube
if m.recorder != nil {
m.recorder.Eventf(allocatedPod, v1.EventTypeNormal, events.ResizeCompleted, podResizeCompletedEventMsg)
}
// TODO(natasha41575): We only need to make this call if any of the resources were decreased.
m.RetryPendingResizes(TriggerReasonPodResized)
}
}

35
pkg/kubelet/kubelet.go
View File

@@ -70,6 +70,7 @@ import (
cloudprovider "k8s.io/cloud-provider"
"k8s.io/component-base/zpages/flagz"
"k8s.io/component-helpers/apimachinery/lease"
resourcehelper "k8s.io/component-helpers/resource"
internalapi "k8s.io/cri-api/pkg/apis"
runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1"
remote "k8s.io/cri-client/pkg"
@@ -1900,10 +1901,9 @@ func (kl *Kubelet) SyncPod(ctx context.Context, updateType kubetypes.SyncPodType
if utilfeature.DefaultFeatureGate.Enabled(features.InPlacePodVerticalScaling) {
// Check whether a resize is in progress so we can set the PodResizeInProgressCondition accordingly.
kl.allocationManager.CheckPodResizeInProgress(pod, podStatus)
// TODO(natasha41575): There is a race condition here, where the goroutine in the
// TODO(#132851): There is a race condition here, where the goroutine in the
// allocation manager may allocate a new resize and unconditionally set the
// PodResizeInProgressCondition before we set the status below.
// See https://github.com/kubernetes/kubernetes/issues/132851.
}
// Generate final API pod status with pod and status manager status
@@ -2804,9 +2804,12 @@ func (kl *Kubelet) HandlePodRemoves(pods []*v1.Pod) {
// pod is updated in the API.
func (kl *Kubelet) HandlePodReconcile(pods []*v1.Pod) {
start := kl.clock.Now()
retryPendingResizes := false
hasPendingResizes := kl.allocationManager.HasPendingResizes()
for _, pod := range pods {
// Update the pod in pod manager, status manager will do periodically reconcile according
// to the pod manager.
oldPod, _ := kl.podManager.GetPodByUID(pod.UID)
kl.podManager.UpdatePod(pod)
pod, mirrorPod, wasMirror := kl.podManager.GetPodAndMirrorPod(pod)
@@ -2818,6 +2821,28 @@ func (kl *Kubelet) HandlePodReconcile(pods []*v1.Pod) {
// Static pods should be reconciled the same way as regular pods
}
if utilfeature.DefaultFeatureGate.Enabled(features.InPlacePodVerticalScaling) {
// If there are pending resizes, check whether the requests shrank as a result of the status
// resources changing.
if hasPendingResizes && !retryPendingResizes && oldPod != nil {
opts := resourcehelper.PodResourcesOptions{
UseStatusResources: true,
SkipPodLevelResources: !utilfeature.DefaultFeatureGate.Enabled(features.PodLevelResources),
}
// Ignore desired resources when aggregating the resources.
allocatedOldPod, _ := kl.allocationManager.UpdatePodFromAllocation(oldPod)
allocatedPod, _ := kl.allocationManager.UpdatePodFromAllocation(pod)
oldRequest := resourcehelper.PodRequests(allocatedOldPod, opts)
newRequest := resourcehelper.PodRequests(allocatedPod, opts)
// If cpu or memory requests shrank, then retry the pending resizes.
retryPendingResizes = newRequest.Memory().Cmp(*oldRequest.Memory()) < 0 ||
newRequest.Cpu().Cmp(*oldRequest.Cpu()) < 0
}
}
// TODO: reconcile being calculated in the config manager is questionable, and avoiding
// extra syncs may no longer be necessary. Reevaluate whether Reconcile and Sync can be
// merged (after resolving the next two TODOs).
@@ -2846,6 +2871,12 @@ func (kl *Kubelet) HandlePodReconcile(pods []*v1.Pod) {
}
}
}
if utilfeature.DefaultFeatureGate.Enabled(features.InPlacePodVerticalScaling) {
if retryPendingResizes {
kl.allocationManager.RetryPendingResizes(allocation.TriggerReasonPodResized)
}
}
}
// HandlePodSyncs is the callback in the syncHandler interface for pods

4
pkg/kubelet/kubelet_node_status_test.go
View File

@@ -213,7 +213,7 @@ func TestUpdateNewNodeStatus(t *testing.T) {
}
inputImageList, expectedImageList := generateTestingImageLists(numTestImages, int(tc.nodeStatusMaxImages))
testKubelet := newTestKubeletWithImageList(
t, inputImageList, false /* controllerAttachDetachEnabled */, true /*initFakeVolumePlugin*/, true /* localStorageCapacityIsolation */, false /*excludePodAdmitHandlers*/)
t, inputImageList, false /* controllerAttachDetachEnabled */, true /*initFakeVolumePlugin*/, true /* localStorageCapacityIsolation */, false /*excludePodAdmitHandlers*/, false /*enableResizing*/)
defer testKubelet.Cleanup()
kubelet := testKubelet.kubelet
kubelet.nodeStatusMaxImages = tc.nodeStatusMaxImages
@@ -1588,7 +1588,7 @@ func TestUpdateNewNodeStatusTooLargeReservation(t *testing.T) {
// generate one more in inputImageList than we configure the Kubelet to report
inputImageList, _ := generateTestingImageLists(nodeStatusMaxImages+1, nodeStatusMaxImages)
testKubelet := newTestKubeletWithImageList(
t, inputImageList, false /* controllerAttachDetachEnabled */, true /* initFakeVolumePlugin */, true /*localStorageCapacityIsolation*/, false /*excludePodAdmitHandlers*/)
t, inputImageList, false /* controllerAttachDetachEnabled */, true /* initFakeVolumePlugin */, true /*localStorageCapacityIsolation*/, false /*excludePodAdmitHandlers*/, false /*enableResizing*/)
defer testKubelet.Cleanup()
kubelet := testKubelet.kubelet
kubelet.nodeStatusMaxImages = nodeStatusMaxImages

149
pkg/kubelet/kubelet_test.go
View File

@@ -177,10 +177,10 @@ func newTestKubelet(t *testing.T, controllerAttachDetachEnabled bool) *TestKubel
Size: 456,
},
}
return newTestKubeletWithImageList(t, imageList, controllerAttachDetachEnabled, true /*initFakeVolumePlugin*/, true /*localStorageCapacityIsolation*/, false /*excludePodAdmitHandlers*/)
return newTestKubeletWithImageList(t, imageList, controllerAttachDetachEnabled, true /*initFakeVolumePlugin*/, true /*localStorageCapacityIsolation*/, false /*excludePodAdmitHandlers*/, false /*enableResizing*/)
}
func newTestKubeletExcludeAdmitHandlers(t *testing.T, controllerAttachDetachEnabled bool) *TestKubelet {
func newTestKubeletExcludeAdmitHandlers(t *testing.T, controllerAttachDetachEnabled, sourcesReady bool) *TestKubelet {
imageList := []kubecontainer.Image{
{
ID: "abc",
@@ -193,7 +193,7 @@ func newTestKubeletExcludeAdmitHandlers(t *testing.T, controllerAttachDetachEnab
Size: 456,
},
}
return newTestKubeletWithImageList(t, imageList, controllerAttachDetachEnabled, true /*initFakeVolumePlugin*/, true /*localStorageCapacityIsolation*/, true /*excludePodAdmitHandlers*/)
return newTestKubeletWithImageList(t, imageList, controllerAttachDetachEnabled, true /*initFakeVolumePlugin*/, true /*localStorageCapacityIsolation*/, true /*excludePodAdmitHandlers*/, sourcesReady)
}
func newTestKubeletWithImageList(
@@ -203,6 +203,7 @@ func newTestKubeletWithImageList(
initFakeVolumePlugin bool,
localStorageCapacityIsolation bool,
excludeAdmitHandlers bool,
enableResizing bool,
) *TestKubelet {
logger, _ := ktesting.NewTestContext(t)
@@ -328,7 +329,7 @@ func newTestKubeletWithImageList(
func(pod *v1.Pod) { kubelet.HandlePodSyncs([]*v1.Pod{pod}) },
kubelet.GetActivePods,
kubelet.podManager.GetPodByUID,
config.NewSourcesReady(func(_ sets.Set[string]) bool { return false }),
config.NewSourcesReady(func(_ sets.Set[string]) bool { return enableResizing }),
)
kubelet.allocationManager.SetContainerRuntime(fakeRuntime)
volumeStatsAggPeriod := time.Second * 10
@@ -1073,7 +1074,7 @@ func TestHandleMemExceeded(t *testing.T) {
// Tests that we handle result of interface UpdatePluginResources correctly
// by setting corresponding status in status map.
func TestHandlePluginResources(t *testing.T) {
testKubelet := newTestKubeletExcludeAdmitHandlers(t, false /* controllerAttachDetachEnabled */)
testKubelet := newTestKubeletExcludeAdmitHandlers(t, false /* controllerAttachDetachEnabled */, false /*enableResizing*/)
defer testKubelet.Cleanup()
kl := testKubelet.kubelet
@@ -4225,3 +4226,141 @@ func TestHandlePodUpdates_RecordContainerRequestedResizes(t *testing.T) {
})
}
}
func TestHandlePodReconcile_RetryPendingResizes(t *testing.T) {
if goruntime.GOOS == "windows" {
t.Skip("InPlacePodVerticalScaling is not currently supported for Windows")
}
featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.InPlacePodVerticalScaling, true)
testKubelet := newTestKubeletExcludeAdmitHandlers(t, false /* controllerAttachDetachEnabled */, true /*enableResizing*/)
defer testKubelet.Cleanup()
kubelet := testKubelet.kubelet
lowCPU := resource.MustParse("500m")
highCPU := resource.MustParse("1")
lowMem := resource.MustParse("500Mi")
highMem := resource.MustParse("1Gi")
// Set desired resources to some huge value to verify that they are being ignored in the aggregate check.
enormousCPU := resource.MustParse("2000m")
enormousMem := resource.MustParse("2000Mi")
makePodWithResources := func(name string, requests v1.ResourceList, status v1.ResourceList) *v1.Pod {
return &v1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: name,
UID: types.UID(name),
},
Spec: v1.PodSpec{
Containers: []v1.Container{
{
Name: "c1",
Image: "i1",
Resources: v1.ResourceRequirements{
Requests: requests,
},
},
},
},
Status: v1.PodStatus{
ContainerStatuses: []v1.ContainerStatus{
{
Name: "c1",
Resources: &v1.ResourceRequirements{
Requests: status,
},
},
},
},
}
}
pendingResizeAllocated := &v1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: "pod-pending-resize",
UID: types.UID("pod-pending-resize"),
},
Spec: v1.PodSpec{
Containers: []v1.Container{
{
Name: "c1",
Image: "i1",
Resources: v1.ResourceRequirements{
Requests: v1.ResourceList{v1.ResourceCPU: highCPU, v1.ResourceMemory: highMem},
},
},
},
},
}
pendingResizeDesired := &v1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: "pod-pending-resize",
UID: types.UID("pod-pending-resize"),
},
Spec: v1.PodSpec{
Containers: []v1.Container{
{
Name: "c1",
Image: "i1",
Resources: v1.ResourceRequirements{
Requests: v1.ResourceList{v1.ResourceCPU: lowCPU, v1.ResourceMemory: lowMem},
},
},
},
},
}
testCases := []struct {
name string
oldPod *v1.Pod
newPod *v1.Pod
shouldRetryPendingResize bool
}{
{
name: "requests are increasing",
oldPod: makePodWithResources("updated-pod", v1.ResourceList{v1.ResourceCPU: highCPU, v1.ResourceMemory: highMem}, v1.ResourceList{v1.ResourceCPU: lowCPU, v1.ResourceMemory: lowMem}),
newPod: makePodWithResources("updated-pod", v1.ResourceList{v1.ResourceCPU: enormousCPU, v1.ResourceMemory: enormousMem}, v1.ResourceList{v1.ResourceCPU: highCPU, v1.ResourceMemory: highMem}),
shouldRetryPendingResize: false,
},
{
name: "requests are unchanged",
oldPod: makePodWithResources("updated-pod", v1.ResourceList{v1.ResourceCPU: lowCPU, v1.ResourceMemory: lowMem}, v1.ResourceList{v1.ResourceCPU: lowCPU, v1.ResourceMemory: lowMem}),
newPod: makePodWithResources("updated-pod", v1.ResourceList{v1.ResourceCPU: enormousCPU, v1.ResourceMemory: enormousMem}, v1.ResourceList{v1.ResourceCPU: lowCPU, v1.ResourceMemory: lowMem}),
shouldRetryPendingResize: false,
},
{
name: "requests are decreasing",
oldPod: makePodWithResources("updated-pod", v1.ResourceList{v1.ResourceCPU: lowCPU, v1.ResourceMemory: lowMem}, v1.ResourceList{v1.ResourceCPU: highCPU, v1.ResourceMemory: highMem}),
newPod: makePodWithResources("updated-pod", v1.ResourceList{v1.ResourceCPU: enormousCPU, v1.ResourceMemory: enormousMem}, v1.ResourceList{v1.ResourceCPU: lowCPU, v1.ResourceMemory: lowMem}),
shouldRetryPendingResize: true,
},
}
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
// For the sake of this test, just reject all resize requests.
handler := &testPodAdmitHandler{podsToReject: []*v1.Pod{pendingResizeAllocated}}
kubelet.allocationManager.AddPodAdmitHandlers(lifecycle.PodAdmitHandlers{handler})
require.NoError(t, kubelet.allocationManager.SetAllocatedResources(pendingResizeAllocated))
require.NoError(t, kubelet.allocationManager.SetAllocatedResources(tc.oldPod))
// We only expect status resources to change in HandlePodReconcile.
tc.oldPod.Spec = tc.newPod.Spec
kubelet.podManager.AddPod(pendingResizeDesired)
kubelet.podManager.AddPod(tc.oldPod)
kubelet.allocationManager.PushPendingResize(pendingResizeDesired.UID)
kubelet.statusManager.ClearPodResizePendingCondition(pendingResizeDesired.UID)
kubelet.HandlePodReconcile([]*v1.Pod{tc.newPod})
require.Equal(t, tc.shouldRetryPendingResize, kubelet.statusManager.IsPodResizeDeferred(pendingResizeDesired.UID))
kubelet.allocationManager.RemovePod(pendingResizeDesired.UID)
kubelet.podManager.RemovePod((pendingResizeDesired))
kubelet.podManager.RemovePod(tc.oldPod)
})
}
}

4
pkg/kubelet/kubelet_volumes_test.go
View File

@@ -220,7 +220,7 @@ func TestPodVolumeDeadlineAttachAndMount(t *testing.T) {
}
testKubelet := newTestKubeletWithImageList(t, nil /*imageList*/, false, /* controllerAttachDetachEnabled */
false /*initFakeVolumePlugin*/, true /*localStorageCapacityIsolation*/, false /*excludePodAdmitHandlers*/)
false /*initFakeVolumePlugin*/, true /*localStorageCapacityIsolation*/, false /*excludePodAdmitHandlers*/, false /*enableResizing*/)
defer testKubelet.Cleanup()
kubelet := testKubelet.kubelet
@@ -282,7 +282,7 @@ func TestPodVolumeDeadlineUnmount(t *testing.T) {
}
testKubelet := newTestKubeletWithImageList(t, nil /*imageList*/, false, /* controllerAttachDetachEnabled */
true /*initFakeVolumePlugin*/, true /*localStorageCapacityIsolation*/, false /*excludePodAdmitHandlers*/)
true /*initFakeVolumePlugin*/, true /*localStorageCapacityIsolation*/, false /*excludePodAdmitHandlers*/, false /*enableResizing*/)
defer testKubelet.Cleanup()
kubelet := testKubelet.kubelet