[CASCL-1386] Implement EKS managed node group 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/eks_mng.go

@@ -3,18 +3,86 @@ package evict
 import (
 	"context"
 	"errors"
+	"fmt"
+	"log"
+	"time"
 
+	awssdk "github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/eks"
+	ekstypes "github.com/aws/aws-sdk-go-v2/service/eks/types"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/pager"
+
+	commonaws "github.com/DataDog/datadog-operator/cmd/kubectl-datadog/autoscaling/cluster/common/aws"
 )
 
 var errEKSDrainIncomplete = errors.New("EKS managed node group drain did not complete within the timeout")
 
 type EKSManagedNodeGroupAPI interface {
-	DescribeNodegroup(ctx context.Context, in *eks.DescribeNodegroupInput, opts ...func(*eks.Options)) (*eks.DescribeNodegroupOutput, error)
 	UpdateNodegroupConfig(ctx context.Context, in *eks.UpdateNodegroupConfigInput, opts ...func(*eks.Options)) (*eks.UpdateNodegroupConfigOutput, error)
 }
 
-func evictEKSManagedNodeGroup(ctx context.Context, eksAPI EKSManagedNodeGroupAPI, clientset kubernetes.Interface, clusterName, nodegroupName string, drainOpts nodeDrainOptions) error {
-	panic("TODO: evictEKSManagedNodeGroup — implemented in PR https://github.com/DataDog/datadog-operator/pull/3174")
+func evictEKSManagedNodeGroup(ctx context.Context, eksAPI EKSManagedNodeGroupAPI, clientset kubernetes.Interface, clusterName, nodegroupName string, nodes []string, drainOpts nodeDrainOptions) error {
+	cordoned, errs := cordonNodes(ctx, clientset, nodes, drainOpts.DryRun)
+	for _, node := range cordoned {
+		if err := drainNode(ctx, clientset, node.Name, drainOpts); err != nil {
+			errs = append(errs, fmt.Errorf("drain node %s: %w", node.Name, err))
+		}
+	}
+	if len(errs) > 0 {
+		// A node still holds workloads: do NOT scale to zero, which would
+		// terminate instances with pods still on them and bypass their PDBs.
+		// Wrapping errEKSDrainIncomplete keeps the temporary PDBs in place; the
+		// group is left untouched for a re-run.
+		return fmt.Errorf("EKS node group %s/%s drain incomplete: %w: %w", clusterName, nodegroupName, errors.Join(errs...), errEKSDrainIncomplete)
+	}
+
+	if drainOpts.DryRun {
+		log.Printf("[dry-run] would scale EKS node group %s/%s to min=desired=0 (max preserved)", clusterName, nodegroupName)
+		return nil
+	}
+
+	if _, err := eksAPI.UpdateNodegroupConfig(ctx, &eks.UpdateNodegroupConfigInput{
+		ClusterName:   awssdk.String(clusterName),
+		NodegroupName: awssdk.String(nodegroupName),
+		ScalingConfig: &ekstypes.NodegroupScalingConfig{
+			MinSize:     awssdk.Int32(0),
+			DesiredSize: awssdk.Int32(0),
+		},
+	}); err != nil {
+		return fmt.Errorf("UpdateNodegroupConfig: %w", err)
+	}
+	log.Printf("EKS node group %s/%s drained; scaling config set to min=desired=0 (max preserved).", clusterName, nodegroupName)
+
+	return waitEKSNodegroupEmpty(ctx, clientset, clusterName, nodegroupName, drainOpts.NodeTimeout, drainOpts.PollInterval)
+}
+
+// waitEKSNodegroupEmpty polls the Kubernetes API for nodes carrying the
+// `eks.amazonaws.com/nodegroup=<name>` label until none remain or the deadline
+// expires. The EKS drain is bounded by drainOpts.NodeTimeout — for large node
+// groups, callers should raise this above the default.
+func waitEKSNodegroupEmpty(ctx context.Context, clientset kubernetes.Interface, clusterName, nodegroupName string, timeout, pollInterval time.Duration) error {
+	selector := commonaws.LabelEKSNodegroup + "=" + nodegroupName
+	p := pager.New(func(ctx context.Context, opts metav1.ListOptions) (runtime.Object, error) {
+		return clientset.CoreV1().Nodes().List(ctx, opts)
+	})
+	var remaining int
+	err := wait.PollUntilContextTimeout(ctx, pollInterval, timeout, true, func(ctx context.Context) (bool, error) {
+		remaining = 0
+		if err := p.EachListItem(ctx, metav1.ListOptions{LabelSelector: selector}, func(runtime.Object) error {
+			remaining++
+			return nil
+		}); err != nil {
+			return false, fmt.Errorf("list EKS managed node group %s/%s nodes: %w", clusterName, nodegroupName, err)
+		}
+		return remaining == 0, nil
+	})
+	if err != nil {
+		return fmt.Errorf("EKS node group %s/%s drain incomplete, %d node(s) still present: %w: %w", clusterName, nodegroupName, remaining, err, errEKSDrainIncomplete)
+	}
+	log.Printf("EKS node group %s/%s fully drained.", clusterName, nodegroupName)
+	return nil
 }

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

@@ -0,0 +1,202 @@
+package evict
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	awssdk "github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/eks"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	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"
+
+	commonaws "github.com/DataDog/datadog-operator/cmd/kubectl-datadog/autoscaling/cluster/common/aws"
+)
+
+type stubEKS struct {
+	gotInputs []*eks.UpdateNodegroupConfigInput
+	err       error
+}
+
+func (s *stubEKS) UpdateNodegroupConfig(_ context.Context, in *eks.UpdateNodegroupConfigInput, _ ...func(*eks.Options)) (*eks.UpdateNodegroupConfigOutput, error) {
+	s.gotInputs = append(s.gotInputs, in)
+	return &eks.UpdateNodegroupConfigOutput{}, s.err
+}
+
+func mngDrainOpts() nodeDrainOptions {
+	return nodeDrainOptions{
+		EvictionTimeout: time.Second,
+		NodeTimeout:     5 * time.Second,
+		PollInterval:    10 * time.Millisecond,
+	}
+}
+
+func TestEvictEKSManagedNodeGroup(t *testing.T) {
+	const cluster, mng, nodeName = "my-cluster", "my-mng", "ip-1"
+	newMngNode := func() *corev1.Node {
+		return &corev1.Node{ObjectMeta: metav1.ObjectMeta{
+			Name:   nodeName,
+			Labels: map[string]string{commonaws.LabelEKSNodegroup: mng},
+		}}
+	}
+	podOnNode := corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Name: "app", Namespace: "default"},
+		Spec:       corev1.PodSpec{NodeName: nodeName},
+	}
+
+	// installEvictionReactor makes the fake accept Eviction creates (echoing
+	// the object) or fail them with evictErr.
+	installEvictionReactor := func(client *fake.Clientset, evictErr error) {
+		client.PrependReactor("create", "pods", func(a clienttesting.Action) (bool, runtime.Object, error) {
+			ca, ok := a.(clienttesting.CreateAction)
+			if !ok || ca.GetSubresource() != "eviction" {
+				return false, nil, nil
+			}
+			if evictErr != nil {
+				return true, nil, evictErr
+			}
+			return true, ca.GetObject(), nil
+		})
+	}
+	// installPodListReactor returns the pod list for each 1-based List call, so a
+	// case can drain a node across successive polls.
+	installPodListReactor := func(client *fake.Clientset, lists func(call int) []corev1.Pod) {
+		var n int
+		client.PrependReactor("list", "pods", func(_ clienttesting.Action) (bool, runtime.Object, error) {
+			n++
+			return true, &corev1.PodList{Items: lists(n)}, nil
+		})
+	}
+	// installNodeListReactor drives waitEKSNodegroupEmpty's Nodes().List (by
+	// nodegroup label). It does not affect cordonNodes, which uses Get + Update.
+	installNodeListReactor := func(client *fake.Clientset, items []corev1.Node) {
+		client.PrependReactor("list", "nodes", func(_ clienttesting.Action) (bool, runtime.Object, error) {
+			return true, &corev1.NodeList{Items: items}, nil
+		})
+	}
+	assertScaledToZero := func(t *testing.T, in *eks.UpdateNodegroupConfigInput) {
+		t.Helper()
+		assert.Equal(t, cluster, awssdk.ToString(in.ClusterName))
+		assert.Equal(t, mng, awssdk.ToString(in.NodegroupName))
+		require.NotNil(t, in.ScalingConfig)
+		assert.Equal(t, int32(0), awssdk.ToInt32(in.ScalingConfig.MinSize))
+		assert.Equal(t, int32(0), awssdk.ToInt32(in.ScalingConfig.DesiredSize))
+		// MaxSize is deliberately omitted so EKS preserves the existing ceiling
+		// (partial ScalingConfig update).
+		assert.Nil(t, in.ScalingConfig.MaxSize)
+	}
+
+	t.Run("dry-run cordons nothing and skips Update", func(t *testing.T) {
+		stub := &stubEKS{}
+		client := fake.NewClientset(newMngNode())
+		opts := mngDrainOpts()
+		opts.DryRun = true
+
+		err := evictEKSManagedNodeGroup(t.Context(), stub, client, cluster, mng, []string{nodeName}, opts)
+		require.NoError(t, err)
+		assert.Empty(t, stub.gotInputs)
+		got, getErr := client.CoreV1().Nodes().Get(t.Context(), nodeName, metav1.GetOptions{})
+		require.NoError(t, getErr)
+		assert.False(t, got.Spec.Unschedulable, "dry-run must not cordon")
+	})
+
+	t.Run("cordons, drains gracefully, then scales to zero", func(t *testing.T) {
+		stub := &stubEKS{}
+		client := fake.NewClientset(newMngNode())
+		installEvictionReactor(client, nil)
+		// The node still hosts the pod when drainNode enumerates it (call 1);
+		// the eviction takes effect by the time waitForNodeEmpty polls (call 2+).
+		installPodListReactor(client, func(call int) []corev1.Pod {
+			if call == 1 {
+				return []corev1.Pod{podOnNode}
+			}
+			return nil
+		})
+		// EKS has terminated the instance by the time we poll for nodes.
+		installNodeListReactor(client, nil)
+
+		err := evictEKSManagedNodeGroup(t.Context(), stub, client, cluster, mng, []string{nodeName}, mngDrainOpts())
+		require.NoError(t, err)
+		require.Len(t, stub.gotInputs, 1)
+		assertScaledToZero(t, stub.gotInputs[0])
+		got, getErr := client.CoreV1().Nodes().Get(t.Context(), nodeName, metav1.GetOptions{})
+		require.NoError(t, getErr)
+		assert.True(t, got.Spec.Unschedulable, "the node must be cordoned before draining")
+	})
+
+	t.Run("drain failure keeps PDBs and skips scale-down", func(t *testing.T) {
+		stub := &stubEKS{}
+		// The seeded pod is never removed, so the node never empties.
+		client := fake.NewClientset(newMngNode(), &podOnNode)
+		installEvictionReactor(client, apierrors.NewTooManyRequests("PDB blocked", 1))
+		opts := mngDrainOpts()
+		opts.EvictionTimeout = 30 * time.Millisecond
+		opts.NodeTimeout = 60 * time.Millisecond
+
+		err := evictEKSManagedNodeGroup(t.Context(), stub, client, cluster, mng, []string{nodeName}, opts)
+		require.Error(t, err)
+		assert.ErrorIs(t, err, errEKSDrainIncomplete)
+		assert.Empty(t, stub.gotInputs, "must not scale down while a node still holds workloads")
+	})
+
+	t.Run("cordon failure keeps PDBs and skips scale-down", func(t *testing.T) {
+		stub := &stubEKS{}
+		client := fake.NewClientset(newMngNode())
+		client.PrependReactor("update", "nodes", func(_ clienttesting.Action) (bool, runtime.Object, error) {
+			return true, nil, apierrors.NewInternalError(errors.New("cordon boom"))
+		})
+
+		err := evictEKSManagedNodeGroup(t.Context(), stub, client, cluster, mng, []string{nodeName}, mngDrainOpts())
+		require.Error(t, err)
+		assert.ErrorIs(t, err, errEKSDrainIncomplete)
+		assert.Empty(t, stub.gotInputs)
+	})
+
+	t.Run("UpdateNodegroupConfig failure after drain does not wrap the sentinel", func(t *testing.T) {
+		stub := &stubEKS{err: errors.New("api error")}
+		client := fake.NewClientset(newMngNode()) // no pods ⇒ the node drains immediately
+
+		err := evictEKSManagedNodeGroup(t.Context(), stub, client, cluster, mng, []string{nodeName}, mngDrainOpts())
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "UpdateNodegroupConfig")
+		assert.NotErrorIs(t, err, errEKSDrainIncomplete)
+		require.Len(t, stub.gotInputs, 1)
+	})
+
+	t.Run("post-scale wait timeout wraps the sentinel", func(t *testing.T) {
+		stub := &stubEKS{}
+		client := fake.NewClientset(newMngNode()) // no pods ⇒ the node drains immediately
+		// The instance never terminates, so waitEKSNodegroupEmpty times out.
+		installNodeListReactor(client, []corev1.Node{*newMngNode()})
+		opts := mngDrainOpts()
+		opts.NodeTimeout = 60 * time.Millisecond
+
+		err := evictEKSManagedNodeGroup(t.Context(), stub, client, cluster, mng, []string{nodeName}, opts)
+		require.Error(t, err)
+		assert.ErrorIs(t, err, errEKSDrainIncomplete)
+		assert.Contains(t, err.Error(), mng)
+		require.Len(t, stub.gotInputs, 1)
+	})
+
+	t.Run("post-scale node-list error keeps the PDBs (wraps the sentinel)", func(t *testing.T) {
+		stub := &stubEKS{}
+		client := fake.NewClientset(newMngNode()) // no pods ⇒ the node drains immediately
+		// The scale-down succeeded, but the Nodes().List used to confirm the
+		// group emptied errors out, so the drain cannot be confirmed complete.
+		client.PrependReactor("list", "nodes", func(_ clienttesting.Action) (bool, runtime.Object, error) {
+			return true, nil, apierrors.NewInternalError(errors.New("apiserver unreachable"))
+		})
+
+		err := evictEKSManagedNodeGroup(t.Context(), stub, client, cluster, mng, []string{nodeName}, mngDrainOpts())
+		require.Error(t, err)
+		assert.ErrorIs(t, err, errEKSDrainIncomplete)
+		require.Len(t, stub.gotInputs, 1)
+	})
+}

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

