add clusterapi IsNodeCandidateForDeletion impl

The initial implementation of this function for clusterapi will return
that a node is not a good candidate for deletion when it belongs to a
MachineDeployment that is currently rolling out an upgrade.

Author: elmiko

cluster-autoscaler/cloudprovider/clusterapi/clusterapi_controller.go

@@ -849,6 +849,27 @@ func (c *machineController) listMachinesForScalableResource(r *unstructured.Unst
 	}
 }
 
+func (c *machineController) listMachineSetsForMachineDeployment(r *unstructured.Unstructured) ([]*unstructured.Unstructured, error) {
+	selector := labels.SelectorFromSet(map[string]string{
+		"cluster.x-k8s.io/deployment-name": r.GetName(),
+	})
+	objs, err := c.machineSetInformer.Lister().ByNamespace(r.GetNamespace()).List(selector)
+	if err != nil {
+		return nil, err
+	}
+
+	results := make([]*unstructured.Unstructured, 0, len(objs))
+	for _, x := range objs {
+		u, ok := x.(*unstructured.Unstructured)
+		if !ok {
+			return nil, fmt.Errorf("expected unstructured resource from lister, not %T", x)
+		}
+		results = append(results, u.DeepCopy())
+	}
+
+	return results, nil
+}
+
 func (c *machineController) listScalableResources() ([]*unstructured.Unstructured, error) {
 	scalableResources, err := c.listResources(c.machineSetInformer.Lister())
 	if err != nil {

cluster-autoscaler/cloudprovider/clusterapi/clusterapi_nodegroup.go

@@ -458,6 +458,68 @@ func (ng *nodegroup) GetOptions(defaults config.NodeGroupAutoscalingOptions) (*c
 	return &defaults, nil
 }
 
+func (ng *nodegroup) IsMachineDeploymentAndRollingOut() (bool, error) {
+	if ng.scalableResource.Kind() != machineDeploymentKind {
+		// Not a MachineDeployment.
+		return false, nil
+	}
+
+	machineSets, err := ng.machineController.listMachineSetsForMachineDeployment(ng.scalableResource.unstructured)
+	if err != nil {
+		return false, err
+	}
+
+	if len(machineSets) == 0 {
+		// No MachineSets => MD is not rolling out.
+		return false, nil
+	}
+
+	// Find the latest revision, the MachineSet with the latest revision is the MachineSet that
+	// matches the MachineDeployment spec.
+	var latestMSRevisionInt int64
+	for _, ms := range machineSets {
+		msRevision, ok := ms.GetAnnotations()[machineDeploymentRevisionAnnotation]
+		if !ok {
+			continue
+		}
+
+		msRevisionInt, err := strconv.ParseInt(msRevision, 10, 64)
+		if err != nil {
+			return false, errors.Wrapf(err, "failed to parse current revision on MachineSet %s", klog.KObj(ms))
+		}
+		latestMSRevisionInt = max(latestMSRevisionInt, msRevisionInt)
+	}
+	maxMSRevision := strconv.FormatInt(latestMSRevisionInt, 10)
+
+	for _, ms := range machineSets {
+		if ms.GetAnnotations()[machineDeploymentRevisionAnnotation] == maxMSRevision {
+			// Ignore the MachineSet with the latest revision
+			continue
+		}
+
+		// Check if any of the old MachineSets still have replicas
+		replicas, found, err := unstructured.NestedInt64(ms.UnstructuredContent(), "spec", "replicas")
+		if err != nil {
+			return false, errors.Wrapf(err, "failed to find spec replicas on MachineSet %s", klog.KObj(ms))
+		}
+		if found && replicas > 0 {
+			// Found old MachineSets that still has replicas => MD is still rolling out.
+			return true, nil
+		}
+		replicas, found, err = unstructured.NestedInt64(ms.UnstructuredContent(), "status", "replicas")
+		if err != nil {
+			return false, errors.Wrapf(err, "failed to find status replicas on MachineSet %s", klog.KObj(ms))
+		}
+		if found && replicas > 0 {
+			// Found old MachineSets that still has replicas => MD is still rolling out.
+			return true, nil
+		}
+	}
+
+	// Didn't find any old MachineSets that still have replicas => MD is not rolling out.
+	return false, nil
+}
+
 func newNodeGroupFromScalableResource(controller *machineController, unstructuredScalableResource *unstructured.Unstructured) (*nodegroup, error) {
 	// Ensure that the resulting node group would be allowed based on the autodiscovery specs if defined
 	if !controller.allowedByAutoDiscoverySpecs(unstructuredScalableResource) {

cluster-autoscaler/cloudprovider/clusterapi/clusterapi_provider.go

@@ -21,6 +21,7 @@ import (
 	"path"
 	"reflect"
 
+	"github.com/pkg/errors"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/client-go/discovery"
@@ -34,7 +35,7 @@ import (
 
 	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
 	"k8s.io/autoscaler/cluster-autoscaler/config"
-	"k8s.io/autoscaler/cluster-autoscaler/utils/errors"
+	caserrors "k8s.io/autoscaler/cluster-autoscaler/utils/errors"
 	"k8s.io/autoscaler/cluster-autoscaler/utils/gpu"
 )
 
@@ -92,7 +93,7 @@ func (p *provider) HasInstance(node *corev1.Node) (bool, error) {
 	return false, fmt.Errorf("machine not found for node %s: %v", node.Name, err)
 }
 
-func (*provider) Pricing() (cloudprovider.PricingModel, errors.AutoscalerError) {
+func (*provider) Pricing() (cloudprovider.PricingModel, caserrors.AutoscalerError) {
 	return nil, cloudprovider.ErrNotImplemented
 }
 
@@ -142,7 +143,20 @@ func (p *provider) GetNodeGpuConfig(node *corev1.Node) *cloudprovider.GpuConfig
 
 // IsNodeCandidateForDeletion returns whether the node is a good candidate for deletion.
 func (p *provider) IsNodeCandidateForDeletion(node *corev1.Node) (bool, error) {
-	return true, cloudprovider.ErrNotImplemented
+	ng, err := p.controller.nodeGroupForNode(node)
+	if err != nil {
+		return false, errors.Wrapf(err, "failed to determine node group for node %s", klog.KObj(node))
+	}
+	if ng == nil {
+		klog.V(5).Infof("node %s is not part of a node group", klog.KObj(node))
+		return false, nil
+	}
+	rollingout, err := ng.IsMachineDeploymentAndRollingOut()
+	if err != nil {
+		return false, errors.Wrapf(err, "failed to determine rolling out status for MachineDeployment %s", ng.scalableResource.ID())
+	}
+	// A node is a good candidate for deletion if it is not currently part of a MachineDeployment that is rolling out.
+	return !rollingout, nil
 }
 
 func newProvider(

cluster-autoscaler/cloudprovider/clusterapi/clusterapi_utils.go

@@ -32,15 +32,16 @@ import (
 )
 
 const (
-	cpuKey          = "capacity.cluster-autoscaler.kubernetes.io/cpu"
-	memoryKey       = "capacity.cluster-autoscaler.kubernetes.io/memory"
-	diskCapacityKey = "capacity.cluster-autoscaler.kubernetes.io/ephemeral-disk"
-	gpuTypeKey      = "capacity.cluster-autoscaler.kubernetes.io/gpu-type"
-	gpuCountKey     = "capacity.cluster-autoscaler.kubernetes.io/gpu-count"
-	maxPodsKey      = "capacity.cluster-autoscaler.kubernetes.io/maxPods"
-	taintsKey       = "capacity.cluster-autoscaler.kubernetes.io/taints"
-	labelsKey       = "capacity.cluster-autoscaler.kubernetes.io/labels"
-	draDriverKey    = "capacity.cluster-autoscaler.kubernetes.io/dra-driver"
+	cpuKey                              = "capacity.cluster-autoscaler.kubernetes.io/cpu"
+	memoryKey                           = "capacity.cluster-autoscaler.kubernetes.io/memory"
+	diskCapacityKey                     = "capacity.cluster-autoscaler.kubernetes.io/ephemeral-disk"
+	gpuTypeKey                          = "capacity.cluster-autoscaler.kubernetes.io/gpu-type"
+	gpuCountKey                         = "capacity.cluster-autoscaler.kubernetes.io/gpu-count"
+	maxPodsKey                          = "capacity.cluster-autoscaler.kubernetes.io/maxPods"
+	taintsKey                           = "capacity.cluster-autoscaler.kubernetes.io/taints"
+	labelsKey                           = "capacity.cluster-autoscaler.kubernetes.io/labels"
+	draDriverKey                        = "capacity.cluster-autoscaler.kubernetes.io/dra-driver"
+	machineDeploymentRevisionAnnotation = "machinedeployment.clusters.x-k8s.io/revision"
 	// UnknownArch is used if the Architecture is Unknown
 	UnknownArch SystemArchitecture = ""
 	// Amd64 is used if the Architecture is x86_64