Ray Orphan Reaper

ray_reaper

Reclaim Ray clusters whose run is over — the backstop for a launcher that never cleaned up.

Every Ray cluster this product creates is torn down in a finally by the process that created it. That is not a guarantee, it is a hope that nothing kills the process: kill -9, a preempted VM, a Composer worker eviction, an OOM and a laptop that sleeps all defeat it, and each one leaves a head node plus workers — possibly an accelerator — billing against a run that no longer exists. This was not theoretical. A dev machine restarted mid-provision on 2026-09-12 and left a head node, a CPU worker and a T4 running for twenty-five minutes with nobody watching.

The Dataproc path does not have this problem, which is what makes the gap worth a module. dataproc_cluster.build_lifecycle_config attaches an idle TTL and a max age to every cluster we build, so the identical kill leaves something the service reclaims without us. Vertex's PersistentResource has no such field to set — there is no TTL, no idle timeout, no auto-delete anywhere on the resource — so the only mechanism available is to come back later and look. That is this module.

What makes "later" safe is that garbage is decidable, not guessed at. Three independent facts have to line up before anything is deleted:

• the cluster carries our app label and a registry label naming this registry (ray_cluster.cluster_labels), so a second deployment sharing the project is invisible to us;
• its name starts with sf-ray- (engines.ray_io.EPHEMERAL_PREFIX), which every run-derived ephemeral cluster has and nothing else is checked against;
• the run_id embedded in that name resolves to a run header that is terminal, or to no header at all and the cluster is older than DEFAULT_MIN_AGE_SECONDS.

A standing reuse target is safe because of the first fact, not the second. The reuse path skips create and teardown (ray_submit.submit_ray), so a cluster an operator provisioned themselves was never labelled by us and ray_cluster.list_clusters does not even return it — which holds however they chose to name it. The name check is the second line, for a labelled cluster of ours that a config then pointed a reuse run at.

The age floor exists for exactly one race: a cluster that has just been created by a run whose header is not yet written reads as "unknown run" for a few seconds. It applies only to the no-header case — a cluster whose run is already COMPLETED is garbage the moment it is seen, however young, because the only thing that could have produced it is a teardown that did not happen.

Name-to-run matching is by prefix, not equality, because ray_io.cluster_name clamps to 63 characters and a long enough run_name loses the tail of its id. A prefix that matches several runs is treated as matching all of them: one live match keeps the cluster. Ambiguity resolves toward leaving a machine running, never toward deleting one.

Preview is the default and yes=True executes, the same shape every destructive verb in registry.ops uses — and the plan prints the kept clusters as prominently as the doomed ones, because "why is this one still here?" is the question an operator actually arrives with.

Two ways in, because a verb alone is not a ceiling. reap_clusters is the operator's command and previews by default. sweep_on_launch is the same decision run automatically, immediately before this deployment creates a Ray cluster — the one moment it is guaranteed to be awake, and the moment a leaked cluster is holding the quota the new one needs. It swallows its own failures, because no cleanup is worth a launch that will not start, and it can be switched off with SF_REAP_ON_LAUNCH=0.

Split along the usual pure/I-O seam: classify_clusters and format_reap_plan are pure and decide everything, while plan_reap_clusters, reap_clusters and sweep_on_launch only fetch and act.

Public surface: reap_clusters, sweep_on_launch, plan_reap_clusters, classify_clusters, format_reap_plan, ReapPlan, ReapCandidate, DEFAULT_MIN_AGE_SECONDS, SWEEP_ON_LAUNCH_ENV. Reachable as registry-ops reap-clusters and as sdk.Registry.reap_clusters; the Vertex listing and the delete it drives are ray_cluster.list_clusters / ray_cluster.teardown_shared_cluster.

ReapCandidate dataclass

ReapCandidate(
    cluster, run_id_prefix, matched_runs, reap, reason
)

One cluster the reaper looked at, and why it lived or died.

