kubelet: updated logic of verifying a static critical pod

pkg/kubelet/eviction/eviction_manager.go

@@ -57,6 +57,8 @@ type managerImpl struct {
 	config Config
 	// the function to invoke to kill a pod
 	killPodFunc KillPodFunc
+	// the function to get the mirror pod by a given statid pod
+	mirrorPodFunc MirrorPodFunc
 	// the interface that knows how to do image gc
 	imageGC ImageGC
 	// the interface that knows how to do container gc
@@ -99,6 +101,7 @@ func NewManager(
 	summaryProvider stats.SummaryProvider,
 	config Config,
 	killPodFunc KillPodFunc,
+	mirrorPodFunc MirrorPodFunc,
 	imageGC ImageGC,
 	containerGC ContainerGC,
 	recorder record.EventRecorder,
@@ -108,6 +111,7 @@ func NewManager(
 	manager := &managerImpl{
 		clock:                        clock,
 		killPodFunc:                  killPodFunc,
+		mirrorPodFunc:                mirrorPodFunc,
 		imageGC:                      imageGC,
 		containerGC:                  containerGC,
 		config:                       config,
@@ -544,9 +548,19 @@ func (m *managerImpl) evictPod(pod *v1.Pod, gracePeriodOverride int64, evictMsg
 	// If the pod is marked as critical and static, and support for critical pod annotations is enabled,
 	// do not evict such pods. Static pods are not re-admitted after evictions.
 	// https://github.com/kubernetes/kubernetes/issues/40573 has more details.
-	if kubelettypes.IsCriticalPod(pod) && kubepod.IsStaticPod(pod) {
-		klog.Errorf("eviction manager: cannot evict a critical static pod %s", format.Pod(pod))
-		return false
+	if kubepod.IsStaticPod(pod) {
+		// need mirrorPod to check its "priority" value; static pod doesn't carry it
+		if mirrorPod, ok := m.mirrorPodFunc(pod); ok && mirrorPod != nil {
+			// skip only when it's a static and critical pod
+			if kubelettypes.IsCriticalPod(mirrorPod) {
+				klog.Errorf("eviction manager: cannot evict a critical static pod %s", format.Pod(pod))
+				return false
+			}
+		} else {
+			// we should never hit this
+			klog.Errorf("eviction manager: cannot get mirror pod from static pod %s, so cannot evict it", format.Pod(pod))
+			return false
+		}
 	}
 	status := v1.PodStatus{
 		Phase:   v1.PodFailed,

pkg/kubelet/eviction/eviction_manager_test.go

@@ -1165,6 +1165,11 @@ func TestCriticalPodsAreNotEvicted(t *testing.T) {
 	activePodsFunc := func() []*v1.Pod {
 		return pods
 	}
+	mirrorPodFunc := func(staticPod *v1.Pod) (*v1.Pod, bool) {
+		mirrorPod := staticPod.DeepCopy()
+		mirrorPod.Annotations[kubelettypes.ConfigSourceAnnotationKey] = kubelettypes.ApiserverSource
+		return mirrorPod, true
+	}
 
 	fakeClock := clock.NewFakeClock(time.Now())
 	podKiller := &mockPodKiller{}
@@ -1199,6 +1204,7 @@ func TestCriticalPodsAreNotEvicted(t *testing.T) {
 	manager := &managerImpl{
 		clock:                        fakeClock,
 		killPodFunc:                  podKiller.killPodNow,
+		mirrorPodFunc:                mirrorPodFunc,
 		imageGC:                      diskGC,
 		containerGC:                  diskGC,
 		config:                       config,

pkg/kubelet/eviction/types.go

@@ -94,6 +94,10 @@ type ContainerGC interface {
 // gracePeriodOverride - the grace period override to use instead of what is on the pod spec
 type KillPodFunc func(pod *v1.Pod, status v1.PodStatus, gracePeriodOverride *int64) error
 
+// MirrorPodFunc returns the mirror pod for the given static pod and
+// whether it was known to the pod manager.
+type MirrorPodFunc func(*v1.Pod) (*v1.Pod, bool)
+
 // ActivePodsFunc returns pods bound to the kubelet that are active (i.e. non-terminal state)
 type ActivePodsFunc func() []*v1.Pod
 

pkg/kubelet/kubelet.go

@@ -827,7 +827,7 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 	klet.podKillingCh = make(chan *kubecontainer.PodPair, podKillingChannelCapacity)
 
 	// setup eviction manager
-	evictionManager, evictionAdmitHandler := eviction.NewManager(klet.resourceAnalyzer, evictionConfig, killPodNow(klet.podWorkers, kubeDeps.Recorder), klet.imageManager, klet.containerGC, kubeDeps.Recorder, nodeRef, klet.clock)
+	evictionManager, evictionAdmitHandler := eviction.NewManager(klet.resourceAnalyzer, evictionConfig, killPodNow(klet.podWorkers, kubeDeps.Recorder), klet.podManager.GetMirrorPodByPod, klet.imageManager, klet.containerGC, kubeDeps.Recorder, nodeRef, klet.clock)
 
 	klet.evictionManager = evictionManager
 	klet.admitHandlers.AddPodAdmitHandler(evictionAdmitHandler)

pkg/kubelet/kubelet_test.go

@@ -301,7 +301,7 @@ func newTestKubeletWithImageList(
 		Namespace: "",
 	}
 	// setup eviction manager
-	evictionManager, evictionAdmitHandler := eviction.NewManager(kubelet.resourceAnalyzer, eviction.Config{}, killPodNow(kubelet.podWorkers, fakeRecorder), kubelet.imageManager, kubelet.containerGC, fakeRecorder, nodeRef, kubelet.clock)
+	evictionManager, evictionAdmitHandler := eviction.NewManager(kubelet.resourceAnalyzer, eviction.Config{}, killPodNow(kubelet.podWorkers, fakeRecorder), kubelet.podManager.GetMirrorPodByPod, kubelet.imageManager, kubelet.containerGC, fakeRecorder, nodeRef, kubelet.clock)
 
 	kubelet.evictionManager = evictionManager
 	kubelet.admitHandlers.AddPodAdmitHandler(evictionAdmitHandler)

pkg/kubelet/runonce_test.go

@@ -120,7 +120,8 @@ func TestRunOnce(t *testing.T) {
 	fakeKillPodFunc := func(pod *v1.Pod, podStatus v1.PodStatus, gracePeriodOverride *int64) error {
 		return nil
 	}
-	evictionManager, evictionAdmitHandler := eviction.NewManager(kb.resourceAnalyzer, eviction.Config{}, fakeKillPodFunc, nil, nil, kb.recorder, nodeRef, kb.clock)
+	fakeMirrodPodFunc := func(*v1.Pod) (*v1.Pod, bool) { return nil, false }
+	evictionManager, evictionAdmitHandler := eviction.NewManager(kb.resourceAnalyzer, eviction.Config{}, fakeKillPodFunc, fakeMirrodPodFunc, nil, nil, kb.recorder, nodeRef, kb.clock)
 
 	kb.evictionManager = evictionManager
 	kb.admitHandlers.AddPodAdmitHandler(evictionAdmitHandler)

test/e2e_node/BUILD

@@ -103,6 +103,7 @@ go_test(
         "runtime_conformance_test.go",
         "security_context_test.go",
         "summary_test.go",
+        "system_node_critical_test.go",
         "volume_manager_test.go",
     ],
     embed = [":go_default_library"],
@@ -121,6 +122,7 @@ go_test(
         "//pkg/kubelet/cm/cpuset:go_default_library",
         "//pkg/kubelet/container:go_default_library",
         "//pkg/kubelet/eviction:go_default_library",
+        "//pkg/kubelet/eviction/api:go_default_library",
         "//pkg/kubelet/images:go_default_library",
         "//pkg/kubelet/kubeletconfig:go_default_library",
         "//pkg/kubelet/kubeletconfig/status:go_default_library",

test/e2e_node/system_node_critical_test.go

+/*
+Copyright 2019 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 e2e_node
+
+import (
+	"fmt"
+	"os"
+	"time"
+
+	"k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+	"k8s.io/apimachinery/pkg/util/uuid"
+	kubeapi "k8s.io/kubernetes/pkg/apis/core"
+	kubeletconfig "k8s.io/kubernetes/pkg/kubelet/apis/config"
+	evictionapi "k8s.io/kubernetes/pkg/kubelet/eviction/api"
+	"k8s.io/kubernetes/test/e2e/framework"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+var _ = framework.KubeDescribe("SystemNodeCriticalPod [Slow] [Serial] [Disruptive] [NodeFeature:SystemNodeCriticalPod]", func() {
+	f := framework.NewDefaultFramework("system-node-critical-pod-test")
+	// this test only manipulates pods in kube-system
+	f.SkipNamespaceCreation = true
+
+	Context("when create a system-node-critical pod", func() {
+		tempSetCurrentKubeletConfig(f, func(initialConfig *kubeletconfig.KubeletConfiguration) {
+			diskConsumed := resource.MustParse("200Mi")
+			summary := eventuallyGetSummary()
+			availableBytes := *(summary.Node.Fs.AvailableBytes)
+			initialConfig.EvictionHard = map[string]string{string(evictionapi.SignalNodeFsAvailable): fmt.Sprintf("%d", availableBytes-uint64(diskConsumed.Value()))}
+			initialConfig.EvictionMinimumReclaim = map[string]string{}
+		})
+
+		// Place the remainder of the test within a context so that the kubelet config is set before and after the test.
+		Context("", func() {
+			var staticPodName, mirrorPodName, podPath string
+			ns := kubeapi.NamespaceSystem
+
+			BeforeEach(func() {
+				By("create a static system-node-critical pod")
+				staticPodName = "static-disk-hog-" + string(uuid.NewUUID())
+				mirrorPodName = staticPodName + "-" + framework.TestContext.NodeName
+				podPath = framework.TestContext.KubeletConfig.StaticPodPath
+				// define a static pod consuming disk gradually
+				// the upper limit is 1024 (iterations) * 10485760 bytes (10MB) = 10GB
+				err := createStaticSystemNodeCriticalPod(
+					podPath, staticPodName, ns, busyboxImage, v1.RestartPolicyNever, 1024,
+					"dd if=/dev/urandom of=file${i} bs=10485760 count=1 2>/dev/null; sleep .1;",
+				)
+				Expect(err).ShouldNot(HaveOccurred())
+
+				By("wait for the mirror pod to be running")
+				Eventually(func() error {
+					return checkMirrorPodRunning(f.ClientSet, mirrorPodName, ns)
+				}, time.Minute, time.Second*2).Should(BeNil())
+			})
+
+			It("should not be evicted upon DiskPressure", func() {
+				By("wait for the node to have DiskPressure condition")
+				Eventually(func() error {
+					if hasNodeCondition(f, v1.NodeDiskPressure) {
+						return nil
+					}
+					msg := fmt.Sprintf("NodeCondition: %s not encountered yet", v1.NodeDiskPressure)
+					framework.Logf(msg)
+					return fmt.Errorf(msg)
+				}, time.Minute*2, time.Second*4).Should(BeNil())
+
+				By("check if it's running all the time")
+				Consistently(func() error {
+					err := checkMirrorPodRunning(f.ClientSet, mirrorPodName, ns)
+					if err == nil {
+						framework.Logf("mirror pod %q is running", mirrorPodName)
+					} else {
+						framework.Logf(err.Error())
+					}
+					return err
+				}, time.Minute*8, time.Second*4).ShouldNot(HaveOccurred())
+			})
+			AfterEach(func() {
+				By("delete the static pod")
+				err := deleteStaticPod(podPath, staticPodName, ns)
+				Expect(err).ShouldNot(HaveOccurred())
+
+				By("wait for the mirror pod to disappear")
+				Eventually(func() error {
+					return checkMirrorPodDisappear(f.ClientSet, mirrorPodName, ns)
+				}, time.Minute, time.Second*2).Should(BeNil())
+			})
+		})
+	})
+})
+
+func createStaticSystemNodeCriticalPod(dir, name, namespace, image string, restart v1.RestartPolicy,
+	iterations int, command string) error {
+	template := `
+apiVersion: v1
+kind: Pod
+metadata:
+  name: %s
+  namespace: %s
+spec:
+  priorityClassName: system-node-critical
+  containers:
+  - name: %s
+    image: %s
+    restartPolicy: %s
+    command: ["sh", "-c", "i=0; while [ $i -lt %d ]; do %s i=$(($i+1)); done; while true; do sleep 5; done"]
+`
+	file := staticPodPath(dir, name, namespace)
+	podYaml := fmt.Sprintf(template, name, namespace, name, image, string(restart), iterations, command)
+
+	f, err := os.OpenFile(file, os.O_RDWR|os.O_TRUNC|os.O_CREATE, 0666)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+
+	_, err = f.WriteString(podYaml)
+	return err
+}