diff --git a/pkg/kubelet/allocation/allocation_manager.go b/pkg/kubelet/allocation/allocation_manager.go index 03624cb151e..be6ffa4721b 100644 --- a/pkg/kubelet/allocation/allocation_manager.go +++ b/pkg/kubelet/allocation/allocation_manager.go @@ -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) } } diff --git a/pkg/kubelet/kubelet.go b/pkg/kubelet/kubelet.go index 4c537c56f81..f0725b7e365 100644 --- a/pkg/kubelet/kubelet.go +++ b/pkg/kubelet/kubelet.go @@ -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 diff --git a/pkg/kubelet/kubelet_node_status_test.go b/pkg/kubelet/kubelet_node_status_test.go index a6f587d55ea..86f03304701 100644 --- a/pkg/kubelet/kubelet_node_status_test.go +++ b/pkg/kubelet/kubelet_node_status_test.go @@ -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 diff --git a/pkg/kubelet/kubelet_test.go b/pkg/kubelet/kubelet_test.go index e3251ae2062..892d56302fd 100644 --- a/pkg/kubelet/kubelet_test.go +++ b/pkg/kubelet/kubelet_test.go @@ -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) + }) + } +} diff --git a/pkg/kubelet/kubelet_volumes_test.go b/pkg/kubelet/kubelet_volumes_test.go index 774223606bb..9ba12c2f31a 100644 --- a/pkg/kubelet/kubelet_volumes_test.go +++ b/pkg/kubelet/kubelet_volumes_test.go @@ -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