runpod(phase3-5): pod pool + scaler + cost ledger, budgets, global-gate bypass, spillover, stats UI

Completes the RunPod backend for testing.

Pods pool (runpod_worker.py):
- RunpodPodPool: provision via _select_gpu (cheaper=lowest $/hr, faster=SECURE+
  throughput, across allowed pools, spot-aware) → create_pod → wait_ready →
  vLLM health-check; least-loaded per-request acquire, provision-on-demand up to
  max_pods; background scaler thread tears down idle pods to min after
  idle_timeout_s, keeps min warm, reaps. get_pod_pool registry + pods_status().
- Budget enforcement at provision: global $/hr rate cap + per-model & global
  cumulative spend caps (rolling period) via the ledger; refuses over budget.

Cost ledger (runpod_ledger.py): persistent per-model + global spend, trailing
rolling windows (hour/day/week/month), totals() for stats. Final pod cost booked
on teardown; live pod cost added on top for enforcement/stats (no double count).

Backend (backends/runpod.py): pods mode wired to the pool (per-request acquire/
release; async stream acquires off the event loop); serverless unchanged. Config
adds global_cost_limit_usd + global_cost_period.

Global-gate bypass (app.py _model_capacity): runpod models get an effectively
unlimited LOCAL queue cap — remote GPUs aren't bound by the local concurrency
gate; RunPod's max_pods/workers are the real limit.

Spillover (app.py): a local model can burst to a RunPod SERVERLESS target when
local can't serve it — front reverse-proxies to the serverless OpenAI URL, guarded
to fall back to local on any error. Wired at the direct path for on_no_gpu
(no engine) + on_concurrency_full (queue full). NOTE partial: broker path,
streaming-concurrency, pods-target and on_local_error triggers are follow-ups.

UI: dedicated /admin/runpod stats page (pods, live $/hr vs caps, spend by period +
model) + nav link; Tasks page RunPod box; Settings split into category tabs
(JS-only, IDs/save untouched) + global spend cap fields.

