Chore: QoL and Test fixes

Makefile

@@ -50,6 +50,22 @@ build-dracpu: ## build dracpu
 clean: ## clean
 	rm -rf "$(OUT_DIR)/"
 
+with-kind: ## run a command with a temporary kind cluster (create if missing, delete on exit if created)
+	@if [ -z "$$CMD" ]; then \
+		echo "CMD is required. Example: CMD='echo hello' $(MAKE) with-kind"; \
+		exit 1; \
+	fi; \
+	created=0; \
+	while read -r name; do \
+		if [ "$$name" = "$(CLUSTER_NAME)" ]; then created=2; fi; \
+	done < <(kind get clusters 2>/dev/null || true); \
+	if [ "$$created" -eq 0 ]; then \
+		kind create cluster --name ${CLUSTER_NAME} --config hack/kind.yaml; \
+		created=1; \
+	fi; \
+	trap 'if [ "$$created" -eq 1 ]; then kind delete cluster --name ${CLUSTER_NAME}; fi' EXIT; \
+	bash -c "$$CMD"
+
 test-unit: ## run tests
 	CGO_ENABLED=1 go test -v -race -count 1 -coverprofile=coverage.out ./pkg/...
 
@@ -96,7 +112,16 @@ export DOCKER_CLI_EXPERIMENTAL=enabled
 image: ## docker build load
 	docker build . -t ${STAGING_IMAGE_NAME} --load
 
-build-image: ## build image
+build-image: ## build image (skip if already exists unless FORCE_BUILD=1)
+	@if [ "$(FORCE_BUILD)" = "1" ]; then \
+		$(MAKE) build-image-force; \
+	elif docker image inspect "${IMAGE}" >/dev/null 2>&1; then \
+		echo "Image ${IMAGE} already exists; skipping build."; \
+	else \
+		$(MAKE) build-image-force; \
+	fi
+
+build-image-force: ## force build image
 	docker buildx build . \
 		--platform="${PLATFORMS}" \
 		--tag="${IMAGE}" \
@@ -116,6 +141,9 @@ kind-cluster:  ## create kind cluster
 kind-load-image: build-image  ## load the current container image into kind
 	kind load docker-image ${IMAGE} ${IMAGE_LATEST} --name ${CLUSTER_NAME}
 
+kind-load-test-image: build-test-image ## load the test image into kind
+	kind load docker-image ${IMAGE_TEST} --name ${CLUSTER_NAME}
+
 kind-uninstall-cpu-dra: ## remove cpu dra from kind cluster
 	kubectl delete -f install.yaml || true
 
@@ -162,7 +190,16 @@ ifneq ($(DRACPU_E2E_VERBOSE),)
 endif
 	$(call kind_setup,$(CI_MANIFEST_FILE))
 
-build-test-image: ## build tests image
+build-test-image: ## build tests image (skip if already exists unless FORCE_BUILD=1)
+	@if [ "$(FORCE_BUILD)" = "1" ]; then \
+		$(MAKE) build-test-image-force; \
+	elif docker image inspect "${IMAGE_TEST}" >/dev/null 2>&1; then \
+		echo "Image ${IMAGE_TEST} already exists; skipping build."; \
+	else \
+		$(MAKE) build-test-image-force; \
+	fi
+
+build-test-image-force: ## force build tests image
 	docker buildx build . \
 		--file test/image/Dockerfile \
 		--platform="${PLATFORMS}" \
@@ -175,11 +212,36 @@ build-test-dracputester: ## build helper to serve as entry point and report cpu
 build-test-dracpuinfo: ## build helper to expose hardware info in the internal dracpu format
 	go build -v -o "$(OUT_DIR)/dracpuinfo" ./test/image/dracpuinfo
 
