diff --git a/charts/tau-core-controller/kustomize/rbac.yaml b/charts/tau-core-controller/kustomize/rbac.yaml index 8010381fd..8b4a05d25 100644 --- a/charts/tau-core-controller/kustomize/rbac.yaml +++ b/charts/tau-core-controller/kustomize/rbac.yaml @@ -110,6 +110,15 @@ rules: - apiGroups: ["scheduling.k8s.io"] resources: ["priorityclasses"] verbs: ["get", "list", "watch"] + - apiGroups: ["batch"] + resources: ["jobs"] + verbs: ["get", "list", "watch"] + - apiGroups: ["apps"] + resources: ["deployments"] + verbs: ["get", "list", "watch"] + - apiGroups: ["ray.io"] + resources: ["rayjobs", "rayservices"] + verbs: ["get", "list", "watch"] - apiGroups: ["kueue.x-k8s.io"] resources: ["workloads"] verbs: ["get", "list", "watch"] diff --git a/charts/tau-core-controller/templates/rbac.yaml b/charts/tau-core-controller/templates/rbac.yaml index f043e8484..5998a570c 100644 --- a/charts/tau-core-controller/templates/rbac.yaml +++ b/charts/tau-core-controller/templates/rbac.yaml @@ -107,6 +107,15 @@ rules: - apiGroups: ["scheduling.k8s.io"] resources: ["priorityclasses"] verbs: ["get", "list", "watch"] + - apiGroups: ["batch"] + resources: ["jobs"] + verbs: ["get", "list", "watch"] + - apiGroups: ["apps"] + resources: ["deployments"] + verbs: ["get", "list", "watch"] + - apiGroups: ["ray.io"] + resources: ["rayjobs", "rayservices"] + verbs: ["get", "list", "watch"] - apiGroups: ["kueue.x-k8s.io"] resources: ["workloads"] verbs: ["get", "list", "watch"] diff --git a/charts/tau-core-controller/tests/rbac_test.yaml b/charts/tau-core-controller/tests/rbac_test.yaml index 18a0a85e5..6d0dec002 100644 --- a/charts/tau-core-controller/tests/rbac_test.yaml +++ b/charts/tau-core-controller/tests/rbac_test.yaml @@ -2,6 +2,30 @@ suite: Researcher RayService permissions templates: - templates/rbac.yaml tests: + - it: grants the controller read-only source Job access for workspace drain checks + asserts: + - contains: + path: rules + content: + apiGroups: ["apps"] + resources: ["deployments"] + verbs: ["get", "list", "watch"] + documentIndex: 2 + - contains: + path: rules + content: + apiGroups: ["batch"] + resources: ["jobs"] + verbs: ["get", "list", "watch"] + documentIndex: 2 + - contains: + path: rules + content: + apiGroups: ["ray.io"] + resources: ["rayjobs", "rayservices"] + verbs: ["get", "list", "watch"] + documentIndex: 2 + - it: grants RayJob and RayService lifecycle permissions to the researcher role documentSelector: path: metadata.name diff --git a/controllers/tau-core/internal/controller/controller_test.go b/controllers/tau-core/internal/controller/controller_test.go index a688a0076..f51e4fc9c 100644 --- a/controllers/tau-core/internal/controller/controller_test.go +++ b/controllers/tau-core/internal/controller/controller_test.go @@ -28,6 +28,7 @@ import ( ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" "sigs.k8s.io/controller-runtime/pkg/event" "sigs.k8s.io/controller-runtime/pkg/reconcile" @@ -1257,9 +1258,14 @@ func TestWorkspaceDeleteCleansWorkspaceAccess(t *testing.T) { now := metav1.Now() workspace.Finalizers = []string{workspaceFinalizer} workspace.DeletionTimestamp = &now + workspace.Spec.Target.Namespace = "new-aurora" workspace.Status.Target.ResolvedNamespace = "aurora" targetNamespace := testNamespace("aurora") - targetNamespace.Labels = map[string]string{labelWorkspace: "aurora"} + targetNamespace.Labels = workspaceNamespaceLabels(workspace.Name, workspace.Spec.Queue) + targetNamespace.Labels[labelTeam] = "vision" + targetNamespace.Labels["preserved"] = "true" + targetNamespace.Annotations[annotationResultScope] = workspace.Spec.Defaults.OutputRoot + targetNamespace.Annotations["preserved"] = "true" targetRole := testRole("aurora", defaultRoleName, "aurora") targetBinding := testRoleBinding("aurora", defaultRoleName, "aurora") targetServiceAccount := testServiceAccount("aurora", "tau-workload", "aurora") @@ -1298,6 +1304,35 @@ func TestWorkspaceDeleteCleansWorkspaceAccess(t *testing.T) { if err := c.Get(ctx, client.ObjectKey{Name: "aurora"}, targetNamespace); err != nil { t.Fatalf("target namespace was deleted: %v", err) } + for _, key := range []string{ + labelManagedBy, + labelTeam, + labelWorkspace, + labelWorkspaceLocalQueue, + labelKueueDefaultLocalQueue, + } { + if _, ok := targetNamespace.Labels[key]; ok { + t.Fatalf("target namespace retained workspace label %q: %#v", key, targetNamespace.Labels) + } + } + for _, key := range []string{annotationOwnerUID, annotationResultScope} { + if _, ok := targetNamespace.Annotations[key]; ok { + t.Fatalf("target namespace retained workspace annotation %q: %#v", key, targetNamespace.Annotations) + } + } + for key, want := range map[string]string{ + labelPSAEnforce: "baseline", + labelPSAAudit: "restricted", + labelPSAWarn: "restricted", + "preserved": "true", + } { + if got := targetNamespace.Labels[key]; got != want { + t.Fatalf("target namespace label %q = %q, want %q", key, got, want) + } + } + if got := targetNamespace.Annotations["preserved"]; got != "true" { + t.Fatalf("target namespace preserved annotation = %q, want true", got) + } if err := c.Get(ctx, client.ObjectKey{Name: clusterQueueReaderBindingName("aurora")}, &rbacv1.ClusterRoleBinding{}); !apierrors.IsNotFound(err) { t.Fatalf("expected ClusterQueue reader ClusterRoleBinding deleted, got err=%v", err) } @@ -1308,6 +1343,500 @@ func TestWorkspaceDeleteCleansWorkspaceAccess(t *testing.T) { } } +func TestWorkspaceDeletePreservesNamespaceMetadataForDifferentOwnerUID(t *testing.T) { + ctx := context.Background() + scheme := testScheme(t) + workspace := testWorkspace("aurora") + now := metav1.Now() + workspace.Finalizers = []string{workspaceFinalizer} + workspace.DeletionTimestamp = &now + workspace.Status.Target.ResolvedNamespace = "aurora" + targetNamespace := testNamespace("aurora") + targetNamespace.Labels = workspaceNamespaceLabels(workspace.Name, workspace.Spec.Queue) + targetNamespace.Labels[labelTeam] = "vision" + targetNamespace.Labels["preserved"] = "true" + targetNamespace.Annotations[annotationOwnerUID] = "different-workspace-uid" + targetNamespace.Annotations[annotationResultScope] = workspace.Spec.Defaults.OutputRoot + targetNamespace.Annotations["preserved"] = "true" + c := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(workspace, targetNamespace). + Build() + reconciler := newTestWorkspaceReconciler(c) + + if _, err := reconciler.Reconcile(ctx, ctrl.Request{NamespacedName: types.NamespacedName{Name: "aurora", Namespace: tauv1alpha1.SystemNamespace}}); err != nil { + t.Fatalf("Reconcile delete error = %v", err) + } + if err := c.Get(ctx, client.ObjectKey{Name: "aurora"}, targetNamespace); err != nil { + t.Fatalf("target namespace was deleted: %v", err) + } + for key, want := range map[string]string{ + labelManagedBy: labelManagedByValue, + labelTeam: "vision", + labelWorkspace: workspace.Name, + labelWorkspaceLocalQueue: workspace.Spec.Queue, + labelKueueDefaultLocalQueue: workspace.Spec.Queue, + labelPSAEnforce: "baseline", + labelPSAAudit: "restricted", + labelPSAWarn: "restricted", + "preserved": "true", + } { + if got := targetNamespace.Labels[key]; got != want { + t.Fatalf("target namespace label %q = %q, want %q", key, got, want) + } + } + for key, want := range map[string]string{ + annotationOwnerUID: "different-workspace-uid", + annotationResultScope: workspace.Spec.Defaults.OutputRoot, + "preserved": "true", + } { + if got := targetNamespace.Annotations[key]; got != want { + t.Fatalf("target namespace annotation %q = %q, want %q", key, got, want) + } + } +} + +func TestWorkspaceDeletePreservesNamespaceRoutingMetadataForExternalQueueWorkloads(t *testing.T) { + for _, test := range []struct { + name string + workloadStatus map[string]any + queueStatus tauv1alpha1.WorkspaceQueueStatus + teamRef *tauv1alpha1.TauClusterObjectReference + wantBlock bool + }{ + { + name: "pending", + queueStatus: tauv1alpha1.WorkspaceQueueStatus{ + LocalQueue: "jobqueue", + ClusterQueue: "jobqueue", + ClusterQueueUID: "jobqueue-uid", + }, + wantBlock: true, + }, + { + name: "admitted", + workloadStatus: map[string]any{ + "admission": map[string]any{"clusterQueue": "jobqueue"}, + }, + queueStatus: tauv1alpha1.WorkspaceQueueStatus{ + LocalQueue: "jobqueue", + ClusterQueue: "jobqueue", + ClusterQueueUID: "jobqueue-uid", + }, + wantBlock: true, + }, + {name: "pending with empty queue status", wantBlock: true}, + { + name: "pending on compatibility alias", + teamRef: &tauv1alpha1.TauClusterObjectReference{Name: "vision"}, + queueStatus: tauv1alpha1.WorkspaceQueueStatus{ + LocalQueue: workspaceClusterQueueName("aurora"), + ClusterQueue: workspaceClusterQueueName("aurora"), + }, + wantBlock: true, + }, + { + name: "finished", + workloadStatus: map[string]any{ + "conditions": []any{map[string]any{"type": "Finished", "status": "True"}}, + }, + queueStatus: tauv1alpha1.WorkspaceQueueStatus{ + LocalQueue: "jobqueue", + ClusterQueue: "jobqueue", + ClusterQueueUID: "jobqueue-uid", + }, + }, + } { + t.Run(test.name, func(t *testing.T) { + ctx := context.Background() + workspace := testWorkspace("aurora") + workspace.Finalizers = []string{workspaceFinalizer} + workspace.Annotations = map[string]string{annotationOwnerUID: string(workspace.UID)} + now := metav1.Now() + workspace.DeletionTimestamp = &now + workspace.Spec.Target.CreateNamespace = false + workspace.Spec.Queue = "jobqueue" + workspace.Spec.TeamRef = test.teamRef + workspace.Status.Target.ResolvedNamespace = workspace.Name + workspace.Status.Queue = test.queueStatus + namespace := testNamespace(workspace.Name) + namespace.Labels = workspaceNamespaceLabels(workspace.Name, workspace.Spec.Queue) + localQueue := testLocalQueue(namespace.Name, "jobqueue", "jobqueue") + localQueue.SetAnnotations(nil) + clusterQueue := clusterQueueWithUID("jobqueue", "jobqueue-uid") + clusterQueue.Object["spec"] = map[string]any{ + "namespaceSelector": map[string]any{ + "matchExpressions": []any{map[string]any{ + "key": labelWorkspace, + "operator": "Exists", + }}, + }, + } + workload := newQueueObject(workloadGVK) + workload.SetName("training") + workload.SetNamespace(namespace.Name) + workload.Object["spec"] = map[string]any{"queueName": localQueue.GetName()} + if test.workloadStatus != nil { + workload.Object["status"] = test.workloadStatus + } + c := fake.NewClientBuilder(). + WithScheme(testScheme(t)). + WithObjects(workspace, namespace, localQueue, clusterQueue, workload). + Build() + reconciler := newTestWorkspaceReconciler(c) + + _, err := reconciler.Reconcile(ctx, ctrl.Request{NamespacedName: client.ObjectKeyFromObject(workspace)}) + if test.wantBlock { + if err == nil || !strings.Contains(err.Error(), "before removing namespace routing metadata") { + t.Fatalf("Reconcile() error = %v, want routing metadata retention", err) + } + } else if err != nil { + t.Fatalf("Reconcile() error = %v", err) + } + + var gotNamespace corev1.Namespace + if err := c.Get(ctx, client.ObjectKeyFromObject(namespace), &gotNamespace); err != nil { + t.Fatal(err) + } + if test.wantBlock { + if gotNamespace.Labels[labelWorkspace] != workspace.Name || + gotNamespace.Annotations[annotationOwnerUID] != string(workspace.UID) { + t.Fatalf("namespace routing metadata was removed while workload remained: labels=%#v annotations=%#v", gotNamespace.Labels, gotNamespace.Annotations) + } + var remaining tauv1alpha1.TauWorkspace + if err := c.Get(ctx, client.ObjectKeyFromObject(workspace), &remaining); err != nil { + t.Fatal(err) + } + if !controllerutil.ContainsFinalizer(&remaining, workspaceFinalizer) { + t.Fatal("workspace finalizer was removed while routing metadata remained in use") + } + } else { + if gotNamespace.Labels[labelWorkspace] != "" || + gotNamespace.Annotations[annotationOwnerUID] != "" { + t.Fatalf("drained namespace retained stale metadata: labels=%#v annotations=%#v", gotNamespace.Labels, gotNamespace.Annotations) + } + if err := c.Get(ctx, client.ObjectKeyFromObject(workspace), &tauv1alpha1.TauWorkspace{}); !apierrors.IsNotFound(err) { + t.Fatalf("workspace finalizer was not removed after workload drain: %v", err) + } + } + for _, queue := range []*unstructured.Unstructured{localQueue, clusterQueue} { + got := newQueueObject(queue.GroupVersionKind()) + if err := c.Get(ctx, client.ObjectKeyFromObject(queue), got); err != nil { + t.Fatalf("external %s was mutated or deleted: %v", queue.GetKind(), err) + } + } + }) + } +} + +func TestWorkspaceDeletePreservesWorkloadIdentityWhileExternalQueueWorkloadDrains(t *testing.T) { + ctx := context.Background() + workspace := testWorkspace("aurora") + workspace.Finalizers = []string{workspaceFinalizer} + workspace.Annotations = map[string]string{annotationOwnerUID: string(workspace.UID)} + now := metav1.Now() + workspace.DeletionTimestamp = &now + workspace.Spec.Target.CreateNamespace = false + workspace.Spec.Queue = "jobqueue" + workspace.Status.Target.ResolvedNamespace = workspace.Name + workspace.Status.Queue = tauv1alpha1.WorkspaceQueueStatus{ + LocalQueue: "jobqueue", + ClusterQueue: "jobqueue", + ClusterQueueUID: "jobqueue-uid", + } + namespace := testNamespace(workspace.Name) + namespace.Labels = workspaceNamespaceLabels(workspace.Name, workspace.Spec.Queue) + account := testServiceAccount(namespace.Name, workspace.Spec.WorkloadIdentity.ServiceAccountName, workspace.Name) + binding := testRoleBinding(namespace.Name, defaultRoleName, workspace.Name) + job := &batchv1.Job{ + ObjectMeta: metav1.ObjectMeta{ + Name: "training", + Namespace: namespace.Name, + UID: types.UID("training-job-uid"), + }, + Spec: batchv1.JobSpec{ + Template: corev1.PodTemplateSpec{ + Spec: corev1.PodSpec{ + ServiceAccountName: account.Name, + RestartPolicy: corev1.RestartPolicyNever, + Containers: []corev1.Container{{ + Name: "trainer", + Image: "example.invalid/trainer", + Command: []string{"train"}, + }}, + }, + }, + }, + } + localQueue := testLocalQueue(namespace.Name, "jobqueue", "jobqueue") + localQueue.SetAnnotations(nil) + clusterQueue := clusterQueueWithUID("jobqueue", "jobqueue-uid") + clusterQueue.Object["spec"] = map[string]any{ + "namespaceSelector": map[string]any{ + "matchExpressions": []any{map[string]any{ + "key": labelWorkspace, + "operator": "Exists", + }}, + }, + } + workload := newQueueObject(workloadGVK) + workload.SetName("job-training") + workload.SetNamespace(namespace.Name) + workload.SetOwnerReferences([]metav1.OwnerReference{{ + APIVersion: batchv1.SchemeGroupVersion.String(), + Kind: "Job", + Name: job.Name, + UID: job.UID, + }}) + workload.Object["spec"] = map[string]any{"queueName": localQueue.GetName()} + c := fake.NewClientBuilder(). + WithScheme(testScheme(t)). + WithObjects(workspace, namespace, account, binding, job, localQueue, clusterQueue, workload). + Build() + reconciler := newTestWorkspaceReconciler(c) + request := ctrl.Request{NamespacedName: client.ObjectKeyFromObject(workspace)} + + if _, err := reconciler.Reconcile(ctx, request); err == nil || + !strings.Contains(err.Error(), "before removing namespace routing metadata") { + t.Fatalf("Reconcile() error = %v, want workload drain wait", err) + } + if err := c.Get(ctx, client.ObjectKeyFromObject(account), &corev1.ServiceAccount{}); err != nil { + t.Fatalf("workspace workload identity was removed before Workload drain: %v", err) + } + if err := c.Get(ctx, client.ObjectKeyFromObject(binding), &rbacv1.RoleBinding{}); !apierrors.IsNotFound(err) { + t.Fatalf("researcher RoleBinding survived deletion drain: %v", err) + } + if err := c.Get(ctx, client.ObjectKeyFromObject(job), &batchv1.Job{}); err != nil { + t.Fatalf("pending Job was removed during Workload drain: %v", err) + } + + workload.Object["status"] = map[string]any{"conditions": []any{ + map[string]any{"type": "Finished", "status": "True"}, + }} + if err := c.Update(ctx, workload); err != nil { + t.Fatal(err) + } + if _, err := reconciler.Reconcile(ctx, request); err != nil { + t.Fatalf("Reconcile() after Workload drain = %v", err) + } + if err := c.Get(ctx, client.ObjectKeyFromObject(account), &corev1.ServiceAccount{}); !apierrors.IsNotFound(err) { + t.Fatalf("workspace workload identity survived after Workload drain: %v", err) + } +} + +func TestLegacyWorkspaceDeletePreservesRoutingMetadataWithEmptyQueueStatus(t *testing.T) { + ctx := context.Background() + workspace := testWorkspace("aurora") + workspace.Finalizers = []string{workspaceFinalizer} + workspace.Annotations = map[string]string{annotationV0Primary: "true"} + now := metav1.Now() + workspace.DeletionTimestamp = &now + workspace.Spec.Target.CreateNamespace = false + workspace.Spec.Queue = "jobqueue" + namespace := testNamespace(workspace.Name) + namespace.Labels = workspaceNamespaceLabels(workspace.Name, workspace.Spec.Queue) + namespace.Annotations = nil + localQueue := testLocalQueue(namespace.Name, "jobqueue", "jobqueue") + localQueue.SetAnnotations(nil) + clusterQueue := clusterQueueWithUID("jobqueue", "jobqueue-uid") + clusterQueue.Object["spec"] = map[string]any{ + "namespaceSelector": map[string]any{ + "matchExpressions": []any{map[string]any{ + "key": labelWorkspace, + "operator": "Exists", + }}, + }, + } + workload := newQueueObject(workloadGVK) + workload.SetName("training") + workload.SetNamespace(namespace.Name) + workload.Object["spec"] = map[string]any{"queueName": localQueue.GetName()} + c := fake.NewClientBuilder(). + WithScheme(testScheme(t)). + WithObjects(workspace, namespace, localQueue, clusterQueue, workload). + WithStatusSubresource(&tauv1alpha1.TauWorkspace{}). + Build() + reconciler := newTestWorkspaceReconciler(c) + request := ctrl.Request{NamespacedName: client.ObjectKeyFromObject(workspace)} + + var cleanupErr error + for range 4 { + if _, err := reconciler.Reconcile(ctx, request); err != nil { + cleanupErr = err + break + } + } + if cleanupErr == nil || !strings.Contains(cleanupErr.Error(), "before removing namespace routing metadata") { + t.Fatalf("Reconcile() error = %v, want routing metadata retention after ownership migration", cleanupErr) + } + var gotNamespace corev1.Namespace + if err := c.Get(ctx, client.ObjectKeyFromObject(namespace), &gotNamespace); err != nil { + t.Fatal(err) + } + if gotNamespace.Labels[labelWorkspace] != workspace.Name || + gotNamespace.Annotations[annotationOwnerUID] != string(workspace.UID) { + t.Fatalf("legacy deletion removed migrated routing metadata: labels=%#v annotations=%#v", gotNamespace.Labels, gotNamespace.Annotations) + } + var remaining tauv1alpha1.TauWorkspace + if err := c.Get(ctx, client.ObjectKeyFromObject(workspace), &remaining); err != nil { + t.Fatal(err) + } + if remaining.Status.Queue != (tauv1alpha1.WorkspaceQueueStatus{}) { + t.Fatalf("legacy deletion unexpectedly populated queue status: %#v", remaining.Status.Queue) + } + if !controllerutil.ContainsFinalizer(&remaining, workspaceFinalizer) { + t.Fatal("workspace finalizer was removed while migrated routing metadata remained in use") + } +} + +func TestWorkspaceDeleteIgnoresUnrelatedWorkloadsWhenCleaningNamespaceMetadata(t *testing.T) { + ctx := context.Background() + workspace := testWorkspace("aurora") + workspace.Finalizers = []string{workspaceFinalizer} + workspace.Annotations = map[string]string{annotationOwnerUID: string(workspace.UID)} + now := metav1.Now() + workspace.DeletionTimestamp = &now + workspace.Spec.Target.CreateNamespace = false + workspace.Spec.Queue = "jobqueue" + workspace.Status.Target.ResolvedNamespace = workspace.Name + workspace.Status.Queue = tauv1alpha1.WorkspaceQueueStatus{ + LocalQueue: "jobqueue", + ClusterQueue: "jobqueue", + } + namespace := testNamespace(workspace.Name) + namespace.Labels = workspaceNamespaceLabels(workspace.Name, workspace.Spec.Queue) + namespace.Labels["queue-scope"] = "shared" + localQueue := testLocalQueue(namespace.Name, "jobqueue", "jobqueue") + localQueue.SetAnnotations(nil) + clusterQueue := testClusterQueue("jobqueue") + clusterQueue.Object["spec"] = map[string]any{ + "namespaceSelector": map[string]any{ + "matchLabels": map[string]any{"queue-scope": "shared"}, + }, + } + selectorIndependent := newQueueObject(workloadGVK) + selectorIndependent.SetName("selector-independent") + selectorIndependent.SetNamespace(namespace.Name) + selectorIndependent.Object["spec"] = map[string]any{"queueName": localQueue.GetName()} + unrelated := newQueueObject(workloadGVK) + unrelated.SetName("unrelated") + unrelated.SetNamespace(namespace.Name) + unrelated.Object["spec"] = map[string]any{"queueName": "missing-queue"} + foreign := newQueueObject(workloadGVK) + foreign.SetName("foreign") + foreign.SetNamespace("other") + foreign.Object["spec"] = map[string]any{"queueName": "jobqueue"} + c := fake.NewClientBuilder(). + WithScheme(testScheme(t)). + WithObjects(workspace, namespace, localQueue, clusterQueue, selectorIndependent, unrelated, foreign). + Build() + + if _, err := newTestWorkspaceReconciler(c).Reconcile(ctx, ctrl.Request{NamespacedName: client.ObjectKeyFromObject(workspace)}); err != nil { + t.Fatalf("Reconcile() error = %v", err) + } + var gotNamespace corev1.Namespace + if err := c.Get(ctx, client.ObjectKeyFromObject(namespace), &gotNamespace); err != nil { + t.Fatal(err) + } + if gotNamespace.Labels[labelWorkspace] != "" || + gotNamespace.Annotations[annotationOwnerUID] != "" { + t.Fatalf("unrelated workloads retained stale namespace metadata: labels=%#v annotations=%#v", gotNamespace.Labels, gotNamespace.Annotations) + } +} + +func TestWorkspaceDeleteFailsClosedWhenExternalQueueRoutingCannotBeVerified(t *testing.T) { + for _, test := range []struct { + name string + failGVK schema.GroupVersionKind + failKey client.ObjectKey + mutateQueue func(*unstructured.Unstructured) + wantError string + }{ + { + name: "local queue read", + failGVK: localQueueGVK, + failKey: client.ObjectKey{Name: "jobqueue", Namespace: "aurora"}, + wantError: "read LocalQueue aurora/jobqueue", + }, + { + name: "cluster queue read", + failGVK: clusterQueueGVK, + failKey: client.ObjectKey{Name: "jobqueue"}, + wantError: "read ClusterQueue jobqueue", + }, + { + name: "malformed selector", + mutateQueue: func(queue *unstructured.Unstructured) { + queue.Object["spec"] = map[string]any{ + "namespaceSelector": map[string]any{ + "matchExpressions": "invalid", + }, + } + }, + wantError: "read ClusterQueue \"jobqueue\" namespaceSelector", + }, + } { + t.Run(test.name, func(t *testing.T) { + ctx := context.Background() + workspace := testWorkspace("aurora") + workspace.Finalizers = []string{workspaceFinalizer} + workspace.Annotations = map[string]string{annotationOwnerUID: string(workspace.UID)} + now := metav1.Now() + workspace.DeletionTimestamp = &now + workspace.Spec.Target.CreateNamespace = false + workspace.Spec.Queue = "jobqueue" + workspace.Status.Target.ResolvedNamespace = workspace.Name + workspace.Status.Queue = tauv1alpha1.WorkspaceQueueStatus{ + LocalQueue: "jobqueue", + ClusterQueue: "jobqueue", + } + namespace := testNamespace(workspace.Name) + namespace.Labels = workspaceNamespaceLabels(workspace.Name, workspace.Spec.Queue) + localQueue := testLocalQueue(namespace.Name, "jobqueue", "jobqueue") + localQueue.SetAnnotations(nil) + clusterQueue := testClusterQueue("jobqueue") + clusterQueue.Object["spec"] = map[string]any{ + "namespaceSelector": map[string]any{ + "matchLabels": map[string]any{labelWorkspace: workspace.Name}, + }, + } + if test.mutateQueue != nil { + test.mutateQueue(clusterQueue) + } + workload := newQueueObject(workloadGVK) + workload.SetName("training") + workload.SetNamespace(namespace.Name) + workload.Object["spec"] = map[string]any{"queueName": localQueue.GetName()} + base := fake.NewClientBuilder(). + WithScheme(testScheme(t)). + WithObjects(workspace, namespace, localQueue, clusterQueue, workload). + Build() + reconciler := newTestWorkspaceReconciler(base) + if !test.failGVK.Empty() { + reconciler.APIReader = &objectReadFailingClient{ + Client: base, + gvk: test.failGVK, + key: test.failKey, + err: errors.New("access denied"), + } + } + + _, err := reconciler.Reconcile(ctx, ctrl.Request{NamespacedName: client.ObjectKeyFromObject(workspace)}) + if err == nil || !strings.Contains(err.Error(), test.wantError) { + t.Fatalf("Reconcile() error = %v, want %q", err, test.wantError) + } + var gotNamespace corev1.Namespace + if err := base.Get(ctx, client.ObjectKeyFromObject(namespace), &gotNamespace); err != nil { + t.Fatal(err) + } + if gotNamespace.Labels[labelWorkspace] != workspace.Name || + gotNamespace.Annotations[annotationOwnerUID] != string(workspace.UID) { + t.Fatalf("failed routing verification removed namespace metadata: labels=%#v annotations=%#v", gotNamespace.Labels, gotNamespace.Annotations) + } + }) + } +} + func TestWorkspaceDeleteWaitsForClusterQueueFinalizer(t *testing.T) { ctx := context.Background() scheme := testScheme(t) @@ -2288,9 +2817,233 @@ func TestUnmigratedWorkspaceRetargetMigratesHistoryBeforeDeletion(t *testing.T) if err := deletionClient.Get(ctx, client.ObjectKeyFromObject(newNamespace), &gotNew); err != nil { t.Fatalf("Get new desired Namespace: %v", err) } - if gotNew.Annotations[annotationOwnerUID] != string(workspace.UID) || - gotNew.Labels[labelWorkspace] != workspace.Name { - t.Fatalf("normal retarget deletion did not preserve adopted Namespace ownership: labels=%#v annotations=%#v", gotNew.Labels, gotNew.Annotations) + if gotNew.Annotations[annotationOwnerUID] != "" || + gotNew.Labels[labelWorkspace] != "" { + t.Fatalf("normal retarget deletion retained adopted Namespace ownership: labels=%#v annotations=%#v", gotNew.Labels, gotNew.Annotations) + } +} + +func TestWorkspaceRetargetPreservesOldWorkloadIdentityWhileExternalQueueWorkloadDrains(t *testing.T) { + for _, identityChange := range []string{"unchanged", "renamed", "client-id-changed", "removed", "deletion-fails", "deletion-pending"} { + t.Run(identityChange, func(t *testing.T) { + ctx := context.Background() + workspace := testWorkspace("aurora") + workspace.Finalizers = []string{workspaceFinalizer} + workspace.Annotations = map[string]string{annotationOwnerUID: string(workspace.UID)} + workspace.Spec.Target.Namespace = "new-target" + workspace.Spec.Target.CreateNamespace = false + workspace.Spec.Queue = "jobqueue" + workspace.Status.Phase = tauv1alpha1.WorkspacePhaseReady + workspace.Status.Target.ResolvedNamespace = "old-target" + workspace.Status.Queue = tauv1alpha1.WorkspaceQueueStatus{ + LocalQueue: "jobqueue", + ClusterQueue: "jobqueue", + ClusterQueueUID: "jobqueue-uid", + } + oldNamespace := testNamespace("old-target") + oldNamespace.Labels = workspaceNamespaceLabels(workspace.Name, workspace.Spec.Queue) + oldNamespace.Annotations = map[string]string{annotationOwnerUID: string(workspace.UID)} + newNamespace := testNamespace("new-target") + newNamespace.Labels = map[string]string{labelWorkspace: workspace.Name} + newNamespace.Annotations = map[string]string{annotationOwnerUID: string(workspace.UID)} + account := testServiceAccount(oldNamespace.Name, workspace.Spec.WorkloadIdentity.ServiceAccountName, workspace.Name) + account.Labels[labelAzureWIUse] = "true" + account.Annotations[annotationAzureWIClientID] = workspace.Spec.WorkloadIdentity.ClientID + switch identityChange { + case "renamed", "deletion-fails", "deletion-pending": + workspace.Spec.WorkloadIdentity.ServiceAccountName = "replacement-workload" + case "client-id-changed": + workspace.Spec.WorkloadIdentity.ClientID = "replacement-client-id" + case "removed": + workspace.Spec.WorkloadIdentity = nil + } + if identityChange == "deletion-pending" { + account.Finalizers = []string{"example.invalid/retain"} + } + deletionBlocked := identityChange == "deletion-fails" || identityChange == "deletion-pending" + binding := testRoleBinding(oldNamespace.Name, defaultRoleName, workspace.Name) + binding.Subjects = []rbacv1.Subject{{ + Kind: rbacv1.GroupKind, + APIGroup: rbacv1.GroupName, + Name: "former-researchers", + }} + job := &batchv1.Job{ + ObjectMeta: metav1.ObjectMeta{ + Name: "training", + Namespace: oldNamespace.Name, + UID: types.UID("training-job-uid"), + }, + Spec: batchv1.JobSpec{ + Template: corev1.PodTemplateSpec{ + Spec: corev1.PodSpec{ + ServiceAccountName: account.Name, + RestartPolicy: corev1.RestartPolicyNever, + Containers: []corev1.Container{{ + Name: "trainer", + Image: "example.invalid/trainer", + Command: []string{"train"}, + }}, + }, + }, + }, + } + localQueue := testLocalQueue(oldNamespace.Name, "jobqueue", "jobqueue") + localQueue.SetAnnotations(nil) + clusterQueue := clusterQueueWithUID("jobqueue", "jobqueue-uid") + clusterQueue.Object["spec"] = map[string]any{ + "namespaceSelector": map[string]any{ + "matchExpressions": []any{map[string]any{ + "key": labelWorkspace, + "operator": "Exists", + }}, + }, + } + workload := newQueueObject(workloadGVK) + workload.SetName("job-training") + workload.SetNamespace(oldNamespace.Name) + workload.SetOwnerReferences([]metav1.OwnerReference{{ + APIVersion: batchv1.SchemeGroupVersion.String(), + Kind: "Job", + Name: job.Name, + UID: job.UID, + }}) + workload.Object["spec"] = map[string]any{"queueName": localQueue.GetName()} + c := fake.NewClientBuilder(). + WithScheme(testScheme(t)). + WithObjects(workspace, oldNamespace, newNamespace, account, binding, job, localQueue, clusterQueue, workload). + WithStatusSubresource(&tauv1alpha1.TauWorkspace{}). + WithInterceptorFuncs(interceptor.Funcs{ + Create: func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.CreateOption) error { + if _, ok := obj.(*rbacv1.RoleBinding); ok && obj.GetNamespace() == oldNamespace.Name && identityChange != "unchanged" { + if err := c.Get(ctx, client.ObjectKeyFromObject(account), &corev1.ServiceAccount{}); !apierrors.IsNotFound(err) { + t.Fatal("researcher access was restored before obsolete identity revocation") + } + } + return c.Create(ctx, obj, opts...) + }, + Update: func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.UpdateOption) error { + if _, ok := obj.(*rbacv1.RoleBinding); ok && obj.GetNamespace() == oldNamespace.Name && identityChange != "unchanged" { + if err := c.Get(ctx, client.ObjectKeyFromObject(account), &corev1.ServiceAccount{}); !apierrors.IsNotFound(err) { + t.Fatal("researcher access was reassigned before obsolete identity revocation") + } + } + return c.Update(ctx, obj, opts...) + }, + Delete: func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.DeleteOption) error { + if deletionBlocked && identityChange == "deletion-fails" { + if _, ok := obj.(*corev1.ServiceAccount); ok && client.ObjectKeyFromObject(obj) == client.ObjectKeyFromObject(account) { + return errors.New("obsolete identity deletion denied") + } + } + return c.Delete(ctx, obj, opts...) + }, + }). + Build() + reconciler := newTestWorkspaceReconciler(c) + request := ctrl.Request{NamespacedName: client.ObjectKeyFromObject(workspace)} + + for attempt := 0; attempt < 3; attempt++ { + if _, err := reconciler.Reconcile(ctx, request); err != nil { + t.Fatalf("Reconcile() during retarget drain = %v", err) + } + err := c.Get(ctx, client.ObjectKeyFromObject(account), &corev1.ServiceAccount{}) + if identityChange == "unchanged" || deletionBlocked { + if err != nil { + t.Fatalf("current workload identity was removed before Workload drain: %v", err) + } + } else if !apierrors.IsNotFound(err) { + t.Fatalf("obsolete workload identity survived retarget drain: %v", err) + } + } + var retainedJob batchv1.Job + if err := c.Get(ctx, client.ObjectKeyFromObject(job), &retainedJob); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(retainedJob.Spec, job.Spec) { + t.Fatalf("retarget changed the submitted Job: %#v", retainedJob.Spec) + } + if workspace.Spec.WorkloadIdentity != nil { + var currentAccount corev1.ServiceAccount + key := client.ObjectKey{Namespace: newNamespace.Name, Name: workspace.Spec.WorkloadIdentity.ServiceAccountName} + if err := c.Get(ctx, key, ¤tAccount); err != nil { + t.Fatal(err) + } + if currentAccount.Annotations[annotationAzureWIClientID] != workspace.Spec.WorkloadIdentity.ClientID { + t.Fatalf("new target identity = %#v, want current client ID", currentAccount.Annotations) + } + } + var gotBinding rbacv1.RoleBinding + if deletionBlocked { + if err := c.Get(ctx, client.ObjectKeyFromObject(binding), &gotBinding); !apierrors.IsNotFound(err) { + t.Fatalf("old researcher binding survived incomplete identity revocation: %v", err) + } + } else { + if err := c.Get(ctx, client.ObjectKeyFromObject(binding), &gotBinding); err != nil { + t.Fatalf("Get old researcher RoleBinding: %v", err) + } + wantSubjects := []rbacv1.Subject{rbacSubject(*workspace.Spec.KubernetesSubject, oldNamespace.Name)} + if !reflect.DeepEqual(gotBinding.Subjects, wantSubjects) { + t.Fatalf("old researcher subjects = %#v, want current subject %#v", gotBinding.Subjects, wantSubjects) + } + } + var waiting tauv1alpha1.TauWorkspace + if err := c.Get(ctx, client.ObjectKeyFromObject(workspace), &waiting); err != nil { + t.Fatal(err) + } + if waiting.Status.Target.ResolvedNamespace != oldNamespace.Name { + t.Fatalf("resolved namespace = %q, want old target retained during drain", waiting.Status.Target.ResolvedNamespace) + } + if deletionBlocked { + assertCondition(t, waiting.Status.Conditions, tauv1alpha1.ConditionDriftDetected, metav1.ConditionTrue) + deletionBlocked = false + } + if identityChange == "deletion-pending" { + var retainedAccount corev1.ServiceAccount + if err := c.Get(ctx, client.ObjectKeyFromObject(account), &retainedAccount); err != nil { + t.Fatal(err) + } + retainedAccount.Finalizers = nil + if err := c.Update(ctx, &retainedAccount); err != nil { + t.Fatal(err) + } + } + if identityChange == "deletion-fails" || identityChange == "deletion-pending" { + reconcileWorkspaceTimes(t, reconciler, ctx, workspace.Name, 3) + if err := c.Get(ctx, client.ObjectKeyFromObject(binding), &gotBinding); err != nil { + t.Fatalf("researcher access was not restored after identity revocation recovered: %v", err) + } + wantSubjects := []rbacv1.Subject{rbacSubject(*workspace.Spec.KubernetesSubject, oldNamespace.Name)} + if !reflect.DeepEqual(gotBinding.Subjects, wantSubjects) { + t.Fatalf("recovered researcher subjects = %#v, want %#v", gotBinding.Subjects, wantSubjects) + } + if err := c.Get(ctx, client.ObjectKeyFromObject(account), &corev1.ServiceAccount{}); !apierrors.IsNotFound(err) { + t.Fatalf("obsolete identity survived successful recovery: %v", err) + } + if err := c.Get(ctx, client.ObjectKeyFromObject(workspace), &waiting); err != nil { + t.Fatal(err) + } + if waiting.Status.Target.ResolvedNamespace != oldNamespace.Name { + t.Fatal("recovery advanced the target before the Workload finished") + } + } + + workload.Object["status"] = map[string]any{"conditions": []any{ + map[string]any{"type": "Finished", "status": "True"}, + }} + if err := c.Update(ctx, workload); err != nil { + t.Fatal(err) + } + reconcileWorkspaceTimes(t, reconciler, ctx, workspace.Name, 3) + if err := c.Get(ctx, client.ObjectKeyFromObject(account), &corev1.ServiceAccount{}); !apierrors.IsNotFound(err) { + t.Fatalf("old workload identity survived completed retarget: %v", err) + } + if err := c.Get(ctx, client.ObjectKeyFromObject(workspace), &waiting); err != nil { + t.Fatal(err) + } + if waiting.Status.Target.ResolvedNamespace != newNamespace.Name { + t.Fatalf("resolved namespace = %q, want completed retarget %q", waiting.Status.Target.ResolvedNamespace, newNamespace.Name) + } + }) } } @@ -4613,6 +5366,25 @@ type statusUpdateFailingClient struct { updateErr error } +type objectReadFailingClient struct { + client.Client + gvk schema.GroupVersionKind + key client.ObjectKey + err error +} + +func (c *objectReadFailingClient) Get( + ctx context.Context, + key client.ObjectKey, + obj client.Object, + opts ...client.GetOption, +) error { + if obj.GetObjectKind().GroupVersionKind() == c.gvk && key == c.key { + return c.err + } + return c.Client.Get(ctx, key, obj, opts...) +} + func (c *statusUpdateFailingClient) Status() client.SubResourceWriter { return &statusUpdateFailingWriter{SubResourceWriter: c.Client.Status(), client: c} } diff --git a/controllers/tau-core/internal/controller/workspace_cleanup.go b/controllers/tau-core/internal/controller/workspace_cleanup.go index 6009f4ba0..8cb321be2 100644 --- a/controllers/tau-core/internal/controller/workspace_cleanup.go +++ b/controllers/tau-core/internal/controller/workspace_cleanup.go @@ -5,17 +5,23 @@ package controller import ( "context" + "errors" "fmt" "strings" tauv1alpha1 "github.com/Azure/taugrid/controllers/tau-core/api/v1alpha1" + appsv1 "k8s.io/api/apps/v1" + batchv1 "k8s.io/api/batch/v1" corev1 "k8s.io/api/core/v1" rbacv1 "k8s.io/api/rbac/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" "sigs.k8s.io/controller-runtime/pkg/client" ) @@ -77,20 +83,90 @@ func (r *TauWorkspaceReconciler) cleanupStaleNamespaceMetadata( workspace *tauv1alpha1.TauWorkspace, namespaceName string, ) error { + namespace, err := r.workspaceNamespaceForMetadataCleanup(ctx, workspace, namespaceName) + if err != nil || namespace == nil { + return err + } + inUse, message, err := r.namespaceRoutingMetadataInUse(ctx, workspace, namespace) + if err != nil { + return err + } + if inUse { + return errors.New(message) + } + for _, key := range []string{ + labelManagedBy, + labelTeam, + labelWorkspace, + labelWorkspaceLocalQueue, + labelKueueDefaultLocalQueue, + } { + delete(namespace.Labels, key) + } + if namespace.Annotations != nil { + delete(namespace.Annotations, annotationResultScope) + delete(namespace.Annotations, annotationOwnerUID) + } + return r.Update(ctx, namespace) +} + +func (r *TauWorkspaceReconciler) workspaceNamespaceForMetadataCleanup( + ctx context.Context, + workspace *tauv1alpha1.TauWorkspace, + namespaceName string, +) (*corev1.Namespace, error) { if reservedNamespaceReason(namespaceName, systemNamespace(r.SystemNamespace)) != "" { - return nil + return nil, nil } var namespace corev1.Namespace - if err := r.Get(ctx, client.ObjectKey{Name: namespaceName}, &namespace); err != nil { + if err := r.apiReader().Get(ctx, client.ObjectKey{Name: namespaceName}, &namespace); err != nil { if apierrors.IsNotFound(err) { - return nil + return nil, nil } - return err + return nil, err } if namespace.Labels[labelWorkspace] != workspace.Name || namespace.Annotations[annotationOwnerUID] != string(workspace.UID) || workspace.UID == "" { - return nil + return nil, nil + } + return &namespace, nil +} + +func (r *TauWorkspaceReconciler) waitForNamespaceRoutingMetadataDrain( + ctx context.Context, + workspace *tauv1alpha1.TauWorkspace, + namespaceName string, +) error { + namespace, err := r.workspaceNamespaceForMetadataCleanup(ctx, workspace, namespaceName) + if err != nil || namespace == nil { + return err + } + inUse, message, err := r.namespaceRoutingMetadataInUse(ctx, workspace, namespace) + if err != nil { + return err + } + if inUse { + return errors.New(message) + } + return nil +} + +func (r *TauWorkspaceReconciler) namespaceRoutingMetadataInUse( + ctx context.Context, + workspace *tauv1alpha1.TauWorkspace, + namespace *corev1.Namespace, +) (bool, string, error) { + workloads := &unstructured.UnstructuredList{} + workloads.SetGroupVersionKind(schema.GroupVersionKind{ + Group: workloadGVK.Group, Version: workloadGVK.Version, Kind: workloadGVK.Kind + "List", + }) + if err := r.apiReader().List(ctx, workloads, client.InNamespace(namespace.Name)); err != nil { + return false, "", fmt.Errorf("list Workloads in namespace %q before routing metadata cleanup: %w", namespace.Name, err) + } + cleanedLabels := make(map[string]string, len(namespace.Labels)) + for key, value := range namespace.Labels { + cleanedLabels[key] = value } for _, key := range []string{ labelManagedBy, @@ -99,16 +175,222 @@ func (r *TauWorkspaceReconciler) cleanupStaleNamespaceMetadata( labelWorkspaceLocalQueue, labelKueueDefaultLocalQueue, } { - delete(namespace.Labels, key) + delete(cleanedLabels, key) } - if namespace.Annotations != nil { - delete(namespace.Annotations, annotationResultScope) - delete(namespace.Annotations, annotationOwnerUID) + for i := range workloads.Items { + workload := &workloads.Items[i] + if workloadFinished(workload) { + continue + } + queueName, found, err := unstructured.NestedString(workload.Object, "spec", "queueName") + if err != nil { + return false, "", fmt.Errorf("read Workload %s queueName: %w", client.ObjectKeyFromObject(workload), err) + } + queueName = strings.TrimSpace(queueName) + if !found || queueName == "" { + continue + } + admittedClusterQueue, _, err := unstructured.NestedString(workload.Object, "status", "admission", "clusterQueue") + if err != nil { + return false, "", fmt.Errorf("read Workload %s admitted ClusterQueue: %w", client.ObjectKeyFromObject(workload), err) + } + inUse, message, err := r.queueRoutingMetadataInUse( + ctx, workspace, namespace, cleanedLabels, queueName, admittedClusterQueue, + fmt.Sprintf("Workload %s", client.ObjectKeyFromObject(workload)), + ) + if err != nil || inUse { + return inUse, message, err + } + } + var jobs batchv1.JobList + if err := r.apiReader().List(ctx, &jobs, client.InNamespace(namespace.Name)); err != nil { + return false, "", fmt.Errorf("list Jobs in namespace %q before routing metadata cleanup: %w", namespace.Name, err) + } + workloadJobs := make(map[types.UID]string) + workloadRayJobs := make(map[types.UID]string) + for i := range workloads.Items { + for _, owner := range workloads.Items[i].GetOwnerReferences() { + if owner.APIVersion == batchv1.SchemeGroupVersion.String() && owner.Kind == "Job" && owner.UID != "" { + workloadJobs[owner.UID] = owner.Name + } + if owner.APIVersion == "ray.io/v1" && owner.Kind == "RayJob" && owner.UID != "" { + workloadRayJobs[owner.UID] = owner.Name + } + } + } + for i := range jobs.Items { + job := &jobs.Items[i] + if job.UID != "" && workloadJobs[job.UID] == job.Name { + continue + } + finished := false + for _, condition := range job.Status.Conditions { + if condition.Status == corev1.ConditionTrue && + (condition.Type == batchv1.JobComplete || condition.Type == batchv1.JobFailed) { + finished = true + break + } + } + queueName := strings.TrimSpace(job.Labels["kueue.x-k8s.io/queue-name"]) + if finished || queueName == "" { + continue + } + inUse, message, err := r.queueRoutingMetadataInUse( + ctx, workspace, namespace, cleanedLabels, queueName, "", + fmt.Sprintf("Job %s", client.ObjectKeyFromObject(job)), + ) + if err != nil || inUse { + return inUse, message, err + } + } + var deployments appsv1.DeploymentList + if err := r.apiReader().List(ctx, &deployments, client.InNamespace(namespace.Name)); err != nil { + return false, "", fmt.Errorf("list Deployments in namespace %q before routing metadata cleanup: %w", namespace.Name, err) + } + for i := range deployments.Items { + deployment := &deployments.Items[i] + queueName := strings.TrimSpace(deployment.Spec.Template.Labels["kueue.x-k8s.io/queue-name"]) + if deployment.Spec.Template.Labels["kueue.x-k8s.io/managed"] != "true" || queueName == "" { + continue + } + inUse, message, err := r.queueRoutingMetadataInUse( + ctx, workspace, namespace, cleanedLabels, queueName, "", + fmt.Sprintf("Deployment %s", client.ObjectKeyFromObject(deployment)), + ) + if err != nil || inUse { + return inUse, message, err + } + } + for _, kind := range []string{"RayJob", "RayService"} { + sources := &unstructured.UnstructuredList{} + sources.SetGroupVersionKind(schema.GroupVersionKind{Group: "ray.io", Version: "v1", Kind: kind + "List"}) + if err := r.apiReader().List(ctx, sources, client.InNamespace(namespace.Name)); err != nil { + if meta.IsNoMatchError(err) || apierrors.IsNotFound(err) { + continue + } + return false, "", fmt.Errorf("list %ss in namespace %q before routing metadata cleanup: %w", kind, namespace.Name, err) + } + for i := range sources.Items { + source := &sources.Items[i] + queueName := strings.TrimSpace(source.GetLabels()["kueue.x-k8s.io/queue-name"]) + if queueName == "" { + continue + } + if kind == "RayJob" { + if source.GetUID() != "" && workloadRayJobs[source.GetUID()] == source.GetName() { + continue + } + finished, err := sourceRayJobFinished(source) + if err != nil { + return false, "", err + } + if finished { + continue + } + } + inUse, message, err := r.queueRoutingMetadataInUse( + ctx, workspace, namespace, cleanedLabels, queueName, "", + fmt.Sprintf("%s %s", kind, client.ObjectKeyFromObject(source)), + ) + if err != nil || inUse { + return inUse, message, err + } + } + } + return false, "", nil +} + +func sourceRayJobFinished(rayJob *unstructured.Unstructured) (bool, error) { + value, _, err := unstructured.NestedString(rayJob.Object, "status", "jobDeploymentStatus") + if err != nil { + return false, fmt.Errorf("read RayJob %s jobDeploymentStatus before routing metadata cleanup: %w", client.ObjectKeyFromObject(rayJob), err) } - return r.Update(ctx, &namespace) + return value == "Complete" || value == "Failed", nil +} + +func (r *TauWorkspaceReconciler) queueRoutingMetadataInUse( + ctx context.Context, + workspace *tauv1alpha1.TauWorkspace, + namespace *corev1.Namespace, + cleanedLabels map[string]string, + queueName, admittedClusterQueue, source string, +) (bool, string, error) { + localQueue := newQueueObject(localQueueGVK) + localQueueKey := client.ObjectKey{Name: queueName, Namespace: namespace.Name} + if err := r.apiReader().Get(ctx, localQueueKey, localQueue); err != nil { + if apierrors.IsNotFound(err) { + return false, "", nil + } + return false, "", fmt.Errorf("read LocalQueue %s before routing metadata cleanup: %w", localQueueKey, err) + } + clusterQueueName, found, err := unstructured.NestedString(localQueue.Object, "spec", "clusterQueue") + if err != nil { + return false, "", fmt.Errorf("read LocalQueue %s backing ClusterQueue: %w", localQueueKey, err) + } + if !found || strings.TrimSpace(clusterQueueName) == "" { + return false, "", fmt.Errorf("LocalQueue %s has no backing ClusterQueue", localQueueKey) + } + if strings.TrimSpace(admittedClusterQueue) != "" { + clusterQueueName = admittedClusterQueue + } + clusterQueue := newQueueObject(clusterQueueGVK) + if err := r.apiReader().Get(ctx, client.ObjectKey{Name: clusterQueueName}, clusterQueue); err != nil { + return false, "", fmt.Errorf("read ClusterQueue %s before routing metadata cleanup: %w", clusterQueueName, err) + } + if expectedUID := strings.TrimSpace(workspace.Status.Queue.ClusterQueueUID); expectedUID != "" && + clusterQueueName == strings.TrimSpace(workspace.Status.Queue.ClusterQueue) && + string(clusterQueue.GetUID()) != expectedUID { + return false, "", fmt.Errorf( + "ClusterQueue %q UID changed from %q to %q before routing metadata cleanup", + clusterQueueName, + expectedUID, + clusterQueue.GetUID(), + ) + } + selectorMap, found, err := unstructured.NestedMap(clusterQueue.Object, "spec", "namespaceSelector") + if err != nil { + return false, "", fmt.Errorf("read ClusterQueue %q namespaceSelector: %w", clusterQueueName, err) + } + labelSelector := metav1.LabelSelector{} + if found { + if err := runtime.DefaultUnstructuredConverter.FromUnstructured(selectorMap, &labelSelector); err != nil { + return false, "", fmt.Errorf("read ClusterQueue %q namespaceSelector: %w", clusterQueueName, err) + } + } + selector, err := metav1.LabelSelectorAsSelector(&labelSelector) + if err != nil { + return false, "", fmt.Errorf("read ClusterQueue %q namespaceSelector: %w", clusterQueueName, err) + } + if !selector.Matches(labels.Set(namespace.Labels)) { + return false, "", fmt.Errorf( + "ClusterQueue %q namespaceSelector does not match namespace %q for unfinished %s", + clusterQueueName, + namespace.Name, + source, + ) + } + if selector.Matches(labels.Set(cleanedLabels)) { + return false, "", nil + } + return true, fmt.Sprintf( + "waiting for %s using retained LocalQueue %q and ClusterQueue %q before removing namespace routing metadata", + source, + queueName, + clusterQueueName, + ), nil } func (r *TauWorkspaceReconciler) cleanupWorkspaceAccess(ctx context.Context, workspace *tauv1alpha1.TauWorkspace) error { + targetNamespace := strings.TrimSpace(workspace.Status.Target.ResolvedNamespace) + if targetNamespace == "" { + targetNamespace = resolvedNamespace(workspace) + } + if err := r.cleanupOwnedResearcherRoleBindings(ctx, workspace); err != nil { + return err + } + if err := r.waitForNamespaceRoutingMetadataDrain(ctx, workspace, targetNamespace); err != nil { + return err + } if err := r.cleanupStaleTargetRBAC(ctx, workspace, "", "", false); err != nil { return err } @@ -121,7 +403,10 @@ func (r *TauWorkspaceReconciler) cleanupWorkspaceAccess(ctx context.Context, wor if err := r.cleanupWorkspaceClusterQueue(ctx, workspace); err != nil { return err } - return r.cleanupSystemReaderRBAC(ctx, workspace) + if err := r.cleanupSystemReaderRBAC(ctx, workspace); err != nil { + return err + } + return r.cleanupStaleNamespaceMetadata(ctx, workspace, targetNamespace) } func (r *TauWorkspaceReconciler) cleanupWorkspaceClusterQueue(ctx context.Context, workspace *tauv1alpha1.TauWorkspace) error { @@ -234,10 +519,10 @@ func (r *TauWorkspaceReconciler) cleanupStaleTargetRBAC( return nil } -func (r *TauWorkspaceReconciler) cleanupOwnedNamespaceAccess( +func (r *TauWorkspaceReconciler) cleanupOwnedNamespaceWorkloadIdentity( ctx context.Context, workspace *tauv1alpha1.TauWorkspace, - targetNamespace, keepServiceAccount string, + keepServiceAccount string, ) error { var namespaces corev1.NamespaceList if err := r.List(ctx, &namespaces, client.MatchingLabels{ @@ -260,7 +545,7 @@ func (r *TauWorkspaceReconciler) cleanupOwnedNamespaceAccess( return err } } - return r.reconcileOwnedNamespaceResearcherAccess(ctx, workspace, targetNamespace) + return nil } func canDeleteLegacyWorkspaceRole(role *rbacv1.Role, workspace *tauv1alpha1.TauWorkspace) bool { @@ -413,6 +698,18 @@ func (r *TauWorkspaceReconciler) cleanupStaleWorkloadIdentityServiceAccounts( if err := r.deleteWithPreconditions(ctx, account); err != nil { return err } + var remaining corev1.ServiceAccount + if err := r.apiReader().Get(ctx, client.ObjectKeyFromObject(account), &remaining); err != nil { + if apierrors.IsNotFound(err) { + continue + } + return err + } + return fmt.Errorf( + "waiting for obsolete workload identity ServiceAccount %s deletion to complete; finalizers=%v", + client.ObjectKeyFromObject(&remaining), + remaining.Finalizers, + ) } return nil } diff --git a/controllers/tau-core/internal/controller/workspace_cleanup_test.go b/controllers/tau-core/internal/controller/workspace_cleanup_test.go new file mode 100644 index 000000000..7baa3b173 --- /dev/null +++ b/controllers/tau-core/internal/controller/workspace_cleanup_test.go @@ -0,0 +1,592 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +package controller + +import ( + "context" + "errors" + "reflect" + "strings" + "testing" + + tauv1alpha1 "github.com/Azure/taugrid/controllers/tau-core/api/v1alpha1" + appsv1 "k8s.io/api/apps/v1" + batchv1 "k8s.io/api/batch/v1" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" + "k8s.io/utils/ptr" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" +) + +func TestWorkspacePreservesRoutingForJobBeforeWorkloadCreation(t *testing.T) { + for _, kind := range []string{"Job", "RayJob", "RayService", "Deployment"} { + t.Run(kind, func(t *testing.T) { + for _, lifecycle := range []string{"delete", "retarget"} { + t.Run(lifecycle, func(t *testing.T) { + ctx := context.Background() + workspace := testWorkspace("aurora") + workspace.Finalizers = []string{workspaceFinalizer} + workspace.Annotations = map[string]string{annotationOwnerUID: string(workspace.UID)} + now := metav1.Now() + if lifecycle == "delete" { + workspace.DeletionTimestamp = &now + } + workspace.Spec.Target.CreateNamespace = false + workspace.Spec.Queue = "jobqueue" + workspace.Status.Target.ResolvedNamespace = workspace.Name + namespace := testNamespace(workspace.Name) + namespace.Labels = workspaceNamespaceLabels(workspace.Name, workspace.Spec.Queue) + account := testServiceAccount(namespace.Name, workspace.Spec.WorkloadIdentity.ServiceAccountName, workspace.Name) + account.Annotations[annotationAzureWIClientID] = workspace.Spec.WorkloadIdentity.ClientID + newNamespace := testNamespace("new-target") + newNamespace.Labels = map[string]string{labelWorkspace: workspace.Name} + newNamespace.Annotations = map[string]string{annotationOwnerUID: string(workspace.UID)} + if lifecycle == "retarget" { + workspace.Spec.Target.Namespace = newNamespace.Name + } + job := &batchv1.Job{ + ObjectMeta: metav1.ObjectMeta{ + Name: "training", Namespace: namespace.Name, UID: types.UID("training-job-uid"), + Labels: map[string]string{"kueue.x-k8s.io/queue-name": "jobqueue"}, + }, + Spec: batchv1.JobSpec{ + Suspend: ptr.To(true), + Template: corev1.PodTemplateSpec{Spec: corev1.PodSpec{ + ServiceAccountName: account.Name, + RestartPolicy: corev1.RestartPolicyNever, + Containers: []corev1.Container{{Name: "trainer", Image: "example.invalid/trainer"}}, + }}, + }, + } + var source client.Object = job + sourceAPIVersion := batchv1.SchemeGroupVersion.String() + if kind == "Deployment" { + sourceAPIVersion = appsv1.SchemeGroupVersion.String() + source = &appsv1.Deployment{ + ObjectMeta: job.ObjectMeta, + Spec: appsv1.DeploymentSpec{ + Replicas: ptr.To(int32(1)), + Selector: &metav1.LabelSelector{MatchLabels: map[string]string{"app": job.Name}}, + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{ + Labels: map[string]string{ + "app": job.Name, "kueue.x-k8s.io/queue-name": "jobqueue", + "kueue.x-k8s.io/managed": "true", + }, + Annotations: map[string]string{"kueue.x-k8s.io/pod-suspending-parent": "deployment"}, + }, + Spec: job.Spec.Template.Spec, + }, + }, + } + } else if kind != "Job" { + sourceAPIVersion = "ray.io/v1" + rayJob := newQueueObject(schema.GroupVersionKind{Group: "ray.io", Version: "v1", Kind: kind}) + rayJob.SetName(job.Name) + rayJob.SetNamespace(job.Namespace) + rayJob.SetUID(job.UID) + rayJob.SetLabels(job.Labels) + rayJob.Object["spec"] = map[string]any{ + "suspend": true, "entrypoint": "python train.py", + "shutdownAfterJobFinishes": true, + "rayClusterSpec": map[string]any{"headGroupSpec": map[string]any{ + "template": map[string]any{"spec": map[string]any{"serviceAccountName": account.Name}}, + }}, + } + if kind == "RayService" { + rayJob.Object["spec"] = map[string]any{ + "serveConfigV2": "applications:\n - name: default\n import_path: serve:app\n", + "rayClusterConfig": map[string]any{ + "headGroupSpec": map[string]any{ + "template": map[string]any{"spec": map[string]any{"serviceAccountName": account.Name}}, + }, + }, + } + } + source = rayJob + } + localQueue := testLocalQueue(namespace.Name, "jobqueue", "jobqueue") + localQueue.SetAnnotations(nil) + clusterQueue := clusterQueueWithUID("jobqueue", "jobqueue-uid") + clusterQueue.Object["spec"] = map[string]any{"namespaceSelector": map[string]any{ + "matchLabels": map[string]any{labelWorkspace: workspace.Name}, + }} + c := fake.NewClientBuilder().WithScheme(testScheme(t)). + WithObjects(workspace, namespace, newNamespace, account, source, localQueue, clusterQueue). + WithStatusSubresource(&tauv1alpha1.TauWorkspace{}).Build() + reconciler := newTestWorkspaceReconciler(c) + request := ctrl.Request{NamespacedName: client.ObjectKeyFromObject(workspace)} + for range 3 { + _, err := reconciler.Reconcile(ctx, request) + if lifecycle == "delete" && (err == nil || + !strings.Contains(err.Error(), "before removing namespace routing metadata")) { + t.Fatalf("Reconcile() = %v, want source Job drain wait", err) + } + if lifecycle == "retarget" && err != nil { + t.Fatal(err) + } + var waiting tauv1alpha1.TauWorkspace + if err := c.Get(ctx, client.ObjectKeyFromObject(workspace), &waiting); err != nil { + t.Fatal(err) + } + if waiting.Status.Target.ResolvedNamespace != namespace.Name { + t.Fatal("resolved target advanced before source Job drain") + } + if err := c.Get(ctx, client.ObjectKeyFromObject(account), &corev1.ServiceAccount{}); err != nil { + t.Fatalf("execution identity removed before source Job drain: %v", err) + } + var retained corev1.Namespace + if err := c.Get(ctx, client.ObjectKeyFromObject(namespace), &retained); err != nil { + t.Fatal(err) + } + if retained.Labels[labelWorkspace] != workspace.Name { + t.Fatal("source Job lost namespace routing metadata") + } + } + workload := newQueueObject(workloadGVK) + workload.SetName("job-training") + workload.SetNamespace(namespace.Name) + workload.SetOwnerReferences([]metav1.OwnerReference{{ + APIVersion: sourceAPIVersion, Kind: kind, Name: source.GetName(), UID: source.GetUID(), + }}) + if kind == "Deployment" { + workload.SetOwnerReferences([]metav1.OwnerReference{{ + APIVersion: "v1", Kind: "Pod", Name: "service-pod", UID: types.UID("service-pod-uid"), + }}) + } + workload.Object["spec"] = map[string]any{"queueName": localQueue.GetName()} + if err := c.Create(ctx, workload); err != nil { + t.Fatal(err) + } + if _, err := reconciler.Reconcile(ctx, request); lifecycle == "delete" && err == nil { + t.Fatal("workspace stopped draining after Kueue created the Workload") + } + workload.Object["status"] = map[string]any{"conditions": []any{ + map[string]any{"type": "Finished", "status": "True"}, + }} + if err := c.Update(ctx, workload); err != nil { + t.Fatal(err) + } + if kind == "RayService" || kind == "Deployment" { + _, err := reconciler.Reconcile(ctx, request) + if lifecycle == "delete" && err == nil { + t.Fatalf("live %s stopped retaining routing after its Workload finished", kind) + } + if lifecycle == "retarget" && err != nil { + t.Fatal(err) + } + if err := c.Get(ctx, client.ObjectKeyFromObject(account), &corev1.ServiceAccount{}); err != nil { + t.Fatalf("live %s lost execution identity: %v", kind, err) + } + if err := c.Delete(ctx, source); err != nil { + t.Fatal(err) + } + } + for range 3 { + if _, err := reconciler.Reconcile(ctx, request); err != nil { + t.Fatalf("Reconcile() after drain = %v", err) + } + } + if err := c.Get(ctx, client.ObjectKeyFromObject(account), &corev1.ServiceAccount{}); !apierrors.IsNotFound(err) { + t.Fatalf("execution account survived drain: %v", err) + } + var finished tauv1alpha1.TauWorkspace + err := c.Get(ctx, client.ObjectKeyFromObject(workspace), &finished) + if lifecycle == "delete" && !apierrors.IsNotFound(err) { + t.Fatalf("workspace survived completed source Job drain: %v", err) + } + if lifecycle == "retarget" && (err != nil || finished.Status.Target.ResolvedNamespace != newNamespace.Name) { + t.Fatalf("retarget did not complete after source Job drain: %v, target=%q", err, finished.Status.Target.ResolvedNamespace) + } + }) + } + }) + } +} + +func TestNamespaceRoutingMetadataSourceDeployments(t *testing.T) { + for _, scenario := range []string{ + "managed", "unmanaged", "missing-managed-label", "root-label-only", "template-label-only", + "template-queue-differs", "zero-replicas", "paused", "deleting", "other-namespace", + "missing-queue", "selector-independent", "list-forbidden", "queue-read-error", + } { + t.Run(scenario, func(t *testing.T) { + ctx := context.Background() + workspace := testWorkspace("aurora") + namespace := testNamespace(workspace.Name) + namespace.Labels = workspaceNamespaceLabels(workspace.Name, "jobqueue") + deployment := &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Name: "service", Namespace: namespace.Name, + Labels: map[string]string{"kueue.x-k8s.io/queue-name": "jobqueue"}, + }, + Spec: appsv1.DeploymentSpec{ + Replicas: ptr.To(int32(1)), + Template: corev1.PodTemplateSpec{ObjectMeta: metav1.ObjectMeta{ + Labels: map[string]string{ + "kueue.x-k8s.io/managed": "true", "kueue.x-k8s.io/queue-name": "jobqueue", + }, + }}, + }, + } + localQueue := testLocalQueue(namespace.Name, "jobqueue", "jobqueue") + localQueue.SetAnnotations(nil) + clusterQueue := clusterQueueWithUID("jobqueue", "jobqueue-uid") + clusterQueue.Object["spec"] = map[string]any{"namespaceSelector": map[string]any{ + "matchLabels": map[string]any{labelWorkspace: workspace.Name}, + }} + wantInUse, wantError := true, false + switch scenario { + case "unmanaged": + deployment.Spec.Template.Labels["kueue.x-k8s.io/managed"] = "false" + wantInUse = false + case "missing-managed-label": + delete(deployment.Spec.Template.Labels, "kueue.x-k8s.io/managed") + wantInUse = false + case "root-label-only": + delete(deployment.Spec.Template.Labels, "kueue.x-k8s.io/queue-name") + wantInUse = false + case "template-label-only": + deployment.Labels = nil + case "template-queue-differs": + deployment.Labels["kueue.x-k8s.io/queue-name"] = "unrelated" + case "zero-replicas": + deployment.Spec.Replicas = ptr.To(int32(0)) + case "paused": + deployment.Spec.Paused = true + case "deleting": + now := metav1.Now() + deployment.DeletionTimestamp = &now + deployment.Finalizers = []string{"example.invalid/drain"} + case "other-namespace": + deployment.Namespace = "unrelated" + wantInUse = false + case "missing-queue": + deployment.Spec.Template.Labels["kueue.x-k8s.io/queue-name"] = "missing" + wantInUse = false + case "selector-independent": + clusterQueue.Object["spec"] = map[string]any{"namespaceSelector": map[string]any{}} + wantInUse = false + case "list-forbidden", "queue-read-error": + wantInUse, wantError = false, true + } + baseClient := fake.NewClientBuilder().WithScheme(testScheme(t)). + WithObjects(namespace, deployment, localQueue, clusterQueue).Build() + c := interceptor.NewClient(baseClient, interceptor.Funcs{ + List: func(ctx context.Context, c client.WithWatch, list client.ObjectList, opts ...client.ListOption) error { + if _, ok := list.(*appsv1.DeploymentList); ok && scenario == "list-forbidden" { + return apierrors.NewForbidden(schema.GroupResource{Group: "apps", Resource: "deployments"}, "", errors.New("discovery denied")) + } + return c.List(ctx, list, opts...) + }, + Get: func(ctx context.Context, c client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { + if scenario == "queue-read-error" && obj.GetObjectKind().GroupVersionKind() == localQueueGVK { + return errors.New("queue unavailable") + } + return c.Get(ctx, key, obj, opts...) + }, + }) + inUse, message, err := newTestWorkspaceReconciler(c).namespaceRoutingMetadataInUse(ctx, workspace, namespace) + if inUse != wantInUse || (err != nil) != wantError { + t.Fatalf("namespaceRoutingMetadataInUse() = %v, %q, %v; want inUse=%v error=%v", inUse, message, err, wantInUse, wantError) + } + if inUse && !strings.Contains(message, "Deployment "+client.ObjectKeyFromObject(deployment).String()) { + t.Fatalf("drain message does not identify source Deployment: %q", message) + } + }) + } +} + +func TestNamespaceRoutingMetadataSourceRayResources(t *testing.T) { + for _, kind := range []string{"RayJob", "RayService"} { + t.Run(kind, func(t *testing.T) { + for _, scenario := range []string{ + "suspended", "running", "complete", "failed", "retrying", "succeeded-driver", + "unlabeled", "other-namespace", "missing-queue", "selector-independent", + "workload-finished", "workload-wrong-uid", "workload-wrong-name", "workload-wrong-kind", + "api-not-installed", "api-not-found", "list-forbidden", "list-error", + "malformed-status", "queue-read-error", "invalid-selector", + "rayjob-api-not-installed", "rayjob-api-not-found", + } { + t.Run(scenario, func(t *testing.T) { + ctx := context.Background() + workspace := testWorkspace("aurora") + namespace := testNamespace(workspace.Name) + namespace.Labels = workspaceNamespaceLabels(workspace.Name, "jobqueue") + rayJob := newQueueObject(schema.GroupVersionKind{Group: "ray.io", Version: "v1", Kind: kind}) + rayJob.SetName("training") + rayJob.SetNamespace(namespace.Name) + rayJob.SetUID(types.UID("rayjob-uid")) + rayJob.SetLabels(map[string]string{"kueue.x-k8s.io/queue-name": "jobqueue"}) + if kind == "RayService" { + rayJob.Object["spec"] = map[string]any{"rayClusterConfig": map[string]any{}} + } else { + rayJob.Object["spec"] = map[string]any{"suspend": true} + } + localQueue := testLocalQueue(namespace.Name, "jobqueue", "jobqueue") + localQueue.SetAnnotations(nil) + clusterQueue := clusterQueueWithUID("jobqueue", "jobqueue-uid") + clusterQueue.Object["spec"] = map[string]any{"namespaceSelector": map[string]any{ + "matchLabels": map[string]any{labelWorkspace: workspace.Name}, + }} + workload := newQueueObject(workloadGVK) + workload.SetName("rayjob-training") + workload.SetNamespace(namespace.Name) + workload.SetOwnerReferences([]metav1.OwnerReference{{ + APIVersion: "ray.io/v1", Kind: kind, Name: rayJob.GetName(), UID: rayJob.GetUID(), + }}) + workload.Object["spec"] = map[string]any{"queueName": "jobqueue"} + workload.Object["status"] = map[string]any{"conditions": []any{ + map[string]any{"type": "Finished", "status": "True"}, + }} + objects := []client.Object{namespace, rayJob, localQueue, clusterQueue} + wantInUse, wantError := true, false + switch scenario { + case "running": + rayJob.Object["spec"] = map[string]any{"suspend": false} + rayJob.Object["status"] = map[string]any{"jobDeploymentStatus": "Running", "jobStatus": "RUNNING"} + case "complete", "failed": + state := "Complete" + if scenario == "failed" { + state = "Failed" + } + rayJob.Object["status"] = map[string]any{"jobDeploymentStatus": state} + wantInUse = kind == "RayService" + case "retrying": + rayJob.Object["status"] = map[string]any{"jobDeploymentStatus": "Retrying", "jobStatus": "FAILED"} + case "succeeded-driver": + rayJob.Object["status"] = map[string]any{"jobDeploymentStatus": "Running", "jobStatus": "SUCCEEDED"} + case "unlabeled": + rayJob.SetLabels(nil) + wantInUse = false + case "other-namespace": + rayJob.SetNamespace("unrelated") + wantInUse = false + case "missing-queue": + rayJob.SetLabels(map[string]string{"kueue.x-k8s.io/queue-name": "missing"}) + wantInUse = false + case "selector-independent": + clusterQueue.Object["spec"] = map[string]any{"namespaceSelector": map[string]any{}} + wantInUse = false + case "workload-finished", "workload-wrong-uid", "workload-wrong-name", "workload-wrong-kind": + owner := workload.GetOwnerReferences()[0] + switch scenario { + case "workload-finished": + wantInUse = kind == "RayService" + case "workload-wrong-uid": + owner.UID = types.UID("another-uid") + case "workload-wrong-name": + owner.Name = "another-job" + case "workload-wrong-kind": + owner.Kind = "RayService" + if kind == "RayService" { + owner.Kind = "RayJob" + } + } + workload.SetOwnerReferences([]metav1.OwnerReference{owner}) + objects = append(objects, workload) + case "api-not-installed", "api-not-found": + wantInUse = false + case "list-forbidden", "list-error", "queue-read-error": + wantInUse, wantError = false, true + case "malformed-status": + rayJob.Object["status"] = map[string]any{"jobDeploymentStatus": int64(42)} + wantInUse, wantError = kind == "RayService", kind == "RayJob" + case "rayjob-api-not-installed", "rayjob-api-not-found": + wantInUse = kind == "RayService" + case "invalid-selector": + clusterQueue.Object["spec"] = map[string]any{"namespaceSelector": map[string]any{ + "matchExpressions": []any{map[string]any{"key": labelWorkspace, "operator": "Invalid"}}, + }} + wantInUse, wantError = false, true + } + baseClient := fake.NewClientBuilder().WithScheme(testScheme(t)).WithObjects(objects...).Build() + c := interceptor.NewClient(baseClient, interceptor.Funcs{ + List: func(ctx context.Context, c client.WithWatch, list client.ObjectList, opts ...client.ListOption) error { + listKind := list.GetObjectKind().GroupVersionKind().Kind + if listKind == "RayJobList" { + if scenario == "rayjob-api-not-installed" { + return &meta.NoKindMatchError{GroupKind: schema.GroupKind{Group: "ray.io", Kind: "RayJob"}} + } + if scenario == "rayjob-api-not-found" { + return apierrors.NewNotFound(schema.GroupResource{Group: "ray.io", Resource: "rayjobs"}, "") + } + } + if listKind == kind+"List" { + resource := schema.GroupResource{Group: "ray.io", Resource: strings.ToLower(kind) + "s"} + switch scenario { + case "api-not-installed": + return &meta.NoKindMatchError{GroupKind: schema.GroupKind{Group: "ray.io", Kind: kind}} + case "api-not-found": + return apierrors.NewNotFound(resource, "") + case "list-forbidden": + return apierrors.NewForbidden(resource, "", errors.New("RayJob discovery denied")) + case "list-error": + return errors.New("RayJob discovery unavailable") + } + } + return c.List(ctx, list, opts...) + }, + Get: func(ctx context.Context, c client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { + if scenario == "queue-read-error" && obj.GetObjectKind().GroupVersionKind() == localQueueGVK { + return errors.New("queue unavailable") + } + return c.Get(ctx, key, obj, opts...) + }, + }) + inUse, message, err := newTestWorkspaceReconciler(c).namespaceRoutingMetadataInUse(ctx, workspace, namespace) + if inUse != wantInUse || (err != nil) != wantError { + t.Fatalf("namespaceRoutingMetadataInUse() = %v, %q, %v; want inUse=%v error=%v", inUse, message, err, wantInUse, wantError) + } + if inUse && !strings.Contains(message, kind+" "+client.ObjectKeyFromObject(rayJob).String()) { + t.Fatalf("drain message does not identify the source %s: %q", kind, message) + } + for _, queue := range []*unstructured.Unstructured{localQueue, clusterQueue} { + got := newQueueObject(queue.GroupVersionKind()) + if err := baseClient.Get(ctx, client.ObjectKeyFromObject(queue), got); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(got.Object, queue.Object) { + t.Fatalf("external queue was mutated: %#v", got.Object) + } + } + }) + } + }) + } +} + +func TestNamespaceRoutingMetadataSourceJobs(t *testing.T) { + for _, scenario := range []string{ + "suspended", "running", "complete", "failed", "false-completion", + "unlabeled", "other-namespace", "missing-queue", "selector-independent", + "workload-finished", "workload-wrong-uid", "workload-wrong-name", + "job-list-error", "queue-read-error", "cluster-queue-read-error", + "invalid-selector", "cluster-queue-uid-changed", + } { + t.Run(scenario, func(t *testing.T) { + ctx := context.Background() + workspace := testWorkspace("aurora") + workspace.Status.Queue = tauv1alpha1.WorkspaceQueueStatus{ + ClusterQueue: "jobqueue", ClusterQueueUID: "jobqueue-uid", + } + namespace := testNamespace(workspace.Name) + namespace.Labels = workspaceNamespaceLabels(workspace.Name, "jobqueue") + job := &batchv1.Job{ObjectMeta: metav1.ObjectMeta{ + Name: "training", Namespace: namespace.Name, UID: types.UID("training-job-uid"), + Labels: map[string]string{"kueue.x-k8s.io/queue-name": "jobqueue"}, + }, Spec: batchv1.JobSpec{Suspend: ptr.To(true)}} + localQueue := testLocalQueue(namespace.Name, "jobqueue", "jobqueue") + localQueue.SetAnnotations(nil) + clusterQueue := clusterQueueWithUID("jobqueue", "jobqueue-uid") + clusterQueue.Object["spec"] = map[string]any{"namespaceSelector": map[string]any{ + "matchLabels": map[string]any{labelWorkspace: workspace.Name}, + }} + workload := newQueueObject(workloadGVK) + workload.SetName("job-training") + workload.SetNamespace(namespace.Name) + workload.SetOwnerReferences([]metav1.OwnerReference{{ + APIVersion: batchv1.SchemeGroupVersion.String(), Kind: "Job", Name: job.Name, UID: job.UID, + }}) + workload.Object["spec"] = map[string]any{"queueName": "jobqueue"} + workload.Object["status"] = map[string]any{"conditions": []any{ + map[string]any{"type": "Finished", "status": "True"}, + }} + wantInUse, wantError := true, false + objects := []client.Object{namespace, job, localQueue, clusterQueue} + switch scenario { + case "running": + job.Spec.Suspend = ptr.To(false) + case "complete", "failed", "false-completion": + conditionType, status := batchv1.JobComplete, corev1.ConditionTrue + if scenario == "failed" { + conditionType = batchv1.JobFailed + } + if scenario == "false-completion" { + status = corev1.ConditionFalse + } else { + wantInUse = false + } + job.Status.Conditions = []batchv1.JobCondition{{Type: conditionType, Status: status}} + case "unlabeled": + job.Labels = nil + wantInUse = false + case "other-namespace": + job.Namespace = "unrelated" + wantInUse = false + case "missing-queue": + job.Labels["kueue.x-k8s.io/queue-name"] = "missing" + wantInUse = false + case "selector-independent": + clusterQueue.Object["spec"] = map[string]any{"namespaceSelector": map[string]any{}} + wantInUse = false + case "workload-finished", "workload-wrong-uid", "workload-wrong-name": + if scenario == "workload-finished" { + wantInUse = false + } else { + owner := workload.GetOwnerReferences()[0] + if scenario == "workload-wrong-uid" { + owner.UID = types.UID("different-job-uid") + } else { + owner.Name = "different-job" + } + workload.SetOwnerReferences([]metav1.OwnerReference{owner}) + } + objects = append(objects, workload) + case "job-list-error", "queue-read-error", "cluster-queue-read-error": + wantInUse, wantError = false, true + case "invalid-selector": + clusterQueue.Object["spec"] = map[string]any{"namespaceSelector": map[string]any{ + "matchExpressions": []any{map[string]any{"key": labelWorkspace, "operator": "Invalid"}}, + }} + wantInUse, wantError = false, true + case "cluster-queue-uid-changed": + clusterQueue.SetUID(types.UID("replacement-uid")) + wantInUse, wantError = false, true + } + baseClient := fake.NewClientBuilder().WithScheme(testScheme(t)).WithObjects(objects...).Build() + c := interceptor.NewClient(baseClient, interceptor.Funcs{ + List: func(ctx context.Context, c client.WithWatch, list client.ObjectList, opts ...client.ListOption) error { + if _, ok := list.(*batchv1.JobList); ok && scenario == "job-list-error" { + return errors.New("source Jobs unavailable") + } + return c.List(ctx, list, opts...) + }, + Get: func(ctx context.Context, c client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { + if queue, ok := obj.(*unstructured.Unstructured); ok && + (scenario == "queue-read-error" && queue.GroupVersionKind() == localQueueGVK || + scenario == "cluster-queue-read-error" && queue.GroupVersionKind() == clusterQueueGVK) { + return errors.New("queue unavailable") + } + return c.Get(ctx, key, obj, opts...) + }, + }) + inUse, message, err := newTestWorkspaceReconciler(c).namespaceRoutingMetadataInUse(ctx, workspace, namespace) + if inUse != wantInUse || (err != nil) != wantError { + t.Fatalf("namespaceRoutingMetadataInUse() = %v, %q, %v; want inUse=%v error=%v", inUse, message, err, wantInUse, wantError) + } + if inUse && !strings.Contains(message, "Job "+client.ObjectKeyFromObject(job).String()) { + t.Fatalf("drain message does not identify the source Job: %q", message) + } + for _, queue := range []*unstructured.Unstructured{localQueue, clusterQueue} { + var got unstructured.Unstructured + got.SetGroupVersionKind(queue.GroupVersionKind()) + if err := baseClient.Get(ctx, client.ObjectKeyFromObject(queue), &got); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(got.Object, queue.Object) { + t.Fatalf("external queue was mutated: %#v", got.Object) + } + } + }) + } +} diff --git a/controllers/tau-core/internal/controller/workspace_rbac.go b/controllers/tau-core/internal/controller/workspace_rbac.go index 75ce902fd..dcc04f0af 100644 --- a/controllers/tau-core/internal/controller/workspace_rbac.go +++ b/controllers/tau-core/internal/controller/workspace_rbac.go @@ -11,6 +11,7 @@ import ( corev1 "k8s.io/api/core/v1" rbacv1 "k8s.io/api/rbac/v1" apiequality "k8s.io/apimachinery/pkg/api/equality" + apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" @@ -32,18 +33,7 @@ func (r *TauWorkspaceReconciler) reconcileRBAC(ctx context.Context, workspace *t if workspace.Spec.PrincipalRef == nil || workspace.Spec.KubernetesSubject == nil { return false, "workspace-rbac authorization requires principalRef and kubernetesSubject", nil } - binding := &rbacv1.RoleBinding{ObjectMeta: metav1.ObjectMeta{Name: defaultRoleName, Namespace: targetNamespace}} - if err := r.getAndValidateWorkspaceOwnership(ctx, binding, workspace); err != nil { - return false, "refusing to adopt existing researcher RoleBinding", err - } - _, err := controllerutil.CreateOrUpdate(ctx, r.Client, binding, func() error { - binding.Labels = workspaceLabels(workspace.Name) - setOwnerUIDAnnotation(binding, workspace.UID) - binding.RoleRef = rbacv1.RoleRef{APIGroup: rbacv1.GroupName, Kind: "ClusterRole", Name: defaultRoleName} - binding.Subjects = []rbacv1.Subject{rbacSubject(*workspace.Spec.KubernetesSubject, targetNamespace)} - return nil - }) - if err != nil { + if err := r.reconcileResearcherRoleBinding(ctx, workspace, targetNamespace); err != nil { return false, "failed to reconcile researcher RoleBinding", err } if err := r.reconcileClusterQueueReaderRBAC(ctx, workspace, targetNamespace); err != nil { @@ -55,6 +45,25 @@ func (r *TauWorkspaceReconciler) reconcileRBAC(ctx context.Context, workspace *t return true, "researcher namespace and ClusterQueue reader bindings are reconciled", nil } +func (r *TauWorkspaceReconciler) reconcileResearcherRoleBinding( + ctx context.Context, + workspace *tauv1alpha1.TauWorkspace, + namespace string, +) error { + binding := &rbacv1.RoleBinding{ObjectMeta: metav1.ObjectMeta{Name: defaultRoleName, Namespace: namespace}} + if err := r.getAndValidateWorkspaceOwnership(ctx, binding, workspace); err != nil { + return fmt.Errorf("refusing to adopt existing researcher RoleBinding: %w", err) + } + _, err := controllerutil.CreateOrUpdate(ctx, r.Client, binding, func() error { + binding.Labels = workspaceLabels(workspace.Name) + setOwnerUIDAnnotation(binding, workspace.UID) + binding.RoleRef = rbacv1.RoleRef{APIGroup: rbacv1.GroupName, Kind: "ClusterRole", Name: defaultRoleName} + binding.Subjects = []rbacv1.Subject{rbacSubject(*workspace.Spec.KubernetesSubject, namespace)} + return nil + }) + return err +} + func authorizationMode(workspace *tauv1alpha1.TauWorkspace) string { if workspace.Spec.Authorization == nil || workspace.Spec.Authorization.Mode == "" { return tauv1alpha1.AuthorizationModeWorkspaceRBAC @@ -94,6 +103,12 @@ func (r *TauWorkspaceReconciler) reconcileExistingResearcherAccess( ) error { binding := &rbacv1.RoleBinding{ObjectMeta: metav1.ObjectMeta{Name: defaultRoleName, Namespace: namespace}} if err := r.Get(ctx, client.ObjectKeyFromObject(binding), binding); err != nil { + if apierrors.IsNotFound(err) && + namespace == workspace.Status.Target.ResolvedNamespace && + authorizationMode(workspace) != tauv1alpha1.AuthorizationModeClusterWide && + workspace.Spec.PrincipalRef != nil && workspace.Spec.KubernetesSubject != nil { + return r.reconcileResearcherRoleBinding(ctx, workspace, namespace) + } return client.IgnoreNotFound(err) } if !ownedByWorkspace(binding.Labels, workspace.Name) { @@ -126,6 +141,29 @@ func (r *TauWorkspaceReconciler) cleanupResearcherRBAC( return r.deleteOwnedObject(ctx, binding, workspace) } +func (r *TauWorkspaceReconciler) cleanupOwnedResearcherRoleBindings( + ctx context.Context, + workspace *tauv1alpha1.TauWorkspace, +) error { + var bindings rbacv1.RoleBindingList + if err := r.List(ctx, &bindings, client.MatchingLabels{ + labelManagedBy: labelManagedByValue, + labelWorkspace: workspace.Name, + }); err != nil { + return err + } + for i := range bindings.Items { + binding := &bindings.Items[i] + if binding.Name != defaultRoleName { + continue + } + if err := r.deleteOwnedObject(ctx, binding, workspace); err != nil { + return err + } + } + return nil +} + func clusterQueueReaderBindingName(workspaceName string) string { return "tau-clusterqueue-reader-" + workspaceName } diff --git a/controllers/tau-core/internal/controller/workspace_rbac_test.go b/controllers/tau-core/internal/controller/workspace_rbac_test.go index 0cfdf68fa..4566be1c6 100644 --- a/controllers/tau-core/internal/controller/workspace_rbac_test.go +++ b/controllers/tau-core/internal/controller/workspace_rbac_test.go @@ -10,7 +10,9 @@ import ( "reflect" "testing" + tauv1alpha1 "github.com/Azure/taugrid/controllers/tau-core/api/v1alpha1" rbacv1 "k8s.io/api/rbac/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/util/yaml" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" @@ -96,6 +98,82 @@ func TestKustomizeControllerRoleAllowsLegacyRoleDeletionWithoutMutation(t *testi t.Fatalf("Kustomize controller role must permit identity-checked Role deletion without Role mutation: %#v", role.Rules) } +func TestKustomizeControllerRoleAllowsReadOnlySourceJobDiscovery(t *testing.T) { + for _, source := range []struct { + group string + resources []string + }{ + {"batch", []string{"jobs"}}, + {"apps", []string{"deployments"}}, + {"ray.io", []string{"rayjobs", "rayservices"}}, + } { + t.Run(source.group, func(t *testing.T) { + role := kustomizeClusterRole(t, "tau-core-controller") + want := rbacv1.PolicyRule{ + APIGroups: []string{source.group}, + Resources: source.resources, + Verbs: []string{"get", "list", "watch"}, + } + for _, rule := range role.Rules { + if reflect.DeepEqual(rule, want) { + return + } + } + t.Fatalf("controller role must allow read-only source Job discovery: %#v", role.Rules) + }) + } +} + +func TestRetainedNamespaceRestoresOnlyCurrentAuthorizedResearcherBinding(t *testing.T) { + for _, scenario := range []string{"group", "service-account", "cluster-wide", "missing-subject", "missing-principal", "abandoned-namespace", "foreign-namespace"} { + t.Run(scenario, func(t *testing.T) { + ctx := context.Background() + workspace := testWorkspace("research") + workspace.Status.Target.ResolvedNamespace = "retained" + namespace := testNamespace("retained") + namespace.Labels = workspaceLabels(workspace.Name) + namespace.Annotations[annotationOwnerUID] = string(workspace.UID) + wantBinding := scenario == "group" || scenario == "service-account" + switch scenario { + case "service-account": + workspace.Spec.KubernetesSubject = &tauv1alpha1.KubernetesSubject{Kind: rbacv1.ServiceAccountKind, Name: "current-researcher"} + case "cluster-wide": + workspace.Spec.Authorization = &tauv1alpha1.WorkspaceAuthorization{Mode: tauv1alpha1.AuthorizationModeClusterWide} + case "missing-subject": + workspace.Spec.KubernetesSubject = nil + case "missing-principal": + workspace.Spec.PrincipalRef = nil + case "abandoned-namespace": + workspace.Status.Target.ResolvedNamespace = "another-namespace" + case "foreign-namespace": + namespace.Annotations[annotationOwnerUID] = "another-workspace-uid" + } + c := fake.NewClientBuilder().WithScheme(testScheme(t)).WithObjects(namespace).Build() + if err := newTestWorkspaceReconciler(c).reconcileOwnedNamespaceResearcherAccess(ctx, workspace, "new-target"); err != nil { + t.Fatal(err) + } + var binding rbacv1.RoleBinding + err := c.Get(ctx, client.ObjectKey{Namespace: namespace.Name, Name: defaultRoleName}, &binding) + if !wantBinding { + if !apierrors.IsNotFound(err) { + t.Fatalf("unexpected researcher access was created: %v, %#v", err, binding) + } + return + } + if err != nil { + t.Fatal(err) + } + wantSubjects := []rbacv1.Subject{rbacSubject(*workspace.Spec.KubernetesSubject, namespace.Name)} + if !reflect.DeepEqual(binding.Subjects, wantSubjects) || + binding.RoleRef != (rbacv1.RoleRef{APIGroup: rbacv1.GroupName, Kind: "ClusterRole", Name: defaultRoleName}) || + !ownedByWorkspace(binding.Labels, workspace.Name) || + binding.Annotations[annotationOwnerUID] != string(workspace.UID) { + t.Fatalf("restored researcher access does not match current workspace: %#v", binding) + } + }) + } +} + func kustomizeClusterRole(t *testing.T, name string) rbacv1.ClusterRole { t.Helper() file, err := os.Open("../../../../charts/tau-core-controller/kustomize/rbac.yaml") diff --git a/controllers/tau-core/internal/controller/workspace_sync.go b/controllers/tau-core/internal/controller/workspace_sync.go index d8e1a4059..1fe1c8f5e 100644 --- a/controllers/tau-core/internal/controller/workspace_sync.go +++ b/controllers/tau-core/internal/controller/workspace_sync.go @@ -103,9 +103,24 @@ func (r *TauWorkspaceReconciler) syncWorkspace(ctx context.Context, workspace *t } previousNamespace := workspace.Status.Target.ResolvedNamespace previousAccessReady := true - if err := r.cleanupOwnedNamespaceAccess(ctx, workspace, targetNamespace, keepServiceAccount); err != nil { + if err := r.cleanupOwnedNamespaceWorkloadIdentity(ctx, workspace, keepServiceAccount); err != nil { previousAccessReady = false conditions = append(conditions, boolCondition(tauv1alpha1.ConditionDriftDetected, true, "RBACCleanupFailed", err.Error(), workspace.Generation)) + if err := r.cleanupOwnedResearcherRoleBindings(ctx, workspace); err != nil { + conditions = append(conditions, boolCondition(tauv1alpha1.ConditionDriftDetected, true, "RBACCleanupFailed", err.Error(), workspace.Generation)) + } + } + if previousAccessReady { + if err := r.reconcileOwnedNamespaceResearcherAccess(ctx, workspace, targetNamespace); err != nil { + previousAccessReady = false + conditions = append(conditions, boolCondition(tauv1alpha1.ConditionDriftDetected, true, "RBACCleanupFailed", err.Error(), workspace.Generation)) + } + } + if previousNamespace != "" && previousNamespace != targetNamespace { + if err := r.waitForNamespaceRoutingMetadataDrain(ctx, workspace, previousNamespace); err != nil { + previousAccessReady = false + conditions = append(conditions, boolCondition(tauv1alpha1.ConditionDriftDetected, true, "NamespaceCleanupFailed", err.Error(), workspace.Generation)) + } } currentTargetReady := namespaceReady && rbacReady && rbacErr == nil && queueReady && previousAccessReady diff --git a/site/content/en/docs/reference/workspace.md b/site/content/en/docs/reference/workspace.md index a865f7416..28ac0fa6c 100644 --- a/site/content/en/docs/reference/workspace.md +++ b/site/content/en/docs/reference/workspace.md @@ -49,6 +49,45 @@ The overall phase currently requires `RBACReady` and `QueueReady`; `DriftDetecte | `QueueReady=False` | Restore the named LocalQueue and its accessible backing ClusterQueue; fix that queue directly rather than creating a second queue with a different name to bypass the workspace spec. | | `DriftDetected=True` | Fix the dependency or workspace spec named in the message, then let the controller restore its owned namespace/RBAC objects. | +During namespace retargeting, unfinished Workloads on retained external queues +can delay cleanup of the old namespace's routing metadata. Submitted, unfinished +Jobs, RayJobs, and RayServices that have not received a derived Workload yet retain routing +metadata on deletion or retargeting, using their queue label and the live queue +selector. Completed or failed batch sources do not block cleanup. A RayJob is finished +only when `status.jobDeploymentStatus` is `Complete` or `Failed`, matching Kueue; +a failed driver with a retry still pending retains its route. RayJob discovery +is skipped when the optional RayJob API is not installed; RayService discovery +is checked independently and also tolerates an absent API. Other discovery errors +block cleanup until the source state can be read. RayServices have no terminal +job state: a queued service retains required routing metadata and unchanged +execution identity until the service is deleted, even if a derived Workload +reports `Finished`. Delete the service before completing workspace +deletion or retargeting. Both Helm and Kustomize grant the controller read-only +discovery of Jobs, RayJobs, and RayServices. +Kueue-managed Deployments submitted with `tau serve deploy --kind=deployment` +also retain routing before their Pod Workloads exist. Discovery uses the Pod +template's managed and queue labels, not just the Deployment's own labels. +A Deployment has no terminal job state: a present managed Deployment retains +its required route even at zero replicas or after a Pod Workload finishes. +Delete the Deployment before completing workspace deletion or retargeting. +Both controller RBAC manifests include read-only Deployment discovery; +unreadable Deployment lists block cleanup. +An unchanged workload identity ServiceAccount is retained until those Workloads finish, while researcher +authorization is reconciled independently. Changing the ServiceAccount name or +client ID, or removing `workloadIdentity`, revokes the obsolete ServiceAccount +before granting the current researcher access to the old namespace, even during +the drain. Pending Jobs that use the revoked account can no longer create Pods; +the operator must cancel or replace affected work before the drain can complete. +Keep the identity unchanged until draining finishes when uninterrupted execution +is required. + +If identity cleanup fails or deletion is delayed by a finalizer, researcher +bindings are removed until obsolete identity revocation is confirmed. After +cleanup recovers, the current researcher's binding is restored in the retained +resolved namespace even while its Workloads are unfinished, so the researcher +can inspect or cancel the remaining work. Cluster-wide authorization does not +create a researcher binding. + ### Storage readiness is platform-owned `Ready` proves RBAC and queue reconciliation only, not storage. TauGrid 0.1 has no `StorageReady` condition on `TauWorkspace` or `TauCluster`, so a workspace can reach `Ready` while a workload's configured PVC is missing or unbound. Verify the platform-managed claim directly before handing the workspace to a researcher: