Move node drain call to Drain controller

controllers/drain_controller.go

@@ -3,8 +3,12 @@ package controllers
 import (
 	"context"
 	"fmt"
+	"github.com/golang/glog"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/kubectl/pkg/drain"
 	"sort"
 	"strings"
+	"time"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -24,9 +28,15 @@ import (
 	"github.com/k8snetworkplumbingwg/sriov-network-operator/pkg/utils"
 )
 
+const (
+	maxParallelNodeConfiguration = 1
+)
+
 type DrainReconciler struct {
 	client.Client
 	Scheme *runtime.Scheme
+
+	Drainer *drain.Helper
 }
 
 //+kubebuilder:rbac:groups="",resources=nodes,verbs=get;list;watch;update;patch
@@ -72,7 +82,7 @@ func (dr *DrainReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
 	}
 
 	reqLogger.Info("Count of draining", "drainingNodes", drainingNodes)
-	if config.Spec.MaxParallelNodeConfiguration != 0 && drainingNodes >= config.Spec.MaxParallelNodeConfiguration {
+	if maxParallelNodeConfiguration != 0 && drainingNodes >= maxParallelNodeConfiguration {
 		reqLogger.Info("MaxParallelNodeConfiguration limit reached for draining nodes")
 		return reconcile.Result{}, nil
 	}
@@ -89,6 +99,7 @@ func (dr *DrainReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
 				reqLogger.Error(err, "Failed to patch node annotations")
 				return reconcile.Result{}, err
 			}
+
 			drainingNodes++
 		} else {
 			reqLogger.Info("Too many nodes to be draining at the moment. Skipping node %s", "node", node.Name)
@@ -99,6 +110,43 @@ func (dr *DrainReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
 	return reconcile.Result{}, nil
 }
 
+func (dr *DrainReconciler) drainNode(node *corev1.Node) error {
+	glog.Info("drainNode(): Update prepared")
+	var err error
+
+	backoff := wait.Backoff{
+		Steps:    5,
+		Duration: 10 * time.Second,
+		Factor:   2,
+	}
+	var lastErr error
+
+	glog.Info("drainNode(): Start draining")
+	if err = wait.ExponentialBackoff(backoff, func() (bool, error) {
+		err := drain.RunCordonOrUncordon(dr.Drainer, node, true)
+		if err != nil {
+			lastErr = err
+			glog.Infof("Cordon failed with: %v, retrying", err)
+			return false, nil
+		}
+		err = drain.RunNodeDrain(dr.Drainer, node.Name)
+		if err == nil {
+			return true, nil
+		}
+		lastErr = err
+		glog.Infof("Draining failed with: %v, retrying", err)
+		return false, nil
+	}); err != nil {
+		if err == wait.ErrWaitTimeout {
+			glog.Errorf("drainNode(): failed to drain node (%d tries): %v :%v", backoff.Steps, err, lastErr)
+		}
+		glog.Errorf("drainNode(): failed to drain node: %v", err)
+		return err
+	}
+	glog.Info("drainNode(): drain complete")
+	return nil
+}
+
 // SetupWithManager sets up the controller with the Manager.
 func (dr *DrainReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	// we always add object with a same(static) key to the queue to reduce

main.go

@@ -20,7 +20,11 @@ import (
 	"context"
 	"flag"
 	"fmt"
+	"github.com/golang/glog"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/kubectl/pkg/drain"
 	"os"
+	"time"
 
 	netattdefv1 "github.com/k8snetworkplumbingwg/network-attachment-definition-client/pkg/apis/k8s.cni.cncf.io/v1"
 	openshiftconfigv1 "github.com/openshift/api/config/v1"
@@ -95,6 +99,7 @@ func main() {
 		setupLog.Error(err, "couldn't create openshift context")
 		os.Exit(1)
 	}
+	kubeclient := kubernetes.NewForConfigOrDie(ctrl.GetConfigOrDie())
 
 	le := leaderelection.GetLeaderElectionConfig(kubeClient, enableLeaderElection)
 
@@ -124,7 +129,7 @@ func main() {
 		os.Exit(1)
 	}
 
-	if err := initNicIDMap(); err != nil {
+	if err := initNicIDMap(kubeclient); err != nil {
 		setupLog.Error(err, "unable to init NicIdMap")
 		os.Exit(1)
 	}
@@ -169,6 +174,24 @@ func main() {
 	if err = (&controllers.DrainReconciler{
 		Client: mgr.GetClient(),
 		Scheme: mgr.GetScheme(),
+		Drainer: &drain.Helper{
+			Client:              kubeclient,
+			Force:               true,
+			IgnoreAllDaemonSets: true,
+			DeleteEmptyDirData:  true,
+			GracePeriodSeconds:  -1,
+			Timeout:             90 * time.Second,
+			OnPodDeletedOrEvicted: func(pod *corev1.Pod, usingEviction bool) {
+				verbStr := "Deleted"
+				if usingEviction {
+					verbStr = "Evicted"
+				}
+				glog.Info(fmt.Sprintf("%s pod from Node %s/%s", verbStr, pod.Namespace, pod.Name))
+			},
+			//Out:    writer{glog.Info},
+			//ErrOut: writer{glog.Error},
+			Ctx: context.Background(),
+		},
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "DrainReconciler")
 		os.Exit(1)
@@ -216,9 +239,8 @@ func main() {
 	}
 }
 
-func initNicIDMap() error {
+func initNicIDMap(kubeclient kubernetes.Interface) error {
 	namespace := os.Getenv("NAMESPACE")
-	kubeclient := kubernetes.NewForConfigOrDie(ctrl.GetConfigOrDie())
 	if err := sriovnetworkv1.InitNicIDMapFromConfigMap(kubeclient, namespace); err != nil {
 		return err
 	}

pkg/consts/constants.go

@@ -28,6 +28,7 @@ const (
 	AnnoDrainRequired   = "Drain_Required"
 	AnnoMcpPaused       = "Draining_MCP_Paused"
 	AnnoDraining        = "Draining"
+	AnnoDrainComplete   = "DrainComplete"
 
 	LinkTypeEthernet   = "ether"
 	LinkTypeInfiniband = "infiniband"