Fix device plugin admission failure after container restart

When a container restarts before kubelet restarts, containerMap has
multiple entries (old exited + new running). GetContainerID() may
return the exited container, causing the running check to fail. Fixed
by checking if ANY container for the pod/name is running.

Also filter terminal pods from podresources since they no longer
consume resources, and fix test error handling to avoid exiting
Eventually immediately on transient errors.

Signed-off-by: Sascha Grunert <sgrunert@redhat.com>

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
 }

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))
+
 		})
 	})
 }