diff --git a/pkg/kubelet/kubelet.go b/pkg/kubelet/kubelet.go index fecdcc6a45b..f6d6d2c37e7 100644 --- a/pkg/kubelet/kubelet.go +++ b/pkg/kubelet/kubelet.go @@ -2107,11 +2107,10 @@ func (kl *Kubelet) canAdmitPod(pods []*v1.Pod, pod *v1.Pod) (bool, string, strin if utilfeature.DefaultFeatureGate.Enabled(features.InPlacePodVerticalScaling) { // Use allocated resources values from checkpoint store (source of truth) to determine fit otherPods := make([]*v1.Pod, 0, len(pods)) - checkpointState := kl.statusManager.State() for _, p := range pods { op := p.DeepCopy() for _, c := range op.Spec.Containers { - resourcesAllocated, found := checkpointState.GetContainerResourceAllocation(string(p.UID), c.Name) + resourcesAllocated, found := kl.statusManager.GetContainerResourceAllocation(string(p.UID), c.Name) if c.Resources.Requests != nil && found { c.Resources.Requests[v1.ResourceCPU] = resourcesAllocated[v1.ResourceCPU] c.Resources.Requests[v1.ResourceMemory] = resourcesAllocated[v1.ResourceMemory] @@ -2406,22 +2405,19 @@ func (kl *Kubelet) HandlePodAdditions(pods []*v1.Pod) { if utilfeature.DefaultFeatureGate.Enabled(features.InPlacePodVerticalScaling) { // To handle kubelet restarts, test pod admissibility using ResourcesAllocated values // (for cpu & memory) from checkpoint store. If found, that is the source of truth. - checkpointState := kl.statusManager.State() podCopy := pod.DeepCopy() for _, c := range podCopy.Spec.Containers { - resourcesAllocated, found := checkpointState.GetContainerResourceAllocation(string(pod.UID), c.Name) + resourcesAllocated, found := kl.statusManager.GetContainerResourceAllocation(string(pod.UID), c.Name) if c.Resources.Requests != nil && found { c.Resources.Requests[v1.ResourceCPU] = resourcesAllocated[v1.ResourceCPU] c.Resources.Requests[v1.ResourceMemory] = resourcesAllocated[v1.ResourceMemory] } } - // Check if we can admit the pod; if not, reject it. if ok, reason, message := kl.canAdmitPod(activePods, podCopy); !ok { kl.rejectPod(pod, reason, message) continue } - // For new pod, checkpoint the resource values at which the Pod has been admitted if err := kl.statusManager.SetPodAllocation(podCopy); err != nil { //TODO(vinaykul,InPlacePodVerticalScaling): Can we recover from this in some way? Investigate diff --git a/pkg/kubelet/kubelet_pods.go b/pkg/kubelet/kubelet_pods.go index 5ee3b9f4bec..335af666f5c 100644 --- a/pkg/kubelet/kubelet_pods.go +++ b/pkg/kubelet/kubelet_pods.go @@ -1472,8 +1472,7 @@ func (kl *Kubelet) determinePodResizeStatus(pod *v1.Pod, podStatus *v1.PodStatus klog.ErrorS(err, "SetPodResizeStatus failed", "pod", pod.Name) } } else { - checkpointState := kl.statusManager.State() - if resizeStatus, found := checkpointState.GetPodResizeStatus(string(pod.UID)); found { + if resizeStatus, found := kl.statusManager.GetPodResizeStatus(string(pod.UID)); found { podResizeStatus = resizeStatus } } @@ -1769,8 +1768,7 @@ func (kl *Kubelet) convertToAPIContainerStatuses(pod *v1.Pod, podStatus *kubecon container := kubecontainer.GetContainerSpec(pod, cName) // ResourcesAllocated values come from checkpoint. It is the source-of-truth. found := false - checkpointState := kl.statusManager.State() - status.ResourcesAllocated, found = checkpointState.GetContainerResourceAllocation(string(pod.UID), cName) + status.ResourcesAllocated, found = kl.statusManager.GetContainerResourceAllocation(string(pod.UID), cName) if !(container.Resources.Requests == nil && container.Resources.Limits == nil) && !found { // Log error and fallback to ResourcesAllocated in oldStatus if it exists klog.ErrorS(nil, "resource allocation not found in checkpoint store", "pod", pod.Name, "container", cName) diff --git a/pkg/kubelet/status/fake_status_manager.go b/pkg/kubelet/status/fake_status_manager.go index ee4b3f5f36f..42cd611984e 100644 --- a/pkg/kubelet/status/fake_status_manager.go +++ b/pkg/kubelet/status/fake_status_manager.go @@ -63,9 +63,14 @@ func (m *fakeManager) RemoveOrphanedStatuses(podUIDs map[types.UID]bool) { return } -func (m *fakeManager) State() state.Reader { - klog.InfoS("State()") - return m.state +func (m *fakeManager) GetContainerResourceAllocation(podUID string, containerName string) (v1.ResourceList, bool) { + klog.InfoS("GetContainerResourceAllocation()") + return m.state.GetContainerResourceAllocation(podUID, containerName) +} + +func (m *fakeManager) GetPodResizeStatus(podUID string) (v1.PodResizeStatus, bool) { + klog.InfoS("GetPodResizeStatus()") + return "", false } func (m *fakeManager) SetPodAllocation(pod *v1.Pod) error { diff --git a/pkg/kubelet/status/status_manager.go b/pkg/kubelet/status/status_manager.go index fdf99b04074..7ee44059cb4 100644 --- a/pkg/kubelet/status/status_manager.go +++ b/pkg/kubelet/status/status_manager.go @@ -137,8 +137,11 @@ type Manager interface { // the provided podUIDs. RemoveOrphanedStatuses(podUIDs map[types.UID]bool) - // State returns a read-only interface to the internal status manager state. - State() state.Reader + // GetContainerResourceAllocation returns checkpointed ResourcesAllocated value for the container + GetContainerResourceAllocation(podUID string, containerName string) (v1.ResourceList, bool) + + // GetPodResizeStatus returns checkpointed PodStatus.Resize value + GetPodResizeStatus(podUID string) (v1.PodResizeStatus, bool) // SetPodAllocation checkpoints the resources allocated to a pod's containers. SetPodAllocation(pod *v1.Pod) error @@ -183,6 +186,17 @@ func isPodStatusByKubeletEqual(oldStatus, status *v1.PodStatus) bool { } func (m *manager) Start() { + // Create pod allocation checkpoint manager even if client is nil so as to allow local get/set of ResourcesAllocated & Resize + if utilfeature.DefaultFeatureGate.Enabled(features.InPlacePodVerticalScaling) { + stateImpl, err := state.NewStateCheckpoint(m.stateFileDirectory, podStatusManagerStateFile) + if err != nil { + // This is a crictical, non-recoverable failure. + klog.ErrorS(err, "Could not initialize pod allocation checkpoint manager, please drain node and remove policy state file") + panic(err) + } + m.state = stateImpl + } + // Don't start the status manager if we don't have a client. This will happen // on the master, where the kubelet is responsible for bootstrapping the pods // of the master components. @@ -191,15 +205,6 @@ func (m *manager) Start() { return } - if utilfeature.DefaultFeatureGate.Enabled(features.InPlacePodVerticalScaling) { - stateImpl, err := state.NewStateCheckpoint(m.stateFileDirectory, podStatusManagerStateFile) - if err != nil { - klog.ErrorS(err, "Could not initialize pod allocation checkpoint manager, please drain node and remove policy state file") - return - } - m.state = stateImpl - } - klog.InfoS("Starting to sync pod status with apiserver") //nolint:staticcheck // SA1015 Ticker can leak since this is only called once and doesn't handle termination. @@ -227,9 +232,20 @@ func (m *manager) Start() { }, 0) } -// State returns the pod resources checkpoint state of the pod status manager -func (m *manager) State() state.Reader { - return m.state +// GetContainerResourceAllocation returns the last checkpointed ResourcesAllocated values +// If checkpoint manager has not been initialized, it returns nil, false +func (m *manager) GetContainerResourceAllocation(podUID string, containerName string) (v1.ResourceList, bool) { + m.podStatusesLock.RLock() + defer m.podStatusesLock.RUnlock() + return m.state.GetContainerResourceAllocation(podUID, containerName) +} + +// GetPodResizeStatus returns the last checkpointed ResizeStaus value +// If checkpoint manager has not been initialized, it returns nil, false +func (m *manager) GetPodResizeStatus(podUID string) (v1.PodResizeStatus, bool) { + m.podStatusesLock.RLock() + defer m.podStatusesLock.RUnlock() + return m.state.GetPodResizeStatus(podUID) } // SetPodAllocation checkpoints the resources allocated to a pod's containers diff --git a/pkg/kubelet/status/testing/mock_pod_status_provider.go b/pkg/kubelet/status/testing/mock_pod_status_provider.go index 3112a4ab625..2de1bee1691 100644 --- a/pkg/kubelet/status/testing/mock_pod_status_provider.go +++ b/pkg/kubelet/status/testing/mock_pod_status_provider.go @@ -27,7 +27,6 @@ import ( v1 "k8s.io/api/core/v1" types "k8s.io/apimachinery/pkg/types" container "k8s.io/kubernetes/pkg/kubelet/container" - state "k8s.io/kubernetes/pkg/kubelet/status/state" ) // MockPodStatusProvider is a mock of PodStatusProvider interface. @@ -189,6 +188,36 @@ func (m *MockManager) EXPECT() *MockManagerMockRecorder { return m.recorder } +// GetContainerResourceAllocation mocks base method. +func (m *MockManager) GetContainerResourceAllocation(podUID, containerName string) (v1.ResourceList, bool) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetContainerResourceAllocation", podUID, containerName) + ret0, _ := ret[0].(v1.ResourceList) + ret1, _ := ret[1].(bool) + return ret0, ret1 +} + +// GetContainerResourceAllocation indicates an expected call of GetContainerResourceAllocation. +func (mr *MockManagerMockRecorder) GetContainerResourceAllocation(podUID, containerName interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetContainerResourceAllocation", reflect.TypeOf((*MockManager)(nil).GetContainerResourceAllocation), podUID, containerName) +} + +// GetPodResizeStatus mocks base method. +func (m *MockManager) GetPodResizeStatus(podUID string) (v1.PodResizeStatus, bool) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetPodResizeStatus", podUID) + ret0, _ := ret[0].(v1.PodResizeStatus) + ret1, _ := ret[1].(bool) + return ret0, ret1 +} + +// GetPodResizeStatus indicates an expected call of GetPodResizeStatus. +func (mr *MockManagerMockRecorder) GetPodResizeStatus(podUID interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetPodResizeStatus", reflect.TypeOf((*MockManager)(nil).GetPodResizeStatus), podUID) +} + // GetPodStatus mocks base method. func (m *MockManager) GetPodStatus(uid types.UID) (v1.PodStatus, bool) { m.ctrl.T.Helper() @@ -292,20 +321,6 @@ func (mr *MockManagerMockRecorder) Start() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Start", reflect.TypeOf((*MockManager)(nil).Start)) } -// State mocks base method. -func (m *MockManager) State() state.Reader { - m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "State") - ret0, _ := ret[0].(state.Reader) - return ret0 -} - -// State indicates an expected call of State. -func (mr *MockManagerMockRecorder) State() *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "State", reflect.TypeOf((*MockManager)(nil).State)) -} - // TerminatePod mocks base method. func (m *MockManager) TerminatePod(pod *v1.Pod) { m.ctrl.T.Helper()