[iris] Consolidate feasibility + drop worker.attributes region/zone override

Fixes a regression in marin-itest where child jobs submitted by zephyr
coordinators were rejected with "no groups in region local" and the
integration pipeline hung. The fake GCP provider synthesized
region=local on workers while groups derived region from the GCP
slice_template zone (europe-west4) — two sources of truth that the new
feasibility gate caught disagreeing.

Removes the synthetic region entirely. Group region now always derives
from slice_template.gcp.zone / coreweave.region; the worker.attributes
override path in ScalingGroup.region/zone and the duplicate writes in
_expand_tpu_pools / _expand_multi_zone_groups were 100% redundant.
_validate_worker_settings now explicitly rejects REGION/ZONE in
worker.attributes so stale configs fail loudly.

Collapses the feasibility logic that PR 4681 introduced into a single
predicate: job_feasibility() + _diagnose() in routing.py, exposed on
Autoscaler as one job_feasibility(constraints, replicas=None).
service.launch_job calls it once instead of chaining two checks. The
two near-identical diagnose functions become one.

Net -206 LOC.

Author: rjpower

lib/iris/examples/coreweave.yaml

@@ -83,7 +83,6 @@ scale_groups:
       capacity_type: on-demand
     worker:
       attributes:
-        region: US-WEST-04A
         pool: cpu-erapids
     buffer_slices: 1
     max_slices: 4
@@ -106,7 +105,6 @@ scale_groups:
       capacity_type: on-demand
     worker:
       attributes:
-        region: US-WEST-04A
         pool: h100-8x
     buffer_slices: 0
     max_slices: 8

lib/iris/src/iris/cluster/config.py

@@ -224,44 +224,26 @@ def _validate_worker_settings(config: config_pb2.IrisClusterConfig) -> None:
         WellKnownAttribute.PREEMPTIBLE,
         WellKnownAttribute.DEVICE_TYPE,
         WellKnownAttribute.DEVICE_VARIANT,
+        WellKnownAttribute.REGION,
+        WellKnownAttribute.ZONE,
     }
     for name, sg_config in config.scale_groups.items():
         if not sg_config.HasField("worker"):
             continue
 
         attributes = sg_config.worker.attributes
 
-        # Reject well-known keys that are now derived from resources
+        # Reject well-known keys that are now derived from resources or slice_template.
+        # region/zone are derived from slice_template.gcp.zone; device fields from resources.
         for attr_key in _well_known_resource_attrs:
             if attr_key in attributes:
                 raise ValueError(
-                    f"Scale group '{name}': worker.attributes.{attr_key} is now derived from "
-                    f"resources and must not be set explicitly. Remove it from worker.attributes."
-                )
-
-        region = attributes.get(WellKnownAttribute.REGION, "").strip()
-        if WellKnownAttribute.REGION in attributes and not region:
-            raise ValueError(f"Scale group '{name}': worker.attributes.region must be non-empty.")
-
-        zone_attr = attributes.get(WellKnownAttribute.ZONE, "").strip()
-        if WellKnownAttribute.ZONE in attributes and not zone_attr:
-            raise ValueError(f"Scale group '{name}': worker.attributes.zone must be non-empty.")
-        if zone_attr and sg_config.slice_template.HasField("gcp") and sg_config.slice_template.gcp.zone:
-            if zone_attr != sg_config.slice_template.gcp.zone:
-                raise ValueError(
-                    f"Scale group '{name}': worker.attributes.zone={zone_attr!r} must match "
-                    f"slice_template.gcp.zone={sg_config.slice_template.gcp.zone!r}."
+                    f"Scale group '{name}': worker.attributes.{attr_key} is derived automatically "
+                    f"(from resources or slice_template.gcp.zone) and must not be set explicitly. "
+                    f"Remove it from worker.attributes."
                 )
 
         template = sg_config.slice_template
-        if region and template.HasField("gcp") and template.gcp.zone:
-            zone_region = template.gcp.zone.rsplit("-", 1)[0]
-            if region != zone_region:
-                raise ValueError(
-                    f"Scale group '{name}': worker.attributes.region={region!r} must match "
-                    f"slice_template.gcp.zone region {zone_region!r}."
-                )
-
         if (
             template.HasField("coreweave")
             and sg_config.resources.device_type == config_pb2.ACCELERATOR_TYPE_GPU
@@ -676,7 +658,6 @@ def _expand_tpu_pools(data: dict) -> None:
             topo = get_tpu_topology(variant)
 
             for zone in zones:
-                region = zone.rsplit("-", 1)[0]
                 sg_name = f"tpu_{pool_name}_{size_int}-{zone}"
 
                 if sg_name in scale_groups:
@@ -705,12 +686,6 @@ def _expand_tpu_pools(data: dict) -> None:
                     "buffer_slices": size_overrides.get("buffer_slices", 0),
                     "max_slices": size_overrides["max_slices"],
                     "slice_template": st,
-                    "worker": {
-                        "attributes": {
-                            WellKnownAttribute.ZONE: zone,
-                            WellKnownAttribute.REGION: region,
-                        }
-                    },
                 }
 
                 scale_groups[sg_name] = sg
@@ -737,15 +712,14 @@ def _expand_multi_zone_groups(data: dict) -> None:
     creates a copy of the scale group with:
     - name suffixed with -{zone} (e.g. tpu_v5e_16-europe-west4-b)
     - slice_template.gcp.zone set to the zone
-    - worker.attributes.zone and worker.attributes.region set automatically
     - buffer_slices defaulted to 0 if not explicitly set
 
     Also merges all expanded zones into platform.gcp.zones.
 
     Raises:
         ValueError: If zones is not a non-empty list of unique non-empty strings,
             if an expanded name collides with an existing scale group, or if
-            user-provided zone/region fields conflict with the expansion.
+            slice_template.gcp.zone is set while zones is also specified.
     """
     scale_groups = data.get("scale_groups")
     if not isinstance(scale_groups, dict):
@@ -785,28 +759,14 @@ def _expand_multi_zone_groups(data: dict) -> None:
 
         # Detect conflicts with user-provided fields that expansion will set
         existing_gcp_zone = (sg.get("slice_template") or {}).get("gcp", {}).get("zone")
-        existing_worker_attrs = (sg.get("worker") or {}).get("attributes", {})
-        existing_zone_attr = existing_worker_attrs.get(WellKnownAttribute.ZONE)
-        existing_region_attr = existing_worker_attrs.get(WellKnownAttribute.REGION)
 
         if existing_gcp_zone:
             raise ValueError(
                 f"Scale group '{name}': cannot set both 'zones' and 'slice_template.gcp.zone'. "
                 f"Remove slice_template.gcp.zone — it is set automatically by zone expansion."
             )
-        if existing_zone_attr:
-            raise ValueError(
-                f"Scale group '{name}': cannot set both 'zones' and 'worker.attributes.zone'. "
-                f"Remove worker.attributes.zone — it is set automatically by zone expansion."
-            )
-        if existing_region_attr:
-            raise ValueError(
-                f"Scale group '{name}': cannot set both 'zones' and 'worker.attributes.region'. "
-                f"Remove worker.attributes.region — it is set automatically by zone expansion."
-            )
 
         for zone in zones:
-            region = zone.rsplit("-", 1)[0]
             expanded_name = f"{name}-{zone}"
 
             if expanded_name in scale_groups:
@@ -826,12 +786,6 @@ def _expand_multi_zone_groups(data: dict) -> None:
             gcp = st.setdefault("gcp", {})
             gcp["zone"] = zone
 
-            # Set worker.attributes.zone and .region
-            worker = expanded_sg.setdefault("worker", {})
-            attrs = worker.setdefault("attributes", {})
-            attrs[WellKnownAttribute.ZONE] = zone
-            attrs[WellKnownAttribute.REGION] = region
-
             if "buffer_slices" not in expanded_sg:
                 expanded_sg["buffer_slices"] = 0
 

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

@@ -14,6 +14,7 @@
 from iris.cluster.constraints import (
     ConstraintIndex,
     DeviceType,
+    PlacementRequirements,
     extract_placement_requirements,
     get_device_type_enum,
     routing_constraints,
@@ -228,52 +229,6 @@ def _format_variants(variants: frozenset[str] | None) -> str:
     return ",".join(sorted(variants))
 
 
-def _diagnose_no_matching_group(entry: DemandEntry, groups: list[ScalingGroup]) -> str:
-    """Produce a concise, actionable reason when no group matches a demand entry."""
-
-    normalized = entry.normalized
-    device_type = normalized.device_type or DeviceType.CPU
-    device_matches = []
-    for group in groups:
-        if group.matches_device_requirement(device_type, normalized.device_variants):
-            device_matches.append(group)
-
-    variants_str = _format_variants(normalized.device_variants)
-    if not device_matches:
-        return f"no_matching_group: no groups with device {device_type.value}:{variants_str}"
-
-    if normalized.preemptible is not None:
-        preempt_matches = [
-            group
-            for group in device_matches
-            if (group.config.resources.capacity_type == config_pb2.CAPACITY_TYPE_PREEMPTIBLE) == normalized.preemptible
-        ]
-        if not preempt_matches:
-            want = "preemptible" if normalized.preemptible else "non-preemptible"
-            return f"no_matching_group: no {want} groups for device {device_type.value}:{variants_str}"
-        device_matches = preempt_matches
-
-    if normalized.required_zones:
-        available_zones = {group.zone for group in device_matches} - {None}
-        requested = sorted(normalized.required_zones)
-        message = f"no_matching_group: no groups in zone {', '.join(requested)}"
-        for requested_zone in requested:
-            close = difflib.get_close_matches(requested_zone, available_zones, n=1, cutoff=0.7)
-            if close:
-                message += f" (did you mean {close[0]}?)"
-        return message
-
-    if normalized.required_regions:
-        requested = sorted(normalized.required_regions)
-        region_message = f"no_matching_group: no groups in region {', '.join(requested)}"
-        return region_message
-
-    return (
-        "no_matching_group: no groups match device="
-        f"{device_type.value}:{_format_variants(normalized.device_variants)}"
-    )
-
-
 # GCP zones end with -{single letter}, e.g. us-central1-a.
 _ZONE_PATTERN = re.compile(r".+-[a-z]$")
 
@@ -282,42 +237,41 @@ def _looks_like_zone(value: str) -> bool:
     return bool(_ZONE_PATTERN.fullmatch(value))
 
 
-def diagnose_unsatisfiable_constraints(
-    constraints: Sequence[job_pb2.Constraint],
-    groups: list[ScalingGroup],
+def _diagnose(
+    placement: PlacementRequirements,
+    groups: Sequence[ScalingGroup],
 ) -> str:
-    """Produce a user-facing error when no scaling group can satisfy constraints.
+    """Explain why no scaling group satisfies a placement requirement.
 
-    Performs layered diagnosis (device, preemptible, zone, region) and
-    detects zone/region value confusion.
+    Layered analysis (device → preemptible → zone → region) with zone/region
+    confusion heuristics and fuzzy-match hints. Returned string has no prefix;
+    callers prepend their own (e.g. "no_matching_group: ") when needed.
     """
