[iris] Address review: bump node overcommit, delete pods by name

- Bump DIRECT_PROVIDER_NODE_OVERCOMMIT from 2 to 16 to match the
  worker provider's tasks-per-worker ratio.
- Replace delete_by_labels with delete_many: deletes pods by name
  instead of label selector. Simpler, more precise, no new protocol
  method needed beyond what kubectl natively supports.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: yoblin

lib/iris/src/iris/cluster/controller/transitions.py

@@ -84,8 +84,12 @@ class ReservationClaim:
 DIRECT_PROVIDER_BOOTSTRAP_BATCH = 64
 """Max tasks promoted per sync cycle when no capacity info is available yet."""
 
-DIRECT_PROVIDER_NODE_OVERCOMMIT = 2
-"""Pods per schedulable node allowed for the direct provider scheduler."""
+DIRECT_PROVIDER_NODE_OVERCOMMIT = 16
+"""Pods per schedulable node allowed for the direct provider scheduler.
+
+Matches the worker provider's tasks-per-worker ratio so that the direct
+provider can keep a similar number of tasks in-flight relative to cluster size.
+"""
 
 
 @dataclass(frozen=True)

lib/iris/src/iris/cluster/providers/k8s/fake.py

@@ -688,28 +688,10 @@ def delete(
         if normalized == "pod":
             self._release_pod_resources(name)
 
-    def delete_by_labels(
-        self,
-        resource: str,
-        labels: dict[str, str],
-        *,
-        cluster_scoped: bool = False,
-        wait: bool = False,
-    ) -> int:
-        self._check_failure("delete_by_labels")
-        normalized = _normalize_resource(resource)
-        to_delete = []
-        for (kind, name), manifest in self._resources.items():
-            if kind != normalized:
-                continue
-            res_labels = manifest.get("metadata", {}).get("labels", {})
-            if all(res_labels.get(k) == v for k, v in labels.items()):
-                to_delete.append(name)
-        for name in to_delete:
-            self._resources.pop((normalized, name), None)
-            if normalized == "pod":
-                self._release_pod_resources(name)
-        return len(to_delete)
+    def delete_many(self, resource: str, names: list[str], *, cluster_scoped: bool = False, wait: bool = False) -> None:
+        """Delete multiple resources by name."""
+        for name in names:
+            self.delete(resource, name)
 
     def logs(self, pod_name: str, *, container: str | None = None, tail: int = 50, previous: bool = False) -> str:
         self._check_failure("logs")

lib/iris/src/iris/cluster/providers/k8s/service.py

@@ -57,15 +57,8 @@ def delete(
         self, resource: str, name: str, *, cluster_scoped: bool = False, force: bool = False, wait: bool = True
     ) -> None: ...
 
-    def delete_by_labels(
-        self,
-        resource: str,
-        labels: dict[str, str],
-        *,
-        cluster_scoped: bool = False,
-        wait: bool = False,
-    ) -> int:
-        """Delete all resources matching label selector. Returns count deleted."""
+    def delete_many(self, resource: str, names: list[str], *, cluster_scoped: bool = False, wait: bool = False) -> None:
+        """Delete multiple resources by name in a single kubectl call."""
         ...
 
     def logs(self, pod_name: str, *, container: str | None = None, tail: int = 50, previous: bool = False) -> str: ...
@@ -240,24 +233,16 @@ def delete(
         if result.returncode != 0:
             raise KubectlError(f"kubectl delete {resource}/{name} failed: {result.stderr.strip()}")
 
-    def delete_by_labels(
-        self,
-        resource: str,
-        labels: dict[str, str],
-        *,
-        cluster_scoped: bool = False,
-        wait: bool = False,
-    ) -> int:
-        """Delete all resources matching label selector in a single kubectl call."""
-        selector = ",".join(f"{k}={v}" for k, v in labels.items())
-        args = ["delete", resource, "-l", selector, "--ignore-not-found"]
+    def delete_many(self, resource: str, names: list[str], *, cluster_scoped: bool = False, wait: bool = False) -> None:
+        """Delete multiple resources by name in a single kubectl call."""
+        if not names:
+            return
+        args = ["delete", resource, *names, "--ignore-not-found"]
         if not wait:
             args.append("--wait=false")
         result = self._run(args, namespaced=not cluster_scoped)
         if result.returncode != 0:
-            raise KubectlError(f"kubectl delete {resource} -l {selector} failed: {result.stderr.strip()}")
-        # kubectl prints one "deleted" line per resource
-        return result.stdout.strip().count("deleted")
+            raise KubectlError(f"kubectl delete {resource} failed: {result.stderr.strip()}")
 
     def set_image(self, resource: str, name: str, container: str, image: str, *, namespaced: bool = False) -> None:
         """Set the container image on a resource via ``kubectl set image``."""

lib/iris/src/iris/cluster/providers/k8s/tasks.py

@@ -960,29 +960,41 @@ def _apply_pod(self, run_req: cluster_pb2.Worker.RunTaskRequest) -> None:
         )
 
     def _bulk_delete_task_pods(self, task_ids: list[str]) -> None:
-        """Delete pods and configmaps for multiple tasks via one kubectl call per job."""
-        # Group by job_id (task_id format: /{user}/{job}/{task_index})
-        jobs: dict[str, list[str]] = {}
-        for task_id in task_ids:
-            job_id, _, _ = task_id.rpartition("/")
-            assert job_id, f"malformed task_id (expected /<user>/<job>/<index>): {task_id}"
-            jobs.setdefault(job_id, []).append(task_id)
-
-        for job_id, job_task_ids in jobs.items():
-            labels = {
+        """Delete pods and configmaps for multiple tasks in one kubectl call per resource type."""
+        task_hashes = {_task_hash(tid) for tid in task_ids}
+
+        all_pods = self.kubectl.list_json(
+            "pods",
+            labels={
                 _LABEL_MANAGED: "true",
                 _LABEL_RUNTIME: _RUNTIME_LABEL_VALUE,
-                _LABEL_JOB_ID: _sanitize_label_value(job_id),
-            }
-            pod_count = self.kubectl.delete_by_labels("pods", labels, wait=False)
-            cm_count = self.kubectl.delete_by_labels("configmaps", labels, wait=False)
-            logger.info(
-                "Bulk deleted %d pods, %d configmaps for job %s (%d tasks)",
-                pod_count,
-                cm_count,
-                job_id,
-                len(job_task_ids),
-            )
+            },
+        )
+        pod_names = [
+            p["metadata"]["name"]
+            for p in all_pods
+            if p.get("metadata", {}).get("labels", {}).get(_LABEL_TASK_HASH) in task_hashes
+        ]
+
+        all_cms = self.kubectl.list_json(
+            "configmaps",
+            labels={
+                _LABEL_MANAGED: "true",
+                _LABEL_RUNTIME: _RUNTIME_LABEL_VALUE,
+            },
+        )
+        cm_names = [
+            c["metadata"]["name"]
+            for c in all_cms
+            if c.get("metadata", {}).get("labels", {}).get(_LABEL_TASK_HASH) in task_hashes
+        ]
+
+        if pod_names:
+            self.kubectl.delete_many("pods", pod_names, wait=False)
+        if cm_names:
+            self.kubectl.delete_many("configmaps", cm_names, wait=False)
+
+        logger.info("Deleted %d pods, %d configmaps for %d tasks", len(pod_names), len(cm_names), len(task_ids))
 
     def _delete_pods_by_task_id(self, task_id: str) -> None:
         """Delete all pods for a given task_id (any attempt).

lib/iris/tests/cluster/controller/test_transitions.py

@@ -3259,18 +3259,18 @@ def test_drain_caps_promotions_and_refills_after_completion(state):
 def test_drain_capacity_limits_promotions(state):
     """With capacity, budget is schedulable_nodes * OVERCOMMIT; active tasks reduce it."""
     capacity = ClusterCapacity(
-        schedulable_nodes=100,
-        total_cpu_millicores=400000,
-        available_cpu_millicores=200000,
-        total_memory_bytes=800 * 1024**3,
-        available_memory_bytes=400 * 1024**3,
+        schedulable_nodes=10,
+        total_cpu_millicores=40000,
+        available_cpu_millicores=20000,
+        total_memory_bytes=80 * 1024**3,
+        available_memory_bytes=40 * 1024**3,
     )
     _submit_job_direct(state, "/user/cap-job", replicas=250)
 
     batch1 = state.drain_for_direct_provider(capacity=capacity)
-    assert len(batch1.tasks_to_run) == 200  # 100 nodes * 2
+    assert len(batch1.tasks_to_run) == 160  # 10 nodes * 16
 
-    # 200 active → budget exhausted.
+    # 160 active → budget exhausted.
     assert len(state.drain_for_direct_provider(capacity=capacity).tasks_to_run) == 0