kubelet: use local status manager as source of truth for pod phase

pkg/kubelet/kubelet.go

@@ -752,6 +752,7 @@ func NewMainKubelet(ctx context.Context,
 		klet.resyncInterval,
 		backOffPeriod,
 		klet.podCache,
+		klet.statusManager,
 		klet.allocationManager,
 	)
 

pkg/kubelet/kubelet_test.go

@@ -798,7 +798,7 @@ func TestVolumeAttachLimitExceededCleanup(t *testing.T) {
 	kl.podWorkers = newPodWorkers(
 		kl, kl.recorder, kl.workQueue,
 		kl.resyncInterval, backOffPeriod,
-		kl.podCache, kl.allocationManager,
+		kl.podCache, kl.statusManager, kl.allocationManager,
 	)
 
 	kl.volumeManager = kubeletvolume.NewFakeVolumeManager(nil, 0, nil, true /* volumeAttachLimitExceededError  */)

pkg/kubelet/pod_workers.go

@@ -39,6 +39,7 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet/events"
 	"k8s.io/kubernetes/pkg/kubelet/eviction"
 	"k8s.io/kubernetes/pkg/kubelet/metrics"
+	"k8s.io/kubernetes/pkg/kubelet/status"
 	kubetypes "k8s.io/kubernetes/pkg/kubelet/types"
 	"k8s.io/kubernetes/pkg/kubelet/util/queue"
 	"k8s.io/utils/clock"
@@ -604,6 +605,9 @@ type podWorkers struct {
 	// podCache stores kubecontainer.PodStatus for all pods.
 	podCache kubecontainer.ROCache
 
+	// statusProvider provides the local authoritative status of a pod
+	statusProvider status.PodStatusProvider
+
 	// allocationManager is used to allocate resources for pods
 	allocationManager allocation.Manager
 
@@ -617,6 +621,7 @@ func newPodWorkers(
 	workQueue queue.WorkQueue,
 	resyncInterval, backOffPeriod time.Duration,
 	podCache kubecontainer.ROCache,
+	statusProvider status.PodStatusProvider,
 	allocationManager allocation.Manager,
 ) PodWorkers {
 	return &podWorkers{
@@ -630,6 +635,7 @@ func newPodWorkers(
 		resyncInterval:                     resyncInterval,
 		backOffPeriod:                      backOffPeriod,
 		podCache:                           podCache,
+		statusProvider:                     statusProvider,
 		allocationManager:                  allocationManager,
 		clock:                              clock.RealClock{},
 	}
@@ -790,7 +796,7 @@ func (p *podWorkers) UpdatePod(ctx context.Context, options UpdatePodOptions) {
 			fullname: kubecontainer.BuildPodFullName(name, ns),
 		}
 		// if this pod is being synced for the first time, we need to make sure it is an active pod
-		if options.Pod != nil && (options.Pod.Status.Phase == v1.PodFailed || options.Pod.Status.Phase == v1.PodSucceeded) {
+		if options.Pod != nil && (p.getPodPhase(options.Pod) == v1.PodFailed || p.getPodPhase(options.Pod) == v1.PodSucceeded) {
 			// Check to see if the pod is not running and the pod is terminal; if this succeeds then record in the podWorker that it is terminated.
 			// This is needed because after a kubelet restart, we need to ensure terminal pods will NOT be considered active in Pod Admission. See http://issues.k8s.io/105523
 			// However, `filterOutInactivePods`, considers pods that are actively terminating as active. As a result, `IsPodKnownTerminated()` needs to return true and thus `terminatedAt` needs to be set.
@@ -873,7 +879,7 @@ func (p *podWorkers) UpdatePod(ctx context.Context, options UpdatePodOptions) {
 			status.deleted = true
 			status.terminatingAt = now
 			becameTerminating = true
-		case pod.Status.Phase == v1.PodFailed, pod.Status.Phase == v1.PodSucceeded:
+		case p.getPodPhase(pod) == v1.PodFailed, p.getPodPhase(pod) == v1.PodSucceeded:
 			updateLogger.V(4).Info("Pod is in a terminal phase (success/failed), begin teardown")
 			status.terminatingAt = now
 			becameTerminating = true
@@ -1756,3 +1762,17 @@ func (p *podWorkers) requeueLastPodUpdate(podUID types.UID, status *podSyncStatu
 	default:
 	}
 }
+
+// getPodPhase returns the authoritative pod phase from statusProvider if available,
+// falling back to pod.Status.Phase.
+func (p *podWorkers) getPodPhase(pod *v1.Pod) v1.PodPhase {
+	if pod == nil {
+		return ""
+	}
+	if p.statusProvider != nil {
+		if status, ok := p.statusProvider.GetPodStatus(pod.UID); ok {
+			return status.Phase
+		}
+	}
+	return pod.Status.Phase
+}

pkg/kubelet/pod_workers_test.go

@@ -45,6 +45,15 @@ import (
 	clocktesting "k8s.io/utils/clock/testing"
 )
 
+type fakePodStatusProvider struct {
+	statuses map[types.UID]v1.PodStatus
+}
+
+func (f *fakePodStatusProvider) GetPodStatus(uid types.UID) (v1.PodStatus, bool) {
+	status, ok := f.statuses[uid]
+	return status, ok
+}
+
 // fakePodWorkers runs sync pod function in serial, so we can have
 // deterministic behaviour in testing.
 type fakePodWorkers struct {
@@ -460,6 +469,7 @@ func createPodWorkersWithLogger(logger klog.Logger) (*podWorkers, *containertest
 		time.Second,
 		time.Millisecond,
 		fakeCache,
+		nil,
 		allocation.NewInMemoryManager(logger, nil, nil, nil, nil, nil, nil),
 	)
 	workers := w.(*podWorkers)
@@ -840,6 +850,41 @@ func TestUpdatePod(t *testing.T) {
 			},
 			expect: nil,
 		},
+		{
+			name: "a running pod is transitioned to terminating if the status manager says it is terminal, even if the API pod phase is running",
+			update: UpdatePodOptions{
+				UpdateType: kubetypes.SyncPodUpdate,
+				Pod:        newPodWithPhase("1", "running-pod", v1.PodRunning),
+			},
+			prepare: func(t *testing.T, tCtx context.Context, w *timeIncrementingWorkers) func() {
+				w.UpdatePod(tCtx, UpdatePodOptions{
+					UpdateType: kubetypes.SyncPodCreate,
+					Pod:        newPodWithPhase("1", "running-pod", v1.PodRunning),
+				})
+				// Set the status manager phase to PodFailed
+				w.w.statusProvider = &fakePodStatusProvider{
+					statuses: map[types.UID]v1.PodStatus{
+						"1": {Phase: v1.PodFailed},
+					},
+				}
+				return nil
+			},
+			expect: hasCancelFn(&podSyncStatus{
+				fullname:           "running-pod_ns",
+				syncedAt:           time.Unix(1, 0),
+				startedAt:          time.Unix(3, 0),
+				terminatingAt:      time.Unix(3, 0),
+				terminatedAt:       time.Unix(5, 0),
+				gracePeriod:        30,
+				startedTerminating: true,
+				finished:           true,
+				activeUpdate: &UpdatePodOptions{
+					Pod:            newPodWithPhase("1", "running-pod", v1.PodRunning),
+					KillPodOptions: &KillPodOptions{PodTerminationGracePeriodSecondsOverride: intp(30)},
+				},
+			}),
+			expectKnownTerminated: true,
+		},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			var uid types.UID
@@ -2168,6 +2213,7 @@ func TestFakePodWorkers(t *testing.T) {
 		time.Second,
 		time.Second,
 		fakeCache,
+		nil,
 		allocation.NewInMemoryManager(logger, nil, nil, nil, nil, nil, nil),
 	)
 	fakePodWorkers := &fakePodWorkers{