[CASCL-1386] Implement node drain primitives and ASG eviction

Part of a stack splitting #3026 (too large to review in one piece) into
small pieces that each build and pass tests on their own. The command is
fully functional only once the whole stack lands.

Author: L3n41c

cmd/kubectl-datadog/autoscaling/cluster/evict/asg.go

@@ -2,9 +2,15 @@ package evict
 
 import (
 	"context"
+	"errors"
+	"fmt"
+	"log"
 
+	awssdk "github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/autoscaling"
 	"k8s.io/client-go/kubernetes"
+
+	commonaws "github.com/DataDog/datadog-operator/cmd/kubectl-datadog/autoscaling/cluster/common/aws"
 )
 
 type AutoscalingAPI interface {
@@ -14,5 +20,114 @@ type AutoscalingAPI interface {
 }
 
 func evictASG(ctx context.Context, clientset kubernetes.Interface, asg AutoscalingAPI, asgName string, nodes []string, drainOpts nodeDrainOptions) error {
-	panic("TODO: evictASG — implemented in PR https://github.com/DataDog/datadog-operator/pull/3175")
+	if drainOpts.DryRun {
+		log.Printf("[dry-run] would suspend AZRebalance and set MinSize=0 on ASG %s", asgName)
+	} else if err := prepareASGForTermination(ctx, asg, asgName); err != nil {
+		return fmt.Errorf("prepare ASG %s for termination: %w", asgName, err)
+	}
+
+	// Cordon every node up front so a pod evicted from one node is never
+	// rescheduled onto another node of the same ASG that is itself about to be
+	// drained. A node that fails to cordon is left undrained; treat that as a
+	// drain failure so the ASG keeps its original MaxSize for a re-run instead
+	// of being locked away with workloads still on it.
+	cordoned, errs := cordonNodes(ctx, clientset, nodes, drainOpts.DryRun)
+
+	for _, node := range cordoned {
+		nodeName := node.Name
+		id, hasInstanceID := commonaws.ExtractEC2InstanceID(node)
+		if !hasInstanceID {
+			log.Printf("Warning: node %s has unexpected providerID %q; its instance will be terminated by the final scale-to-zero instead", nodeName, node.Spec.ProviderID)
+		}
+		if err := drainNode(ctx, clientset, nodeName, drainOpts); err != nil {
+			errs = append(errs, fmt.Errorf("drain node %s: %w", nodeName, err))
+			continue // do NOT terminate this instance: workloads are still on it
+		}
+		// The node drained cleanly: terminate its instance now, decrementing
+		// the ASG's desired capacity so it is not relaunched. Nodes with an
+		// unexpected providerID are left for the final scale-to-zero.
+		if !hasInstanceID {
+			continue
+		}
+		if drainOpts.DryRun {
+			log.Printf("[dry-run] would terminate instance %s in ASG %s (decrementing desired capacity)", id, asgName)
+			continue
+		}
+		if err := terminateASGInstance(ctx, asg, id); err != nil {
+			errs = append(errs, fmt.Errorf("terminate instance %s in ASG %s: %w", id, asgName, err))
+		}
+	}
+
+	if len(errs) > 0 {
+		log.Printf("ASG %s: at least one node failed to cordon, drain, or terminate; leaving the ASG at MinSize=0 without locking MaxSize. Re-run after addressing the errors above.", asgName)
+		return errors.Join(errs...)
+	}
+
+	// Every node drained and its instance was terminated. Lock the ASG at
+	// min=max=desired=0 so nothing is ever relaunched, and to clean up any
+	// instance that couldn't be terminated per-node (unexpected providerID).
+	if drainOpts.DryRun {
+		log.Printf("[dry-run] would scale ASG %s to min=max=desired=0", asgName)
+		return nil
+	}
+	// All instances are now terminated; only the MaxSize=0 lock remains. A
+	// crash here would leave the ASG at desired=0 but unlocked and no longer
+	// rediscoverable by a re-run (see the crash-window note on evictASG).
+	log.Printf("ASG %s: all nodes drained; locking the ASG at min=max=desired=0.", asgName)
+	if err := scaleASGToZero(ctx, asg, asgName); err != nil {
+		errs = append(errs, fmt.Errorf("scale ASG %s to 0: %w", asgName, err))
+	}
+	return errors.Join(errs...)
+}
+
+// prepareASGForTermination makes the ASG safe for the per-instance termination
+// performed during the drain loop:
+//
+//   - AZRebalance is suspended so that decrementing desired capacity one
+//     instance at a time cannot trigger AZ rebalancing — which would terminate
+//     a not-yet-drained instance in another availability zone.
+//   - MinSize is set to 0 so that TerminateInstanceInAutoScalingGroup may
+//     decrement DesiredCapacity (AWS rejects the decrement while
+//     MinSize == DesiredCapacity).
+func prepareASGForTermination(ctx context.Context, asg AutoscalingAPI, asgName string) error {
+	if _, err := asg.SuspendProcesses(ctx, &autoscaling.SuspendProcessesInput{
+		AutoScalingGroupName: awssdk.String(asgName),
+		ScalingProcesses:     []string{"AZRebalance"},
+	}); err != nil {
+		return fmt.Errorf("suspend AZRebalance: %w", err)
+	}
+	if _, err := asg.UpdateAutoScalingGroup(ctx, &autoscaling.UpdateAutoScalingGroupInput{
+		AutoScalingGroupName: awssdk.String(asgName),
+		MinSize:              awssdk.Int32(0),
+	}); err != nil {
+		return fmt.Errorf("set MinSize=0: %w", err)
+	}
+	log.Printf("Prepared ASG %s for termination (AZRebalance suspended, MinSize=0).", asgName)
+	return nil
+}
+
+// terminateASGInstance terminates a single (already drained) instance and
+// decrements the ASG's desired capacity so it is not relaunched.
+func terminateASGInstance(ctx context.Context, asg AutoscalingAPI, instanceID string) error {
+	if _, err := asg.TerminateInstanceInAutoScalingGroup(ctx, &autoscaling.TerminateInstanceInAutoScalingGroupInput{
+		InstanceId:                     awssdk.String(instanceID),
+		ShouldDecrementDesiredCapacity: awssdk.Bool(true),
+	}); err != nil {
+		return err
+	}
+	log.Printf("Terminated instance %s and decremented ASG desired capacity.", instanceID)
+	return nil
+}
+
+func scaleASGToZero(ctx context.Context, asg AutoscalingAPI, asgName string) error {
+	if _, err := asg.UpdateAutoScalingGroup(ctx, &autoscaling.UpdateAutoScalingGroupInput{
+		AutoScalingGroupName: awssdk.String(asgName),
+		MinSize:              awssdk.Int32(0),
+		MaxSize:              awssdk.Int32(0),
+		DesiredCapacity:      awssdk.Int32(0),
+	}); err != nil {
+		return err
+	}
+	log.Printf("Scaled ASG %s to min=max=desired=0.", asgName)
+	return nil
 }