Non-Graceful Node Shutdown before deleting k8s node

cluster.go

@@ -320,6 +320,16 @@ const DefaultRepairEvictionTimeoutSeconds = 600
 const DefaultRepairHealthCheckCommandTimeoutSeconds = 30
 const DefaultRepairCommandTimeoutSeconds = 30
 
+type Retire struct {
+	ShutdownCommand       []string `json:"shutdown_command"`
+	CheckCommand          []string `json:"check_command"`
+	CommandTimeoutSeconds *int     `json:"command_timeout_seconds,omitempty"`
+	CheckTimeoutSeconds   *int     `json:"check_timeout_seconds,omitempty"`
+}
+
+const DefaultRetireCommandTimeoutSeconds = 30
+const DefaultRetireCheckTimeoutSeconds = 300
+
 // Options is a set of optional parameters for k8s components.
 type Options struct {
 	Etcd              EtcdParams      `json:"etcd"`
@@ -343,6 +353,7 @@ type Cluster struct {
 	DNSService    string   `json:"dns_service"`
 	Reboot        Reboot   `json:"reboot"`
 	Repair        Repair   `json:"repair"`
+	Retire        Retire   `json:"retire"`
 	Options       Options  `json:"options"`
 }
 

op/kube_node_remove.go

@@ -1,11 +1,15 @@
 package op
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/cybozu-go/cke"
+	"github.com/cybozu-go/log"
+	"github.com/cybozu-go/well"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -16,12 +20,13 @@ import (
 type kubeNodeRemove struct {
 	apiserver *cke.Node
 	nodes     []*corev1.Node
+	config    *cke.Retire
 	done      bool
 }
 
 // KubeNodeRemoveOp removes k8s Node resources.
-func KubeNodeRemoveOp(apiserver *cke.Node, nodes []*corev1.Node) cke.Operator {
-	return &kubeNodeRemove{apiserver: apiserver, nodes: nodes}
+func KubeNodeRemoveOp(apiserver *cke.Node, nodes []*corev1.Node, config *cke.Retire) cke.Operator {
+	return &kubeNodeRemove{apiserver: apiserver, nodes: nodes, config: config}
 }
 
 func (o *kubeNodeRemove) Name() string {
@@ -34,7 +39,14 @@ func (o *kubeNodeRemove) NextCommand() cke.Commander {
 	}
 
 	o.done = true
-	return nodeRemoveCommand{o.apiserver, o.nodes}
+	return nodeRemoveCommand{
+		o.apiserver,
+		o.nodes,
+		o.config.ShutdownCommand,
+		o.config.CheckCommand,
+		o.config.CommandTimeoutSeconds,
+		o.config.CheckTimeoutSeconds,
+	}
 }
 
 func (o *kubeNodeRemove) Targets() []string {
@@ -44,8 +56,12 @@ func (o *kubeNodeRemove) Targets() []string {
 }
 
 type nodeRemoveCommand struct {
-	apiserver *cke.Node
-	nodes     []*corev1.Node
+	apiserver           *cke.Node
+	nodes               []*corev1.Node
+	shutdownCommand     []string
+	checkCommand        []string
+	timeoutSeconds      *int
+	checkTimeoutSeconds *int
 }
 
 func (c nodeRemoveCommand) Run(ctx context.Context, inf cke.Infrastructure, _ string) error {
@@ -77,6 +93,70 @@ func (c nodeRemoveCommand) Run(ctx context.Context, inf cke.Infrastructure, _ st
 				return fmt.Errorf("failed to patch node %s: %v", n.Name, err)
 			}
 		}
+		err := func() error {
+			ctx := ctx
+			timeout := cke.DefaultRetireCommandTimeoutSeconds
+			if c.timeoutSeconds != nil {
+				timeout = *c.timeoutSeconds
+			}
+			if timeout != 0 {
+				var cancel context.CancelFunc
+				ctx, cancel = context.WithTimeout(ctx, time.Second*time.Duration(timeout))
+				defer cancel()
+			}
+			args := append(c.shutdownCommand[1:], n.Name)
+			command := well.CommandContext(ctx, c.shutdownCommand[0], args...)
+			return command.Run()
+		}()
+		if err != nil {
+			return fmt.Errorf("failed to shutdown node %s: %v", n.Name, err)
+		}
+
+		err = func() error {
+			ctx := ctx
+			checkTimeout := cke.DefaultRetireCheckTimeoutSeconds
+			if c.checkTimeoutSeconds != nil {
+				checkTimeout = *c.checkTimeoutSeconds
+			}
+			timeout := time.After(time.Duration(checkTimeout) * time.Second)
+			tick := time.Tick(10 * time.Second)
+			for {
+				select {
+				case <-timeout:
+					return fmt.Errorf("timeout")
+				case <-tick:
+					args := append(c.shutdownCommand[1:], n.Name)
+					command := well.CommandContext(ctx, c.shutdownCommand[0], args...)
+					stdout := bytes.Buffer{}
+					command.Stdout = &stdout
+					err := command.Run()
+					if err != nil {
+						log.Warn("failed to check shutdown status of node", map[string]interface{}{
+							log.FnError: err,
+							"node":      n.Name,
+						})
+						continue
+					}
+					if stdout.String() == "off" {
+						return nil
+					}
+				}
+			}
+		}()
+		if err != nil {
+			return fmt.Errorf("failed to check shutdown status of node %s: %v", n.Name, err)
+		}
+		shutdownTaint := corev1.Taint{
+			Key:    "node.kubernetes.io/out-of-service",
+			Value:  "nodeshutdown",
+			Effect: corev1.TaintEffectNoExecute,
+		}
+		n.Spec.Taints = append(n.Spec.Taints, shutdownTaint)
+		_, err = nodesAPI.Update(ctx, n, metav1.UpdateOptions{})
+		if err != nil {
+			return fmt.Errorf("failed to update node %s: %v", n.Name, err)
+		}
+
 		err = nodesAPI.Delete(ctx, n.Name, metav1.DeleteOptions{})
 		if err != nil {
 			return fmt.Errorf("failed to delete node %s: %v", n.Name, err)

server/strategy.go

@@ -353,7 +353,7 @@ OUTER_ETCD:
 	}
 
 	if nodes := nf.NonClusterNodes(); len(nodes) > 0 {
-		ops = append(ops, op.KubeNodeRemoveOp(apiServer, nodes))
+		ops = append(ops, op.KubeNodeRemoveOp(apiServer, nodes, &c.Retire))
 	}
 
 	return ops