[Iris] Add manual slice CLI (create-slice / delete-slice) (#5078)

Add iris cluster create-slice --scale-group <name> that allocates a
slice tagged iris_manual=true bound to the running controller, and
iris cluster delete-slice <slice_id> to terminate it. The autoscaler
filters iris_manual slices out of list_all_slices and reconcile, so
manual slices do not count toward demand, do not get scaled down on
idle, and survive iris cluster stop.

Fixes #5069

Author: rjpower

lib/iris/src/iris/cli/cluster.py

@@ -22,6 +22,12 @@
 )
 from iris.cli.main import IRIS_CLUSTER_CONFIG_DIRS, require_controller_url, rpc_client
 from iris.cluster.config import IrisConfig, clear_remote_state, make_local_config
+from iris.cluster.controller.autoscaler.scaling_group import (
+    build_worker_config_for_group,
+    prepare_slice_config,
+)
+from iris.cluster.providers.types import Labels
+from iris.rpc import config_pb2
 from iris.rpc import vm_pb2
 from iris.rpc import job_pb2
 from iris.rpc import controller_pb2
@@ -409,6 +415,112 @@ def cluster_restart(ctx):
     ctx.invoke(cluster_start)
 
 
+@cluster.command("create-slice")
+@click.option("--scale-group", "scale_group_name", required=True, help="Scale group whose template to use")
+@click.pass_context
+def cluster_create_slice(ctx, scale_group_name: str):
+    """Create an operator-managed slice bound to the running controller.
+
+    Allocates a slice using the named scale group's template, tags it with
+    ``iris-{prefix}-manual=true``, and bootstraps workers so they connect to
+    the controller. The autoscaler ignores manual slices: they don't count
+    toward demand, won't be scaled down on idle, and survive
+    ``iris cluster stop``. Remove with ``iris cluster delete-slice``.
+    """
+    config = ctx.obj.get("config")
+    if not config:
+        raise click.ClickException("--config is required for cluster create-slice")
+    if config.controller.WhichOneof("controller") == "local":
+        raise click.ClickException("create-slice is not supported for local clusters")
+
+    sg_config = config.scale_groups.get(scale_group_name)
+    if sg_config is None:
+        available = ", ".join(sorted(config.scale_groups.keys())) or "(none)"
+        raise click.ClickException(f"Unknown scale group '{scale_group_name}'. Available: {available}")
+
+    # Verify the controller is reachable before creating the slice. The
+    # returned URL may be a tunnel endpoint that's only reachable from the CLI
+    # host; workers need the cluster-internal address instead, resolved below.
+    require_controller_url(ctx)
+    iris_config = IrisConfig(config)
+    bundle = ctx.obj.get("provider_bundle") or iris_config.provider_bundle()
+
+    # Resolve the address workers will connect to. Prefer an explicit value in
+    # defaults.worker.controller_address, then discover it via the provider
+    # (e.g., GCE label lookup). Never pass the CLI-local tunnel URL here.
+    worker_controller_address = iris_config.controller_address()
+    if not worker_controller_address:
+        worker_controller_address = bundle.controller.discover_controller(config.controller)
+
+    label_prefix = config.platform.label_prefix or "iris"
+    labels = Labels(label_prefix)
+
+    slice_config = prepare_slice_config(sg_config.slice_template, sg_config, label_prefix)
+    slice_config.labels[labels.iris_manual] = "true"
+
+    base_worker_config = config_pb2.WorkerConfig()
+    base_worker_config.CopyFrom(config.defaults.worker)
+    if not base_worker_config.controller_address:
+        base_worker_config.controller_address = worker_controller_address
+    base_worker_config.platform.CopyFrom(config.platform)
+    if config.storage.remote_state_dir:
+        base_worker_config.storage_prefix = config.storage.remote_state_dir
+
+    worker_config = build_worker_config_for_group(base_worker_config, sg_config)
+
+    click.echo(f"Creating manual slice from scale group '{scale_group_name}'...")
+    try:
+        handle = bundle.workers.create_slice(slice_config, worker_config=worker_config)
+    except Exception as e:
+        click.echo(f"Failed to create slice: {e}", err=True)
+        raise SystemExit(1) from e
+
+    click.echo(f"Created manual slice: {handle.slice_id}")
+    click.echo(f"  Scale group: {handle.scale_group or scale_group_name}")
+    click.echo(f"  Zone:        {handle.zone}")
+    click.echo("Workers will register with the controller as they bootstrap.")
+    click.echo(f"Terminate with: iris cluster delete-slice {handle.slice_id}")
+
+
+@cluster.command("delete-slice")
+@click.argument("slice_id")
+@click.pass_context
+def cluster_delete_slice(ctx, slice_id: str):
+    """Terminate an operator-managed slice created via ``create-slice``.
+
+    Only slices tagged ``iris-{prefix}-manual=true`` are eligible —
+    autoscaler-managed slices must go through the autoscaler.
+    """
+    config = ctx.obj.get("config")
+    if not config:
+        raise click.ClickException("--config is required for cluster delete-slice")
+    if config.controller.WhichOneof("controller") == "local":
+        raise click.ClickException("delete-slice is not supported for local clusters")
+
+    iris_config = IrisConfig(config)
+    bundle = ctx.obj.get("provider_bundle") or iris_config.provider_bundle()
+
+    label_prefix = config.platform.label_prefix or "iris"
+    labels = Labels(label_prefix)
+
+    manual_slices = bundle.workers.list_slices(zones=[], labels={labels.iris_manual: "true"})
+    match = next((s for s in manual_slices if s.slice_id == slice_id), None)
+    if match is None:
+        raise click.ClickException(
+            f"No manual slice found with id '{slice_id}'. "
+            "List manual slices with the controller dashboard or "
+            "`iris cluster status` to find the correct id."
+        )
+
+    click.echo(f"Terminating manual slice {slice_id}...")
+    try:
+        match.terminate()
+    except Exception as e:
+        click.echo(f"Failed to terminate slice: {e}", err=True)
+        raise SystemExit(1) from e
+    click.echo("Terminated.")
+
+
 @cluster.command("status")
 @click.pass_context
 def cluster_status_cmd(ctx):