@@ -28,17 +28,6 @@ type nodeDrainOptions struct {
 	PollInterval    time.Duration // interval between empty-checks; default 2s
 }
 
-// TODO(CASCL-1386): remove this block once the EKS managed node group eviction
-// (next PR in the stack) calls these primitives. They are introduced here as
-// shared building blocks; golangci-lint's `unused` runs with `tests = false`,
-// so the test coverage in this package does not count as use. Referencing the
-// two entry points keeps `unused` quiet until the first production caller lands
-// (drainNode/cordonNodes transitively reach every other helper below).
-var (
-	_ = drainNode
-	_ = cordonNodes
-)
-
 // drainNode evicts every evictable pod from the node and waits for the node to
 // become empty. Pods owned by a DaemonSet, mirror pods, terminating pods and
 // completed Job pods are skipped — the kubelet handles their cleanup when the

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

@@ -38,12 +38,12 @@ func printPlan(streams genericclioptions.IOStreams, info *clusterinfo.ClusterInf
 	}
 	if ensurePDBs {
 		fmt.Fprintln(streams.Out, "  • Remove the temporary PodDisruptionBudgets created above")
-		// Cleanup is deferred when an EKS managed node group drain times out
-		// (see run.go): EKS may still be evicting pods, so the temp PDBs are
-		// left in place and reclaimed on a later rerun. Only mention this when
-		// the plan actually targets an EKS managed node group.
+		// Cleanup is deferred when an EKS managed node group does not finish
+		// draining (see run.go): workloads may still be on its nodes, so the temp
+		// PDBs are left in place and reclaimed on a later rerun. Only mention this
+		// when the plan actually targets an EKS managed node group.
 		if _, hasEKS := grouped[clusterinfo.NodeManagerEKSManagedNodeGroup]; hasEKS {
-			fmt.Fprintln(streams.Out, "      ◦ (deferred to a later rerun if an EKS managed node group drain times out)")
+			fmt.Fprintln(streams.Out, "      ◦ (deferred to a later rerun if an EKS managed node group does not finish draining)")
 		}
 	}
 	fmt.Fprintln(streams.Out, color.YellowString("\n⚠ This will drain workloads and terminate the underlying instances of non-Datadog node groups."))

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

