Merge pull request kubernetes#130599 from tallclair/acknowledged-resources

[FG:InPlacePodVerticalScaling] Track actuated resources to trigger resizes

Author: k8s-ci-robot

pkg/kubelet/allocation/allocation_manager.go

@@ -17,6 +17,8 @@ limitations under the License.
 package allocation
 
 import (
+	"path/filepath"
+
 	v1 "k8s.io/api/core/v1"
 	apiequality "k8s.io/apimachinery/pkg/api/equality"
 	"k8s.io/apimachinery/pkg/types"
@@ -29,7 +31,10 @@ import (
 )
 
 // podStatusManagerStateFile is the file name where status manager stores its state
-const podStatusManagerStateFile = "pod_status_manager_state"
+const (
+	allocatedPodsStateFile = "allocated_pods_state"
+	actuatedPodsStateFile  = "actuated_pods_state"
+)
 
 // AllocationManager tracks pod resource allocations.
 type Manager interface {
@@ -41,57 +46,71 @@ type Manager interface {
 	// Returns the updated (or original) pod, and whether there was an allocation stored.
 	UpdatePodFromAllocation(pod *v1.Pod) (*v1.Pod, bool)
 
-	// SetPodAllocation checkpoints the resources allocated to a pod's containers.
-	SetPodAllocation(pod *v1.Pod) error
+	// SetAllocatedResources checkpoints the resources allocated to a pod's containers.
+	SetAllocatedResources(allocatedPod *v1.Pod) error
+
+	// SetActuatedResources records the actuated resources of the given container (or the entire
+	// pod, if actuatedContainer is nil).
+	SetActuatedResources(allocatedPod *v1.Pod, actuatedContainer *v1.Container) error
 
-	// DeletePodAllocation removes any stored state for the given pod UID.
-	DeletePodAllocation(uid types.UID)
+	// GetActuatedResources returns the stored actuated resources for the container, and whether they exist.
+	GetActuatedResources(podUID types.UID, containerName string) (v1.ResourceRequirements, bool)
+
+	// RemovePod removes any stored state for the given pod UID.
+	RemovePod(uid types.UID)
 
 	// RemoveOrphanedPods removes the stored state for any pods not included in the set of remaining pods.
 	RemoveOrphanedPods(remainingPods sets.Set[types.UID])
 }
 
 type manager struct {
-	state state.State
+	allocated state.State
+	actuated  state.State
 }
 
 func NewManager(checkpointDirectory string) Manager {
-	m := &manager{}
-
-	if utilfeature.DefaultFeatureGate.Enabled(features.InPlacePodVerticalScaling) {
-		stateImpl, err := state.NewStateCheckpoint(checkpointDirectory, podStatusManagerStateFile)
-		if err != nil {
-			// This is a crictical, non-recoverable failure.
-			klog.ErrorS(err, "Failed to initialize allocation checkpoint manager")
-			panic(err)
-		}
-		m.state = stateImpl
-	} else {
-		m.state = state.NewNoopStateCheckpoint()
+	return &manager{
+		allocated: newStateImpl(checkpointDirectory, allocatedPodsStateFile),
+		actuated:  newStateImpl(checkpointDirectory, actuatedPodsStateFile),
+	}
+}
+
+func newStateImpl(checkpointDirectory, checkpointName string) state.State {
+	if !utilfeature.DefaultFeatureGate.Enabled(features.InPlacePodVerticalScaling) {
+		return state.NewNoopStateCheckpoint()
 	}
 
-	return m
+	stateImpl, err := state.NewStateCheckpoint(checkpointDirectory, checkpointName)
+	if err != nil {
+		// This is a critical, non-recoverable failure.
+		klog.ErrorS(err, "Failed to initialize allocation checkpoint manager",
+			"checkpointPath", filepath.Join(checkpointDirectory, checkpointName))
+		panic(err)
+	}
+
+	return stateImpl
 }
 
 // NewInMemoryManager returns an allocation manager that doesn't persist state.
 // For testing purposes only!
 func NewInMemoryManager() Manager {
 	return &manager{
-		state: state.NewStateMemory(nil),
+		allocated: state.NewStateMemory(nil),
+		actuated:  state.NewStateMemory(nil),
 	}
 }
 
 // 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) {
-	return m.state.GetContainerResourceAllocation(podUID, containerName)
+	return m.allocated.GetContainerResourceAllocation(podUID, containerName)
 }
 
 // UpdatePodFromAllocation overwrites the pod spec with the allocation.
 // This function does a deep copy only if updates are needed.
 func (m *manager) UpdatePodFromAllocation(pod *v1.Pod) (*v1.Pod, bool) {
 	// TODO(tallclair): This clones the whole cache, but we only need 1 pod.
-	allocs := m.state.GetPodResourceAllocation()
+	allocs := m.allocated.GetPodResourceAllocation()
 	return updatePodFromAllocation(pod, allocs)
 }
 
@@ -132,8 +151,12 @@ func updatePodFromAllocation(pod *v1.Pod, allocs state.PodResourceAllocation) (*
 	return pod, updated
 }
 
-// SetPodAllocation checkpoints the resources allocated to a pod's containers
-func (m *manager) SetPodAllocation(pod *v1.Pod) error {
+// SetAllocatedResources checkpoints the resources allocated to a pod's containers
+func (m *manager) SetAllocatedResources(pod *v1.Pod) error {
+	return m.allocated.SetPodResourceAllocation(pod.UID, allocationFromPod(pod))
+}
+
+func allocationFromPod(pod *v1.Pod) map[string]v1.ResourceRequirements {
 	podAlloc := make(map[string]v1.ResourceRequirements)
 	for _, container := range pod.Spec.Containers {
 		alloc := *container.Resources.DeepCopy()
@@ -149,16 +172,35 @@ func (m *manager) SetPodAllocation(pod *v1.Pod) error {
 		}
 	}
 
-	return m.state.SetPodResourceAllocation(pod.UID, podAlloc)
+	return podAlloc
 }
 
-func (m *manager) DeletePodAllocation(uid types.UID) {
-	if err := m.state.Delete(uid, ""); err != nil {
+func (m *manager) RemovePod(uid types.UID) {
+	if err := m.allocated.Delete(uid, ""); err != nil {
+		// If the deletion fails, it will be retried by RemoveOrphanedPods, so we can safely ignore the error.
+		klog.V(3).ErrorS(err, "Failed to delete pod allocation", "podUID", uid)
+	}
+
+	if err := m.actuated.Delete(uid, ""); err != nil {
 		// If the deletion fails, it will be retried by RemoveOrphanedPods, so we can safely ignore the error.
 		klog.V(3).ErrorS(err, "Failed to delete pod allocation", "podUID", uid)
 	}
 }
 
 func (m *manager) RemoveOrphanedPods(remainingPods sets.Set[types.UID]) {
-	m.state.RemoveOrphanedPods(remainingPods)
+	m.allocated.RemoveOrphanedPods(remainingPods)
+	m.actuated.RemoveOrphanedPods(remainingPods)
+}
+
+func (m *manager) SetActuatedResources(allocatedPod *v1.Pod, actuatedContainer *v1.Container) error {
+	if actuatedContainer == nil {
+		alloc := allocationFromPod(allocatedPod)
+		return m.actuated.SetPodResourceAllocation(allocatedPod.UID, alloc)
+	}
+
+	return m.actuated.SetContainerResourceAllocation(allocatedPod.UID, actuatedContainer.Name, actuatedContainer.Resources)
+}
+
+func (m *manager) GetActuatedResources(podUID types.UID, containerName string) (v1.ResourceRequirements, bool) {
+	return m.actuated.GetContainerResourceAllocation(podUID, containerName)
 }

pkg/kubelet/kubelet.go

@@ -740,6 +740,7 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 		kubeDeps.ContainerManager,
 		klet.containerLogManager,
 		klet.runtimeClassManager,
+		klet.allocationManager,
 		seccompDefault,
 		kubeCfg.MemorySwap.SwapBehavior,
 		kubeDeps.ContainerManager.GetNodeAllocatableAbsolute,
@@ -2666,7 +2667,7 @@ func (kl *Kubelet) HandlePodAdditions(pods []*v1.Pod) {
 					continue
 				}
 				// For new pod, checkpoint the resource values at which the Pod has been admitted
-				if err := kl.allocationManager.SetPodAllocation(allocatedPod); err != nil {
+				if err := kl.allocationManager.SetAllocatedResources(allocatedPod); err != nil {
 					//TODO(vinaykul,InPlacePodVerticalScaling): Can we recover from this in some way? Investigate
 					klog.ErrorS(err, "SetPodAllocation failed", "pod", klog.KObj(pod))
 				}
@@ -2722,7 +2723,7 @@ func (kl *Kubelet) HandlePodRemoves(pods []*v1.Pod) {
 	start := kl.clock.Now()
 	for _, pod := range pods {
 		kl.podManager.RemovePod(pod)
-		kl.allocationManager.DeletePodAllocation(pod.UID)
+		kl.allocationManager.RemovePod(pod.UID)
 
 		pod, mirrorPod, wasMirror := kl.podManager.GetPodAndMirrorPod(pod)
 		if wasMirror {
@@ -2886,8 +2887,7 @@ func (kl *Kubelet) handlePodResourcesResize(pod *v1.Pod, podStatus *kubecontaine
 
 	if !updated {
 		// Desired resources == allocated resources. Check whether a resize is in progress.
-		resizeInProgress := !allocatedResourcesMatchStatus(allocatedPod, podStatus)
-		if resizeInProgress {
+		if kl.isPodResizeInProgress(allocatedPod, podStatus) {
 			// If a resize is in progress, make sure the cache has the correct state in case the Kubelet restarted.
 			kl.statusManager.SetPodResizeStatus(pod.UID, v1.PodResizeStatusInProgress)
 		} else {
@@ -2909,7 +2909,7 @@ func (kl *Kubelet) handlePodResourcesResize(pod *v1.Pod, podStatus *kubecontaine
 	fit, resizeStatus, resizeMsg := kl.canResizePod(pod)
 	if fit {
 		// Update pod resource allocation checkpoint
-		if err := kl.allocationManager.SetPodAllocation(pod); err != nil {
+		if err := kl.allocationManager.SetAllocatedResources(pod); err != nil {
 			return nil, err
 		}
 		for i, container := range pod.Spec.Containers {
@@ -2928,11 +2928,11 @@ func (kl *Kubelet) handlePodResourcesResize(pod *v1.Pod, podStatus *kubecontaine
 		}
 		allocatedPod = pod
 
-		// Special case when the updated allocation matches the actual resources. This can occur
+		// Special case when the updated allocation matches the actuated resources. This can occur
 		// when reverting a resize that hasn't been actuated, or when making an equivalent change
 		// (such as CPU requests below MinShares). This is an optimization to clear the resize
 		// status immediately, rather than waiting for the next SyncPod iteration.
-		if allocatedResourcesMatchStatus(allocatedPod, podStatus) {
+		if !kl.isPodResizeInProgress(allocatedPod, podStatus) {
 			// In this case, consider the resize complete.
 			kl.statusManager.SetPodResizeStatus(pod.UID, "")
 			return allocatedPod, nil
@@ -2952,6 +2952,46 @@ func (kl *Kubelet) handlePodResourcesResize(pod *v1.Pod, podStatus *kubecontaine
 	return allocatedPod, nil
 }
 
+// isPodResizingInProgress checks whether the actuated resizable resources differ from the allocated resources
+// for any running containers. Specifically, the following differences are ignored:
+// - Non-resizable containers: non-restartable init containers, ephemeral containers
+// - Non-resizable resources: only CPU & memory are resizable
+// - Non-actuated resources: memory requests are not actuated
+// - Non-running containers: they will be sized correctly when (re)started
+func (kl *Kubelet) isPodResizeInProgress(allocatedPod *v1.Pod, podStatus *kubecontainer.PodStatus) bool {
+	return !podutil.VisitContainers(&allocatedPod.Spec, podutil.InitContainers|podutil.Containers,
+		func(allocatedContainer *v1.Container, containerType podutil.ContainerType) (shouldContinue bool) {
+			if !isResizableContainer(allocatedContainer, containerType) {
+				return true
+			}
+
+			containerStatus := podStatus.FindContainerStatusByName(allocatedContainer.Name)
+			if containerStatus == nil || containerStatus.State != kubecontainer.ContainerStateRunning {
+				// If the container isn't running, it doesn't need to be resized.
+				return true
+			}
+
+			actuatedResources, _ := kl.allocationManager.GetActuatedResources(allocatedPod.UID, allocatedContainer.Name)
+			allocatedResources := allocatedContainer.Resources
+
+			// Memory requests are excluded since they don't need to be actuated.
+			return allocatedResources.Requests[v1.ResourceCPU].Equal(actuatedResources.Requests[v1.ResourceCPU]) &&
+				allocatedResources.Limits[v1.ResourceCPU].Equal(actuatedResources.Limits[v1.ResourceCPU]) &&
+				allocatedResources.Limits[v1.ResourceMemory].Equal(actuatedResources.Limits[v1.ResourceMemory])
+		})
+}
+
+func isResizableContainer(container *v1.Container, containerType podutil.ContainerType) bool {
+	switch containerType {
+	case podutil.InitContainers:
+		return podutil.IsRestartableInitContainer(container)
+	case podutil.Containers:
+		return true
+	default:
+		return false
+	}
+}
+
 // LatestLoopEntryTime returns the last time in the sync loop monitor.
 func (kl *Kubelet) LatestLoopEntryTime() time.Time {
 	val := kl.syncLoopMonitor.Load()

pkg/kubelet/kubelet_pods.go

@@ -60,7 +60,6 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet/metrics"
 	"k8s.io/kubernetes/pkg/kubelet/status"
 	kubetypes "k8s.io/kubernetes/pkg/kubelet/types"
-	"k8s.io/kubernetes/pkg/kubelet/util/format"
 	utilfs "k8s.io/kubernetes/pkg/util/filesystem"
 	utilpod "k8s.io/kubernetes/pkg/util/pod"
 	volumeutil "k8s.io/kubernetes/pkg/volume/util"
@@ -1746,86 +1745,6 @@ func (kl *Kubelet) determinePodResizeStatus(allocatedPod *v1.Pod, podStatus *kub
 	return resizeStatus
 }
 
-// allocatedResourcesMatchStatus tests whether the resizeable resources in the pod spec match the
-// resources reported in the status.
-func allocatedResourcesMatchStatus(allocatedPod *v1.Pod, podStatus *kubecontainer.PodStatus) bool {
-	for _, c := range allocatedPod.Spec.Containers {
-		if !allocatedContainerResourcesMatchStatus(allocatedPod, &c, podStatus) {
-			return false
-		}
-	}
-	if utilfeature.DefaultFeatureGate.Enabled(features.SidecarContainers) {
-		for _, c := range allocatedPod.Spec.InitContainers {
-			if podutil.IsRestartableInitContainer(&c) && !allocatedContainerResourcesMatchStatus(allocatedPod, &c, podStatus) {
-				return false
-			}
-		}
-	}
-	return true
-}
-
-// allocatedContainerResourcesMatchStatus returns true if the container resources matches with the container statuses resources.
-func allocatedContainerResourcesMatchStatus(allocatedPod *v1.Pod, c *v1.Container, podStatus *kubecontainer.PodStatus) bool {
-	if cs := podStatus.FindContainerStatusByName(c.Name); cs != nil {
-		if cs.State != kubecontainer.ContainerStateRunning {
-			// If the container isn't running, it isn't resizing.
-			return true
-		}
-
-		cpuReq, hasCPUReq := c.Resources.Requests[v1.ResourceCPU]
-		cpuLim, hasCPULim := c.Resources.Limits[v1.ResourceCPU]
-		memLim, hasMemLim := c.Resources.Limits[v1.ResourceMemory]
-
-		if cs.Resources == nil {
-			if hasCPUReq || hasCPULim || hasMemLim {
-				// Container status is missing Resources information, but the container does
-				// have resizable resources configured.
-				klog.ErrorS(nil, "Missing runtime resources information for resizing container",
-					"pod", format.Pod(allocatedPod), "container", c.Name)
-				return false // We don't want to clear resize status with insufficient information.
-			} else {
-				// No resizable resources configured; this might be ok.
-				return true
-			}
-		}
-
-		// Only compare resizeable resources, and only compare resources that are explicitly configured.
-		if hasCPUReq {
-			if cs.Resources.CPURequest == nil {
-				if !cpuReq.IsZero() {
-					return false
-				}
-			} else if !cpuReq.Equal(*cs.Resources.CPURequest) &&
-				(cpuReq.MilliValue() > cm.MinShares || cs.Resources.CPURequest.MilliValue() > cm.MinShares) {
-				// If both allocated & status CPU requests are at or below MinShares then they are considered equal.
-				return false
-			}
-		}
-		if hasCPULim {
-			if cs.Resources.CPULimit == nil {
-				if !cpuLim.IsZero() {
-					return false
-				}
-			} else if !cpuLim.Equal(*cs.Resources.CPULimit) &&
-				(cpuLim.MilliValue() > cm.MinMilliCPULimit || cs.Resources.CPULimit.MilliValue() > cm.MinMilliCPULimit) {
-				// If both allocated & status CPU limits are at or below the minimum limit, then they are considered equal.
-				return false
-			}
-		}
-		if hasMemLim {
-			if cs.Resources.MemoryLimit == nil {
-				if !memLim.IsZero() {
-					return false
-				}
-			} else if !memLim.Equal(*cs.Resources.MemoryLimit) {
-				return false
-			}
-		}
-	}
-
-	return true
-}
-
 // generateAPIPodStatus creates the final API pod status for a pod, given the
 // internal pod status. This method should only be called from within sync*Pod methods.
 func (kl *Kubelet) generateAPIPodStatus(pod *v1.Pod, podStatus *kubecontainer.PodStatus, podIsTerminal bool) v1.PodStatus {