-test-e2e: ## run e2e test against an existing configured cluster
-	env DRACPU_E2E_TEST_IMAGE=$(IMAGE_TEST) DRACPU_E2E_RESERVED_CPUS=$(DRACPU_E2E_RESERVED_CPUS) go test -v ./test/e2e/ --ginkgo.v
+# E2E: shared setup and run targets (assume cluster exists; use with-kind for "create if missing, delete if we created").
+define e2e_daemonset_patch
+[{"op":"replace","path":"/spec/template/spec/containers/0/args","value":["/dracpu","--v=4","--cpu-device-mode=$(1)","--reserved-cpus=$(DRACPU_E2E_RESERVED_CPUS)"]}]
+endef
+
+kind-e2e-setup: ## label workers, load test image, install DRA, wait for rollout (cluster must exist)
+	kubectl label nodes -l '!node-role.kubernetes.io/control-plane' node-role.kubernetes.io/worker='' --overwrite
+	$(MAKE) kind-load-test-image kind-install-cpu-dra
+	kubectl -n kube-system rollout status daemonset/dracpu --timeout=120s
+
+test-e2e-grouped-run: ## patch daemonset to grouped and run e2e (requires kind-e2e-setup)
+	kubectl -n kube-system patch daemonset dracpu --type=json -p='$(call e2e_daemonset_patch,grouped)'
+	env DRACPU_E2E_TEST_IMAGE=$(IMAGE_TEST) DRACPU_E2E_RESERVED_CPUS=$(DRACPU_E2E_RESERVED_CPUS) DRACPU_E2E_CPU_DEVICE_MODE=grouped go test -v ./test/e2e/ --ginkgo.v
+
+test-e2e-individual-run: ## patch daemonset to individual and run e2e (requires kind-e2e-setup)
+	kubectl -n kube-system patch daemonset dracpu --type=json -p='$(call e2e_daemonset_patch,individual)'
+	kubectl -n kube-system rollout status daemonset/dracpu --timeout=120s
+	env DRACPU_E2E_TEST_IMAGE=$(IMAGE_TEST) DRACPU_E2E_RESERVED_CPUS=$(DRACPU_E2E_RESERVED_CPUS) DRACPU_E2E_CPU_DEVICE_MODE=individual go test -v ./test/e2e/ --ginkgo.v
+
+test-e2e: ## run e2e in both grouped and individual mode (one cluster: create if missing, run both, delete if we created)
+	CMD='$(MAKE) kind-e2e-setup; $(MAKE) test-e2e-grouped-run; r1=$$?; $(MAKE) test-e2e-individual-run; r2=$$?; exit $$((r1|r2))' $(MAKE) with-kind
 
 test-e2e-kind: ci-kind-setup test-e2e ## run e2e test against a purpose-built kind cluster
 
+test-e2e-grouped-mode: ## run e2e tests in grouped mode (with-kind)
+	CMD='$(MAKE) kind-e2e-setup test-e2e-grouped-run' $(MAKE) with-kind
+
+test-e2e-individual-mode: ## run e2e tests in individual mode (with-kind)
+	CMD='$(MAKE) kind-e2e-setup test-e2e-individual-run' $(MAKE) with-kind
+
 lint:  ## run the linter against the codebase
 	$(GOLANGCI_LINT) run ./...
 

pkg/cpuinfo/cpuinfo.go

@@ -576,3 +576,26 @@ func (t *CPUTopology) CPUsPerUncore() int {
 	// Note: this is an approximation that assumes all uncore caches have the same number of CPUs.
 	return t.NumCPUs / t.NumUncoreCache
 }
+
+// AffinityMaxCPUID is the upper bound when scanning a scheduler affinity mask (e.g. from
+// sched_getaffinity). The number of CPUs visible to the process (e.g. runtime.NumCPU()
+// or cgroup size) may be less than the highest CPU ID in the mask, so the scan must
+// use a fixed limit to avoid missing CPUs (e.g. 9-13 when cpuset is 2-5,9-13 and
+// NumCPU() is 8).
+const AffinityMaxCPUID = 2048
+
+// AffinityMask is satisfied by kernel affinity masks (e.g. unix.CPUSet) and by test doubles.
+type AffinityMask interface {
+	IsSet(i int) bool
+}
+
+// FromAffinityMask scans the mask from 0 to AffinityMaxCPUID and returns the set of set CPU IDs.
+func FromAffinityMask(mask AffinityMask) cpuset.CPUSet {
+	var allowedCPUs []int
+	for i := 0; i < AffinityMaxCPUID; i++ {
+		if mask.IsSet(i) {
+			allowedCPUs = append(allowedCPUs, i)
+		}
+	}
+	return cpuset.New(allowedCPUs...)
+}

pkg/cpuinfo/cpuinfo_test.go

@@ -565,3 +565,59 @@ func TestSMTDetection(t *testing.T) {
 		})
 	}
 }