matched_runs is every run id in the registry whose name the cluster's embedded id is a prefix of — usually one, empty when the run is unknown, and occasionally several when a clamped name is ambiguous. reason is written for an operator reading a preview, so it names the run and the status rather than restating the rule.

ReapPlan dataclass

ReapPlan(
    registry,
    regions=(),
    candidates=(),
    min_age_seconds=DEFAULT_MIN_AGE_SECONDS,
)

Exactly which Ray clusters reap_clusters would delete, and what it is leaving alone.

classify_clusters

classify_clusters(
    clusters,
    status_by_run,
    *,
    registry_dataset_id,
    now,
    min_age_seconds=DEFAULT_MIN_AGE_SECONDS,
)

Decide each cluster's fate (pure). Every cluster comes back with a verdict and a reason.

The checks run cheapest-and-most-exclusionary first — registry label, then ephemeral name, then the STOPPING short-circuit, then the run header — so a cluster that is not ours exits before anything has to reason about run ids. status_by_run is the whole registry's latest header per run; a run id missing from it is a run this registry has never heard of.

Source code in src/scale_forecasting/ray_reaper.py

def classify_clusters(
    clusters: Sequence[RayCluster],
    status_by_run: Mapping[str, str | None],
    *,
    registry_dataset_id: str,
    now: datetime,
    min_age_seconds: float = DEFAULT_MIN_AGE_SECONDS,
) -> tuple[ReapCandidate, ...]:
    """Decide each cluster's fate (pure). Every cluster comes back with a verdict and a reason.

    The checks run cheapest-and-most-exclusionary first — registry label, then ephemeral name, then
    the ``STOPPING`` short-circuit, then the run header — so a cluster that is not ours exits before
    anything has to reason about run ids. ``status_by_run`` is the whole registry's latest header
    per run; a run id missing from it is a run this registry has never heard of.
    """
    out: list[ReapCandidate] = []
    for cluster in clusters:
        prefix = run_id_prefix_from_cluster_name(cluster.name)
        matched = tuple(sorted(r for r in status_by_run if prefix and r.startswith(prefix)))

        owner = cluster.labels.get(REGISTRY_LABEL_KEY)
        if owner is not None and owner != registry_dataset_id:
            reap, reason = False, f"belongs to registry '{owner}', not this one"
        elif prefix is None:
            reap, reason = False, "not a run-derived cluster — a reuse target we must never delete"
        elif cluster.state == _GOING_AWAY:
            reap, reason = False, "already STOPPING"
        else:
            reap, reason = _verdict(
                matched,
                status_by_run,
                age=_age_seconds(cluster, now),
                min_age_seconds=min_age_seconds,
            )
        out.append(
            ReapCandidate(
                cluster=cluster,
                run_id_prefix=prefix,
                matched_runs=matched,
                reap=reap,
                reason=reason,
            )
        )
    return tuple(out)

format_reap_plan

format_reap_plan(plan)

Render a ReapPlan as the operator-facing preview — the exact text a dry run prints.

Kept clusters are listed too, and by design: a reaper that prints only what it will delete cannot answer "so why is that T4 still running?", which is the question that brings an operator here in the first place.

Source code in src/scale_forecasting/ray_reaper.py

def format_reap_plan(plan: ReapPlan) -> str:
    """Render a `ReapPlan` as the operator-facing preview — the exact text a dry run prints.

    Kept clusters are listed too, and by design: a reaper that prints only what it will delete
    cannot answer "so why is that T4 still running?", which is the question that brings an operator
    here in the first place.
    """
    lines = [f"reap-clusters against registry {plan.registry}"]
    lines.append(f"  regions: {', '.join(plan.regions) or '(none)'}")
    lines.append(f"  unknown-run age floor: {plan.min_age_seconds:.0f}s")
    if plan.reapable:
        lines.append(f"  will DELETE ({len(plan.reapable)}):")
        for c in plan.reapable:
            lines.append(f"    {c.cluster.name}  [{c.cluster.region} {c.cluster.state}]")
            lines.append(f"      {c.reason}")
    if plan.kept:
        lines.append(f"  leaving alone ({len(plan.kept)}):")
        for c in plan.kept:
            lines.append(f"    {c.cluster.name}  [{c.cluster.region} {c.cluster.state}]")
            lines.append(f"      {c.reason}")
    if not plan.candidates:
        lines.append("  no scale-forecasting Ray clusters found")
    return "\n".join(lines)

