[CASCL-1386] Implement standalone EC2 node 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/standalone.go

@@ -2,15 +2,49 @@ package evict
 
 import (
 	"context"
+	"errors"
+	"fmt"
+	"log"
 
 	"github.com/aws/aws-sdk-go-v2/service/ec2"
 	"k8s.io/client-go/kubernetes"
+
+	commonaws "github.com/DataDog/datadog-operator/cmd/kubectl-datadog/autoscaling/cluster/common/aws"
 )
 
 type EC2API interface {
 	TerminateInstances(ctx context.Context, in *ec2.TerminateInstancesInput, opts ...func(*ec2.Options)) (*ec2.TerminateInstancesOutput, error)
 }
 
 func evictStandalone(ctx context.Context, clientset kubernetes.Interface, ec2API EC2API, nodes []string, drainOpts nodeDrainOptions) error {
-	panic("TODO: evictStandalone — implemented in PR https://github.com/DataDog/datadog-operator/pull/3176")
+	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; cannot terminate the underlying instance", 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 right away so its
+		// capacity is freed without waiting for the rest of the group.
+		if !hasInstanceID {
+			continue
+		}
+		if drainOpts.DryRun {
+			log.Printf("[dry-run] would terminate standalone instance %s", id)
+			continue
+		}
+		if _, err := ec2API.TerminateInstances(ctx, &ec2.TerminateInstancesInput{
+			InstanceIds: []string{id},
+		}); err != nil {
+			errs = append(errs, fmt.Errorf("terminate instance %s: %w", id, err))
+		} else {
+			log.Printf("Terminated standalone EC2 instance %s.", id)
+		}
+	}
+	return errors.Join(errs...)
 }

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