+
+// affinityMaskFromSet implements AffinityMask for a cpuset (e.g. in tests).
+type affinityMaskFromSet struct{ cpus cpuset.CPUSet }
+
+func (m affinityMaskFromSet) IsSet(i int) bool { return m.cpus.Contains(i) }
+
+func TestFromAffinityMask(t *testing.T) {
+	tests := []struct {
+		name string
+		mask AffinityMask
+		want cpuset.CPUSet
+	}{
+		{
+			name: "empty",
+			mask: affinityMaskFromSet{cpuset.New()},
+			want: cpuset.New(),
+		},
+		{
+			name: "single CPU",
+			mask: affinityMaskFromSet{cpuset.New(3)},
+			want: cpuset.New(3),
+		},
+		{
+			name: "contiguous low CPUs",
+			mask: affinityMaskFromSet{cpuset.New(0, 1, 2, 3)},
+			want: cpuset.New(0, 1, 2, 3),
+		},
+		{
+			name: "split set 2-5 and 9-13 (bug case: high IDs beyond NumCPU())",
+			mask: affinityMaskFromSet{mustParseCPUSet(t, "2-5,9-13")},
+			want: mustParseCPUSet(t, "2-5,9-13"),
+		},
+		{
+			name: "high CPU IDs only",
+			mask: affinityMaskFromSet{cpuset.New(100, 101, 200)},
+			want: cpuset.New(100, 101, 200),
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := FromAffinityMask(tt.mask)
+			if !got.Equals(tt.want) {
+				t.Errorf("FromAffinityMask() = %s, want %s", got.String(), tt.want.String())
+			}
+		})
+	}
+}
+
+func mustParseCPUSet(t *testing.T, s string) cpuset.CPUSet {
+	t.Helper()
+	cpus, err := cpuset.Parse(s)
+	if err != nil {
+		t.Fatalf("cpuset.Parse(%q): %v", s, err)
+	}
+	return cpus
+}

test/e2e/cpu_assignment_test.go

@@ -184,7 +184,8 @@ var _ = ginkgo.Describe("CPU Allocation", ginkgo.Serial, ginkgo.Ordered, ginkgo.
 
 				fxt.Log.Info("Creating pods requesting exclusive CPUs", "numPods", numPods, "cpusPerClaim", cpusPerClaim)
 				var exclPods []*v1.Pod
-				allAllocatedCPUs := cpuset.New()
+				// CPU IDs are per-node; track allocations per node so we don't treat same IDs on different nodes as overlapping.
+				allAllocatedCPUsByNode := make(map[string]cpuset.CPUSet)
 
 				claimTemplate := resourcev1.ResourceClaimTemplate{
 					ObjectMeta: metav1.ObjectMeta{
@@ -205,18 +206,34 @@ var _ = ginkgo.Describe("CPU Allocation", ginkgo.Serial, ginkgo.Ordered, ginkgo.
 
 				fixture.By("Verifying CPU allocations for each exclusive pod")
 				for i, pod := range exclPods {
-					alloc := getTesterPodCPUAllocation(fxt.K8SClientset, ctx, pod)
+					var alloc CPUAllocation
+					nodeName := pod.Spec.NodeName
+					gomega.Eventually(func() error {
+						alloc = getTesterPodCPUAllocation(fxt.K8SClientset, ctx, pod)
+						if alloc.CPUAssigned.Size() != cpusPerClaim {
+							return fmt.Errorf("pod %d: got %d CPUs, want %d", i, alloc.CPUAssigned.Size(), cpusPerClaim)
+						}
+						if !alloc.CPUAssigned.IsSubsetOf(availableCPUs) {
+							return fmt.Errorf("pod %d: CPUs %s outside available set %s", i, alloc.CPUAssigned.String(), availableCPUs.String())
+						}
+						allocatedOnNode := allAllocatedCPUsByNode[nodeName]
+						if allocatedOnNode.Intersection(alloc.CPUAssigned).Size() != 0 {
+							return fmt.Errorf("pod %d: overlapping CPUs with previous pods on node %s", i, nodeName)
+						}
+						return nil
+					}).WithTimeout(2*time.Minute).WithPolling(5*time.Second).Should(gomega.Succeed(), "exclusive pod %d allocation did not stabilize", i)
 					fxt.Log.Info("Checking exclusive CPU allocation", "pod", e2epod.Identify(pod), "cpuAllocated", alloc.CPUAssigned.String())
-					gomega.Expect(alloc.CPUAssigned.Size()).To(gomega.Equal(cpusPerClaim), "Pod %d did not get %d CPUs", i, cpusPerClaim)
-					gomega.Expect(alloc.CPUAssigned.IsSubsetOf(availableCPUs)).To(gomega.BeTrue(), "Pod %d got CPUs outside available set", i)
-					gomega.Expect(allAllocatedCPUs.Intersection(alloc.CPUAssigned).Size()).To(gomega.Equal(0), "Pod %d has overlapping CPUs", i)
-					allAllocatedCPUs = allAllocatedCPUs.Union(alloc.CPUAssigned)
+					allAllocatedCPUsByNode[nodeName] = allAllocatedCPUsByNode[nodeName].Union(alloc.CPUAssigned)
 				}
-				gomega.Expect(allAllocatedCPUs.Size()).To(gomega.Equal(numPods * cpusPerClaim))
-				rootFxt.Log.Info("All exclusive allocation", "pod", "exclusive CPUs", allAllocatedCPUs.String(), "expected Shared CPUs", availableCPUs.Difference(allAllocatedCPUs).String())
+				var totalAllocated int
+				for _, cpus := range allAllocatedCPUsByNode {
+					totalAllocated += cpus.Size()
+				}
+				gomega.Expect(totalAllocated).To(gomega.Equal(numPods * cpusPerClaim))
+				rootFxt.Log.Info("All exclusive allocation", "byNode", allAllocatedCPUsByNode, "expected Shared CPUs on target", availableCPUs.Difference(allAllocatedCPUsByNode[targetNode.Name]).String())
 
 				fixture.By("checking the shared pool does not include anymore the exclusively allocated CPUs")
-				expectedSharedCPUs := availableCPUs.Difference(allAllocatedCPUs)
+				expectedSharedCPUs := availableCPUs.Difference(allAllocatedCPUsByNode[targetNode.Name])
 
 				fixture.By("creating a second best-effort reference pod")
 				shrPod2 := mustCreateBestEffortPod(ctx, fxt, targetNode.Name, dracpuTesterImage)
@@ -248,5 +265,5 @@ func verifySharedPoolMatches(ctx context.Context, fxt *fixture.Fixture, sharedPo
 			return fmt.Errorf("shared CPUs mismatch: expected %v got %v", expectedSharedCPUs.String(), sharedAllocUpdated.CPUAssigned.String())
 		}
 		return nil
-	}).WithTimeout(1*time.Minute).WithPolling(5*time.Second).Should(gomega.Succeed(), "the best-effort tester pod %s does not have access to the exclusively allocated CPUs", e2epod.Identify(sharedPod))
+	}).WithTimeout(2*time.Minute).WithPolling(5*time.Second).Should(gomega.Succeed(), "the best-effort tester pod %s CPU pool did not match expected %s", e2epod.Identify(sharedPod), expectedSharedCPUs.String())
 }