-    normalized = extract_placement_requirements(constraints)
-    device_type = normalized.device_type or DeviceType.CPU
+    device_type = placement.device_type or DeviceType.CPU
+    device_matches = [g for g in groups if g.matches_device_requirement(device_type, placement.device_variants)]
+    variants_str = _format_variants(placement.device_variants)
 
-    device_matches = [g for g in groups if g.matches_device_requirement(device_type, normalized.device_variants)]
-    variants_str = _format_variants(normalized.device_variants)
     if not device_matches:
         available = ", ".join(g.name for g in groups)
         return f"no scaling group provides device {device_type.value}:{variants_str} (available: {available})"
 
-    if normalized.preemptible is not None:
+    if placement.preemptible is not None:
         preempt_matches = [
             g
             for g in device_matches
-            if (g.config.resources.capacity_type == config_pb2.CAPACITY_TYPE_PREEMPTIBLE) == normalized.preemptible
+            if (g.config.resources.capacity_type == config_pb2.CAPACITY_TYPE_PREEMPTIBLE) == placement.preemptible
         ]
         if not preempt_matches:
-            want = "preemptible" if normalized.preemptible else "non-preemptible"
+            want = "preemptible" if placement.preemptible else "non-preemptible"
             return f"no {want} group provides device {device_type.value}:{variants_str}"
         device_matches = preempt_matches
 
-    if normalized.required_zones:
+    if placement.required_zones:
         available_zones = {g.zone for g in device_matches} - {None}
         available_regions = {g.region for g in device_matches} - {None}
-        requested = sorted(normalized.required_zones)
+        requested = sorted(placement.required_zones)
         parts = [f"no groups in zone {', '.join(requested)}"]
         for z in requested:
-            # Prioritize zone/region confusion over fuzzy match
             if not _looks_like_zone(z) and z in available_regions:
                 parts.append(f"'{z}' looks like a region, not a zone; use a region constraint instead")
             else:
@@ -326,13 +280,12 @@ def diagnose_unsatisfiable_constraints(
                     parts.append(f"did you mean {close[0]}?")
         return "; ".join(parts)
 
-    if normalized.required_regions:
+    if placement.required_regions:
         available_regions = {g.region for g in device_matches} - {None}
         available_zones = {g.zone for g in device_matches} - {None}
-        requested = sorted(normalized.required_regions)
+        requested = sorted(placement.required_regions)
         parts = [f"no groups in region {', '.join(requested)}"]
         for r in requested:
-            # Prioritize zone/region confusion over fuzzy match
             if _looks_like_zone(r) and r in available_zones:
                 parts.append(f"'{r}' looks like a zone, not a region; use a zone constraint instead")
             else:
@@ -345,6 +298,70 @@ def diagnose_unsatisfiable_constraints(
     return f"no scaling group matches constraints (available: {available})"
 
 
+@dataclass(frozen=True)
+class GroupFeasibility:
+    """Result of the job_feasibility predicate.
+
+    `feasible` is the subset of groups whose hard routing constraints match
+    and (if coscheduled) have a compatible num_vms. Non-empty means the job
+    can, in principle, be scheduled; an autoscaler tick may still need to
+    grow a group before capacity appears.
+
+    `reason` is populated iff `feasible` is empty, with a user-facing
+    explanation suitable for rejecting the job at submit time.
+    """
+
+    feasible: list[ScalingGroup]
+    reason: str | None
+
+
+def job_feasibility(
+    groups: Sequence[ScalingGroup],
+    constraints: Sequence[job_pb2.Constraint],
+    replicas: int | None = None,
+) -> GroupFeasibility:
+    """Answer: can any scaling group ever host this job shape?
+
+    Ignores runtime availability (quota, cooldown, in-flight capacity) — that
+    is the autoscaler's job on each tick. This predicate gates LaunchJob at
+    submit time so jobs that can never be scheduled fail fast.
+
+    Args:
+        groups: scaling groups to consider.
+        constraints: the job's hard + soft routing constraints.
+        replicas: for coscheduled jobs, the required replica count; None for
+            non-coscheduled jobs. When set, groups must also have num_vms that
+            divides replicas evenly.
+    """
+    groups_list = list(groups)
+    if not groups_list:
+        return GroupFeasibility(feasible=[], reason=None)
+
+    group_attrs = {g.name: g.to_attributes() for g in groups_list}
+    group_index = ConstraintIndex.build(group_attrs)
+    hard_cs, _ = split_hard_soft(routing_constraints(constraints))
+    matching_names = group_index.matching_entities(hard_cs)
+    matching = [g for g in groups_list if g.name in matching_names]
+
+    if not matching:
+        placement = extract_placement_requirements(constraints)
+        return GroupFeasibility(feasible=[], reason=_diagnose(placement, groups_list))
+
+    if replicas is not None:
+        compatible = [g for g in matching if g.num_vms > 0 and replicas % g.num_vms == 0]
+        if not compatible:
+            sizes = {g.name: g.num_vms for g in matching}
+            reason = (
+                f"job requires {replicas} coscheduled replicas but no matching scaling group "
+                f"has a compatible size (replicas must be an exact multiple of num_vms); "
+                f"matching group sizes: {sizes}"
+            )
+            return GroupFeasibility(feasible=[], reason=reason)
+        matching = compatible
+
+    return GroupFeasibility(feasible=matching, reason=None)
+
+
 def _diagnose_no_capacity(
     entry: DemandEntry,
     matching_groups: list[ScalingGroup],
@@ -493,7 +510,7 @@ def route_demand(
             reason = (
                 f"tier_blocked: {pre_tier_count} matching group(s) blocked by quota-pool tier monotonicity"
                 if pre_tier_count > 0
-                else _diagnose_no_matching_group(entry, sorted_groups)
+                else f"no_matching_group: {_diagnose(entry.normalized, sorted_groups)}"
             )
             unmet.append(UnmetDemand(entry=entry, reason=reason))
             continue