lib/iris/src/iris/cluster/controller/autoscaler/runtime.py

@@ -23,7 +23,7 @@
 from collections import deque
 from collections.abc import Sequence
 
-from iris.cluster.constraints import Constraint, WellKnownAttribute
+from iris.cluster.constraints import Constraint
 from iris.cluster.providers.protocols import WorkerInfraProvider
 from iris.cluster.providers.types import (
     CloudSliceState,
@@ -46,7 +46,7 @@
     restore_autoscaler_state,
 )
 from iris.cluster.controller.autoscaler.routing import job_feasibility, route_demand
-from iris.cluster.controller.autoscaler.scaling_group import ScalingGroup
+from iris.cluster.controller.autoscaler.scaling_group import ScalingGroup, build_worker_config_for_group
 from iris.cluster.controller.autoscaler.status import PendingHint, build_job_pending_hints, routing_decision_to_proto
 from iris.cluster.controller.autoscaler.worker_registry import TrackedWorker, WorkerRegistry
 from iris.cluster.controller.db import ControllerDB
@@ -228,6 +228,14 @@ def _log_action(
             status=status,
         )
         self._action_log.append(action)
+        logger.info(
+            "event=autoscaler action=%s entity=%s trigger=- group=%s status=%s reason=%s",
+            action_type,
+            slice_id or scale_group,
+            scale_group,
+            status,
+            reason,
+        )
         return action
 
     def evaluate(
@@ -372,39 +380,7 @@ def _do_scale_up(self, group: ScalingGroup, ts: Timestamp, reason: str = "") ->
 
     def _per_group_worker_config(self, group: ScalingGroup) -> config_pb2.WorkerConfig | None:
         """Build per-group WorkerConfig by merging base config with scale group overrides."""
-        if not self._base_worker_config:
-            return None
-
-        wc = config_pb2.WorkerConfig()
-        wc.CopyFrom(self._base_worker_config)
-
-        # Accelerator config from scale group resources
-        resources = group.config.resources if group.config.HasField("resources") else None
-        if resources is not None:
-            wc.accelerator_type = resources.device_type
-            if resources.device_variant:
-                wc.accelerator_variant = resources.device_variant
-            if resources.device_type == config_pb2.ACCELERATOR_TYPE_GPU and resources.device_count > 0:
-                wc.gpu_count = resources.device_count
-            wc.capacity_type = resources.capacity_type
-
-        # Worker attributes from scale group
-        if group.config.HasField("worker"):
-            for k, v in group.config.worker.attributes.items():
-                wc.worker_attributes[k] = v
-
-        region = group.region
-        if region and not wc.worker_attributes.get(WellKnownAttribute.REGION):
-            wc.worker_attributes[WellKnownAttribute.REGION] = region
-
-        zone = group.zone
-        if zone and not wc.worker_attributes.get(WellKnownAttribute.ZONE):
-            wc.worker_attributes[WellKnownAttribute.ZONE] = zone
-
-        if group.config.name:
-            wc.worker_attributes["scale-group"] = group.config.name
-
-        return wc
+        return build_worker_config_for_group(self._base_worker_config, group.config)
 
     def _register_slice_workers(
         self,

lib/iris/src/iris/cluster/controller/autoscaler/scaling_group.py

@@ -152,6 +152,66 @@ def prepare_slice_config(
     return config
 
 
+def _region_from_template(template: config_pb2.SliceConfig) -> str | None:
+    """Region derived from a scale group's slice template."""
+    if template.HasField("gcp") and template.gcp.zone:
+        return template.gcp.zone.rsplit("-", 1)[0]
+    if template.HasField("coreweave") and template.coreweave.region:
+        return template.coreweave.region
+    return None
+
+
+def _zone_from_template(template: config_pb2.SliceConfig) -> str | None:
+    """Zone derived from a scale group's slice template."""
+    if template.HasField("gcp") and template.gcp.zone:
+        return template.gcp.zone
+    if template.HasField("coreweave") and template.coreweave.region:
+        return template.coreweave.region
+    return None
+
+
+def build_worker_config_for_group(
+    base_worker_config: config_pb2.WorkerConfig | None,
+    group_config: config_pb2.ScaleGroupConfig,
+) -> config_pb2.WorkerConfig | None:
+    """Merge base worker config with per-scale-group overrides.
+
+    Returns None when base_worker_config is None (test/local mode).
+    """
+    if not base_worker_config:
+        return None
+
+    wc = config_pb2.WorkerConfig()
+    wc.CopyFrom(base_worker_config)
+
+    resources = group_config.resources if group_config.HasField("resources") else None
+    if resources is not None:
+        wc.accelerator_type = resources.device_type
+        if resources.device_variant:
+            wc.accelerator_variant = resources.device_variant
+        if resources.device_type == config_pb2.ACCELERATOR_TYPE_GPU and resources.device_count > 0:
+            wc.gpu_count = resources.device_count
+        wc.capacity_type = resources.capacity_type
+
+    if group_config.HasField("worker"):
+        for k, v in group_config.worker.attributes.items():
+            wc.worker_attributes[k] = v
+
+    template = group_config.slice_template
+    region = _region_from_template(template)
+    if region and not wc.worker_attributes.get(WellKnownAttribute.REGION):
+        wc.worker_attributes[WellKnownAttribute.REGION] = region
+
+    zone = _zone_from_template(template)
+    if zone and not wc.worker_attributes.get(WellKnownAttribute.ZONE):
+        wc.worker_attributes[WellKnownAttribute.ZONE] = zone
+
+    if group_config.name:
+        wc.worker_attributes["scale-group"] = group_config.name
+
+    return wc
+
+
 def _zones_from_config(config: config_pb2.ScaleGroupConfig) -> list[str]:
     """Extract zones from ScaleGroupConfig's slice_template.
 
@@ -500,12 +560,16 @@ def reconcile(self) -> None:
 
         Used in tests to populate a scaling group with pre-injected slices.
         Production restore uses prepare_for_restore() + restore_scaling_group().
+        Skips operator-created manual slices (iris_manual=true), which the
+        autoscaler must not track or scale down.
         """
         zones = _zones_from_config(self._config)
         labels = {self._labels.iris_scale_group: self._config.name}
         slice_handles = self._platform.list_slices(zones, labels)
         with self._slices_lock:
             for handle in slice_handles:
+                if handle.labels.get(self._labels.iris_manual) == "true":
+                    continue
                 state = SliceState(handle=handle)
                 self._slices[handle.slice_id] = state
                 self._db_upsert_slice(handle.slice_id, state)

lib/iris/src/iris/cluster/providers/gcp/workers.py

@@ -648,15 +648,20 @@ def list_slices(
         return handles
 
     def list_all_slices(self) -> list[GcpSliceHandle | GcpVmSliceHandle]:
-        """List all slices managed by this cluster.
+        """List all autoscaler-managed slices for this cluster.
 
         Uses project-wide queries (empty zones = all zones) via GcpService,
-        filtered by iris-{prefix}-managed=true.
+        filtered by iris-{prefix}-managed=true. Slices tagged
+        iris-{prefix}-manual=true (operator-created via `iris cluster
+        create-slice`) are excluded: the autoscaler and `cluster stop` must
+        not see or terminate them.
         """
         managed_labels = {self._iris_labels.iris_managed: "true"}
+        manual_label = self._iris_labels.iris_manual
 
         if self._gcp.mode == ServiceMode.LOCAL:
-            return self._gcp.get_local_slices(managed_labels)  # type: ignore[return-value]
+            local_handles = self._gcp.get_local_slices(managed_labels)
+            return [h for h in local_handles if h.labels.get(manual_label) != "true"]  # type: ignore[return-value]
 
         tpu_infos = self._gcp.tpu_list(zones=[], labels=managed_labels)
         vm_infos = self._gcp.vm_list(zones=[], labels=managed_labels)
@@ -666,6 +671,8 @@ def list_all_slices(self) -> list[GcpSliceHandle | GcpVmSliceHandle]:
         for tpu in tpu_infos:
             if tpu.state not in ("READY", "CREATING"):
                 continue
+            if tpu.labels.get(manual_label) == "true":
+                continue
             handles.append(
                 GcpSliceHandle(
                     _slice_id=tpu.name,
@@ -693,6 +700,8 @@ def list_all_slices(self) -> list[GcpSliceHandle | GcpVmSliceHandle]:
                 continue
             if qr.state in ("FAILED", "SUSPENDED", "DELETING"):
                 continue
+            if qr.labels.get(manual_label) == "true":
+                continue
             handles.append(
                 GcpSliceHandle(
                     _slice_id=qr.name,
@@ -715,6 +724,8 @@ def list_all_slices(self) -> list[GcpSliceHandle | GcpVmSliceHandle]:
             slice_id = vm.labels.get(self._iris_labels.iris_slice_id, "")
             if not slice_id:
                 continue
+            if vm.labels.get(manual_label) == "true":
+                continue
             handles.append(
                 GcpVmSliceHandle(
                     _slice_id=slice_id,

lib/iris/src/iris/cluster/providers/manual/provider.py

@@ -334,7 +334,15 @@ def list_slices(
         return results
 
     def list_all_slices(self) -> list[ManualSliceHandle]:
-        return self.list_slices(zones=[], labels={self._iris_labels.iris_managed: "true"})
+        """List autoscaler-managed slices.
+
+        Excludes slices tagged iris_manual=true (operator-created via
+        `iris cluster create-slice`), which the autoscaler and
+        `iris cluster stop` must not see or terminate.
+        """
+        all_managed = self.list_slices(zones=[], labels={self._iris_labels.iris_managed: "true"})
+        manual_label = self._iris_labels.iris_manual
+        return [s for s in all_managed if s.labels.get(manual_label) != "true"]
 
     def list_vms(
         self,

lib/iris/src/iris/cluster/providers/types.py

@@ -51,6 +51,10 @@ def __init__(self, prefix: str):
         self.iris_controller = f"iris-{prefix}-controller"
         self.iris_controller_address = f"iris-{prefix}-controller-address"
         self.iris_slice_id = f"iris-{prefix}-slice-id"
+        # Marks a slice as operator-created via `iris cluster create-slice`.
+        # The autoscaler ignores these: they don't count toward demand, don't
+        # participate in scale-down, and survive `iris cluster stop`.
+        self.iris_manual = f"iris-{prefix}-manual"
 
 
 def find_free_port(start: int = -1) -> int:

lib/iris/tests/cluster/providers/gcp/test_platform.py

@@ -732,6 +732,27 @@ def test_list_all_slices_returns_all_managed(platform_env: PlatformEnv):
     assert len(all_slices) == 2
 
 
+def test_list_all_slices_excludes_manual_slices(platform_env: PlatformEnv):
+    """list_all_slices drops slices labeled iris_manual=true so the autoscaler ignores them."""
+    labels = Labels(platform_env.label_prefix)
+
+    cfg_auto = _make_slice_config(platform_env, "auto-group")
+    handle_auto = platform_env.platform.create_slice(cfg_auto)
+
+    cfg_manual = _make_slice_config(platform_env, "auto-group")
+    cfg_manual.labels[labels.iris_manual] = "true"
+    handle_manual = platform_env.platform.create_slice(cfg_manual)
+
+    all_slices = platform_env.platform.list_all_slices()
+    slice_ids = {s.slice_id for s in all_slices}
+    assert handle_auto.slice_id in slice_ids
+    assert handle_manual.slice_id not in slice_ids
+
+    # Manual slices are still discoverable when explicitly asked for (delete-slice path).
+    manual_only = platform_env.platform.list_slices(zones=[platform_env.zone], labels={labels.iris_manual: "true"})
+    assert {s.slice_id for s in manual_only} == {handle_manual.slice_id}
+
+
 def test_gcp_list_all_slices_multi_zone():
     """GcpWorkerProvider.list_all_slices returns slices across multiple zones."""
     gcp_service = InMemoryGcpService(mode=ServiceMode.DRY_RUN, project_id="test-project")