@@ -65,7 +65,7 @@ func TestPrintPlan(t *testing.T) {
 			ensurePDBs: true,
 			wantContains: []string{
 				"Remove the temporary PodDisruptionBudgets",
-				"deferred to a later rerun if an EKS managed node group drain times out",
+				"deferred to a later rerun if an EKS managed node group does not finish draining",
 			},
 		},
 		{

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

@@ -151,15 +151,15 @@ func Run(ctx context.Context, streams genericclioptions.IOStreams, configFlags *
 	// appear BEFORE the final success/failure summary, so the user does not
 	// experience an apparent hang after seeing the "✅" box.
 	//
-	// When at least one EKS managed node group did not finish draining
-	// within the timeout, we skip the cleanup: EKS may still be evicting
-	// pods on those nodes, and dropping the temp PDBs now would leave
-	// cross-type workloads unprotected mid-drain. The label-based cleanup
-	// on a subsequent run picks the PDBs up once EKS converges.
+	// When at least one EKS managed node group did not finish draining, we skip
+	// the cleanup: workloads may still be on its nodes (our drain did not
+	// complete, or EKS had not finished terminating the drained instances), and
+	// dropping the temp PDBs now would leave cross-type workloads unprotected. A
+	// subsequent run finishes the drain and reclaims the PDBs by label.
 	switch {
 	case !opts.EnsurePDBs:
 	case result.EKSDrainIncomplete:
-		log.Printf("Note: at least one EKS managed node group did not finish draining within --node-timeout; temporary PDBs left in place so EKS can finish safely. Re-run the command once `aws eks describe-nodegroup` reports the group at 0 nodes — the next run will clean them up.")
+		log.Printf("Note: at least one EKS managed node group did not finish draining; temporary PDBs were left in place. Re-run the command to retry the drain — it will finish evicting the remaining pods (or clean up the PDBs once the group is empty).")
 	default:
 		if err := cleanupTempPDBs(ctx, cli.K8sClient, opts.DryRun); err != nil {
 			log.Printf("Warning: failed to cleanup temporary PDBs: %v", err)
@@ -201,9 +201,11 @@ func classify(ctx context.Context, clientset *kubernetes.Clientset, cli *clients
 // evictResult is the structured outcome of evictAllTargets. Errors is the
 // aggregated per-target error list (empty when everything succeeded).
 // EKSDrainIncomplete is true when at least one EKS managed node group target
-// returned an error WRAPPING errEKSDrainIncomplete — the EKS drain may still
-// be in progress, in which case the caller must NOT cleanup the temporary
-// PDBs (EKS would then over-disrupt the still-running workloads).
+// returned an error WRAPPING errEKSDrainIncomplete — workloads may still be on
+// its nodes (the cordon/drain did not complete, or EKS had not finished
+// terminating the drained instances), in which case the caller must NOT clean
+// up the temporary PDBs (they would then over-disrupt the still-running
+// workloads).
 type evictResult struct {
 	Errors             []error
 	EKSDrainIncomplete bool
@@ -232,10 +234,11 @@ func evictAllTargets(ctx context.Context, targets []Target, drainOpts nodeDrainO
 			continue
 		}
 		errs = append(errs, fmt.Errorf("%s/%s: %w", t.Manager, t.Entity, err))
-		// Only treat the drain as "still in progress" when EKS accepted the
-		// scaling change but observation of the drain failed or timed out.
-		// A failed UpdateNodegroupConfig means EKS never started draining,
-		// so cleanup is safe.
+		// errEKSDrainIncomplete means workloads may still be on the node group's
+		// nodes — either our cordon/drain did not complete, or EKS had not
+		// finished terminating the drained instances before the wait timed out —
+		// so the temporary PDBs must stay. A failed UpdateNodegroupConfig after a
+		// successful drain does not wrap the sentinel: cleanup is safe.
 		if errors.Is(err, errEKSDrainIncomplete) {
 			eksDrainIncomplete = true
 		}
@@ -251,7 +254,7 @@ func evictTarget(ctx context.Context, clientset kubernetes.Interface, cli *clien
 	case clusterinfo.NodeManagerASG:
 		return evictASG(ctx, clientset, cli.Autoscaling, t.Entity, t.Nodes, drainOpts)
 	case clusterinfo.NodeManagerEKSManagedNodeGroup:
-		return evictEKSManagedNodeGroup(ctx, cli.EKS, clientset, clusterName, t.Entity, drainOpts)
+		return evictEKSManagedNodeGroup(ctx, cli.EKS, clientset, clusterName, t.Entity, t.Nodes, drainOpts)
 	case clusterinfo.NodeManagerKarpenter:
 		return evictKarpenterUserNodePool(ctx, clientset, t.Entity, t.Nodes, drainOpts)
 	case clusterinfo.NodeManagerStandalone: