Merge pull request #633 from cybozu-go/retry-drain

Retry eviction of pod

Author: zoetrope

.gitignore

@@ -6,6 +6,7 @@
 
 # Project-local glide cache, RE: https://github.com/Masterminds/glide/issues/736
 .glide/
+vendor/
 
 # Editors
 *~

cluster.go

@@ -277,6 +277,8 @@ type Reboot struct {
 	CommandTimeoutSeconds  *int                  `json:"command_timeout_seconds,omitempty"`
 	CommandRetries         *int                  `json:"command_retries"`
 	CommandInterval        *int                  `json:"command_interval"`
+	EvictRetries           *int                  `json:"evict_retries"`
+	EvictInterval          *int                  `json:"evict_interval"`
 	ProtectedNamespaces    *metav1.LabelSelector `json:"protected_namespaces,omitempty"`
 }
 
@@ -495,6 +497,12 @@ func validateReboot(reboot Reboot) error {
 	if reboot.CommandInterval != nil && *reboot.CommandInterval < 0 {
 		return errors.New("command_interval must not be negative")
 	}
+	if reboot.EvictRetries != nil && *reboot.EvictRetries < 0 {
+		return errors.New("evict_retries must not be negative")
+	}
+	if reboot.EvictInterval != nil && *reboot.EvictInterval < 0 {
+		return errors.New("evict_interval must not be negative")
+	}
 	if reboot.MaxConcurrentReboots != nil && *reboot.MaxConcurrentReboots <= 0 {
 		return errors.New("max_concurrent_reboots must be positive")
 	}

cluster_test.go

@@ -110,6 +110,18 @@ func testClusterYAML(t *testing.T) {
 	if *c.Reboot.CommandInterval != 30 {
 		t.Error(`*c.Reboot.CommandInterval != 30`)
 	}
+	if c.Reboot.EvictRetries == nil {
+		t.Fatal(`c.Reboot.EvictRetries == nil`)
+	}
+	if *c.Reboot.EvictRetries != 10 {
+		t.Error(`*c.Reboot.EvictRetries != 10`)
+	}
+	if c.Reboot.EvictInterval == nil {
+		t.Fatal(`c.Reboot.EvictInterval == nil`)
+	}
+	if *c.Reboot.EvictInterval != 3 {
+		t.Error(`*c.Reboot.EvictInterval != 3`)
+	}
 	if c.Reboot.ProtectedNamespaces == nil {
 		t.Fatal(`c.Reboot.ProtectedNamespaces == nil`)
 	}

docs/cluster.md

@@ -68,13 +68,15 @@ Reboot
 ------
 
 | Name                       | Required | Type                             | Description                                                             |
-| -------------------------- | -------- | -------------------------------- | ----------------------------------------------------------------------- |
+|----------------------------| -------- | -------------------------------- |-------------------------------------------------------------------------|
 | `reboot_command`           | true     | array                            | A command to reboot.  List of strings.                                  |
 | `boot_check_command`       | true     | array                            | A command to check nodes booted.  List of strings.                      |
 | `eviction_timeout_seconds` | false    | *int                             | Deadline for eviction. Must be positive. Default: 600 (10 minutes).     |
 | `command_timeout_seconds`  | false    | *int                             | Deadline for rebooting. Zero means infinity. Default: wait indefinitely |
 | `command_retries`          | false    | *int                             | Number of reboot retries, not including initial attempt. Default: 0     |
 | `command_interval`         | false    | *int                             | Interval of time between reboot retries in seconds. Default: 0          |
+| `evict_retries`            | false    | *int                             | Number of eviction retries, not including initial attempt. Default: 0   |
+| `evict_interval`           | false    | *int                             | Interval of time between eviction retries in seconds. Default: 0        |
 | `max_concurrent_reboots`   | false    | *int                             | Maximum number of nodes to be rebooted concurrently. Default: 1         |
 | `protected_namespaces`     | false    | [`LabelSelector`][LabelSelector] | A label selector to protect namespaces.                                 |
 
@@ -89,6 +91,7 @@ If the node is not booted yet, this command should output `false` to stdout and
 If `command_timeout_seconds` is specified, the check command should return within `command_timeout_seconds` seconds, or it is considered failed.
 
 CKE tries to delete Pods in the `protected_namespaces` gracefully with the Kubernetes eviction API.
+If the eviction API has failed, CKE retries it for `evict_retries` times with `evict_interval`-second interval.
 If any of the Pods cannot be deleted, it aborts the operation.
 
 The Pods in the non-protected namespaces are also tried to be deleted gracefully with the Kubernetes eviction API, but they would be simply deleted if eviction is denied.

mtest/run_test.go

@@ -341,7 +341,11 @@ func connectEtcd() (*clientv3.Client, error) {
 }
 
 func getClusterStatus(cluster *cke.Cluster) (*cke.ClusterStatus, []cke.ResourceDefinition, error) {
-	controller := server.NewController(nil, 0, time.Hour, time.Second*2, nil, 10)
+	controller := server.NewController(nil, nil, &server.Config{
+		Interval:             0,
+		CertsGCInterval:      time.Hour,
+		MaxConcurrentUpdates: 10,
+	})
 
 	etcd, err := connectEtcd()
 	if err != nil {

op/reboot.go

@@ -42,6 +42,8 @@ type rebootDrainStartCommand struct {
 	entries             []*cke.RebootQueueEntry
 	protectedNamespaces *metav1.LabelSelector
 	apiserver           *cke.Node
+	evictAttempts       int
+	evictInterval       time.Duration
 
 	notifyFailedNode func(string)
 }
@@ -77,11 +79,22 @@ func (o *rebootDrainStartOp) NextCommand() cke.Commander {
 	}
 	o.finished = true
 
+	attempts := 1
+	if o.config.EvictRetries != nil {
+		attempts = *o.config.EvictRetries + 1
+	}
+	interval := 0 * time.Second
+	if o.config.EvictInterval != nil {
+		interval = time.Second * time.Duration(*o.config.EvictInterval)
+	}
+
 	return rebootDrainStartCommand{
 		entries:             o.entries,
 		protectedNamespaces: o.config.ProtectedNamespaces,
 		apiserver:           o.apiserver,
 		notifyFailedNode:    o.notifyFailedNode,
+		evictAttempts:       attempts,
+		evictInterval:       interval,
 	}
 }
 
@@ -156,7 +169,7 @@ func (c rebootDrainStartCommand) Run(ctx context.Context, inf cke.Infrastructure
 
 	// next, evict pods on each node
 	for _, entry := range evictNodes {
-		err := evictOrDeleteNodePod(ctx, cs, entry.Node, protected)
+		err := evictOrDeleteNodePod(ctx, cs, entry.Node, protected, c.evictAttempts, c.evictInterval)
 		if err != nil {
 			c.notifyFailedNode(entry.Node)
 			err = drainBackOff(ctx, inf, entry, err)

op/reboot_decide.go

@@ -74,14 +74,21 @@ func checkJobPodNotExist(ctx context.Context, cs *kubernetes.Clientset, node str
 // evictOrDeleteNodePod evicts or delete Pods on the specified Node.
 // It first tries eviction. If the eviction failed and the Pod's namespace is not protected, it deletes the Pod.
 // If a running Job Pod exists, this function returns an error.
-func evictOrDeleteNodePod(ctx context.Context, cs *kubernetes.Clientset, node string, protected map[string]bool) error {
+func evictOrDeleteNodePod(ctx context.Context, cs *kubernetes.Clientset, node string, protected map[string]bool, attempts int, interval time.Duration) error {
 	return enumeratePods(ctx, cs, node, func(pod *corev1.Pod) error {
+		evictCount := 0
+	EVICT:
+		log.Info("start evicting pod", map[string]interface{}{
+			"namespace": pod.Namespace,
+			"name":      pod.Name,
+		})
 		err := cs.CoreV1().Pods(pod.Namespace).EvictV1(ctx, &policyv1.Eviction{
 			ObjectMeta: metav1.ObjectMeta{Name: pod.Name, Namespace: pod.Namespace},
 		})
+		evictCount++
 		switch {
 		case err == nil:
-			log.Info("start evicting pod", map[string]interface{}{
+			log.Info("evicted pod", map[string]interface{}{
 				"namespace": pod.Namespace,
 				"name":      pod.Name,
 			})
@@ -105,6 +112,18 @@ func evictOrDeleteNodePod(ctx context.Context, cs *kubernetes.Clientset, node st
 				"name":      pod.Name,
 			})
 		default:
+			if evictCount < attempts {
+				select {
+				case <-ctx.Done():
+					return ctx.Err()
+				case <-time.After(interval):
+				}
+				log.Info("retry eviction of pod", map[string]interface{}{
+					"namespace": pod.Namespace,
+					"name":      pod.Name,
+				})
+				goto EVICT
+			}
 			return fmt.Errorf("failed to evict pod %s/%s due to PDB: %w", pod.Namespace, pod.Name, err)
 		}
 		return nil

pkg/cke/main.go

@@ -116,7 +116,11 @@ func main() {
 	}
 
 	// Controller
-	controller := server.NewController(session, interval, gcInterval, timeout, addon, maxConcurrentUpdates)
+	controller := server.NewController(session, addon, &server.Config{
+		Interval:             interval,
+		CertsGCInterval:      gcInterval,
+		MaxConcurrentUpdates: maxConcurrentUpdates,
+	})
 	well.Go(controller.Run)
 
 	// API server

server/config.go

@@ -0,0 +1,13 @@
+package server
+
+import "time"
+
+// Config is the configuration for cke-server.
+type Config struct {
+	// Interval is the interval of the main loop.
+	Interval time.Duration
+	// CertsGCInterval is the interval of the certificate garbage collection.
+	CertsGCInterval time.Duration
+	// MaxConcurrentUpdates is the maximum number of concurrent updates.
+	MaxConcurrentUpdates int
+}

server/control.go

@@ -23,17 +23,14 @@ var (
 
 // Controller manage operations
 type Controller struct {
-	session              *concurrency.Session
-	interval             time.Duration
-	certsGCInterval      time.Duration
-	timeout              time.Duration
-	addon                Integrator
-	maxConcurrentUpdates int
+	session *concurrency.Session
+	addon   Integrator
+	config  *Config
 }
 
 // NewController construct controller instance
-func NewController(s *concurrency.Session, interval, gcInterval, timeout time.Duration, addon Integrator, maxConcurrentUpdates int) Controller {
-	return Controller{s, interval, gcInterval, timeout, addon, maxConcurrentUpdates}
+func NewController(s *concurrency.Session, addon Integrator, config *Config) Controller {
+	return Controller{s, addon, config}
 }
 
 // Run execute procedures with leader elections
@@ -149,7 +146,7 @@ func (c Controller) runLoop(ctx context.Context, leaderKey string) error {
 		case <-ctx.Done():
 			return ctx.Err()
 		}
-		ticker := time.NewTicker(c.interval)
+		ticker := time.NewTicker(c.config.Interval)
 		defer ticker.Stop()
 		for {
 			select {
@@ -170,7 +167,7 @@ func (c Controller) runLoop(ctx context.Context, leaderKey string) error {
 			return ctx.Err()
 		default:
 		}
-		ticker := time.NewTicker(c.certsGCInterval)
+		ticker := time.NewTicker(c.config.CertsGCInterval)
 		defer ticker.Stop()
 		for {
 			select {
@@ -347,7 +344,7 @@ func (c Controller) runOnce(ctx context.Context, leaderKey string, tick <-chan t
 		DrainCompleted: drainCompleted,
 		DrainTimedout:  drainTimedout,
 		RebootDequeued: rebootDequeued,
-	}, c.maxConcurrentUpdates)
+	}, c.config)
 
 	st := &cke.ServerStatus{
 		Phase:     phase,