Implement Drain controller

controllers/drain_controller.go

@@ -0,0 +1,118 @@
+package controllers
+
+import (
+	"context"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/util/workqueue"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/builder"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+	"sigs.k8s.io/controller-runtime/pkg/source"
+
+	sriovnetworkv1 "github.com/k8snetworkplumbingwg/sriov-network-operator/api/v1"
+	constants "github.com/k8snetworkplumbingwg/sriov-network-operator/pkg/consts"
+	"github.com/k8snetworkplumbingwg/sriov-network-operator/pkg/utils"
+)
+
+type DrainReconciler struct {
+	client.Client
+	Scheme *runtime.Scheme
+}
+
+//+kubebuilder:rbac:groups="",resources=nodes,verbs=get;list;watch;update;patch
+//+kubebuilder:rbac:groups=sriovnetwork.openshift.io,resources=sriovoperatorconfigs,verbs=get;list;watch
+
+// Reconcile is part of the main kubernetes reconciliation loop which aims to
+// move the current state of the cluster closer to the desired state.
+// For more details, check Reconcile and its Result here:
+// - https://pkg.go.dev/sigs.k8s.io/[email protected]/pkg/reconcile
+func (dr *DrainReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	// The SriovNetwork CR shall only be defined in operator namespace.
+	req.Namespace = namespace
+	reqLogger := log.FromContext(ctx).WithValues("drain", req.NamespacedName)
+	reqLogger.Info("Reconciling Drain")
+
+	config := &sriovnetworkv1.SriovOperatorConfig{}
+	err := dr.Get(ctx, types.NamespacedName{
+		Name: constants.DefaultConfigName, Namespace: namespace}, config)
+	if err != nil {
+		reqLogger.Error(err, "Error occurred on GET SriovOperatorConfig request from API server.")
+		return reconcile.Result{}, err
+	}
+
+	nodeList := &corev1.NodeList{}
+	err = dr.List(ctx, nodeList)
+	if err != nil {
+		// Failed to get node list
+		reqLogger.Error(err, "Error occurred on LIST nodes request from API server")
+		return reconcile.Result{}, err
+	}
+
+	reqLogger.Info("Max node allowed to be draining at the same time", "MaxParallelNodeConfiguration", config.Spec.MaxParallelNodeConfiguration)
+
+	drainingNodes := 0
+	for _, node := range nodeList.Items {
+		if utils.NodeHasAnnotation(node, "sriovnetwork.openshift.io/state", "Draining") {
+			drainingNodes++
+		}
+	}
+
+	reqLogger.Info("Count of draining", "drainingNodes", drainingNodes)
+	if drainingNodes == config.Spec.MaxParallelNodeConfiguration {
+		return reconcile.Result{}, nil
+	}
+
+	for _, node := range nodeList.Items {
+		if utils.NodeHasAnnotation(node, "sriovnetwork.openshift.io/state", "Drain_Required") {
+			if drainingNodes < config.Spec.MaxParallelNodeConfiguration {
+				reqLogger.Info("Start draining node", "node", node.Name)
+				node.Annotations["sriovnetwork.openshift.io/state"] = "Draining"
+				if err := dr.Update(ctx, &node); err != nil {
+					return reconcile.Result{}, err
+				}
+				drainingNodes++
+			} else {
+				return reconcile.Result{}, nil
+			}
+		}
+	}
+
+	return reconcile.Result{}, 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
+	// reconciliation count
+	qHandler := func(q workqueue.RateLimitingInterface) {
+		q.Add(reconcile.Request{NamespacedName: types.NamespacedName{
+			Namespace: "drain-reconcile-namespace",
+			Name:      "drain-upgrade-reconcile-name",
+		}})
+	}
+
+	createUpdateEnqueue := handler.Funcs{
+		CreateFunc: func(e event.CreateEvent, q workqueue.RateLimitingInterface) {
+			qHandler(q)
+		},
+		UpdateFunc: func(e event.UpdateEvent, q workqueue.RateLimitingInterface) {
+			qHandler(q)
+		},
+	}
+
+	// Watch for spec and annotation changes
+	nodePredicates := builder.WithPredicates(predicate.AnnotationChangedPredicate{})
+
+	return ctrl.NewControllerManagedBy(mgr).
+		For(&sriovnetworkv1.SriovOperatorConfig{}).
+		Watches(&source.Kind{Type: &corev1.Node{}}, createUpdateEnqueue, nodePredicates).
+		Complete(dr)
+}

main.go

@@ -164,6 +164,13 @@ func main() {
 		setupLog.Error(err, "unable to create controller", "controller", "SriovNetworkPoolConfig")
 		os.Exit(1)
 	}
+	if err = (&controllers.DrainReconciler{
+		Client: mgr.GetClient(),
+		Scheme: mgr.GetScheme(),
+	}).SetupWithManager(mgr); err != nil {
+		setupLog.Error(err, "unable to create controller", "controller", "DrainReconciler")
+		os.Exit(1)
+	}
 	// +kubebuilder:scaffold:builder
 
 	// Create a default SriovNetworkNodePolicy

pkg/utils/cluster.go

@@ -111,3 +111,11 @@ func openshiftControlPlaneTopologyStatus(c client.Client) (configv1.TopologyMode
 	}
 	return infra.Status.ControlPlaneTopology, nil
 }
+
+func NodeHasAnnotation(node corev1.Node, annoKey string, value string) bool {
+	// Check if node already contains annotation
+	if anno, ok := node.Annotations[annoKey]; ok && (anno == value) {
+		return true
+	}
+	return false
+}