test/e2e/e2e_suite_test.go

@@ -66,7 +66,7 @@ func BeFailedToCreate(fxt *fixture.Fixture) types.GomegaMatcher {
 			return false, nil
 		}
 		if cntSt.State.Waiting.Reason != reasonCreateContainerError {
-			lh.Info("container terminated for different reason", "containerName", cntSt.Name, "reason", cntSt.State.Terminated.Reason)
+			lh.Info("container waiting for different reason", "containerName", cntSt.Name, "reason", cntSt.State.Waiting.Reason)
 			return false, nil
 		}
 		lh.Info("container creation error", "containerName", cntSt.Name)

test/e2e/sharing_test.go

@@ -147,7 +147,7 @@ var _ = ginkgo.Describe("Claim sharing", ginkgo.Serial, ginkgo.Ordered, ginkgo.C
 			gomega.Expect(fxt.Teardown(ctx)).To(gomega.Succeed())
 		})
 
-		ginkgo.It("should fail to run pods which share a claim", func(ctx context.Context) {
+		ginkgo.It("should fail to run pods which share a claim", ginkgo.SpecTimeout(5*time.Minute), func(ctx context.Context) {
 			testPod := v1.Pod{
 				ObjectMeta: metav1.ObjectMeta{
 					Namespace:    fxt.Namespace.Name,
@@ -196,10 +196,10 @@ var _ = ginkgo.Describe("Claim sharing", ginkgo.Serial, ginkgo.Ordered, ginkgo.C
 					return nil
 				}
 				return pod
-			}).WithTimeout(time.Minute).WithPolling(2 * time.Second).Should(BeFailedToCreate(fxt))
+			}).WithTimeout(2 * time.Minute).WithPolling(2 * time.Second).Should(BeFailedToCreate(fxt))
 		})
 
