From 11cab58c865a311c59018069a8e33bcaa211c8d2 Mon Sep 17 00:00:00 2001 From: chokevin Date: Thu, 5 Nov 2026 13:03:54 -0800 Subject: [PATCH 1/3] queue: validate CPU topology flavor compatibility --- cli/internal/cli/topology_flags.go | 58 ++++++++++++++++++++- cli/internal/cli/topology_flags_test.go | 64 +++++++++++++++++++++++ cli/internal/queueresolve/validate.go | 68 +++++++++++++++++++++++++ 3 files changed, 189 insertions(+), 1 deletion(-) diff --git a/cli/internal/cli/topology_flags.go b/cli/internal/cli/topology_flags.go index 7c3551269..d7caa348f 100644 --- a/cli/internal/cli/topology_flags.go +++ b/cli/internal/cli/topology_flags.go @@ -260,6 +260,7 @@ type renderedQueueContract struct { QueueName string GPUClass string GPUCount int + CPURequest bool GPUResourceName string NodeSelector map[string]string PodTolerations [][]kueueapi.Toleration @@ -306,6 +307,7 @@ func inspectRenderedQueue(ctx context.Context, r kubeRawRunner, namespace string NodeSelector: nodeSelector, PodTolerations: contract.PodTolerations, GPUCount: contract.GPUCount, + CPURequest: contract.CPURequest, GPUResourceName: gpuResourceName, TopologyRequest: contract.TopologyRequest, TopologyName: policy.TopologyName, @@ -366,6 +368,9 @@ func renderedQueueContractFromManifest(manifest []byte) (renderedQueueContract, if renderedGPUObjectsRequestTopology(obj) { out.TopologyRequest = true } + if renderedObjectsRequestCPU(obj) { + out.CPURequest = true + } resourceNames := map[string]struct{}{} collectRenderedGPUResourceNames(obj, resourceNames) for resourceName := range resourceNames { @@ -429,7 +434,7 @@ func renderedGPUObjectsRequestTopology(value any) bool { func renderedPodTemplateRequestsTopology(value map[string]any) bool { spec, ok := value["spec"].(map[string]any) - if !ok || collectRenderedDirectPodGPUCount(spec) == 0 { + if !ok || !renderedDirectPodRequestsCompute(spec) { return false } metadata, _ := value["metadata"].(map[string]any) @@ -446,6 +451,57 @@ func renderedPodTemplateRequestsTopology(value map[string]any) bool { return false } +func renderedDirectPodRequestsCompute(spec map[string]any) bool { + return collectRenderedDirectPodGPUCount(spec) > 0 || + renderedResourceRequested(spec["containers"], "cpu") || + renderedResourceRequested(spec["initContainers"], "cpu") +} + +func renderedObjectsRequestCPU(value any) bool { + switch typed := value.(type) { + case map[string]any: + if spec, ok := typed["spec"].(map[string]any); ok && + (renderedResourceRequested(spec["containers"], "cpu") || + renderedResourceRequested(spec["initContainers"], "cpu")) { + return true + } + for _, child := range typed { + if renderedObjectsRequestCPU(child) { + return true + } + } + case []any: + for _, child := range typed { + if renderedObjectsRequestCPU(child) { + return true + } + } + } + return false +} + +func renderedResourceRequested(value any, resourceName string) bool { + switch typed := value.(type) { + case map[string]any: + for key, child := range typed { + if key == resourceName && strings.TrimSpace(fmt.Sprint(child)) != "" && + strings.TrimSpace(fmt.Sprint(child)) != "0" { + return true + } + if renderedResourceRequested(child, resourceName) { + return true + } + } + case []any: + for _, child := range typed { + if renderedResourceRequested(child, resourceName) { + return true + } + } + } + return false +} + func collectRenderedDirectPodGPUCount(spec map[string]any) int { count := collectRenderedGPUCount(spec["containers"]) count = maxInt(count, collectRenderedGPUCount(spec["initContainers"])) diff --git a/cli/internal/cli/topology_flags_test.go b/cli/internal/cli/topology_flags_test.go index 9e87e428c..2e519c6fd 100644 --- a/cli/internal/cli/topology_flags_test.go +++ b/cli/internal/cli/topology_flags_test.go @@ -279,6 +279,70 @@ spec: } } +func TestValidateRenderedQueueRejectsCPUTopologyOnTopologyFreeFlavor(t *testing.T) { + runner := &fakeRawRunner{ + outputs: map[string]string{ + fakeRawKey("-n", "workspace", "get", "localqueue.kueue.x-k8s.io", "cpu-queue", "-o", "json"): `{"metadata":{"name":"cpu-queue"},"spec":{"clusterQueue":"cpu-cq"}}`, + fakeRawKey("get", "clusterqueue.kueue.x-k8s.io", "cpu-cq", "-o", "json"): `{"metadata":{"name":"cpu-cq"},"spec":{"resourceGroups":[{"flavors":[{"name":"cpu","resources":[{"name":"cpu","nominalQuota":"100"}]}]}]}}`, + fakeRawKey("get", "resourceflavor.kueue.x-k8s.io", "cpu", "-o", "json"): `{"metadata":{"name":"cpu"},"spec":{}}`, + }, + errors: map[string]error{}, + } + manifest := []byte(`apiVersion: batch/v1 +kind: Job +metadata: + labels: + kueue.x-k8s.io/queue-name: cpu-queue +spec: + template: + metadata: + annotations: + kueue.x-k8s.io/podset-unconstrained-topology: "true" + spec: + containers: + - resources: + requests: + cpu: "1" +`) + + err := validateRenderedQueue(context.Background(), runner, "workspace", manifest, jobrender.Options{}, queueValidationPolicy{}) + if err == nil || !strings.Contains(err.Error(), "CPU workload sets policy.topology") || + !strings.Contains(err.Error(), "do not support TopologyAwareScheduling") { + t.Fatalf("expected actionable CPU topology mismatch, got %v", err) + } +} + +func TestValidateRenderedQueueAcceptsCPUTopologyOnTASFlavor(t *testing.T) { + runner := &fakeRawRunner{ + outputs: map[string]string{ + fakeRawKey("-n", "workspace", "get", "localqueue.kueue.x-k8s.io", "cpu-queue", "-o", "json"): `{"metadata":{"name":"cpu-queue"},"spec":{"clusterQueue":"cpu-cq"}}`, + fakeRawKey("get", "clusterqueue.kueue.x-k8s.io", "cpu-cq", "-o", "json"): `{"metadata":{"name":"cpu-cq"},"spec":{"resourceGroups":[{"flavors":[{"name":"cpu-tas","resources":[{"name":"cpu","nominalQuota":"100"}]}]}]}}`, + fakeRawKey("get", "resourceflavor.kueue.x-k8s.io", "cpu-tas", "-o", "json"): `{"metadata":{"name":"cpu-tas"},"spec":{"topologyName":"default-node-topology"}}`, + }, + errors: map[string]error{}, + } + manifest := []byte(`apiVersion: batch/v1 +kind: Job +metadata: + labels: + kueue.x-k8s.io/queue-name: cpu-queue +spec: + template: + metadata: + annotations: + kueue.x-k8s.io/podset-unconstrained-topology: "true" + spec: + containers: + - resources: + requests: + cpu: "1" +`) + + if err := validateRenderedQueue(context.Background(), runner, "workspace", manifest, jobrender.Options{}, queueValidationPolicy{}); err != nil { + t.Fatalf("CPU topology-capable flavor rejected: %v", err) + } +} + func TestRenderedQueueContractIgnoresTopologyOutsideGPUPodTemplate(t *testing.T) { manifest := []byte(`apiVersion: ray.io/v1 kind: RayJob diff --git a/cli/internal/queueresolve/validate.go b/cli/internal/queueresolve/validate.go index d65402448..85ce8a414 100644 --- a/cli/internal/queueresolve/validate.go +++ b/cli/internal/queueresolve/validate.go @@ -34,6 +34,7 @@ type ValidationOptions struct { NodeSelector map[string]string PodTolerations [][]kueueapi.Toleration GPUCount int + CPURequest bool GPUResourceName string } @@ -61,6 +62,7 @@ type validationTarget struct { NodeSelector map[string]string PodTolerations [][]kueueapi.Toleration GPUCount int + CPURequest bool GPUResourceName string } @@ -109,6 +111,9 @@ func ValidateSelection(ctx context.Context, r RawRunner, opts ValidationOptions) if err := validateOptionalTopologyLabels("ClusterQueue", target.ClusterQueue, cq.Metadata.Labels, target); err != nil { return report, err } + if err := validateCPUQueueTopologyIntent(ctx, r, cq, target); err != nil { + return report, err + } policyTopologyContract := target.CatalogTopologyContract if policyTopologyContract { @@ -291,6 +296,7 @@ func (o ValidationOptions) resolve() validationTarget { NodeSelector: o.NodeSelector, PodTolerations: o.PodTolerations, GPUCount: o.GPUCount, + CPURequest: o.CPURequest, GPUResourceName: strings.TrimSpace(o.GPUResourceName), } if target.Namespace == "" { @@ -622,6 +628,68 @@ func validateQueueTopologyIntent(ctx context.Context, r RawRunner, cq kueueapi.C target.ClusterQueue, strings.Join(names, ", ")) } +func validateCPUQueueTopologyIntent(ctx context.Context, r RawRunner, cq kueueapi.ClusterQueue, target validationTarget) error { + if !target.CPURequest || target.GPUCount > 0 || !target.TopologyRequest { + return nil + } + flavorNames := resourceFlavorNames(cq, "cpu") + if target.ResourceFlavor != "" { + flavorNames = []string{target.ResourceFlavor} + } + if len(flavorNames) == 0 { + return nil + } + var topologyFlavors []string + var unreadable []string + for _, name := range flavorNames { + rf, err := getResourceFlavor(ctx, r, name) + if err != nil { + unreadable = append(unreadable, fmt.Sprintf("%s (%s)", name, err)) + continue + } + if strings.TrimSpace(rf.Spec.TopologyName) != "" { + topologyFlavors = append(topologyFlavors, name) + } + } + if len(unreadable) > 0 { + return fmt.Errorf( + "cannot verify CPU policy.topology compatibility with ClusterQueue %q because ResourceFlavor capabilities could not be read: %s", + target.ClusterQueue, + strings.Join(unreadable, ", "), + ) + } + if len(topologyFlavors) > 0 { + return nil + } + return fmt.Errorf( + "CPU workload sets policy.topology, but CPU ResourceFlavors in ClusterQueue %q do not support TopologyAwareScheduling; use a topology-capable CPU flavor or omit independent/unconstrained topology for this queue", + target.ClusterQueue, + ) +} + +func resourceFlavorNames(cq kueueapi.ClusterQueue, resourceName string) []string { + seen := map[string]bool{} + var names []string + for _, group := range cq.Spec.ResourceGroups { + for _, flavor := range group.Flavors { + if seen[flavor.Name] { + continue + } + for _, resource := range flavor.Resources { + if resource.Name != resourceName || + resource.NominalQuota.Int64()+resource.BorrowingLimit.Int64() <= 0 { + continue + } + seen[flavor.Name] = true + names = append(names, flavor.Name) + break + } + } + } + sort.Strings(names) + return names +} + func compatibleGPUFlavorNames(cq kueueapi.ClusterQueue, target validationTarget) []string { seen := map[string]struct{}{} var names []string From 0e2fc11debcaa0a612d4273c4d868b487bb3f316 Mon Sep 17 00:00:00 2001 From: chokevin Date: Fri, 6 Nov 2026 09:39:54 -0800 Subject: [PATCH 2/3] fix(rbac): allow workspace flavor preflight reads --- charts/tau-core-controller/kustomize/rbac.yaml | 3 +++ charts/tau-core-controller/templates/rbac.yaml | 3 +++ charts/tau-core-controller/tests/rbac_test.yaml | 14 ++++++++++++++ .../internal/controller/workspace_rbac_test.go | 15 +++++++++++++++ controllers/tau-core/scripts/kind-e2e.sh | 2 ++ 5 files changed, 37 insertions(+) diff --git a/charts/tau-core-controller/kustomize/rbac.yaml b/charts/tau-core-controller/kustomize/rbac.yaml index 8b4a05d25..b34eadf8e 100644 --- a/charts/tau-core-controller/kustomize/rbac.yaml +++ b/charts/tau-core-controller/kustomize/rbac.yaml @@ -40,6 +40,9 @@ rules: - apiGroups: ["kueue.x-k8s.io"] resources: ["clusterqueues"] verbs: ["get"] + - apiGroups: ["kueue.x-k8s.io"] + resources: ["resourceflavors"] + verbs: ["get"] - apiGroups: ["tau.azure.com"] resources: ["clusters"] resourceNames: ["cluster"] diff --git a/charts/tau-core-controller/templates/rbac.yaml b/charts/tau-core-controller/templates/rbac.yaml index 5998a570c..e71af7e6f 100644 --- a/charts/tau-core-controller/templates/rbac.yaml +++ b/charts/tau-core-controller/templates/rbac.yaml @@ -38,6 +38,9 @@ rules: - apiGroups: ["kueue.x-k8s.io"] resources: ["clusterqueues"] verbs: ["get"] + - apiGroups: ["kueue.x-k8s.io"] + resources: ["resourceflavors"] + verbs: ["get"] - apiGroups: ["tau.azure.com"] resources: ["clusters"] resourceNames: ["cluster"] diff --git a/charts/tau-core-controller/tests/rbac_test.yaml b/charts/tau-core-controller/tests/rbac_test.yaml index 6d0dec002..3d7e5daf6 100644 --- a/charts/tau-core-controller/tests/rbac_test.yaml +++ b/charts/tau-core-controller/tests/rbac_test.yaml @@ -40,6 +40,20 @@ tests: resources: ["rayjobs", "rayservices"] verbs: ["create", "get", "list", "watch", "delete", "patch", "update"] + - it: grants workspace researchers ResourceFlavor reads for queue preflight + documentSelector: + path: metadata.name + value: tau-clusterqueue-reader-v1 + asserts: + - isKind: + of: ClusterRole + - contains: + path: rules + content: + apiGroups: ["kueue.x-k8s.io"] + resources: ["resourceflavors"] + verbs: ["get"] + - it: grants the controller create-only mutation access to Kueue Topologies asserts: - contains: diff --git a/controllers/tau-core/internal/controller/workspace_rbac_test.go b/controllers/tau-core/internal/controller/workspace_rbac_test.go index 4566be1c6..0b0a9d4b9 100644 --- a/controllers/tau-core/internal/controller/workspace_rbac_test.go +++ b/controllers/tau-core/internal/controller/workspace_rbac_test.go @@ -51,6 +51,21 @@ func TestKustomizeResearcherRoleGrantsRayServicePermissions(t *testing.T) { t.Fatalf("researcher role must grant RayService lifecycle permissions: %#v", role.Rules) } +func TestKustomizeClusterQueueReaderRoleGrantsResourceFlavorRead(t *testing.T) { + role := kustomizeClusterRole(t, clusterQueueReaderRoleName) + want := rbacv1.PolicyRule{ + APIGroups: []string{"kueue.x-k8s.io"}, + Resources: []string{"resourceflavors"}, + Verbs: []string{"get"}, + } + for _, rule := range role.Rules { + if reflect.DeepEqual(rule, want) { + return + } + } + t.Fatalf("ClusterQueue reader role must grant ResourceFlavor reads for queue preflight: %#v", role.Rules) +} + func TestKustomizeControllerRoleMatchesTopologyAndGPUFlavorPermissions(t *testing.T) { role := kustomizeClusterRole(t, "tau-core-controller") for _, want := range []rbacv1.PolicyRule{ diff --git a/controllers/tau-core/scripts/kind-e2e.sh b/controllers/tau-core/scripts/kind-e2e.sh index 3eb52b506..795452a1c 100755 --- a/controllers/tau-core/scripts/kind-e2e.sh +++ b/controllers/tau-core/scripts/kind-e2e.sh @@ -872,8 +872,10 @@ kubectl auth can-i get configmaps -n "${TARGET_NAMESPACE}" --as=researcher@examp kubectl auth can-i get "workspaces.tau.azure.com/${WORKSPACE_NAME}" -n "${SYSTEM_NAMESPACE}" --as=researcher@example.com --as-group="${WORKSPACE_GROUP}" | grep -qx yes kubectl auth can-i create quotarequests.tau.azure.com -n "${SYSTEM_NAMESPACE}" --as=researcher@example.com --as-group="${WORKSPACE_GROUP}" | grep -qx yes kubectl auth can-i get "clusterqueues.kueue.x-k8s.io/${WORKSPACE_NAME}" --as=researcher@example.com --as-group="${WORKSPACE_GROUP}" | grep -qx yes +kubectl auth can-i get resourceflavors.kueue.x-k8s.io --as=researcher@example.com --as-group="${WORKSPACE_GROUP}" | grep -qx yes kubectl auth can-i get "clusters.tau.azure.com/cluster" --as=researcher@example.com --as-group="${WORKSPACE_GROUP}" | grep -qx yes [[ "$(kubectl auth can-i list clusterqueues.kueue.x-k8s.io --as=researcher@example.com --as-group="${WORKSPACE_GROUP}" || true)" == "no" ]] +[[ "$(kubectl auth can-i list resourceflavors.kueue.x-k8s.io --as=researcher@example.com --as-group="${WORKSPACE_GROUP}" || true)" == "no" ]] [[ "$(kubectl auth can-i list clusters.tau.azure.com --as=researcher@example.com --as-group="${WORKSPACE_GROUP}" || true)" == "no" ]] [[ "$(kubectl auth can-i get clusters.tau.azure.com/other --as=researcher@example.com --as-group="${WORKSPACE_GROUP}" || true)" == "no" ]] [[ "$(kubectl auth can-i list workspaces.tau.azure.com -n "${SYSTEM_NAMESPACE}" --as=researcher@example.com --as-group="${WORKSPACE_GROUP}" || true)" == "no" ]] From 54c43179498dc610b47de1bca2618f3f6ae82e77 Mon Sep 17 00:00:00 2001 From: chokevin Date: Wed, 11 Nov 2026 17:08:07 -0800 Subject: [PATCH 3/3] test(queue): explain CPU flavor read denial --- cli/internal/queueresolve/validate.go | 2 +- cli/internal/queueresolve/validate_test.go | 30 ++++++++++++++++++++++ 2 files changed, 31 insertions(+), 1 deletion(-) diff --git a/cli/internal/queueresolve/validate.go b/cli/internal/queueresolve/validate.go index 85ce8a414..e5e998252 100644 --- a/cli/internal/queueresolve/validate.go +++ b/cli/internal/queueresolve/validate.go @@ -653,7 +653,7 @@ func validateCPUQueueTopologyIntent(ctx context.Context, r RawRunner, cq kueueap } if len(unreadable) > 0 { return fmt.Errorf( - "cannot verify CPU policy.topology compatibility with ClusterQueue %q because ResourceFlavor capabilities could not be read: %s", + "cannot verify CPU policy.topology compatibility with ClusterQueue %q because ResourceFlavor capabilities could not be read: %s; grant read access to ResourceFlavors or ask the platform owner to inspect the queue", target.ClusterQueue, strings.Join(unreadable, ", "), ) diff --git a/cli/internal/queueresolve/validate_test.go b/cli/internal/queueresolve/validate_test.go index 4fad14b9b..1eafc4926 100644 --- a/cli/internal/queueresolve/validate_test.go +++ b/cli/internal/queueresolve/validate_test.go @@ -6,6 +6,7 @@ package queueresolve import ( "context" "encoding/json" + "errors" "fmt" "reflect" "strings" @@ -568,6 +569,35 @@ func TestValidateSelectionSkipsTopologyCapabilityForCPUOnlyConfig(t *testing.T) } } +func TestValidateSelectionExplainsCPUResourceFlavorReadDenial(t *testing.T) { + runner := &validationFakeRunner{ + outputs: map[string]string{ + validationKey("-n", "workspace", "get", "localqueue.kueue.x-k8s.io", "cpu-queue", "-o", "json"): localQueueObject("cpu-queue", "cpu-cq", nil), + validationKey("get", "clusterqueue.kueue.x-k8s.io", "cpu-cq", "-o", "json"): `{ + "metadata":{"name":"cpu-cq"}, + "spec":{"resourceGroups":[{"flavors":[ + {"name":"cpu-tas","resources":[{"name":"cpu","nominalQuota":"100"}]} + ]}]} + }`, + }, + errors: map[string]error{ + validationKey("get", "resourceflavor.kueue.x-k8s.io", "cpu-tas", "-o", "json"): errors.New("Error from server (Forbidden): resourceflavors.kueue.x-k8s.io is forbidden"), + }, + } + + _, err := ValidateSelection(context.Background(), runner, ValidationOptions{ + Namespace: "workspace", + QueueName: "cpu-queue", + TopologyRequest: true, + CPURequest: true, + }) + if err == nil || + !strings.Contains(err.Error(), "Forbidden") || + !strings.Contains(err.Error(), "grant read access to ResourceFlavors") { + t.Fatalf("CPU ResourceFlavor read denial should be actionable, got %v", err) + } +} + func TestValidateSelectionAcceptsTopologyCapableWorkspaceQueue(t *testing.T) { runner := &validationFakeRunner{ outputs: map[string]string{