Verified: imports in-image (8b16914ee24b), ledger + GPU selection (cheaper/faster)
unit-tested, all templates Jinja-parse. Live RunPod calls (provision/serverless)
still need an end-to-end test with the user's account.
Co-Authored-By: 's avatarClaude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01R6iugoNeEgt9PkajKrFyq4
parent 17604ab4
......@@ -3617,6 +3617,8 @@ def build_settings_dict(c, gpu_cards):
"default_gpu_type": c.runpod.default_gpu_type,
"data_center": c.runpod.data_center,
"global_max_hourly_usd": c.runpod.global_max_hourly_usd,
"global_cost_limit_usd": c.runpod.global_cost_limit_usd,
"global_cost_period": c.runpod.global_cost_period,
},
"ocr": {
"enabled": c.ocr.enabled,
......@@ -4158,6 +4160,11 @@ async def api_save_settings(request: Request, username: str = Depends(require_ad
if "global_max_hourly_usd" in d:
try: rp.global_max_hourly_usd = max(0.0, float(d.get("global_max_hourly_usd") or 0))
except (TypeError, ValueError): pass
if "global_cost_limit_usd" in d:
try: rp.global_cost_limit_usd = max(0.0, float(d.get("global_cost_limit_usd") or 0))
except (TypeError, ValueError): pass
if "global_cost_period" in d:
rp.global_cost_period = (d.get("global_cost_period") or "unlimited").strip().lower()
if "ocr" in data:
d = data["ocr"]
......@@ -4318,6 +4325,35 @@ async def api_runpod_status(username: str = Depends(require_admin)):
"total_hourly_usd": round(total, 4)}
@router.get("/admin/api/runpod/stats", summary="RunPod pod usage + spend stats")
async def api_runpod_stats(username: str = Depends(require_admin)):
"""coderai-managed pod pool status + spend ledger (local view; no RunPod call).
Runs on the engine hosting the RunPod pool, so it sees the live pods and the
persistent spend ledger."""
from codai.api import runpod_worker, runpod_ledger
try:
pods = runpod_worker.pods_status()
except Exception:
pods = []
try:
ledger = runpod_ledger.totals()
except Exception:
ledger = {"per_model": {}, "global": {}}
live_hourly = sum((p.get("hourly_usd") or 0.0) for p in pods if p.get("healthy"))
live_cost = sum((p.get("live_cost_usd") or 0.0) for p in pods)
cfg = config_manager.config.runpod if config_manager and config_manager.config else None
caps = {}
if cfg is not None:
caps = {"global_max_hourly_usd": cfg.global_max_hourly_usd,
"global_cost_limit_usd": cfg.global_cost_limit_usd,
"global_cost_period": cfg.global_cost_period,
"enabled": cfg.enabled}
return {"success": True, "pods": pods, "ledger": ledger, "caps": caps,
"live_hourly_usd": round(live_hourly, 4),
"live_uncommitted_usd": round(live_cost, 4)}
# =============================================================================
# Archive management
# =============================================================================
......
......@@ -78,6 +78,7 @@
<a href="{{ root_path }}/admin/users" class="nav-link {% if '/users' in request.url.path %}active{% endif %}">Users</a>
<a href="{{ root_path }}/admin/archive" class="nav-link {% if '/archive' in request.url.path %}active{% endif %}">Archive</a>
<a href="{{ root_path }}/admin/settings" class="nav-link {% if '/settings' in request.url.path %}active{% endif %}">Settings</a>
<a href="{{ root_path }}/admin/runpod" class="nav-link {% if '/runpod' in request.url.path %}active{% endif %}">RunPod</a>
{% endif %}
<button class="nav-link nav-donate-btn" onclick="document.getElementById('donateModal').classList.add('show')">&#9829; Donate</button>
</div>
......
{% extends "base.html" %}
{% block title %}RunPod — CoderAI{% endblock %}
{% block content %}
<div class="page-header">
<div>
<h1>RunPod</h1>
<p>Remote-GPU pods, live cost, and spend against your budgets.</p>
</div>
<div class="header-actions">
<span id="rp-updated" class="dim small"></span>
<button class="btn btn-secondary" onclick="rpRefresh()">Refresh</button>
</div>
</div>
<div id="rp-disabled" style="display:none;margin:0 0 1rem;padding:.7rem .9rem;border-radius:8px;
background:rgba(148,163,184,.12);border:1px solid rgba(148,163,184,.4);font-size:13px">
RunPod is not enabled. Turn it on in <a href="{{ root_path }}/admin/settings">Settings → RunPod</a>.
</div>
<!-- Cost tiles -->
<div id="rp-tiles" style="display:grid;grid-template-columns:repeat(auto-fit,minmax(200px,1fr));gap:.75rem;margin-bottom:1rem">
<div class="card"><div class="dim small">Running pods</div><div id="rp-podcount" style="font-size:1.6rem;font-weight:700">—</div></div>
<div class="card"><div class="dim small">Live rate ($/hr)</div><div id="rp-rate" style="font-size:1.6rem;font-weight:700">—</div><div id="rp-rate-cap" class="dim small"></div></div>
<div class="card"><div class="dim small">Uncommitted cost</div><div id="rp-uncommitted" style="font-size:1.6rem;font-weight:700">—</div><div class="dim small">pods running, not yet billed to the ledger</div></div>
<div class="card"><div class="dim small">Global spend (period)</div><div id="rp-global-spend" style="font-size:1.6rem;font-weight:700">—</div><div id="rp-global-cap" class="dim small"></div></div>
</div>
<!-- Running pods -->
<div class="card" style="margin-bottom:1rem">
<div class="card-title">Running pods</div>
<div style="overflow-x:auto">
<table style="width:100%;border-collapse:collapse;font-size:13px">
<thead><tr style="text-align:left;color:var(--text-3)">
<th>Model</th><th>GPU</th><th>Pool</th><th>In-flight</th><th>$/hr</th><th>Uptime</th><th>Cost so far</th><th>Health</th>
</tr></thead>
<tbody id="rp-pods"><tr><td colspan="8" class="dim">No pods running.</td></tr></tbody>
</table>
</div>
</div>
<!-- Spend per model -->
<div class="card">
<div class="card-title">Spend by model (trailing)</div>
<div style="overflow-x:auto">
<table style="width:100%;border-collapse:collapse;font-size:13px">
<thead><tr style="text-align:left;color:var(--text-3)">
<th>Model</th><th>Hour</th><th>Day</th><th>Week</th><th>Month</th>
</tr></thead>
<tbody id="rp-spend"><tr><td colspan="5" class="dim">No spend recorded yet.</td></tr></tbody>
</table>
</div>
</div>
{% endblock %}
{% block scripts %}
<script>
function _fmtDur(s){ s=Math.max(0,s|0); const h=(s/3600)|0, m=((s%3600)/60)|0; return h?`${h}h ${m}m`:`${m}m ${s%60}s`; }
function _usd(v){ return (v==null)?'—':('$'+Number(v).toFixed(v<1?4:2)); }
async function rpRefresh(){
let d;
try {
const r = await fetch(ROOT_PATH + '/admin/api/runpod/stats');
d = await r.json();
if(!r.ok) throw new Error(d.detail || ('HTTP '+r.status));
} catch(e){
document.getElementById('rp-updated').textContent = 'error: ' + e.message;
return;
}
const caps = d.caps || {};
document.getElementById('rp-disabled').style.display = caps.enabled ? 'none' : '';
const pods = d.pods || [];
const running = pods.filter(p=>p.healthy);
document.getElementById('rp-podcount').textContent = running.length + (pods.length>running.length ? ` (+${pods.length-running.length} unhealthy)` : '');
document.getElementById('rp-rate').textContent = _usd(d.live_hourly_usd);
document.getElementById('rp-rate-cap').textContent = caps.global_max_hourly_usd ? ('cap ' + _usd(caps.global_max_hourly_usd) + '/hr') : 'no rate cap';
document.getElementById('rp-uncommitted').textContent = _usd(d.live_uncommitted_usd);
const gperiod = caps.global_cost_period || 'unlimited';
const gspend = (d.ledger && d.ledger.global && gperiod!=='unlimited') ? d.ledger.global[gperiod] : null;
document.getElementById('rp-global-spend').textContent = (gperiod==='unlimited') ? 'unlimited' : _usd(gspend);
document.getElementById('rp-global-cap').textContent = (caps.global_cost_limit_usd && gperiod!=='unlimited') ? ('cap ' + _usd(caps.global_cost_limit_usd) + ' / ' + gperiod) : 'no spend cap';
// pods table
const tb = document.getElementById('rp-pods');
if(!pods.length){ tb.innerHTML = '<tr><td colspan="8" class="dim">No pods running.</td></tr>'; }
else {
tb.innerHTML = pods.map(p => `<tr>
<td>${p.model||''}</td>
<td>${p.gpu||''}${p.is_spot?' <span class="dim">(spot)</span>':''}</td>
<td>${p.is_spot?'spot':'on-demand'}</td>
<td>${p.inflight||0}</td>
<td>${_usd(p.hourly_usd)}</td>
<td>${_fmtDur(p.uptime_s)}</td>
<td>${_usd(p.live_cost_usd)}</td>
<td>${p.healthy?'<span style="color:#22c55e">healthy</span>':'<span style="color:#ef4444">down</span>'}</td>
</tr>`).join('');
}
// spend by model
const sb = document.getElementById('rp-spend');
const per = (d.ledger && d.ledger.per_model) || {};
const keys = Object.keys(per);
if(!keys.length){ sb.innerHTML = '<tr><td colspan="5" class="dim">No spend recorded yet.</td></tr>'; }
else {
sb.innerHTML = keys.map(k => {
const m = per[k] || {};
return `<tr><td>${k}</td><td>${_usd(m.hour)}</td><td>${_usd(m.day)}</td><td>${_usd(m.week)}</td><td>${_usd(m.month)}</td></tr>`;
}).join('');
}
const now = new Date();
document.getElementById('rp-updated').textContent = 'updated ' + now.toLocaleTimeString();
}
rpRefresh();
setInterval(rpRefresh, 10000);
</script>
{% endblock %}
......@@ -15,6 +15,8 @@
<div id="settings-alert" style="display:none"></div>
<div id="settings-tabbar" style="display:flex;flex-wrap:wrap;gap:.4rem;margin:0 0 1rem;border-bottom:1px solid var(--border,#2a2a2a);padding-bottom:.6rem"></div>
<!-- Server -->
<div class="card mb-0">
<div class="card-title">Server</div>
......@@ -844,6 +846,24 @@
<div class="form-group" style="flex:1;min-width:140px">
<label class="form-label">Global $/hr cap</label>
<input type="number" id="s-runpod-budget" class="form-input" min="0" step="0.1" placeholder="0 = no cap">
<span class="form-hint" style="font-size:11px">Instantaneous rate: total $/hr across all running pods.</span>
</div>
</div>
<div class="form-row" style="display:flex;gap:1rem;flex-wrap:wrap">
<div class="form-group" style="flex:1;min-width:150px">
<label class="form-label">Global spend cap ($)</label>
<input type="number" id="s-runpod-cost-limit" class="form-input" min="0" step="1" placeholder="0 = unlimited">
<span class="form-hint" style="font-size:11px">Cumulative spend across all RunPod models over the period below.</span>
</div>
<div class="form-group" style="flex:1;min-width:150px">
<label class="form-label">Spend period</label>
<select id="s-runpod-cost-period" class="form-input">
<option value="unlimited">Unlimited</option>
<option value="hour">Hour</option>
<option value="day">Day</option>
<option value="week">Week</option>
<option value="month">Month</option>
</select>
</div>
</div>
<div class="form-group">
......@@ -1761,6 +1781,8 @@ async function loadSettings(){
document.getElementById('s-runpod-cloud').value = rp.cloud_type || 'SECURE';
document.getElementById('s-runpod-default-gpu').value = rp.default_gpu_type ?? '';
document.getElementById('s-runpod-budget').value = rp.global_max_hourly_usd ?? 0;
document.getElementById('s-runpod-cost-limit').value = rp.global_cost_limit_usd ?? 0;
document.getElementById('s-runpod-cost-period').value = rp.global_cost_period || 'unlimited';
document.getElementById('s-runpod-dc').value = rp.data_center ?? '';
document.getElementById('s-runpod-api-base').value = rp.api_base ?? '';
document.getElementById('s-runpod-rest-base').value = rp.rest_base ?? '';
......@@ -1952,6 +1974,8 @@ async function saveSettings(silent){
cloud_type: document.getElementById('s-runpod-cloud').value || 'SECURE',
default_gpu_type: document.getElementById('s-runpod-default-gpu').value.trim(),
global_max_hourly_usd: parseFloat(document.getElementById('s-runpod-budget').value) || 0,
global_cost_limit_usd: parseFloat(document.getElementById('s-runpod-cost-limit').value) || 0,
global_cost_period: document.getElementById('s-runpod-cost-period').value || 'unlimited',
data_center: document.getElementById('s-runpod-dc').value.trim(),
api_base: document.getElementById('s-runpod-api-base').value.trim(),
rest_base: document.getElementById('s-runpod-rest-base').value.trim(),
......@@ -2036,5 +2060,62 @@ async function saveSettings(silent){
}
loadSettings();
// Group the settings cards into category tabs (JS-only: no markup/ID changes, so
// the single Save still submits every field regardless of the visible tab).
function _initSettingsTabs(){
const TABS = [
["Server & Queue", ["Server","Concurrency","Storage","Background Jobs"]],
["Engines", ["DeepSeek","colibri","Kimi-K3","ktransformers","vLLM"]],
["RunPod", ["RunPod"]],
["OCR", ["OCR"]],
["Media", ["Video Enhancement","Generation Archive"]],
["System", ["Thermal","AISBF","Auto-compact","Credits"]],
];
const bar = document.getElementById('settings-tabbar');
if (!bar) return;
// Assign each card to a tab by its title text.
const cards = Array.from(document.querySelectorAll('.card')).filter(c => c.querySelector('.card-title'));
const tabOf = (title) => {
for (const [name, keys] of TABS)
if (keys.some(k => title.toLowerCase().includes(k.toLowerCase()))) return name;
return "Other";
};
const present = [];
cards.forEach(c => {
const t = (c.querySelector('.card-title').textContent || '').trim();
const tab = tabOf(t);
c.dataset.setTab = tab;
if (!present.includes(tab)) present.push(tab);
});
// Order tabs as declared, then any extras (Other).
const order = TABS.map(t => t[0]).filter(n => present.includes(n))
.concat(present.filter(n => !TABS.some(t => t[0] === n)));
let active = null;
try { active = localStorage.getItem('coderai_settings_tab'); } catch(e){}
if (!order.includes(active)) active = order[0];
const show = (tab) => {
active = tab;
try { localStorage.setItem('coderai_settings_tab', tab); } catch(e){}
cards.forEach(c => { c.style.display = (c.dataset.setTab === tab) ? '' : 'none'; });
bar.querySelectorAll('button').forEach(b => {
const on = b.dataset.tab === tab;
b.style.background = on ? 'var(--accent,#6366f1)' : 'transparent';
b.style.color = on ? '#fff' : 'var(--text-2,#aaa)';
b.style.borderColor = on ? 'var(--accent,#6366f1)' : 'var(--border,#2a2a2a)';
});
};
bar.innerHTML = '';
order.forEach(tab => {
const b = document.createElement('button');
b.type = 'button'; b.dataset.tab = tab; b.textContent = tab;
b.style.cssText = 'padding:.4rem .8rem;border-radius:8px;border:1px solid var(--border,#2a2a2a);'
+ 'background:transparent;color:var(--text-2,#aaa);cursor:pointer;font-size:13px;font-weight:500';
b.onclick = () => show(tab);
bar.appendChild(b);
});
show(active);
}
_initSettingsTabs();
</script>
{% endblock %}
......@@ -47,6 +47,16 @@
<div id="engines-body" style="display:grid;grid-template-columns:repeat(auto-fit,minmax(240px,1fr));gap:.6rem"></div>
</div>
<!-- RunPod remote-GPU box (shown only when RunPod is enabled) -->
<div id="runpod-card" style="display:none;margin:0 0 1.25rem">
<div style="display:flex;align-items:baseline;gap:.5rem;margin-bottom:.5rem">
<h2 style="font-size:14px;margin:0">RunPod (remote GPUs)</h2>
<span class="dim small" id="runpod-rate"></span>
<a href="{{ root_path }}/admin/runpod" class="dim small" style="margin-left:auto">stats &amp; spend →</a>
</div>
<div id="runpod-body" style="display:grid;grid-template-columns:repeat(auto-fit,minmax(240px,1fr));gap:.6rem"></div>
</div>
<style>
.sys-tile{border:1px solid var(--border,#2a2a2a);border-radius:10px;padding:.7rem .85rem;
background:var(--card-bg,rgba(255,255,255,.02))}
......@@ -442,11 +452,42 @@ async function removeTask(id) {
loadTasks();
}
async function loadRunpod(){
let d;
try {
const r = await fetch(ROOT_PATH + '/admin/api/runpod/stats');
if (!r.ok) { document.getElementById('runpod-card').style.display = 'none'; return; }
d = await r.json();
} catch(e) { document.getElementById('runpod-card').style.display = 'none'; return; }
const caps = d.caps || {};
const pods = d.pods || [];
// Show the box only when RunPod is enabled (or there are pods to report).
if (!caps.enabled && !pods.length) { document.getElementById('runpod-card').style.display = 'none'; return; }
document.getElementById('runpod-card').style.display = '';
const rate = d.live_hourly_usd || 0;
document.getElementById('runpod-rate').textContent =
pods.filter(p=>p.healthy).length + ' pod(s) · $' + Number(rate).toFixed(2) + '/hr';
const body = document.getElementById('runpod-body');
if (!pods.length) {
body.innerHTML = '<div class="dim small" style="padding:.5rem">No pods running (scaled to zero — a request will provision one).</div>';
return;
}
body.innerHTML = pods.map(p => `
<div class="sys-tile">
<div style="font-weight:600;font-size:13px">${p.model||''}</div>
<div class="dim small">${p.gpu||''}${p.is_spot?' (spot)':''}</div>
<div class="small" style="margin-top:.3rem">in-flight: <b>${p.inflight||0}</b> · $${Number(p.hourly_usd||0).toFixed(2)}/hr</div>
<div class="dim small">up ${Math.round((p.uptime_s||0)/60)}m · ${p.healthy?'<span style="color:#22c55e">healthy</span>':'<span style="color:#ef4444">down</span>'}</div>
</div>`).join('');
}
loadTasks();
loadSystemStats();
loadEngines();
loadRunpod();
setInterval(loadTasks, 1000);
setInterval(loadSystemStats, 1000);
setInterval(loadEngines, 5000);
setInterval(loadRunpod, 5000);
</script>
{% endblock %}
# CoderAI - OpenAI-compatible API server
# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
"""RunPod spend ledger — persistent per-model + global cost accounting.
Records FINAL pod costs (uptime x hourly rate) as they're torn down. Spend for a
period is a trailing rolling window ("$100 per week" = the last 7 days), which
avoids calendar-boundary edge cases. Live in-flight cost (pods still running) is
NOT stored here — it's computed from the pool and added on top for enforcement
and stats, so nothing is double-counted.
Enforcement (in the pool) refuses to provision a pod when the model's or the
account's period spend would be exceeded. cost_period "unlimited" (or a
cost_limit of 0) means no cap.
"""
import json
import os
import threading
import time
_lock = threading.RLock()
_PERIOD_SECONDS = {
"hour": 3600.0,
"day": 86400.0,
"week": 7 * 86400.0,
"month": 30 * 86400.0,
}
_PRUNE_SECONDS = 31 * 86400.0 # keep at most ~1 month of history
_GLOBAL_KEY = "__global__"
def _store_path() -> str:
p = os.environ.get("CODERAI_RUNPOD_LEDGER")
if p:
return p
for d in ("/config/coderai", os.path.expanduser("~/.coderai")):
if os.path.isdir(d):
return os.path.join(d, "runpod_ledger.json")
return os.path.join(os.path.expanduser("~/.coderai"), "runpod_ledger.json")
def _load() -> dict:
try:
with open(_store_path()) as f:
d = json.load(f)
return d if isinstance(d, dict) else {}
except Exception:
return {}
def _save(data: dict) -> None:
path = _store_path()
try:
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
tmp = path + ".tmp"
with open(tmp, "w") as f:
json.dump(data, f)
os.replace(tmp, path)
except Exception as exc:
print(f"[runpod-ledger] save failed: {exc}", flush=True)
def _prune(entries: list, now: float) -> list:
cutoff = now - _PRUNE_SECONDS
return [e for e in entries if (e.get("ts") or 0) >= cutoff]
def record_spend(model_key: str, usd: float, meta: dict = None) -> None:
"""Append a final spend record for a torn-down pod (and to the global tally)."""
if not usd or usd <= 0:
return
now = time.time()
rec = {"ts": now, "usd": float(usd)}
if meta:
rec.update({k: meta[k] for k in ("pod_id", "gpu", "is_spot") if k in meta})
with _lock:
data = _load()
for key in (model_key or "?", _GLOBAL_KEY):
lst = data.get(key) or []
lst.append(dict(rec))
data[key] = _prune(lst, now)
_save(data)
def _sum_in_window(entries: list, seconds: float, now: float) -> float:
cutoff = now - seconds
return sum((e.get("usd") or 0.0) for e in entries if (e.get("ts") or 0) >= cutoff)
def spend_in_period(model_key: str, period: str) -> float:
"""Trailing spend for one model over the given rolling period (0 if unlimited)."""
secs = _PERIOD_SECONDS.get((period or "unlimited").lower())
if not secs:
return 0.0
now = time.time()
with _lock:
return _sum_in_window(_load().get(model_key or "?") or [], secs, now)
def global_spend_in_period(period: str) -> float:
"""Trailing spend across ALL RunPod models over the given rolling period."""
secs = _PERIOD_SECONDS.get((period or "unlimited").lower())
if not secs:
return 0.0
now = time.time()
with _lock:
return _sum_in_window(_load().get(_GLOBAL_KEY) or [], secs, now)
def totals() -> dict:
"""Snapshot for the stats page: trailing spend per model + global, all periods."""
now = time.time()
out = {"per_model": {}, "global": {}}
with _lock:
data = _load()
for period, secs in _PERIOD_SECONDS.items():
out["global"][period] = round(_sum_in_window(data.get(_GLOBAL_KEY) or [], secs, now), 4)
for key, entries in data.items():
if key == _GLOBAL_KEY:
continue
out["per_model"][key] = {
period: round(_sum_in_window(entries, secs, now), 4)
for period, secs in _PERIOD_SECONDS.items()
}
return out
This diff is collapsed.
......@@ -60,10 +60,12 @@ class RunpodBackend(ModelBackend):
self._acct = cfg
self._model_id = "runpod"
self._served_name = "runpod"
self._url: Optional[str] = None
self._url: Optional[str] = None # serverless: fixed base URL
self._headers: Dict[str, str] = {}
self._ctx = 32768
self._mcfg = None
self._mode = "serverless"
self._pool = None # pods: RunpodPodPool
self._last_usage: Dict = {}
# ------------------------------------------------------------------ #
......@@ -90,18 +92,24 @@ class RunpodBackend(ModelBackend):
if _ctx > 0:
self._ctx = _ctx
mode = (self._mcfg.mode or "pods").lower()
if mode == "serverless":
self._mode = (self._mcfg.mode or "pods").lower()
if self._mode == "serverless":
self._url = runpod_worker.serverless_base_url(self._acct, self._mcfg)
self._headers = runpod_worker.auth_headers(self._acct)
print(f"[runpod] '{model_name}' -> serverless endpoint "
f"{self._mcfg.endpoint_id} ({self._url})", flush=True)
else:
# pods / auto: provisioned pool — lands in the next phase.
raise RuntimeError(
f"RunPod mode '{mode}' is not available yet — the pod pool is "
"built in a later phase. Set the model's runpod mode to 'serverless' "
"for now.")
# pods / auto: coderai-managed pool of remote GPU pods. (auto currently
# behaves as pods; serverless-vs-pods auto-arbitration is a later refinement.)
self._pool = runpod_worker.get_pod_pool(
model_name, self._acct, self._mcfg, self._served_name)
self._headers = {} # the pod's vLLM server needs no auth
# Eagerly warm a pod only when min_pods >= 1; otherwise provision lazily
# on the first request (min_pods=0 = scale-to-zero when idle).
if self._mcfg.min_pods >= 1:
self._pool.ensure_ready()
print(f"[runpod] '{model_name}' -> managed pod pool "
f"(min={self._mcfg.min_pods}, max={self._mcfg.max_pods})", flush=True)
def get_model_name(self) -> str:
return self._model_id
......@@ -125,6 +133,16 @@ class RunpodBackend(ModelBackend):
raise RuntimeError("RunPod endpoint not resolved")
return self._url
def _acquire_endpoint(self):
"""Return (base_url, release_callable) for one request. Serverless yields the
fixed endpoint; pods pick (and provision-on-demand) a pod from the pool."""
if self._mode == "serverless":
return self._base(), (lambda: None)
if not self._pool:
raise RuntimeError("RunPod pod pool not initialised")
pod, url = self._pool.acquire()
return url, (lambda: self._pool.release(pod))
def _store_usage(self, usage: dict) -> None:
if usage:
self._last_usage = {
......@@ -162,33 +180,38 @@ class RunpodBackend(ModelBackend):
top_p=1.0, stop=None, tools=None, response_format=None):
import requests
self._enter_request()
base, release = self._acquire_endpoint()
try:
payload = self._chat_payload(messages, max_tokens, temperature, top_p, stop, False)
if response_format and response_format.get("type") == "json_object":
payload["response_format"] = {"type": "json_object"}
if tools:
payload["tools"] = tools
r = requests.post(self._base() + "/chat/completions", json=payload,
r = requests.post(base + "/chat/completions", json=payload,
headers=self._headers, timeout=3600)
r.raise_for_status()
data = r.json()
self._store_usage(data.get("usage", {}))
return data["choices"][0]["message"].get("content") or ""
finally:
release()
self._exit_request()
async def generate_chat_stream(self, messages: List[Dict], max_tokens=None,
temperature=0.7, top_p=1.0, stop=None, tools=None,
response_format=None) -> AsyncGenerator[str, None]:
self._enter_request()
# Pod acquire may provision (blocking) — do it off the event loop.
base, release = await asyncio.to_thread(self._acquire_endpoint)
try:
payload = self._chat_payload(messages, max_tokens, temperature, top_p, stop, True)
if tools:
payload["tools"] = tools
async for chunk in self._stream(self._base() + "/chat/completions", payload,
async for chunk in self._stream(base + "/chat/completions", payload,
delta_key="delta"):
yield chunk
finally:
release()
self._exit_request()
# ------------------------------------------------------------------ #
......
......@@ -580,6 +580,11 @@ class RunpodConfig:
# Account-wide safety net: the scaler will not let the SUM of all running RunPod
# pods' hourly cost exceed this. 0 = no global cap (per-model $/hr still applies).
global_max_hourly_usd: float = 0.0
# Account-wide CUMULATIVE spend cap over a rolling period (distinct from the $/hr
# rate cap above): the scaler refuses to provision when trailing spend across ALL
# RunPod models would exceed this. 0 / "unlimited" = no cap.
global_cost_limit_usd: float = 0.0
global_cost_period: str = "unlimited" # hour | day | week | month | unlimited
@dataclass
......@@ -1134,6 +1139,8 @@ class ConfigManager:
"default_gpu_type": self.config.runpod.default_gpu_type,
"data_center": self.config.runpod.data_center,
"global_max_hourly_usd": self.config.runpod.global_max_hourly_usd,
"global_cost_limit_usd": self.config.runpod.global_cost_limit_usd,
"global_cost_period": self.config.runpod.global_cost_period,
},
"ocr": {
"enabled": self.config.ocr.enabled,
......
......@@ -638,7 +638,12 @@ class FrontProxy:
"load_status_updates": m.get("load_status_updates"),
# Per-model wait-keepalive mode (None = inherit global
# models.wait_status_mode): silent | invisible | visible.
"wait_status_mode": m.get("wait_status_mode")}
"wait_status_mode": m.get("wait_status_mode"),
# Optional burst-to-RunPod spillover config for a LOCAL model
# (enabled + triggers + serverless target). Used by the front
# to reroute when local can't serve.
"runpod_spillover": (m.get("runpod_spillover")
if isinstance(m.get("runpod_spillover"), dict) else None)}
for field_ in (m.get("path"), m.get("id"), m.get("alias")):
if not field_:
continue
......@@ -668,6 +673,68 @@ class FrontProxy:
except Exception:
pass
async def _spill_to_runpod(self, model, trigger, request, path, body_bytes):
"""Burst a request to a RunPod SERVERLESS spillover target when the local
model can't serve it. Returns a Response/StreamingResponse on spill, or None
to keep normal (local) behaviour.
A serverless endpoint is OpenAI-compatible, so this is a plain reverse-proxy
hop to a remote HTTPS URL with a Bearer header — no request parsing. Fully
guarded: any problem returns None so the caller falls back to local.
(Pods-target spillover routes through the engine and is not handled here.)
"""
try:
if not (model and body_bytes):
return None
sp = self._model_info(model).get("runpod_spillover")
if not isinstance(sp, dict) or not sp.get("enabled"):
return None
if not sp.get(trigger):
return None
target = sp.get("target") if isinstance(sp.get("target"), dict) else {}
if (target.get("mode") or "serverless").lower() != "serverless":
return None # pods-target spill goes through the engine, not here
rp = getattr(self.config, "runpod", None)
if rp is None or not getattr(rp, "enabled", False) or not getattr(rp, "api_key", ""):
return None
eid = (target.get("endpoint_id") or "").strip()
if not eid:
return None
# /v1/chat/completions -> <serverless_base>/<eid>/openai/v1/chat/completions
p = path.split("?", 1)[0]
suffix = p[len("/v1"):] if p.startswith("/v1/") else "/chat/completions"
base = (getattr(rp, "serverless_base", "") or "https://api.runpod.ai/v2").rstrip("/")
url = f"{base}/{eid}/openai/v1{suffix}"
# Rewrite the model to the target's served name, if set.
body = body_bytes
served = (target.get("served_model") or "").strip()
if served:
try:
import json as _j
obj = _j.loads(body_bytes or b"{}")
if isinstance(obj, dict):
obj["model"] = served
body = _j.dumps(obj).encode()
except Exception:
body = body_bytes
headers = {"Authorization": f"Bearer {rp.api_key}",
"Content-Type": "application/json"}
print("[runpod-spill] model=%s trigger=%s -> serverless %s"
% (model, trigger, eid), flush=True)
rp_req = self._long.build_request("POST", url, headers=headers, content=body)
rp_resp = await self._long.send(rp_req, stream=True)
return StreamingResponse(
rp_resp.aiter_raw(), status_code=rp_resp.status_code,
headers=dict(self._filter_headers(rp_resp.headers, _DROP_RESP)),
media_type=rp_resp.headers.get("content-type"),
background=BackgroundTask(rp_resp.aclose))
except Exception as exc:
try:
print("[runpod-spill] failed, falling back to local: %s" % exc, flush=True)
except Exception:
pass
return None
def _queue_key(self, model: Optional[str]) -> str:
"""Stable gate key for a model: its canonical id (so every alias/path of
the same model shares one gate), else the raw model string."""
......@@ -824,6 +891,13 @@ class FrontProxy:
"""Per-model concurrency = its max_instances, falling back to the global
server default. This is the number of front queue slots for the model."""
info = self._model_info(model)
# RunPod-served models run on REMOTE GPUs with their own pod-level scaling
# (max_pods) — they must bypass the LOCAL global concurrency gate, which
# exists to protect local VRAM. Return an effectively-unlimited local cap so
# the front never queues them behind the small local default; RunPod's own
# max_pods / serverless workers are the real limit.
if (info.get("backend") or "").lower() == "runpod":
return 1_000_000
mi = info.get("max_instances")
if mi:
try:
......@@ -1877,6 +1951,11 @@ class FrontProxy:
default_engine=self.default_engine, pinned=self._pin_for(model),
pin_fallback=bool(self._model_info(model).get("engine_fallback")))
if engine is None:
# No local engine can serve it (none ready / no GPU meeting specs).
# If this model has RunPod spillover for that case, burst to the cloud.
spilled = await self._spill_to_runpod(model, "on_no_gpu", request, path, body_bytes)
if spilled is not None:
return spilled
return JSONResponse(
{"error": "No engine is ready yet (still starting/loading)."},
status_code=503)
......@@ -1935,6 +2014,12 @@ class FrontProxy:
rid=engine.name + ":" + (model or ""), model=model or "",
engine=engine.name)
except QueueFull:
# Local model at capacity — burst to RunPod if spillover is enabled
# for the concurrency case; else 503.
spilled = await self._spill_to_runpod(
model, "on_concurrency_full", request, path, body_bytes)
if spilled is not None:
return spilled
return JSONResponse(
{"error": "Server busy: the generation queue is full, "
"please retry shortly."}, status_code=503)
......
......@@ -153,6 +153,13 @@ def register_ui_pages(app: FastAPI, config_dir) -> bool:
return redir
return _tmpl(request, "settings.html", {"username": u, "is_admin": True})
@app.get("/admin/runpod", include_in_schema=False)
async def _runpod_stats_page(request: Request):
u, redir = _auth_or_redirect(request, admin=True)
if redir:
return redir
return _tmpl(request, "runpod.html", {"username": u, "is_admin": True})
@app.get("/admin/archive", include_in_schema=False)
async def _archive_page(request: Request):
u, redir = _auth_or_redirect(request, admin=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