UPSTREAM: 98424: register all pending pod deletions and check for kill

pkg/kubelet/cm/BUILD

@@ -10,7 +10,9 @@ go_library(
         "container_manager_stub.go",
         "container_manager_unsupported.go",
         "container_manager_windows.go",
+        "fake_container_manager.go",
         "fake_internal_container_lifecycle.go",
+        "fake_pod_container_manager.go",
         "helpers.go",
         "helpers_linux.go",
         "helpers_unsupported.go",

pkg/kubelet/cm/fake_container_manager.go

@@ -0,0 +1,202 @@
+/*
+Copyright 2021 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cm
+
+import (
+	"sync"
+
+	v1 "k8s.io/api/core/v1"
+
+	"k8s.io/apimachinery/pkg/api/resource"
+	internalapi "k8s.io/cri-api/pkg/apis"
+	podresourcesapi "k8s.io/kubelet/pkg/apis/podresources/v1"
+	"k8s.io/kubernetes/pkg/kubelet/cm/cpumanager"
+	"k8s.io/kubernetes/pkg/kubelet/cm/topologymanager"
+	"k8s.io/kubernetes/pkg/kubelet/config"
+	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
+	"k8s.io/kubernetes/pkg/kubelet/lifecycle"
+	"k8s.io/kubernetes/pkg/kubelet/pluginmanager/cache"
+	"k8s.io/kubernetes/pkg/kubelet/status"
+	schedulerframework "k8s.io/kubernetes/pkg/scheduler/framework"
+)
+
+type FakeContainerManager struct {
+	sync.Mutex
+	CalledFunctions                     []string
+	PodContainerManager                 *FakePodContainerManager
+	shouldResetExtendedResourceCapacity bool
+}
+
+var _ ContainerManager = &FakeContainerManager{}
+
+func NewFakeContainerManager() *FakeContainerManager {
+	return &FakeContainerManager{
+		PodContainerManager: NewFakePodContainerManager(),
+	}
+}
+
+func (cm *FakeContainerManager) Start(_ *v1.Node, _ ActivePodsFunc, _ config.SourcesReady, _ status.PodStatusProvider, _ internalapi.RuntimeService) error {
+	cm.Lock()
+	defer cm.Unlock()
+	cm.CalledFunctions = append(cm.CalledFunctions, "Start")
+	return nil
+}
+
+func (cm *FakeContainerManager) SystemCgroupsLimit() v1.ResourceList {
+	cm.Lock()
+	defer cm.Unlock()
+	cm.CalledFunctions = append(cm.CalledFunctions, "SystemCgroupsLimit")
+	return v1.ResourceList{}
+}
+
+func (cm *FakeContainerManager) GetNodeConfig() NodeConfig {
+	cm.Lock()
+	defer cm.Unlock()
+	cm.CalledFunctions = append(cm.CalledFunctions, "GetNodeConfig")
+	return NodeConfig{}
+}
+
+func (cm *FakeContainerManager) GetMountedSubsystems() *CgroupSubsystems {
+	cm.Lock()
+	defer cm.Unlock()
+	cm.CalledFunctions = append(cm.CalledFunctions, "GetMountedSubsystems")
+	return &CgroupSubsystems{}
+}
+
+func (cm *FakeContainerManager) GetQOSContainersInfo() QOSContainersInfo {
+	cm.Lock()
+	defer cm.Unlock()
+	cm.CalledFunctions = append(cm.CalledFunctions, "QOSContainersInfo")
+	return QOSContainersInfo{}
+}
+
+func (cm *FakeContainerManager) UpdateQOSCgroups() error {
+	cm.Lock()
+	defer cm.Unlock()
+	cm.CalledFunctions = append(cm.CalledFunctions, "UpdateQOSCgroups")
+	return nil
+}
+
+func (cm *FakeContainerManager) Status() Status {
+	cm.Lock()
+	defer cm.Unlock()
+	cm.CalledFunctions = append(cm.CalledFunctions, "Status")
+	return Status{}
+}
+
+func (cm *FakeContainerManager) GetNodeAllocatableReservation() v1.ResourceList {
+	cm.Lock()
+	defer cm.Unlock()
+	cm.CalledFunctions = append(cm.CalledFunctions, "GetNodeAllocatableReservation")
+	return nil
+}
+
+func (cm *FakeContainerManager) GetCapacity() v1.ResourceList {
+	cm.Lock()
+	defer cm.Unlock()
+	cm.CalledFunctions = append(cm.CalledFunctions, "GetCapacity")
+	c := v1.ResourceList{
+		v1.ResourceEphemeralStorage: *resource.NewQuantity(
+			int64(0),
+			resource.BinarySI),
+	}
+	return c
+}
+
+func (cm *FakeContainerManager) GetPluginRegistrationHandler() cache.PluginHandler {
+	cm.Lock()
+	defer cm.Unlock()
+	cm.CalledFunctions = append(cm.CalledFunctions, "GetPluginRegistrationHandler")
+	return nil
+}
+
+func (cm *FakeContainerManager) GetDevicePluginResourceCapacity() (v1.ResourceList, v1.ResourceList, []string) {
+	cm.Lock()
+	defer cm.Unlock()
+	cm.CalledFunctions = append(cm.CalledFunctions, "GetDevicePluginResourceCapacity")
+	return nil, nil, []string{}
+}
+
+func (cm *FakeContainerManager) NewPodContainerManager() PodContainerManager {
+	cm.Lock()
+	defer cm.Unlock()
+	cm.CalledFunctions = append(cm.CalledFunctions, "PodContainerManager")
+	return cm.PodContainerManager
+}
+
+func (cm *FakeContainerManager) GetResources(pod *v1.Pod, container *v1.Container) (*kubecontainer.RunContainerOptions, error) {
+	cm.Lock()
+	defer cm.Unlock()
+	cm.CalledFunctions = append(cm.CalledFunctions, "GetResources")
+	return &kubecontainer.RunContainerOptions{}, nil
+}
+
+func (cm *FakeContainerManager) UpdatePluginResources(*schedulerframework.NodeInfo, *lifecycle.PodAdmitAttributes) error {
+	cm.Lock()
+	defer cm.Unlock()
+	cm.CalledFunctions = append(cm.CalledFunctions, "UpdatePluginResources")
+	return nil
+}
+
+func (cm *FakeContainerManager) InternalContainerLifecycle() InternalContainerLifecycle {
+	cm.Lock()
+	defer cm.Unlock()
+	cm.CalledFunctions = append(cm.CalledFunctions, "InternalContainerLifecycle")
+	return &internalContainerLifecycleImpl{cpumanager.NewFakeManager(), topologymanager.NewFakeManager()}
+}
+
+func (cm *FakeContainerManager) GetPodCgroupRoot() string {
+	cm.Lock()
+	defer cm.Unlock()
+	cm.CalledFunctions = append(cm.CalledFunctions, "GetPodCgroupRoot")
+	return ""
+}
+
+func (cm *FakeContainerManager) GetDevices(_, _ string) []*podresourcesapi.ContainerDevices {
+	cm.Lock()
+	defer cm.Unlock()
+	cm.CalledFunctions = append(cm.CalledFunctions, "GetDevices")
+	return nil
+}
+
+func (cm *FakeContainerManager) ShouldResetExtendedResourceCapacity() bool {
+	cm.Lock()
+	defer cm.Unlock()
+	cm.CalledFunctions = append(cm.CalledFunctions, "ShouldResetExtendedResourceCapacity")
+	return cm.shouldResetExtendedResourceCapacity
+}
+
+func (cm *FakeContainerManager) GetAllocateResourcesPodAdmitHandler() lifecycle.PodAdmitHandler {
+	cm.Lock()
+	defer cm.Unlock()
+	cm.CalledFunctions = append(cm.CalledFunctions, "GetAllocateResourcesPodAdmitHandler")
+	return topologymanager.NewFakeManager()
+}
+
+func (cm *FakeContainerManager) UpdateAllocatedDevices() {
+	cm.Lock()
+	defer cm.Unlock()
+	cm.CalledFunctions = append(cm.CalledFunctions, "UpdateAllocatedDevices")
+	return
+}
+
+func (cm *FakeContainerManager) GetCPUs(_, _ string) []int64 {
+	cm.Lock()
+	defer cm.Unlock()
+	cm.CalledFunctions = append(cm.CalledFunctions, "GetCPUs")
+	return nil
+}

pkg/kubelet/cm/fake_pod_container_manager.go

@@ -0,0 +1,106 @@
+/*
+Copyright 2021 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cm
+
+import (
+	"reflect"
+	"sync"
+
+	"k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
+	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
+)
+
+type FakePodContainerManager struct {
+	sync.Mutex
+	CalledFunctions []string
+	Cgroups         map[types.UID]CgroupName
+}
+
+var _ PodContainerManager = &FakePodContainerManager{}
+
+func NewFakePodContainerManager() *FakePodContainerManager {
+	return &FakePodContainerManager{
+		Cgroups: make(map[types.UID]CgroupName),
+	}
+}
+
+func (m *FakePodContainerManager) AddPodFromCgroups(pod *kubecontainer.Pod) {
+	m.Lock()
+	defer m.Unlock()
+	m.Cgroups[pod.ID] = []string{pod.Name}
+}
+
+func (m *FakePodContainerManager) Exists(_ *v1.Pod) bool {
+	m.Lock()
+	defer m.Unlock()
+	m.CalledFunctions = append(m.CalledFunctions, "Exists")
+	return true
+}
+
+func (m *FakePodContainerManager) EnsureExists(_ *v1.Pod) error {
+	m.Lock()
+	defer m.Unlock()
+	m.CalledFunctions = append(m.CalledFunctions, "EnsureExists")
+	return nil
+}
+
+func (m *FakePodContainerManager) GetPodContainerName(_ *v1.Pod) (CgroupName, string) {
+	m.Lock()
+	defer m.Unlock()
+	m.CalledFunctions = append(m.CalledFunctions, "GetPodContainerName")
+	return nil, ""
+}
+
+func (m *FakePodContainerManager) Destroy(name CgroupName) error {
+	m.Lock()
+	defer m.Unlock()
+	m.CalledFunctions = append(m.CalledFunctions, "Destroy")
+	for key, cgname := range m.Cgroups {
+		if reflect.DeepEqual(cgname, name) {
+			delete(m.Cgroups, key)
+			return nil
+		}
+	}
+	return nil
+}
+
+func (m *FakePodContainerManager) ReduceCPULimits(_ CgroupName) error {
+	m.Lock()
+	defer m.Unlock()
+	m.CalledFunctions = append(m.CalledFunctions, "ReduceCPULimits")
+	return nil
+}
+
+func (m *FakePodContainerManager) GetAllPodsFromCgroups() (map[types.UID]CgroupName, error) {
+	m.Lock()
+	defer m.Unlock()
+	m.CalledFunctions = append(m.CalledFunctions, "GetAllPodsFromCgroups")
+	// return a copy for the race detector
+	grp := make(map[types.UID]CgroupName)
+	for key, value := range m.Cgroups {
+		grp[key] = value
+	}
+	return grp, nil
+}
+
+func (m *FakePodContainerManager) IsPodCgroup(cgroupfs string) (bool, types.UID) {
+	m.Lock()
+	defer m.Unlock()
+	m.CalledFunctions = append(m.CalledFunctions, "IsPodCgroup")
+	return false, types.UID("")
+}

pkg/kubelet/kubelet.go

@@ -1502,10 +1502,12 @@ func (kl *Kubelet) syncPod(o syncPodOptions) error {
 		return nil
 	}
 
-	// If the pod is a static pod and its mirror pod is still gracefully terminating,
-	// we do not want to start the new static pod until the old static pod is gracefully terminated.
+	// If a pod is still gracefully terminating, then we do not want to
+	// take further action. This mitigates static pods and deleted pods
+	// from getting rerun prematurely or their cgroups being deleted before
+	// the runtime cleans up.
 	podFullName := kubecontainer.GetPodFullName(pod)
-	if kl.podKiller.IsMirrorPodPendingTerminationByPodName(podFullName) {
+	if kl.podKiller.IsPodPendingTerminationByPodName(podFullName) {
 		return fmt.Errorf("pod %q is pending termination", podFullName)
 	}
 
@@ -1773,10 +1775,6 @@ func (kl *Kubelet) deletePod(pod *v1.Pod) error {
 		return fmt.Errorf("pod not found")
 	}
 	podPair := kubecontainer.PodPair{APIPod: pod, RunningPod: &runningPod}
-
-	if _, ok := kl.podManager.GetMirrorPodByPod(pod); ok {
-		kl.podKiller.MarkMirrorPodPendingTermination(pod)
-	}
 	kl.podKiller.KillPod(&podPair)
 
 	// We leave the volume/directory cleanup to the periodic cleanup routine.