[CASCL-1386] Implement Karpenter user NodePool 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/karpenter_user.go

@@ -2,10 +2,25 @@ package evict
 
 import (
 	"context"
+	"errors"
+	"fmt"
+	"log"
 
 	"k8s.io/client-go/kubernetes"
 )
 
 func evictKarpenterUserNodePool(ctx context.Context, clientset kubernetes.Interface, nodePoolName string, nodes []string, drainOpts nodeDrainOptions) error {
-	panic("TODO: evictKarpenterUserNodePool — implemented in PR https://github.com/DataDog/datadog-operator/pull/3177")
+	// Cordon every node up front so a pod evicted from one node is never
+	// rescheduled onto another node of the same NodePool that is itself about
+	// to be drained.
+	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 !drainOpts.DryRun && len(errs) == 0 {
+		log.Printf("Drained %d node(s) from user NodePool %s; Karpenter will terminate their NodeClaims once empty.", len(cordoned), nodePoolName)
+	}
+	return errors.Join(errs...)
 }

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

@@ -0,0 +1,33 @@
+package evict
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/fake"
+)
+
+func TestEvictKarpenterUserNodePool(t *testing.T) {
+	for _, tc := range []struct {
+		name              string
+		dryRun            bool
+		wantUnschedulable bool
+	}{
+		{name: "cordons and drains", dryRun: false, wantUnschedulable: true},
+		{name: "dry-run touches nothing", dryRun: true, wantUnschedulable: false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "n1"}}
+			client := fake.NewClientset(node)
+
+			require.NoError(t, evictKarpenterUserNodePool(t.Context(), client, "user-np", []string{"n1"}, newDrainOpts(tc.dryRun)))
+
+			got, err := client.CoreV1().Nodes().Get(t.Context(), "n1", metav1.GetOptions{})
+			require.NoError(t, err)
+			assert.Equal(t, tc.wantUnschedulable, got.Spec.Unschedulable)
+		})
+	}
+}