-		ginkgo.It("should fail to run a pod with multiple containers which share a claim", ginkgo.Label("negative"), func(ctx context.Context) {
+		ginkgo.It("should fail to run a pod with multiple containers which share a claim", ginkgo.Label("negative"), ginkgo.SpecTimeout(5*time.Minute), func(ctx context.Context) {
 			testPod := v1.Pod{
 				ObjectMeta: metav1.ObjectMeta{
 					Namespace: fxt.Namespace.Name,
@@ -259,7 +259,7 @@ var _ = ginkgo.Describe("Claim sharing", ginkgo.Serial, ginkgo.Ordered, ginkgo.C
 					return nil
 				}
 				return pod
-			}).WithTimeout(time.Minute).WithPolling(2 * time.Second).Should(BeFailedToCreate(fxt))
+			}).WithTimeout(2 * time.Minute).WithPolling(2 * time.Second).Should(BeFailedToCreate(fxt))
 		})
 	})
 })

test/image/dracputester/app.go

@@ -23,10 +23,10 @@ import (
 	"os"
 	"os/signal"
 	"path/filepath"
-	"runtime"
 	"strings"
 	"time"
 
+	"github.com/kubernetes-sigs/dra-driver-cpu/pkg/cpuinfo"
 	"github.com/kubernetes-sigs/dra-driver-cpu/test/pkg/discovery"
 	"golang.org/x/sys/unix"
 	"k8s.io/utils/cpuset"
@@ -55,14 +55,7 @@ func getAffinity() (cpuset.CPUSet, error) {
 	if err != nil {
 		return cpuset.New(), err
 	}
-
-	var allowedCPUs []int
-	for i := 0; i < runtime.NumCPU(); i++ {
-		if unixCS.IsSet(i) {
-			allowedCPUs = append(allowedCPUs, i)
-		}
-	}
-	return cpuset.New(allowedCPUs...), nil
+	return cpuinfo.FromAffinityMask(&unixCS), nil
 }
 
 func main() {

test/pkg/pod/pod.go

@@ -75,11 +75,50 @@ func RunToCompletion(ctx context.Context, cs kubernetes.Interface, pod *v1.Pod)
 func WaitToBeRunning(ctx context.Context, cs kubernetes.Interface, podNamespace, podName string) error {
 	phase, err := WaitForPhase(ctx, cs, podNamespace, podName, v1.PodRunning)
 	if err != nil {
-		return fmt.Errorf("pod=%s/%s is not Running; phase=%q; %w", podNamespace, podName, phase, err)
+		hint := formatPodEventsHint(ctx, cs, podNamespace, podName)
+		return fmt.Errorf("pod=%s/%s is not Running; phase=%q; %w%s", podNamespace, podName, phase, err, hint)
 	}
 	return nil
 }
 
+// formatPodEventsHint returns a short hint (pod events, newest first) for debugging Pending pods. Safe to call on failure.
+func formatPodEventsHint(ctx context.Context, cs kubernetes.Interface, podNamespace, podName string) string {
+	eventList, err := cs.CoreV1().Events(podNamespace).List(ctx, metav1.ListOptions{
+		FieldSelector: "involvedObject.name=" + podName,
+		Limit:         15,
+	})
+	if err != nil || len(eventList.Items) == 0 {
+		return ""
+	}
+	var b strings.Builder
+	b.WriteString("; pod events (newest first): ")
+	const maxHintLen = 600
+	for i := len(eventList.Items) - 1; i >= 0 && b.Len() < maxHintLen; i-- {
+		e := eventList.Items[i]
+		if b.Len() > 30 {
+			b.WriteString("; ")
+		}
+		b.WriteString("[")
+		b.WriteString(string(e.Type))
+		b.WriteString("] ")
+		b.WriteString(e.Reason)
+		if e.Message != "" {
+			msg := e.Message
+			if len(msg) > 120 {
+				msg = msg[:117] + "..."
+			}
+			b.WriteString(": ")
+			b.WriteString(msg)
+		}
+		if !e.LastTimestamp.IsZero() {
+			b.WriteString(" (")
+			b.WriteString(e.LastTimestamp.Format("15:04:05"))
+			b.WriteString(")")
+		}
+	}
+	return b.String()
+}
+
 func WaitToBeDeleted(ctx context.Context, cs kubernetes.Interface, podNamespace, podName string) error {
 	immediate := true
 	err := wait.PollUntilContextTimeout(ctx, PollInterval, PollTimeout, immediate, func(ctx2 context.Context) (done bool, err error) {