11package evict
22
3- import "time"
3+ import (
4+ "context"
5+ "fmt"
6+ "log"
7+ "time"
8+
9+ "github.com/samber/lo"
10+ corev1 "k8s.io/api/core/v1"
11+ policyv1 "k8s.io/api/policy/v1"
12+ apierrors "k8s.io/apimachinery/pkg/api/errors"
13+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
14+ "k8s.io/apimachinery/pkg/fields"
15+ "k8s.io/apimachinery/pkg/runtime"
16+ "k8s.io/apimachinery/pkg/runtime/schema"
17+ "k8s.io/apimachinery/pkg/util/wait"
18+ "k8s.io/client-go/kubernetes"
19+ "k8s.io/client-go/tools/pager"
20+ "k8s.io/client-go/util/retry"
21+ )
422
523// nodeDrainOptions captures the per-call tunables for evicting a node's pods.
624type nodeDrainOptions struct {
@@ -9,3 +27,149 @@ type nodeDrainOptions struct {
927 NodeTimeout time.Duration // total wait for the node to become empty
1028 PollInterval time.Duration // interval between empty-checks; default 2s
1129}
30+
31+ // TODO(CASCL-1386): remove this block once the EKS managed node group eviction
32+ // (next PR in the stack) calls these primitives. They are introduced here as
33+ // shared building blocks; golangci-lint's `unused` runs with `tests = false`,
34+ // so the test coverage in this package does not count as use. Referencing the
35+ // two entry points keeps `unused` quiet until the first production caller lands
36+ // (drainNode/cordonNodes transitively reach every other helper below).
37+ var (
38+ _ = drainNode
39+ _ = cordonNodes
40+ )
41+
42+ // drainNode evicts every evictable pod from the node and waits for the node to
43+ // become empty. Pods owned by a DaemonSet, mirror pods, terminating pods and
44+ // completed Job pods are skipped — the kubelet handles their cleanup when the
45+ // underlying instance disappears.
46+ //
47+ // Pods that cannot be evicted (PDB-blocked beyond EvictionTimeout, etc.) are
48+ // logged as warnings; drainNode then continues with the remaining pods rather
49+ // than aborting the whole run.
50+ func drainNode (ctx context.Context , clientset kubernetes.Interface , nodeName string , opts nodeDrainOptions ) error {
51+ if opts .DryRun {
52+ log .Printf ("[dry-run] would drain node %s" , nodeName )
53+ return nil
54+ }
55+ pods , err := listPodsOnNode (ctx , clientset , nodeName )
56+ if err != nil {
57+ return fmt .Errorf ("failed to list pods on node %s: %w" , nodeName , err )
58+ }
59+ for _ , p := range pods {
60+ if shouldSkipEviction (& p ) {
61+ continue
62+ }
63+ if err := evictPodWithRetry (ctx , clientset , & p , opts .EvictionTimeout , opts .PollInterval ); err != nil {
64+ log .Printf ("Warning: pod %s/%s: %v" , p .Namespace , p .Name , err )
65+ }
66+ }
67+ return waitForNodeEmpty (ctx , clientset , nodeName , opts .NodeTimeout , opts .PollInterval )
68+ }
69+
70+ // listPodsOnNode enumerates pods scheduled on the given node, server-side
71+ // filtered via the spec.nodeName field selector.
72+ func listPodsOnNode (ctx context.Context , clientset kubernetes.Interface , nodeName string ) (pods []corev1.Pod , err error ) {
73+ p := pager .New (func (ctx context.Context , opts metav1.ListOptions ) (runtime.Object , error ) {
74+ return clientset .CoreV1 ().Pods (metav1 .NamespaceAll ).List (ctx , opts )
75+ })
76+ if err = p .EachListItem (ctx , metav1.ListOptions {
77+ FieldSelector : fields .OneTermEqualSelector ("spec.nodeName" , nodeName ).String (),
78+ }, func (obj runtime.Object ) error {
79+ pods = append (pods , * obj .(* corev1.Pod ))
80+ return nil
81+ }); err != nil {
82+ return nil , err
83+ }
84+ return pods , nil
85+ }
86+
87+ // evictPodWithRetry sends a single Eviction request and retries while the
88+ // apiserver returns 429 TooManyRequests (the canonical PDB-blocked signal).
89+ // Retries at retryInterval for up to timeout's worth of attempts; the caller
90+ // then logs and moves to the next pod.
91+ //
92+ // 404 (pod already gone) is treated as success — the eviction goal is met.
93+ // Non-429 errors are returned immediately.
94+ func evictPodWithRetry (ctx context.Context , clientset kubernetes.Interface , p * corev1.Pod , timeout , retryInterval time.Duration ) error {
95+ eviction := & policyv1.Eviction {
96+ ObjectMeta : metav1.ObjectMeta {Name : p .Name , Namespace : p .Namespace },
97+ }
98+ steps := 1
99+ if retryInterval > 0 {
100+ steps = max (1 , int (timeout / retryInterval ))
101+ }
102+ backoff := wait.Backoff {Duration : retryInterval , Factor : 1 , Steps : steps }
103+ err := retry .OnError (backoff , apierrors .IsTooManyRequests , func () error {
104+ return clientset .CoreV1 ().Pods (p .Namespace ).EvictV1 (ctx , eviction )
105+ })
106+ switch {
107+ case err == nil :
108+ log .Printf ("Evicted pod %s/%s." , p .Namespace , p .Name )
109+ return nil
110+ case apierrors .IsNotFound (err ):
111+ return nil // already gone — eviction goal met
112+ case apierrors .IsTooManyRequests (err ):
113+ // OnError exhausted its retries while still 429-blocked.
114+ return fmt .Errorf ("eviction timed out (likely PDB-blocked): %w" , err )
115+ default :
116+ return fmt .Errorf ("eviction failed: %w" , err )
117+ }
118+ }
119+
120+ // waitForNodeEmpty polls the node until no pod still occupies it. A pod is
121+ // considered to still occupy the node until it is actually gone from the API
122+ // server — including pods that are merely terminating (DeletionTimestamp
123+ // set). This matters for callers that terminate the underlying EC2 instance
124+ // next: returning early while a pod is mid-grace-period would kill the
125+ // container before its preStop hook / graceful shutdown finishes.
126+ //
127+ // DaemonSet pods, mirror pods and completed pods are not counted: DS and
128+ // mirror pods are tied to the node's lifecycle and disappear with it,
129+ // completed pods are inert and GC'd promptly.
130+ func waitForNodeEmpty (ctx context.Context , clientset kubernetes.Interface , nodeName string , timeout , pollInterval time.Duration ) error {
131+ return wait .PollUntilContextTimeout (ctx , pollInterval , timeout , true , func (ctx context.Context ) (bool , error ) {
132+ pods , err := listPodsOnNode (ctx , clientset , nodeName )
133+ if err != nil {
134+ return false , fmt .Errorf ("failed to list pods on node %s: %w" , nodeName , err )
135+ }
136+ return lo .NoneBy (pods , func (p corev1.Pod ) bool {
137+ return podOccupiesNode (& p )
138+ }), nil
139+ })
140+ }
141+
142+ // shouldSkipEviction reports whether drainNode should leave a pod alone
143+ // rather than call the Eviction API on it. A pod already terminating is
144+ // skipped because issuing another eviction would 404 once the kubelet
145+ // finishes the deletion; every pod that does not occupy the node for drain
146+ // purposes (mirror, DaemonSet and completed pods) needs no eviction either.
147+ func shouldSkipEviction (p * corev1.Pod ) bool {
148+ return p .DeletionTimestamp != nil || ! podOccupiesNode (p )
149+ }
150+
151+ // podOccupiesNode reports whether a pod still consumes the node from a drain
152+ // perspective. Terminating pods DO occupy the node until they are gone;
153+ // returning false for them prematurely would let the caller terminate the
154+ // instance before the container finished its grace period.
155+ func podOccupiesNode (p * corev1.Pod ) bool {
156+ return ! isMirrorPod (p ) && ! isDaemonSetPod (p ) && ! isCompleted (p )
157+ }
158+
159+ func isMirrorPod (p * corev1.Pod ) bool {
160+ _ , ok := p .Annotations [corev1 .MirrorPodAnnotationKey ]
161+ return ok
162+ }
163+
164+ func isDaemonSetPod (p * corev1.Pod ) bool {
165+ ctrl := metav1 .GetControllerOf (p )
166+ if ctrl == nil || ctrl .Kind != "DaemonSet" {
167+ return false
168+ }
169+ gv , _ := schema .ParseGroupVersion (ctrl .APIVersion )
170+ return gv .Group == "apps"
171+ }
172+
173+ func isCompleted (p * corev1.Pod ) bool {
174+ return p .Status .Phase == corev1 .PodSucceeded || p .Status .Phase == corev1 .PodFailed
175+ }
0 commit comments