Merge pull request #138027 from zxqlxy/local-cherry-pick-135485-v1.34

Cherry pick of #135485: Fix device plugin admission failure after container restart

pkg/kubelet/apis/podresources/server_v1.go

@@ -64,6 +64,7 @@ func (p *v1PodResourcesServer) List(ctx context.Context, req *podresourcesv1.Lis
 
 	var pods []*v1.Pod
 	if p.useActivePods {
+		// GetActivePods already filters out terminal pods, so no need for additional filtering.
 		pods = p.podsProvider.GetActivePods()
 	} else {
 		pods = p.podsProvider.GetPods()

pkg/kubelet/cm/devicemanager/manager.go

@@ -1161,21 +1161,36 @@ func (m *ManagerImpl) ShouldResetExtendedResourceCapacity() bool {
 }
 
 func (m *ManagerImpl) isContainerAlreadyRunning(podUID, cntName string) bool {
-	cntID, err := m.containerMap.GetContainerID(podUID, cntName)
-	if err != nil {
-		klog.ErrorS(err, "Container not found in the initial map, assumed NOT running", "podUID", podUID, "containerName", cntName)
+	// Check if ANY container for this pod/container name is running.
+	// This handles the case where a container restarted before kubelet restart,
+	// so the containerMap might have multiple entries (old exited + new running).
+	// We need to check all of them to see if any are running.
+	foundAnyContainer := false
+	foundRunningContainer := false
+
+	m.containerMap.Visit(func(visitPodUID, visitContainerName, visitContainerID string) {
+		if visitPodUID == podUID && visitContainerName == cntName {
+			foundAnyContainer = true
+			if m.containerRunningSet.Has(visitContainerID) {
+				foundRunningContainer = true
+				klog.V(4).InfoS("Container found in the initial running set", "podUID", podUID, "containerName", cntName, "containerID", visitContainerID)
+			}
+		}
+	})
+
+	if !foundAnyContainer {
+		klog.V(4).InfoS("Container not found in the initial map, assumed NOT running", "podUID", podUID, "containerName", cntName)
 		return false
 	}
 
-	// note that if container runtime is down when kubelet restarts, this set will be empty,
-	// so on kubelet restart containers will again fail admission, hitting https://github.com/kubernetes/kubernetes/issues/118559 again.
-	// This scenario should however be rare enough.
-	if !m.containerRunningSet.Has(cntID) {
-		klog.V(4).InfoS("Container not present in the initial running set", "podUID", podUID, "containerName", cntName, "containerID", cntID)
+	if !foundRunningContainer {
+		// note that if container runtime is down when kubelet restarts, this set will be empty,
+		// so on kubelet restart containers will again fail admission, hitting https://github.com/kubernetes/kubernetes/issues/118559 again.
+		// This scenario should however be rare enough.
+		klog.V(4).InfoS("Container found in map but not in running set", "podUID", podUID, "containerName", cntName)
 		return false
 	}
 
-	// Once we make it here we know we have a running container.
-	klog.V(4).InfoS("Container found in the initial set, assumed running", "podUID", podUID, "containerName", cntName, "containerID", cntID)
+	klog.V(4).InfoS("Container assumed running", "podUID", podUID, "containerName", cntName)
 	return true
 }

pkg/kubelet/kubelet.go

@@ -3066,7 +3066,15 @@ func (pp *kubeletPodsProvider) GetPods() []*v1.Pod {
 }
 
 func (pp *kubeletPodsProvider) GetPodByName(namespace, name string) (*v1.Pod, bool) {
-	return pp.kl.podManager.GetPodByName(namespace, name)
+	pod, found := pp.kl.podManager.GetPodByName(namespace, name)
+	if !found {
+		return nil, false
+	}
+	// use the same filtering logic as GetActivePods
+	if pp.kl.isPodInactive(pod) {
+		return nil, false
+	}
+	return pod, true
 }
 
 // ListenAndServePodResources runs the kubelet podresources grpc service

pkg/kubelet/kubelet_pods.go

@@ -1128,22 +1128,31 @@ func (kl *Kubelet) PodIsFinished(pod *v1.Pod) bool {
 func (kl *Kubelet) filterOutInactivePods(pods []*v1.Pod) []*v1.Pod {
 	filteredPods := make([]*v1.Pod, 0, len(pods))
 	for _, p := range pods {
-		// if a pod is fully terminated by UID, it should be excluded from the
-		// list of pods
-		if kl.podWorkers.IsPodKnownTerminated(p.UID) {
+		if kl.isPodInactive(p) {
 			continue
 		}
-
-		// terminal pods are considered inactive UNLESS they are actively terminating
-		if kl.isAdmittedPodTerminal(p) && !kl.podWorkers.IsPodTerminationRequested(p.UID) {
-			continue
-		}
-
 		filteredPods = append(filteredPods, p)
 	}
 	return filteredPods
 }
 
+// isPodInactive returns true if the pod is in a terminal phase
+// or is known to be fully terminated. This method should only be used
+// when the pod being processed is upstream of the pod worker, i.e.
+// the pods the pod manager is aware of.
+func (kl *Kubelet) isPodInactive(p *v1.Pod) bool {
+	// if a pod is fully terminated by UID, it should be excluded from the
+	// list of pods
+	if kl.podWorkers.IsPodKnownTerminated(p.UID) {
+		return true
+	}
+	// terminal pods are considered inactive UNLESS they are actively terminating
+	if kl.isAdmittedPodTerminal(p) && !kl.podWorkers.IsPodTerminationRequested(p.UID) {
+		return true
+	}
+	return false
+}
+
 // isAdmittedPodTerminal returns true if the provided config source pod is in
 // a terminal phase, or if the Kubelet has already indicated the pod has reached
 // a terminal phase but the config source has not accepted it yet. This method

test/e2e_node/device_plugin_test.go

@@ -40,6 +40,7 @@ import (
 	runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1"
 	kubeletdevicepluginv1beta1 "k8s.io/kubelet/pkg/apis/deviceplugin/v1beta1"
 	admissionapi "k8s.io/pod-security-admission/api"
+	"k8s.io/utils/ptr"
 
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -932,7 +933,13 @@ func testDevicePluginNodeReboot(f *framework.Framework, pluginSockDir string) {
 		// exposing those devices as resource has restarted. The expected behavior is that the application pod fails at admission time.
 		framework.It("Does not keep device plugin assignments across node reboots if fails admission (no pod restart, no device plugin re-registration)", func(ctx context.Context) {
 			podRECMD := fmt.Sprintf("devs=$(ls /tmp/ | egrep '^Dev-[0-9]+$') && echo stub devices: $devs && sleep %s", sleepIntervalForever)
-			pod1 := e2epod.NewPodClient(f).CreateSync(ctx, makeBusyboxPod(SampleDeviceResourceName, podRECMD))
+			testPod := makeBusyboxPod(SampleDeviceResourceName, podRECMD)
+			// Set a short termination grace period to speed up the test.
+			// After kubelet restart, this pod will fail admission and need to be terminated.
+			// The default 30s grace period causes the test to timeout waiting for the pod
+			// to be filtered from the active pods list.
+			testPod.Spec.TerminationGracePeriodSeconds = ptr.To(int64(1))
+			pod1 := e2epod.NewPodClient(f).CreateSync(ctx, testPod)
 			deviceIDRE := "stub devices: (Dev-[0-9]+)"
 			devID1, err := parseLog(ctx, f, pod1.Name, pod1.Name, deviceIDRE)
 			framework.ExpectNoError(err, "getting logs for pod %q", pod1.Name)
@@ -976,22 +983,24 @@ func testDevicePluginNodeReboot(f *framework.Framework, pluginSockDir string) {
 			// never restarted (runs "forever" from this test timescale perspective) hence re-doing this check
 			// is useless.
 			ginkgo.By("Verifying the device assignment after kubelet restart using podresources API")
-			gomega.Eventually(ctx, func() error {
-				v1PodResources, err = getV1NodeDevices(ctx)
-				return err
-			}, 30*time.Second, framework.Poll).ShouldNot(gomega.HaveOccurred(), "cannot fetch the compute resource assignment after kubelet restart")
 
-			// if we got this far, podresources API will now report 2 entries:
-			// - sample device plugin pod, running and doing fine
-			// - our test pod, in failed state. Pods in terminal state will still be reported, see https://github.com/kubernetes/kubernetes/issues/119423
-			// so we care about our test pod, and it will be present in the returned list till 119423 is fixed, but since it failed admission it must not have
-			// any device allocated to it, hence we check for empty device set in the podresources response. So, we check that
-			// A. our test pod must be present in the list response *and*
-			// B. it has no devices assigned to it.
-			// anything else is unexpected and thus makes the test fail. Once 119423 is fixed, a better, simpler and more intuitive check will be for the
-			// test pod to not be present in the podresources list response, but till that time we're stuck with this approach.
-			_, found := checkPodResourcesAssignment(v1PodResources, pod1.Namespace, pod1.Name, pod1.Spec.Containers[0].Name, SampleDeviceResourceName, []string{})
-			gomega.Expect(found).To(gomega.BeTrueBecause("%s/%s/%s failed admission, should not have devices registered", pod1.Namespace, pod1.Name, pod1.Spec.Containers[0].Name))
+			// if we got this far, podresources API will now report only the sample device plugin pod,
+			// because pods that failed admission are no longer reported in the list
+			// (see https://github.com/kubernetes/kubernetes/pull/132028).
+			// Hence, we verify that our test pod is NOT present in the podresources response.
+
+			// We need to poll because there's a delay between the pod failing admission and being
+			// removed from the podresources list.
+			gomega.Eventually(ctx, func(ctx context.Context) bool {
+				v1PodResources, err := getV1NodeDevices(ctx)
+				if err != nil {
+					framework.Logf("Failed to get node devices: %v, will retry", err)
+					return true // Return true to continue polling on error
+				}
+				_, found := checkPodResourcesAssignment(v1PodResources, pod1.Namespace, pod1.Name, pod1.Spec.Containers[0].Name, SampleDeviceResourceName, []string{})
+				return found
+			}, 30*time.Second, framework.Poll).Should(gomega.BeFalseBecause("%s/%s/%s failed admission, so it must not appear in podresources list", pod1.Namespace, pod1.Name, pod1.Spec.Containers[0].Name))
+
 		})
 	})
 }

test/e2e_node/podresources_test.go

@@ -995,23 +995,31 @@ func podresourcesGetTests(ctx context.Context, f *framework.Framework, cli kubel
 
 	restartNever := v1.RestartPolicyNever
 	tpd = newTestPodData()
-	ginkgo.By("checking the output when only pod require CPU is terminated")
-	expected = []podDesc{
-		{
-			podName:        "pod-01",
-			cntName:        "cnt-00",
-			cpuRequest:     1000,
-			restartPolicy:  &restartNever,
-			mainCntCommand: []string{"sh", "-c", "/bin/true"},
-		},
+	ginkgo.By("checking Get() returns an error for a terminated pod")
+
+	completedDesc := podDesc{
+		podName:        "pod-01",
+		cntName:        "cnt-00",
+		cpuRequest:     1000,
+		restartPolicy:  &restartNever,
+		mainCntCommand: []string{"sh", "-c", "/bin/true"},
+	}
+
+	completedPod := e2epod.NewPodClient(f).Create(ctx, makePodResourcesTestPod(completedDesc))
+	framework.Logf("created pod %s", completedDesc.podName)
+	tpd.PodMap[completedDesc.podName] = completedPod
+	// Wait for the pod to complete to avoid races.
+	err = e2epod.WaitForPodCondition(ctx, f.ClientSet, f.Namespace.Name, completedPod.Name, "Pod Succeeded", 2*time.Minute, testutils.PodSucceeded)
+	framework.ExpectNoError(err, "pod did not succeed as expected")
+
+	resp, err = cli.Get(ctx, &kubeletpodresourcesv1.GetPodResourcesRequest{PodName: completedDesc.podName, PodNamespace: f.Namespace.Name})
+	gomega.Expect(err).To(gomega.HaveOccurred(), "expected Get() to return an error for a terminated pod")
+	// Returned PodResources for a terminated pod must be empty.
+	pr := resp.GetPodResources()
+	if pr != nil {
+		gomega.Expect(pr.GetContainers()).To(gomega.BeEmpty(),
+			"expected no container resources in response for terminated pod; got: %#v", pr.GetContainers())
 	}
-	tpd.createPodsForTest(ctx, f, expected)
-	resp, err = cli.Get(ctx, &kubeletpodresourcesv1.GetPodResourcesRequest{PodName: "pod-01", PodNamespace: f.Namespace.Name})
-	podResourceList = []*kubeletpodresourcesv1.PodResources{resp.GetPodResources()}
-	gomega.Expect(err).To(gomega.HaveOccurred(), "pod not found")
-	res = convertToMap(podResourceList)
-	err = matchPodDescWithResources(expected, res)
-	framework.ExpectNoError(err, "matchPodDescWithResources() failed err %v", err)
 	tpd.deletePodsForTest(ctx, f)
 
 	if sidecarContainersEnabled {