@@ -0,0 +1,198 @@
+package evict
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/service/ec2"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	policyv1 "k8s.io/api/policy/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/kubernetes/fake"
+	clienttesting "k8s.io/client-go/testing"
+)
+
+type stubEC2 struct {
+	terminated []string
+	err        error
+}
+
+func (s *stubEC2) TerminateInstances(_ context.Context, in *ec2.TerminateInstancesInput, _ ...func(*ec2.Options)) (*ec2.TerminateInstancesOutput, error) {
+	s.terminated = append(s.terminated, in.InstanceIds...)
+	return &ec2.TerminateInstancesOutput{}, s.err
+}
+
+func TestEvictStandalone(t *testing.T) {
+	ec2Node := func(name, az, id string) *corev1.Node {
+		return &corev1.Node{
+			ObjectMeta: metav1.ObjectMeta{Name: name},
+			Spec:       corev1.NodeSpec{ProviderID: "aws:///" + az + "/" + id},
+		}
+	}
+	stuckPod := func() *corev1.Pod {
+		return &corev1.Pod{
+			ObjectMeta: metav1.ObjectMeta{Name: "blocker", Namespace: "default"},
+			Spec:       corev1.PodSpec{NodeName: "ip-stuck"},
+		}
+	}
+	gceNode := &corev1.Node{
+		ObjectMeta: metav1.ObjectMeta{Name: "ip-1"},
+		Spec:       corev1.NodeSpec{ProviderID: "gce:///zone/instance"},
+	}
+	fastDrain := nodeDrainOptions{
+		EvictionTimeout: 50 * time.Millisecond,
+		NodeTimeout:     100 * time.Millisecond,
+		PollInterval:    10 * time.Millisecond,
+	}
+
+	for _, tc := range []struct {
+		name                   string
+		objects                []runtime.Object
+		installEvictionReactor bool
+		nodes                  []string
+		opts                   nodeDrainOptions
+		// stubErr, when set, makes every TerminateInstances call fail.
+		stubErr error
+		wantErr bool
+		// wantTerminated is the expected set of instance IDs in
+		// stubEC2.terminated. nil ⇒ TerminateInstances must not be called.
+		wantTerminated []string
+		// wantUnschedulable: per-node expected `Spec.Unschedulable`.
+		wantUnschedulable map[string]bool
+	}{
+		{
+			name:              "happy path terminates and cordons two nodes",
+			objects:           []runtime.Object{ec2Node("ip-1", "eu-west-3a", "i-aaa"), ec2Node("ip-2", "eu-west-3b", "i-bbb")},
+			nodes:             []string{"ip-1", "ip-2"},
+			opts:              newDrainOpts(false),
+			wantTerminated:    []string{"i-aaa", "i-bbb"},
+			wantUnschedulable: map[string]bool{"ip-1": true, "ip-2": true},
+		},
+		{
+			name:              "dry-run touches nothing",
+			objects:           []runtime.Object{ec2Node("ip-1", "eu-west-3a", "i-aaa")},
+			nodes:             []string{"ip-1"},
+			opts:              newDrainOpts(true),
+			wantUnschedulable: map[string]bool{"ip-1": false},
+		},
+		{
+			// Node already gone from K8s ⇒ no instance ID extracted ⇒
+			// no TerminateInstances call.
+			name:  "node already gone skips terminate",
+			nodes: []string{"ip-1"},
+			opts:  newDrainOpts(false),
+		},
+		{
+			// Non-EC2 providerID can't yield an instance ID, but the
+			// node is still cordoned and drained.
+			name:              "non-EC2 providerID skips terminate but still cordons",
+			objects:           []runtime.Object{gceNode},
+			nodes:             []string{"ip-1"},
+			opts:              newDrainOpts(false),
+			wantUnschedulable: map[string]bool{"ip-1": true},
+		},
+		{
+			// Safety regression: drain failure must prevent the
+			// instance from being terminated (otherwise the EC2 dies
+			// mid-grace-period).
+			name:                   "drain failure prevents terminate",
+			objects:                []runtime.Object{ec2Node("ip-stuck", "eu-west-3a", "i-aaaaaaaaaaaaaaaaa"), stuckPod()},
+			installEvictionReactor: true,
+			nodes:                  []string{"ip-stuck"},
+			opts:                   fastDrain,
+			wantErr:                true,
+		},
+		{
+			// A TerminateInstances failure on one node must propagate as an
+			// error yet not stop the loop: every drained node's instance is
+			// still attempted.
+			name:           "terminate failure propagates but loop continues",
+			objects:        []runtime.Object{ec2Node("ip-1", "eu-west-3a", "i-aaa"), ec2Node("ip-2", "eu-west-3b", "i-bbb")},
+			nodes:          []string{"ip-1", "ip-2"},
+			opts:           newDrainOpts(false),
+			stubErr:        errors.New("terminate boom"),
+			wantErr:        true,
+			wantTerminated: []string{"i-aaa", "i-bbb"},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			client := fake.NewClientset(tc.objects...)
+			if tc.installEvictionReactor {
+				installPodEvictionReactor(client)
+			}
+			stub := &stubEC2{err: tc.stubErr}
+
+			err := evictStandalone(t.Context(), client, stub, tc.nodes, tc.opts)
+			if tc.wantErr {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+			}
+			if tc.wantTerminated == nil {
+				assert.Empty(t, stub.terminated)
+			} else {
+				assert.ElementsMatch(t, tc.wantTerminated, stub.terminated)
+			}
+			for nodeName, want := range tc.wantUnschedulable {
+				got, getErr := client.CoreV1().Nodes().Get(t.Context(), nodeName, metav1.GetOptions{})
+				require.NoError(t, getErr)
+				assert.Equal(t, want, got.Spec.Unschedulable, "Spec.Unschedulable for %s", nodeName)
+			}
+		})
+	}
+}
+
+// TestEvictStandaloneCordonsAllBeforeDraining locks in the cordon-all-up-front
+// behavior: every node of the group is cordoned before ANY node starts
+// draining, so a pod evicted off one node is never rescheduled onto a sibling
+// node that is itself about to be drained. It asserts that at the moment ip-1's
+// pod is evicted, the sibling ip-2 has already been cordoned.
+func TestEvictStandaloneCordonsAllBeforeDraining(t *testing.T) {
+	ec2Node := func(name, az, id string) *corev1.Node {
+		return &corev1.Node{
+			ObjectMeta: metav1.ObjectMeta{Name: name},
+			Spec:       corev1.NodeSpec{ProviderID: "aws:///" + az + "/" + id},
+		}
+	}
+	pod1 := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Name: "p1", Namespace: "default"},
+		Spec:       corev1.PodSpec{NodeName: "ip-1"},
+	}
+	client := fake.NewClientset(ec2Node("ip-1", "eu-west-3a", "i-aaa"), ec2Node("ip-2", "eu-west-3b", "i-bbb"), pod1)
+
+	var (
+		recorded                     bool
+		node2CordonedAtFirstEviction bool
+	)
+	// On the first pod eviction (ip-1's drain), record whether ip-2 is already
+	// cordoned, then delete the evicted pod so the node becomes empty and the
+	// drain completes. Read/mutate via the tracker, not the typed client:
+	// Fake.Invokes holds a global lock for the whole reaction, so re-entering
+	// the typed client from inside a reactor would deadlock.
+	client.PrependReactor("create", "pods", func(action clienttesting.Action) (bool, runtime.Object, error) {
+		ca, ok := action.(clienttesting.CreateAction)
+		if !ok || ca.GetSubresource() != "eviction" {
+			return false, nil, nil
+		}
+		if !recorded {
+			recorded = true
+			if obj, err := client.Tracker().Get(corev1.SchemeGroupVersion.WithResource("nodes"), "", "ip-2"); err == nil {
+				node2CordonedAtFirstEviction = obj.(*corev1.Node).Spec.Unschedulable
+			}
+		}
+		evic := ca.GetObject().(*policyv1.Eviction)
+		_ = client.Tracker().Delete(corev1.SchemeGroupVersion.WithResource("pods"), evic.Namespace, evic.Name)
+		return true, ca.GetObject(), nil
+	})
+
+	stub := &stubEC2{}
+	err := evictStandalone(t.Context(), client, stub, []string{"ip-1", "ip-2"}, newDrainOpts(false))
+	require.NoError(t, err)
+	assert.True(t, node2CordonedAtFirstEviction, "ip-2 must be cordoned before ip-1 starts draining")
+	assert.ElementsMatch(t, []string{"i-aaa", "i-bbb"}, stub.terminated)
+}