diff --git a/pkg/kubelet/kubelet.go b/pkg/kubelet/kubelet.go index cdeab0a1abb..318fe4adc8b 100644 --- a/pkg/kubelet/kubelet.go +++ b/pkg/kubelet/kubelet.go @@ -752,6 +752,7 @@ func NewMainKubelet(ctx context.Context, klet.resyncInterval, backOffPeriod, klet.podCache, + klet.statusManager, klet.allocationManager, ) diff --git a/pkg/kubelet/kubelet_test.go b/pkg/kubelet/kubelet_test.go index 4ed6a67ac8b..1f03894d7d6 100644 --- a/pkg/kubelet/kubelet_test.go +++ b/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 */) diff --git a/pkg/kubelet/pod_workers.go b/pkg/kubelet/pod_workers.go index 0f3034e2622..83f817f03f7 100644 --- a/pkg/kubelet/pod_workers.go +++ b/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 +} diff --git a/pkg/kubelet/pod_workers_test.go b/pkg/kubelet/pod_workers_test.go index 22837ecb81a..35ab0fdbc42 100644 --- a/pkg/kubelet/pod_workers_test.go +++ b/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{