plan_reap_clusters

plan_reap_clusters(
    *,
    settings=None,
    regions=None,
    min_age_seconds=DEFAULT_MIN_AGE_SECONDS,
)

Which Ray clusters reap_clusters would delete — without deleting anything.

regions defaults to the data-plane region alone. Widen it when a run's capacity walk may have hopped the cluster elsewhere; a region left off the list is one the sweep cannot clean, and ray_cluster.list_clusters raises rather than quietly skipping one it cannot read.

Source code in src/scale_forecasting/ray_reaper.py

def plan_reap_clusters(
    *,
    settings: Settings | None = None,
    regions: Sequence[str] | None = None,
    min_age_seconds: float = DEFAULT_MIN_AGE_SECONDS,
) -> ReapPlan:  # pragma: no cover - GCP I/O, @gcp smoke
    """Which Ray clusters `reap_clusters` would delete — without deleting anything.

    ``regions`` defaults to the data-plane region alone. Widen it when a run's capacity walk may
    have hopped the cluster elsewhere; a region left off the list is one the sweep cannot clean,
    and `ray_cluster.list_clusters` raises rather than quietly skipping one it cannot read.
    """
    from datetime import UTC, datetime

    from .ray_cluster import list_clusters
    from .registry.ops import all_header_statuses

    resolved = _settings(settings)
    where = tuple(regions or (resolved.region,))
    clusters = list_clusters(resolved, where)
    # Read the headers only when there is something to judge — the usual answer is zero clusters,
    # and a preview that costs a BigQuery query every time is a preview nobody runs on a schedule.
    statuses = all_header_statuses(resolved) if clusters else {}
    return ReapPlan(
        registry=resolved.registry_dataset_ref,
        regions=where,
        candidates=classify_clusters(
            clusters,
            statuses,
            registry_dataset_id=resolved.registry_dataset_id,
            now=datetime.now(UTC),
            min_age_seconds=min_age_seconds,
        ),
        min_age_seconds=min_age_seconds,
    )

reap_clusters

reap_clusters(
    *,
    settings=None,
    regions=None,
    min_age_seconds=DEFAULT_MIN_AGE_SECONDS,
    yes=False,
)

Delete the Ray clusters whose run is over. Preview by default; yes=True executes.

Deletes through ray_cluster.teardown_shared_cluster, which is the same verified teardown a normal run uses: the SDK's success line is ignored and the resource is polled until it reads NOT_FOUND. A cluster that refuses to go is logged by name as still billing rather than counted as reclaimed — a reaper that reports a delete it did not achieve is worse than none at all, because it also removes the operator's reason to go and look.

Every cluster is attempted even if one delete fails, for the reason shared_clusters.teardown_spark_clusters gives: stopping at the first failure strands all of the ones behind it, which is precisely the outcome this module exists to prevent.

Source code in src/scale_forecasting/ray_reaper.py

