runpod: let a client lease warm pods up to max_pods (v0.2.93)

The autoscaler reacts to load — a pod is added once the least-loaded one
already has scale_up_inflight_per_pod requests in flight. That is right for
traffic nobody predicted and wrong for a client that KNOWS what is coming: a
batch of a thousand documents otherwise discovers the pool one cold start at a
time.

So a client may now say so: POST /v1/runpod/scale {model, pods}. It is a lease,
bounded three ways, because a client that can raise the floor can raise the
bill:

  * allow_client_scale, off by default, so no model is client-scalable unless
    someone said so;
  * clamped to max_pods — asking for more returns the ceiling with
    "clamped": true rather than an error, since the client asked for "as many
    as you can" and refusing would leave it with none;
  * client_scale_ttl_s (default 30 min), because the failure that matters is
    not a wrong number, it is a client that asks for four A40s and then dies.
    A lease that never expired would be a bill that never stopped, so a 0 in
    the config falls back to the default instead of meaning "forever".

The lease and a warm-pod schedule are MAXED, never summed: both mean "hold this
many ready", so a client cannot lower a configured floor and a closed window
cannot cancel a lease. /v1/runpod/status reports the live lease per model, so an
unexpected bill has a visible cause.

Not admin-scoped, unlike /v1/runpod/spend: the gate is the per-model flag, not
the key, and the account's $/hr and spend caps still refuse to provision.
Co-Authored-By: 's avatarClaude Opus 5 <noreply@anthropic.com>
parent 05802ce0
...@@ -16,7 +16,7 @@ ...@@ -16,7 +16,7 @@
# Canonical product version for CoderAI — single source of truth. Both the API # Canonical product version for CoderAI — single source of truth. Both the API
# metadata and the admin web UI read from here. # metadata and the admin web UI read from here.
__version__ = "0.2.92" __version__ = "0.2.93"
__tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference." __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
# Configure the CUDA caching allocator BEFORE torch is imported anywhere. # Configure the CUDA caching allocator BEFORE torch is imported anywhere.
......
...@@ -27,6 +27,7 @@ from typing import List ...@@ -27,6 +27,7 @@ from typing import List
from fastapi import FastAPI, HTTPException, Request from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import FileResponse, JSONResponse from fastapi.responses import FileResponse, JSONResponse
from pydantic import BaseModel, ConfigDict, Field
from codai.broker.client import BrokerClient from codai.broker.client import BrokerClient
from codai.broker.service import BrokerService from codai.broker.service import BrokerService
...@@ -851,6 +852,41 @@ async def runpod_status(request: Request): ...@@ -851,6 +852,41 @@ async def runpod_status(request: Request):
return await runpod_spend(request) return await runpod_spend(request)
class RunpodScaleRequest(BaseModel):
model: str = Field(..., description="Model id whose pod pool to scale.")
pods: int = Field(..., ge=0,
description="Warm pods to hold, clamped to the model's max_pods. "
"0 releases the lease.")
model_config = ConfigDict(extra="allow")
@app.post("/v1/runpod/scale", summary="Ask for warm pods on a model", tags=["Core"])
async def runpod_scale(body: RunpodScaleRequest, request: Request):
"""Let a client hold N pods warm for a model, up to its configured max_pods.
For a client that KNOWS what is coming — a batch of a thousand documents —
and would otherwise discover the pool one cold start at a time. The
orchestrator already grows a pool under load
(``scale_up_inflight_per_pod``), so this is not needed for ordinary traffic;
it removes the ramp, it does not replace the autoscaler.
Deliberately NOT admin-scoped: the gate is per model
(``allow_client_scale``), the number is clamped to ``max_pods``, and the
lease expires (``client_scale_ttl_s``, default 30 min) so a client that dies
holding four A40s stops paying for them by itself. The account's $/hr and
spend caps still refuse to provision regardless of what is asked here.
"""
from codai.api import runpod_worker
try:
return runpod_worker.set_client_pods(body.model, body.pods)
except LookupError as exc:
raise HTTPException(status_code=404, detail=str(exc))
except PermissionError as exc:
raise HTTPException(status_code=403, detail=str(exc))
except Exception as exc:
raise HTTPException(status_code=500, detail=f"scale failed: {exc}")
@app.get("/coderai/capabilities", summary="Server capability document", tags=["Core"]) @app.get("/coderai/capabilities", summary="Server capability document", tags=["Core"])
async def get_broker_capabilities(): async def get_broker_capabilities():
"""Return broker capability metadata.""" """Return broker capability metadata."""
......
...@@ -174,6 +174,14 @@ class RunpodModelConfig: ...@@ -174,6 +174,14 @@ class RunpodModelConfig:
# block requests: one arriving outside the window still cold-starts a pod, # block requests: one arriving outside the window still cold-starts a pod,
# exactly as min_pods=0 always has. # exactly as min_pods=0 always has.
schedule_enabled: bool = False schedule_enabled: bool = False
# Let a CLIENT ask for a warm floor of its own, up to max_pods, through
# POST /v1/runpod/scale. Off by default: a client that can raise the floor
# can raise the bill, so it is opt-in per model. The client's floor is a
# LEASE — it expires after client_scale_ttl_s unless renewed — because the
# failure that matters is not a wrong number, it is a client that asks for
# four pods and then dies, leaving them billing until someone notices.
allow_client_scale: bool = False
client_scale_ttl_s: int = 1800
schedule_days: list = field(default_factory=list) # 0=Mon … 6=Sun; empty = every day schedule_days: list = field(default_factory=list) # 0=Mon … 6=Sun; empty = every day
schedule_start: str = "" # "HH:MM" local to schedule_tz schedule_start: str = "" # "HH:MM" local to schedule_tz
schedule_end: str = "" # "HH:MM"; end <= start means overnight schedule_end: str = "" # "HH:MM"; end <= start means overnight
...@@ -1514,6 +1522,11 @@ def parse_model_runpod(block: Optional[dict]) -> RunpodModelConfig: ...@@ -1514,6 +1522,11 @@ def parse_model_runpod(block: Optional[dict]) -> RunpodModelConfig:
# two that can disagree. # two that can disagree.
cfg.min_pods = max(1, cfg.min_pods) cfg.min_pods = max(1, cfg.min_pods)
cfg.max_pods = max(1, _as_int(b.get("max_pods"), cfg.max_pods)) cfg.max_pods = max(1, _as_int(b.get("max_pods"), cfg.max_pods))
cfg.allow_client_scale = _as_bool(b.get("allow_client_scale"), cfg.allow_client_scale)
# A zero/absent TTL would mean "never expires", which is the one thing a
# lease must not mean. Keep the default rather than honour it.
cfg.client_scale_ttl_s = max(60, _as_int(b.get("client_scale_ttl_s"),
cfg.client_scale_ttl_s) or 1800)
cfg.scale_up_inflight_per_pod = max(1, _as_int(b.get("scale_up_inflight_per_pod"), cfg.scale_up_inflight_per_pod = max(1, _as_int(b.get("scale_up_inflight_per_pod"),
cfg.scale_up_inflight_per_pod)) cfg.scale_up_inflight_per_pod))
cfg.max_inflight_per_pod = max(0, _as_int(b.get("max_inflight_per_pod"), cfg.max_inflight_per_pod = max(0, _as_int(b.get("max_inflight_per_pod"),
...@@ -1946,6 +1959,11 @@ class RunpodPodPool: ...@@ -1946,6 +1959,11 @@ class RunpodPodPool:
#: Set from the pod plan at provision time; the default suits an LLM pod. #: Set from the pod plan at provision time; the default suits an LLM pod.
self.health_path = mcfg.health_path or "/v1/models" self.health_path = mcfg.health_path or "/v1/models"
self._cv = threading.Condition(threading.RLock()) self._cv = threading.Condition(threading.RLock())
#: A warm floor a client leased through /v1/runpod/scale, and when the
#: lease runs out. 0/0 = nobody asked.
self._client_floor = 0
self._client_floor_until = 0.0
self._client_floor_logged = -1
self._provisioning = False self._provisioning = False
self._closed = False self._closed = False
# One log line per window transition, not one every scaler tick. # One log line per window transition, not one every scaler tick.
...@@ -2490,6 +2508,39 @@ class RunpodPodPool: ...@@ -2490,6 +2508,39 @@ class RunpodPodPool:
print(f"[runpod] pod {pod.pod_id} for {self.model_key!r} terminated ({reason}); " print(f"[runpod] pod {pod.pod_id} for {self.model_key!r} terminated ({reason}); "
f"billed ~${cost:.3f}", flush=True) f"billed ~${cost:.3f}", flush=True)
def client_floor(self) -> int:
"""The warm floor a client leased, 0 once the lease has run out.
Read-only and cheap: the lease is not removed here, so the scaler, the
status page and the endpoint all see the same number within a pass.
"""
if self._client_floor <= 0:
return 0
if time.time() >= self._client_floor_until:
return 0
return min(self._client_floor, max(1, self.mcfg.max_pods))
def set_client_floor(self, pods: int) -> dict:
"""Lease (or release, with 0) a client-requested warm floor.
Clamped to max_pods: the configured ceiling is the budget, and a client
asking for more gets the ceiling rather than an error — it asked for
"as many as you can", and failing the call would leave it with none.
"""
cap = max(1, int(self.mcfg.max_pods))
want = max(0, int(pods))
granted = min(want, cap)
ttl = max(60, int(getattr(self.mcfg, "client_scale_ttl_s", 1800) or 1800))
with self._cv:
self._client_floor = granted
self._client_floor_until = (time.time() + ttl) if granted else 0.0
self._client_floor_logged = -1
print(f"[runpod] {self.model_key!r}: client asked for {want} warm pod(s), "
f"granted {granted} (max_pods {cap}) for {ttl}s", flush=True)
return {"model": self.model_key, "requested": want, "granted": granted,
"max_pods": cap, "expires_in_s": ttl if granted else 0,
"clamped": want > cap}
def effective_min_pods(self) -> int: def effective_min_pods(self) -> int:
"""The warm floor that applies right now. """The warm floor that applies right now.
...@@ -2497,10 +2548,23 @@ class RunpodPodPool: ...@@ -2497,10 +2548,23 @@ class RunpodPodPool:
serving (a request still cold-starts a pod) but stops paying to hold one serving (a request still cold-starts a pod) but stops paying to hold one
ready. Inside the window, or with no schedule, it is the configured ready. Inside the window, or with no schedule, it is the configured
min_pods. min_pods.
A client lease (``/v1/runpod/scale``) is the HIGHER of the two, never the
sum: it says "hold this many ready", which is already satisfied by a
schedule holding that many. So a client cannot lower the configured
floor, and the schedule cannot cancel a lease the client is relying on.
""" """
if not getattr(self.mcfg, "schedule_enabled", False): if not getattr(self.mcfg, "schedule_enabled", False):
return self.mcfg.min_pods scheduled = self.mcfg.min_pods
return self.mcfg.min_pods if schedule_state(self.mcfg)["in_window"] else 0 else:
scheduled = (self.mcfg.min_pods
if schedule_state(self.mcfg)["in_window"] else 0)
leased = self.client_floor()
if leased > scheduled and leased != self._client_floor_logged:
print(f"[runpod] {self.model_key!r}: warm floor {leased} from a client "
f"lease (configured {scheduled})", flush=True)
self._client_floor_logged = leased
return max(scheduled, min(leased, max(1, self.mcfg.max_pods)))
def maintain(self): def maintain(self):
"""One scaler pass: reap idle/dead pods to min_pods, keep min warm.""" """One scaler pass: reap idle/dead pods to min_pods, keep min warm."""
...@@ -2825,12 +2889,46 @@ def pools_schedule_status() -> list: ...@@ -2825,12 +2889,46 @@ def pools_schedule_status() -> list:
"next_change": st["next_change"], "next_change": st["next_change"],
"min_pods": cfg.min_pods, "min_pods": cfg.min_pods,
"effective_min_pods": pool.effective_min_pods(), "effective_min_pods": pool.effective_min_pods(),
"max_pods": cfg.max_pods,
"healthy_pods": healthy, "healthy_pods": healthy,
"data_center": cfg.data_center or "", "data_center": cfg.data_center or "",
# The client lease, so a bill nobody expected has a visible cause.
"client_scale_allowed": bool(getattr(cfg, "allow_client_scale", False)),
"client_pods": pool.client_floor(),
"client_lease_expires_in_s": max(
0, int(pool._client_floor_until - time.time())) if pool.client_floor() else 0,
}) })
return out return out
def set_client_pods(model: str, pods: int) -> dict:
"""Lease a warm floor for one model on a client's behalf.
Raises LookupError when the model has no RunPod block, and PermissionError
when it has one that did not opt in — told apart so the caller can answer
404 and 403, which are different problems for whoever is integrating.
"""
key = str(model or "").strip()
if not key:
raise LookupError("no model named")
block = model_runpod_block(key)
if not block:
raise LookupError(f"{key!r} has no runpod block")
mcfg = parse_model_runpod(block)
if not mcfg.allow_client_scale:
raise PermissionError(
f"{key!r} does not allow client scaling — set "
'"allow_client_scale": true on its runpod block')
if mcfg.is_serverless:
raise PermissionError(
f"{key!r} runs serverless — RunPod owns its scaling; set "
"min_workers/max_workers on the endpoint instead")
# Creating the pool is the point: asking for a warm floor on a model nobody
# has called yet is exactly when it is worth pre-warming.
pool = get_model_pod_pool(key, _model_entry(key), block)
return pool.set_client_floor(pods)
def pods_status() -> list: def pods_status() -> list:
"""Snapshot of all pods across pools for the stats/tasks pages.""" """Snapshot of all pods across pools for the stats/tasks pages."""
now = time.time() now = time.time()
......
...@@ -604,6 +604,8 @@ Four numbers decide the shape of a pool: ...@@ -604,6 +604,8 @@ Four numbers decide the shape of a pool:
| `max_pods` | hard ceiling on concurrent pods for this model. | | `max_pods` | hard ceiling on concurrent pods for this model. |
| `scale_up_inflight_per_pod` | add a pod once the least-loaded one already has this many requests in flight. **Without this the pool only grows when it has no healthy pod at all**, so fifty concurrent requests would pile onto pod #1. | | `scale_up_inflight_per_pod` | add a pod once the least-loaded one already has this many requests in flight. **Without this the pool only grows when it has no healthy pod at all**, so fifty concurrent requests would pile onto pod #1. |
| `idle_timeout_s` | destroy a pod this long after its last request. | | `idle_timeout_s` | destroy a pod this long after its last request. |
| `allow_client_scale` | let a client lease a warm floor up to `max_pods` through `POST /v1/runpod/scale`. Off by default. |
| `client_scale_ttl_s` | how long such a lease lasts unless renewed (default 1800). A lease that never expired would be a bill that never stopped. |
**Then the two that are easy to miss**, and the reason a pod can look slow while **Then the two that are easy to miss**, and the reason a pod can look slow while
its GPU is idle: its GPU is idle:
...@@ -646,6 +648,52 @@ start runs overnight and belongs to the day it started on, so `22:00`–`06:00` ...@@ -646,6 +648,52 @@ start runs overnight and belongs to the day it started on, so `22:00`–`06:00`
`fri` covers Friday night into Saturday morning. Leave `schedule_days` empty for `fri` covers Friday night into Saturday morning. Leave `schedule_days` empty for
every day. every day.
### Let a client ask for pods
The autoscaler reacts to load: it adds a pod once the least-loaded one already
has `scale_up_inflight_per_pod` requests in flight. That is the right behaviour
for traffic nobody predicted, and the wrong one for a client that KNOWS what is
coming — a batch of a thousand documents discovers the pool one cold start at a
time.
So a client can say so, on a model that opted in:
```json
"allow_client_scale": true,
"client_scale_ttl_s": 1800
```
```bash
curl -fsS -X POST http://127.0.0.1:8776/v1/runpod/scale \
-H "Authorization: Bearer sk-coderai-…" -H "Content-Type: application/json" \
-d '{"model": "qwen38-awq", "pods": 3}'
```
```json
{"model": "qwen38-awq", "requested": 3, "granted": 3,
"max_pods": 3, "expires_in_s": 1800, "clamped": false}
```
It is a **lease**, bounded three ways, because a client that can raise the floor
can raise the bill:
| bound | what it does |
|---|---|
| `allow_client_scale` | off by default — no model is scalable by a client unless you said so |
| `max_pods` | asking for more returns the ceiling with `"clamped": true`, not an error: the client asked for "as many as you can" |
| `client_scale_ttl_s` | the lease expires (default 30 min) unless renewed, so a client that dies holding four A40s stops paying for them by itself |
`"pods": 0` releases it early. The lease and a warm-pod schedule are **maxed,
never summed** — both mean "hold this many ready" — so a client cannot lower a
floor you configured, and a closed window cannot cancel a lease the client is
relying on. The account's $/hr and spend caps still refuse to provision whatever
is asked here, and `GET /v1/runpod/status` reports the live lease per model
(`client_pods`, `client_lease_expires_in_s`) so an unexpected bill has a visible
cause.
Deliberately not admin-scoped: the gate is the per-model flag, not the key. An
ordinary API key may scale a model you marked scalable, and nothing else.
--- ---
## 6. Watch the cost ## 6. Watch the cost
......
This diff is collapsed.
...@@ -128,6 +128,12 @@ def test_effective_min_pods_drops_to_zero_outside_the_window(monkeypatch): ...@@ -128,6 +128,12 @@ def test_effective_min_pods_drops_to_zero_outside_the_window(monkeypatch):
"schedule_start": "08:00", "schedule_start": "08:00",
"schedule_end": "20:00"}) "schedule_end": "20:00"})
effective_min_pods = rw.RunpodPodPool.effective_min_pods effective_min_pods = rw.RunpodPodPool.effective_min_pods
# The floor is also the max of any client lease (/v1/runpod/scale), so
# the real method needs the lease state. None here: this is the schedule.
client_floor = rw.RunpodPodPool.client_floor
_client_floor = 0
_client_floor_until = 0.0
_client_floor_logged = -1
pool = _Pool() pool = _Pool()
monkeypatch.setattr(rw, "schedule_state", lambda cfg, now=None: {"in_window": True}) monkeypatch.setattr(rw, "schedule_state", lambda cfg, now=None: {"in_window": True})
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment