add support to filter out certain pods during node draining process

**What this PR does / why we need it**:
The PR adds support to filter out certain pods during node draining process

**Which issue(s) this PR fixes** *(optional, in `fixes #<issue number>(, fixes #<issue_number>, ...)` format, will close the issue(s) when PR gets merged)*:
Fixes kubernetes-sigs#11024

<!--
Please label this pull request according to what area(s) you are addressing. For reference on PR/issue labels, see: https://github.com/kubernetes-sigs/cluster-api/labels?q=area+

Area example:
/area runtime-sdk
-->

api/v1beta1/machine_types.go

@@ -132,6 +132,10 @@ type MachineSpec struct {
 	// Defaults to 10 seconds.
 	// +optional
 	NodeDeletionTimeout *metav1.Duration `json:"nodeDeletionTimeout,omitempty"`
+
+	// NodeDrainPodFilters allows to specify filters for pods to be excluded during node drain
+	// +optional
+	NodeDrainPodFilters *metav1.LabelSelector `json:"nodeDrainPodFilters,omitempty"`
 }
 
 // ANCHOR_END: MachineSpec

internal/controllers/machine/machine_controller.go

@@ -26,6 +26,7 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
 	kerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/apimachinery/pkg/util/wait"
@@ -386,7 +387,7 @@ func (r *Reconciler) reconcileDelete(ctx context.Context, cluster *clusterv1.Clu
 				return ctrl.Result{}, errors.Wrap(err, "failed to patch Machine")
 			}
 
-			if result, err := r.drainNode(ctx, cluster, m.Status.NodeRef.Name); !result.IsZero() || err != nil {
+			if result, err := r.drainNode(ctx, cluster, m); !result.IsZero() || err != nil {
 				if err != nil {
 					conditions.MarkFalse(m, clusterv1.DrainingSucceededCondition, clusterv1.DrainingFailedReason, clusterv1.ConditionSeverityWarning, err.Error())
 					r.recorder.Eventf(m, corev1.EventTypeWarning, "FailedDrainNode", "error draining Machine's node %q: %v", m.Status.NodeRef.Name, err)
@@ -615,7 +616,8 @@ func (r *Reconciler) isDeleteNodeAllowed(ctx context.Context, cluster *clusterv1
 	return nil
 }
 
-func (r *Reconciler) drainNode(ctx context.Context, cluster *clusterv1.Cluster, nodeName string) (ctrl.Result, error) {
+func (r *Reconciler) drainNode(ctx context.Context, cluster *clusterv1.Cluster, m *clusterv1.Machine) (ctrl.Result, error) {
+	nodeName := m.Status.NodeRef.Name
 	log := ctrl.LoggerFrom(ctx, "Node", klog.KRef("", nodeName))
 
 	restConfig, err := r.Tracker.GetRESTConfig(ctx, util.ObjectKey(cluster))
@@ -667,6 +669,9 @@ func (r *Reconciler) drainNode(ctx context.Context, cluster *clusterv1.Cluster,
 		ErrOut: writer{func(msg string, keysAndValues ...interface{}) {
 			log.Error(nil, msg, keysAndValues...)
 		}},
+		AdditionalFilters: []kubedrain.PodFilter{
+			SkipFuncGenerator(m.Spec.NodeDrainPodFilters),
+		},
 	}
 
 	if noderefutil.IsNodeUnreachable(node) {
@@ -700,6 +705,18 @@ func (r *Reconciler) drainNode(ctx context.Context, cluster *clusterv1.Cluster,
 	return ctrl.Result{}, nil
 }
 
+func SkipFuncGenerator(labelSelector *metav1.LabelSelector) func(pod corev1.Pod) kubedrain.PodDeleteStatus {
+	return func(pod corev1.Pod) kubedrain.PodDeleteStatus {
+		if pod.Labels == nil {
+			return kubedrain.MakePodDeleteStatusOkay()
+		}
+		if labels.Equals(labelSelector.MatchLabels, pod.ObjectMeta.Labels) {
+			return kubedrain.MakePodDeleteStatusSkip()
+		}
+		return kubedrain.MakePodDeleteStatusOkay()
+	}
+}
+
 // shouldWaitForNodeVolumes returns true if node status still have volumes attached and the node is reachable
 // pod deletion and volume detach happen asynchronously, so pod could be deleted before volume detached from the node
 // this could cause issue for some storage provisioner, for example, vsphere-volume this is problematic