def reap_clusters(
    *,
    settings: Settings | None = None,
    regions: Sequence[str] | None = None,
    min_age_seconds: float = DEFAULT_MIN_AGE_SECONDS,
    yes: bool = False,
) -> ReapPlan:  # pragma: no cover - GCP I/O, @gcp smoke
    """Delete the Ray clusters whose run is over. Preview by default; ``yes=True`` executes.

    Deletes through `ray_cluster.teardown_shared_cluster`, which is the same verified teardown a
    normal run uses: the SDK's success line is ignored and the resource is polled until it reads
    ``NOT_FOUND``. A cluster that refuses to go is logged by name as still billing rather than
    counted as reclaimed — a reaper that reports a delete it did not achieve is worse than none at
    all, because it also removes the operator's reason to go and look.

    Every cluster is attempted even if one delete fails, for the reason
    `shared_clusters.teardown_spark_clusters` gives: stopping at the first failure strands all of
    the ones behind it, which is precisely the outcome this module exists to prevent.
    """
    resolved = _settings(settings)
    plan = plan_reap_clusters(settings=resolved, regions=regions, min_age_seconds=min_age_seconds)
    _log.warning("%s", format_reap_plan(plan))
    if plan.is_empty:
        _log.warning("nothing to reap in %s", plan.registry)
        return plan
    if not yes:
        _log.warning("DRY RUN — no cluster deleted. Re-run with yes=True to execute.")
        return plan

    _delete_all(plan, resolved)
    return plan

sweep_on_launch

sweep_on_launch(settings, regions)

Reclaim finished-run clusters just before creating another one. Best effort; never raises.

The verb above only runs when a person runs it, and a leak is by definition something that happened while nobody was watching. Vertex has no cluster-side ceiling to lean on instead (no TTL, no idle timeout), so the closest thing to "always" available here is every time this deployment provisions a Ray cluster — which is also when it matters most, since the leaked cluster is holding the regional quota the create is about to ask for.

Two deliberate choices:

• Failure is swallowed. A launch must not die because a cleanup read failed; the worst case of skipping the sweep is the status quo before it existed, while the worst case of raising is a run that never starts. Anything that goes wrong is logged and the create proceeds.
• It is an environment switch, not a config field. SF_REAP_ON_LAUNCH=0 turns it off. A RunConfig field would change every run's run_id — the id is a digest of the config — and a deployment-level cleanup policy has no business changing the identity of a run's results. Same reasoning as SF_SERVERLESS_DEPS living on the infra object.

Returns the names it reclaimed, for the caller that wants to say so.

Source code in src/scale_forecasting/ray_reaper.py

def sweep_on_launch(settings: Settings, regions: Sequence[str]) -> tuple[str, ...]:
    """Reclaim finished-run clusters just before creating another one. Best effort; never raises.

    The verb above only runs when a person runs it, and a leak is by definition something that
    happened while nobody was watching. Vertex has no cluster-side ceiling to lean on instead (no
    TTL, no idle timeout), so the closest thing to "always" available here is *every time this
    deployment provisions a Ray cluster* — which is also when it matters most, since the leaked
    cluster is holding the regional quota the create is about to ask for.

    Two deliberate choices:

    * **Failure is swallowed.** A launch must not die because a cleanup read failed; the worst case
      of skipping the sweep is the status quo before it existed, while the worst case of raising is
      a run that never starts. Anything that goes wrong is logged and the create proceeds.
    * **It is an environment switch, not a config field.** ``SF_REAP_ON_LAUNCH=0`` turns it off. A
      `RunConfig` field would change every run's ``run_id`` — the id is a digest of the config — and
      a deployment-level cleanup policy has no business changing the identity of a run's results.
      Same reasoning as ``SF_SERVERLESS_DEPS`` living on the infra object.

    Returns the names it reclaimed, for the caller that wants to say so.
    """
    import os

    if os.environ.get(SWEEP_ON_LAUNCH_ENV, "1").strip().lower() in _OFF:
        return ()
    try:
        plan = plan_reap_clusters(settings=settings, regions=regions)
        if plan.is_empty:
            return ()
        _log.warning("%s", format_reap_plan(plan))
        return _delete_all(plan, settings)
    except Exception as exc:  # noqa: BLE001 - a cleanup must never be the reason a run fails
        _log.warning("launch-time Ray cluster sweep skipped (%r); provisioning anyway", exc)
        return ()