Skip to content

C
coderai
Commit 0d4a338b authored by Stefy Lanza (nextime / spora )'s avatar Stefy Lanza (nextime / spora )
Browse files
Options

cluster: mDNS discovery with a shared token, prefix-cache routing, Prometheus... 

cluster: mDNS discovery with a shared token, prefix-cache routing, Prometheus /metrics + per-key usage, node logs and re-push on return, GGUF row split, cosign-signed images

- codai/cluster/discovery.py: _coderai._tcp announce + browse (zeroconf), HMAC
  fingerprint of cluster.token as membership; members auto-join as nodes,
  listed nodes with a blank key get the shared token; nodes and engines accept
  the cluster token (front /cluster/*, engine BearerAuth); env overrides for
  capability images (CODERAI_CLUSTER_TOKEN/DISCOVERY/NODE_NAME/ADVERTISE_HOST);
  coderai-docker --host-network
- frontproxy/affinity.py + router prefer=: a conversation goes back to the
  engine/node that served it when the model is resident in several places;
  engine_for_model now picks the least busy holder
- frontproxy/metrics.py: /metrics (requests per key/model/kind/status/engine,
  engines/nodes up, VRAM, in-flight, RPC, discovery, RunPod spend), usage
  tables on the Cluster page, server.metrics_public
- node logs: /cluster/logs on the node, /admin/api/cluster/logs on the head,
  log viewer on the Cluster page; a node that comes back gets its assigned
  models pushed again
- per-model split_mode=row (llama.cpp tensor parallel over local + RPC devices);
  vLLM TP may span nodes (documented)
- packaging/sign-images.sh + packaging/cosign.pub: all 21 GHCR images signed
- docs/cluster.md §§ discovery, 7 prefix routing, 9 operating, 10 accelerators
  evaluation; tests/test_cluster_ops.py (11)
Co-Authored-By: 's avatarClaude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FKm7Z56iKjHgNkHc6VhWKz
parent b0184216
Show whitespace changes
Inline Side-by-side
Showing with 1536 additions and 74 deletions
README.md
View file @ 0d4a338b
......@@ -148,13 +148,19 @@ remote placement of every model type, the capability images and the `host` backe
one — polled, assigned models, pinned by name from the model page (*Engine / card*),
routed by capability and load; the node's own front picks the card. A node needs only
an API token. Load/Unload from the head reach the node; the Cluster page shows every
node's engines, VRAM and reachability
node's engines, VRAM, reachability and log tail
- **Zero-config on a LAN**: one **cluster token** typed on every box, **mDNS discovery**
(`_coderai._tcp`) and auto-join — members find each other and become nodes with no
URL and no per-node token; a node that restarts gets its models pushed again
- **Prefix-cache-aware routing**: a conversation's next turn goes back to the engine or
node that served it (warm KV cache) when the model is loaded in several places
- **One GGUF over several machines**: llama.cpp RPC — this machine runs `rpc-server`
processes (Settings → Cluster) that lend its cards; a model names them under *Cards on
other machines* and its layer split spreads over local and remote cards alike. Built
from the same llama.cpp as the bundled llama-cpp-python (`packaging/build-rpc-server.sh`)
- **One HF model over several machines**: vLLM on Ray (tensor parallel inside a node,
pipeline parallel across nodes) and SGLang `--nnodes` — coderai starts the peers by
other machines* and its layer split — or *row* split, llama.cpp's tensor parallelism,
for a fast fabric — spreads over local and remote cards alike. Built from the same
llama.cpp as the bundled llama-cpp-python (`packaging/build-rpc-server.sh`)
- **One HF model over several machines**: vLLM on Ray (pipeline parallel across nodes on
Ethernet; tensor parallel may span them on 10 GbE/RDMA) and SGLang `--nnodes` — coderai starts the peers by
their configured commands, waits for their GPUs, launches, tears down; per model or
as defaults
- **Distributed generation and training**: a model's *Distribute* switch fans `n` images,
......@@ -166,6 +172,9 @@ remote placement of every model type, the capability images and the `host` backe
healthy + least busy wins, a dead one is skipped and the request moves on
- **No impossible pairs**: the model form greys out engines that cannot run the chosen
backend and backends the pinned engine cannot run, and refuses to save a wrong pair
- **Operable**: `GET /metrics` for Prometheus (requests per API key / model / engine,
engines and nodes up, VRAM, RunPod spend), usage tables on the Cluster page, node logs
from the head; every published image is cosign-signed (`packaging/cosign.pub`)
Full guide: [`docs/cluster.md`](docs/cluster.md).
......@@ -596,7 +605,7 @@ The [`docs/`](docs/) directory carries the deep dives — one per engine and sub
| [`runpod.md`](docs/runpod.md) | Renting remote GPUs: pods, serverless, budgets, the reaper |
| [`remote-execution.md`](docs/remote-execution.md) | Running any model type elsewhere: the capability gateway, per-model placement, weights and adapters on a pod, test runs |
| [`install-from-packages.md`](docs/install-from-packages.md) | The published images (full + nineteen capability images), Linux and Windows installs, the offline tarball |
| [`cluster.md`](docs/cluster.md) | Several machines as one: cluster nodes as engines, one GGUF over llama.cpp RPC, vLLM/SGLang multi-node, pools of hosts and remotes |
| [`cluster.md`](docs/cluster.md) | Several machines as one: cluster nodes as engines, mDNS + shared token, one GGUF over llama.cpp RPC, vLLM/SGLang multi-node, prefix-cache routing, pools, Prometheus, node logs |
| [`vllm.md`](docs/vllm.md) | vLLM as a first-class engine node |
| [`deepseek-ds4.md`](docs/deepseek-ds4.md) · [`glm-colibri.md`](docs/glm-colibri.md) · [`kimi-k3.md`](docs/kimi-k3.md) · [`ktransformers.md`](docs/ktransformers.md) | The native MoE engines |
| [`ocr.md`](docs/ocr.md) | The OCR subsystem and schema store |
......@@ -710,7 +719,7 @@ Other top-level blocks, each with its own tab on the Settings page:
| `server.engine_specs` / `engines` / `engine_gpus` | The front/engine split — how many engines, which GPU each owns, and their capabilities |
| `ds4`, `colibri`, `k3`, `ktransformers`, `vllm` | Per-engine enablement, install dirs, model ids and build settings |
| `runpod` | RunPod account settings and global cost caps — see [`docs/runpod.md`](docs/runpod.md) |
| `cluster` | Other installs used as engines (`nodes`), the `rpc_servers` this machine lends, `advertise_host` — see [`docs/cluster.md`](docs/cluster.md) |
| `cluster` | Other installs used as engines (`nodes`), the shared `token`, mDNS `discovery` + `auto_join`, `prefix_affinity`, the `rpc_servers` this machine lends, `advertise_host` — see [`docs/cluster.md`](docs/cluster.md) |
| `ocr` | OCR engines, DPI, concurrency, detection mode, structured-extraction model and venv paths |
| `broker` | AISBF broker client (below) |
......@@ -1037,9 +1046,13 @@ To force Vulkan to use only the AMD GPU:
### Multi-machine (cluster nodes, llama.cpp RPC, vLLM on Ray)
See [`docs/cluster.md`](docs/cluster.md). In short: a node is another coderai with an
API token; an RPC server is `cluster.rpc_servers` on the lending machine plus
`rpc_servers` on the model; vLLM spans machines with `pipeline_parallel_size` and
`nodes` on the model's vLLM block.
API token — or, on one LAN, every box with the same `cluster.token` and
`"discovery": true` finds the others over mDNS and joins (`coderai-docker
--host-network` so the container sees the LAN); an RPC server is `cluster.rpc_servers`
on the lending machine plus `rpc_servers` on the model (`"split_mode": "row"` for
tensor parallel on a fast fabric); vLLM spans machines with `pipeline_parallel_size`
(and/or a `tensor_parallel_size` larger than the local cards) and `nodes` on the
model's vLLM block. `GET /metrics` is the Prometheus page.
### Low VRAM
......
codai/__init__.py
View file @ 0d4a338b
......@@ -16,7 +16,7 @@
# Canonical product version for CoderAI — single source of truth. Both the API
# metadata and the admin web UI read from here.
__version__ = "0.2.20"
__version__ = "0.2.21"
__tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
# Configure the CUDA caching allocator BEFORE torch is imported anywhere.
......
codai/admin/routes.py
View file @ 0d4a338b
......@@ -2753,6 +2753,9 @@ async def api_model_configure(request: Request, username: str = Depends(require_
# and where the weights are on a cluster node this model is
# pinned to (a path, or {node: path}).
"rpc_servers", "node_path", "node_paths",
# llama.cpp split mode over those devices: "" / "layer" | "row"
# (row = tensor parallel; see backends/vulkan.py).
"split_mode",
):
if key in data:
entry[key] = data[key]
......@@ -3612,6 +3615,7 @@ def build_settings_dict(c, gpu_cards):
"https": c.server.https,
"https_key_path": c.server.https_key_path,
"https_cert_path": c.server.https_cert_path,
"metrics_public": c.server.metrics_public,
"queue_max_size": c.server.queue_max_size,
"max_parallel_requests": c.server.max_parallel_requests,
"max_parallel_requests_overrides": c.server.max_parallel_requests_overrides,
......@@ -3806,6 +3810,12 @@ def build_settings_dict(c, gpu_cards):
"poll_timeout_s": c.cluster.poll_timeout_s,
"advertise_host": c.cluster.advertise_host,
"rpc_bin": c.cluster.rpc_bin,
"discovery": c.cluster.discovery,
"auto_join": c.cluster.auto_join,
"prefix_affinity": c.cluster.prefix_affinity,
"prefix_affinity_ttl_s": c.cluster.prefix_affinity_ttl_s,
# The shared cluster token is a secret: only whether it is set.
"token": "", "token_set": bool(c.cluster.token),
# The node token is a secret of the NODE: masked like the RunPod key.
"nodes": [{**n, "api_key": "", "api_key_set": bool(n.get("api_key"))}
for n in (c.cluster.nodes or []) if isinstance(n, dict)],
......@@ -3918,6 +3928,8 @@ async def api_save_settings(request: Request, username: str = Depends(require_ad
c.server.https = bool(srv.get("https", c.server.https))
c.server.https_key_path = srv.get("https_key_path") or None
c.server.https_cert_path = srv.get("https_cert_path") or None
if "metrics_public" in srv:
c.server.metrics_public = bool(srv["metrics_public"])
if "queue_max_size" in srv:
c.server.queue_max_size = max(1, int(srv["queue_max_size"]))
from codai.queue.manager import queue_manager
......@@ -4422,6 +4434,16 @@ async def api_save_settings(request: Request, username: str = Depends(require_ad
if "node_name" in d: cl.node_name = (d.get("node_name") or "").strip()
if "advertise_host" in d: cl.advertise_host = (d.get("advertise_host") or "").strip()
if "rpc_bin" in d: cl.rpc_bin = (d.get("rpc_bin") or "").strip()
if "discovery" in d: cl.discovery = bool(d["discovery"])
if "auto_join" in d: cl.auto_join = bool(d["auto_join"])
if "prefix_affinity" in d: cl.prefix_affinity = bool(d["prefix_affinity"])
if "prefix_affinity_ttl_s" in d:
try: cl.prefix_affinity_ttl_s = max(0.0, float(d.get("prefix_affinity_ttl_s") or 0))
except (TypeError, ValueError): pass
# Blank keeps the stored token; the literal "-" clears it.
tok = d.get("token")
if isinstance(tok, str) and tok.strip():
cl.token = "" if tok.strip() == "-" else tok.strip()
if "poll_timeout_s" in d:
try: cl.poll_timeout_s = max(1.0, float(d.get("poll_timeout_s") or 4.0))
except (TypeError, ValueError): pass
......
codai/admin/templates/cluster.html
View file @ 0d4a338b
......@@ -29,15 +29,28 @@
<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>Node</th><th>URL</th><th>State</th><th>Capabilities</th><th>VRAM</th><th>Its engines</th><th>Loaded</th><th>Assigned</th><th>In-flight</th>
<th>Node</th><th>URL</th><th>State</th><th>Capabilities</th><th>VRAM</th><th>Its engines</th><th>Loaded</th><th>Assigned</th><th>In-flight</th><th></th>
</tr></thead>
<tbody id="cl-nodes"><tr><td colspan="9" class="dim">No nodes configured — Settings → Cluster.</td></tr></tbody>
<tbody id="cl-nodes"><tr><td colspan="10" class="dim">No nodes configured — Settings → Cluster.</td></tr></tbody>
</table>
</div>
</div>
<div class="card" style="margin-bottom:1rem">
<div class="card-title">Engines on this machine</div>
<div class="card-title">Seen on the network <span class="dim small">— mDNS <code>_coderai._tcp</code>; members hold this cluster's token and join by themselves</span></div>
<div id="cl-disc-note" class="dim small" style="margin-bottom:.4rem"></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>Name</th><th>URL</th><th>Version</th><th>Capabilities</th><th>Member</th><th>Joined</th><th>Last seen</th>
</tr></thead>
<tbody id="cl-disc"><tr><td colspan="7" class="dim">—</td></tr></tbody>
</table>
</div>
</div>
<div class="card" style="margin-bottom:1rem">
<div class="card-title">Engines on this machine <button class="btn btn-secondary" style="float:right;padding:.15rem .5rem;font-size:12px" onclick="clLogs('')">This front's log</button></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)">
......@@ -77,6 +90,29 @@
</div>
</div>
<div style="display:grid;grid-template-columns:1fr 1fr;gap:1rem;margin-bottom:1rem">
<div class="card">
<div class="card-title">Usage by API key <span class="dim small">— since the front started; the same numbers <code>/metrics</code> exports</span></div>
<table style="width:100%;border-collapse:collapse;font-size:13px">
<thead><tr style="text-align:left;color:var(--text-3)"><th>Key</th><th>Requests</th><th>Errors</th><th>Time</th><th>Kinds</th><th>Where</th></tr></thead>
<tbody id="cl-usage-key"><tr><td colspan="6" class="dim">No requests yet.</td></tr></tbody>
</table>
</div>
<div class="card">
<div class="card-title">Usage by model</div>
<table style="width:100%;border-collapse:collapse;font-size:13px">
<thead><tr style="text-align:left;color:var(--text-3)"><th>Model</th><th>Requests</th><th>Errors</th><th>Time</th><th>Where</th></tr></thead>
<tbody id="cl-usage-model"><tr><td colspan="5" class="dim">No requests yet.</td></tr></tbody>
</table>
</div>
</div>
<div id="cl-logbox" class="card" style="display:none;margin-bottom:1rem">
<div class="card-title"><span id="cl-log-title">Log</span> <button class="btn btn-secondary" style="float:right;padding:.15rem .5rem;font-size:12px" onclick="document.getElementById('cl-logbox').style.display='none'">close</button>
<button class="btn btn-secondary" style="float:right;margin-right:.4rem;padding:.15rem .5rem;font-size:12px" onclick="clLogs(_logNode)">refresh</button></div>
<pre id="cl-log" style="max-height:420px;overflow:auto;font-size:11px;line-height:1.35;margin:0;white-space:pre-wrap"></pre>
</div>
<div class="card">
<div class="card-title">RunPod pods <span class="dim small">— live list, costs on the <a href="{{ root_path }}/admin/runpod">RunPod page</a></span></div>
<table style="width:100%;border-collapse:collapse;font-size:13px">
......@@ -86,7 +122,7 @@
</div>
<style>
#cl-nodes td, #cl-engines td, #cl-rpc td, #cl-hosts td, #cl-remotes td, #cl-pods td { padding:.35rem .4rem; border-top:1px solid var(--border,#333); vertical-align:top }
#cl-nodes td, #cl-engines td, #cl-rpc td, #cl-hosts td, #cl-remotes td, #cl-pods td, #cl-disc td, #cl-usage-key td, #cl-usage-model td { padding:.35rem .4rem; border-top:1px solid var(--border,#333); vertical-align:top }
.cl-ok { color:#2ecc71; font-weight:600 } .cl-bad { color:#e74c3c; font-weight:600 } .cl-dim { color:var(--text-3) }
.cl-pill { display:inline-block; padding:.05rem .4rem; border-radius:999px; border:1px solid var(--border,#444); font-size:11px; margin:0 .15rem .15rem 0 }
</style>
......@@ -124,12 +160,41 @@ async function clRefresh(){
document.getElementById('cl-n-pods').textContent = pods.length;
document.getElementById('cl-nodes').innerHTML = nodes.length ? nodes.map(n => `<tr>
<td><b>${esc(n.name)}</b></td><td class="cl-dim">${esc(n.url)}</td>
<td><b>${esc(n.name)}</b>${n.discovered ? ' <span class="cl-pill" title="found over mDNS with the cluster token">mDNS</span>' : ''}</td><td class="cl-dim">${esc(n.url)}</td>
<td>${ok(n.healthy, 'reachable', 'unreachable')}${n.last_error ? '<br><span class="cl-dim" style="font-size:11px">' + esc(n.last_error) + '</span>' : ''}</td>
<td>${pills(n.capabilities)}</td><td>${vram(n.vram)}</td>
<td>${(n.node_engines || []).map(e => '<span class="cl-pill" title="' + esc(vram(e.vram)) + '">' + esc(e.name) + (e.healthy ? '' : ' ✗') + '</span>').join('') || '<span class="cl-dim">—</span>'}</td>
<td>${short(n.loaded_models)}</td><td>${short(n.assigned_models)}</td><td>${n.inflight || 0}</td></tr>`).join('')
: '<tr><td colspan="9" class="dim">No nodes configured — Settings → Cluster.</td></tr>';
<td>${short(n.loaded_models)}</td><td>${short(n.assigned_models)}</td><td>${n.inflight || 0}</td>
<td><button class="btn btn-secondary" style="padding:.1rem .45rem;font-size:11px" onclick="clLogs(${JSON.stringify(n.name)})">log</button></td></tr>`).join('')
: '<tr><td colspan="10" class="dim">No nodes configured — Settings → Cluster (or turn on discovery and share a cluster token).</td></tr>';
const disc = d.discovery || {};
const dn = [];
if (!disc.enabled) dn.push('Discovery is <b>off</b> — Settings → Cluster → <em>Find nodes on the LAN</em>.');
else if (!disc.available) dn.push('Discovery is on but the <code>zeroconf</code> package is missing in this install.');
else if (disc.error) dn.push('Discovery error: ' + esc(disc.error));
else if (disc.running) dn.push('Announcing as <b>' + esc(disc.name) + '</b> on port ' + esc(disc.port) + (disc.token_set ? '' : ' — <b>no cluster token set</b>: this install is visible but joins nothing and nobody joins it') + (disc.auto_join ? '' : '; auto-join is off (members are listed, not used)') + '.');
document.getElementById('cl-disc-note').innerHTML = dn.join(' ');
const peers = disc.peers || [];
document.getElementById('cl-disc').innerHTML = peers.length ? peers.map(p => `<tr>
<td><b>${esc(p.name)}</b></td><td class="cl-dim">${esc(p.url)}</td><td>${esc(p.version)}</td><td>${pills(p.capabilities)}</td>
<td>${p.member ? '<span class="cl-ok">same token</span>' : '<span class="cl-dim">other token</span>'}${p.serve === false ? ' <span class="cl-dim">(does not serve)</span>' : ''}</td>
<td>${p.joined ? '<span class="cl-ok">node</span>' : (p.member ? (disc.auto_join ? '<span class="cl-dim">joining…</span>' : '<span class="cl-dim">auto-join off</span>') : '<span class="cl-dim">—</span>')}</td>
<td class="cl-dim">${Math.max(0, Math.round((Date.now()/1000) - p.seen))}s ago</td></tr>`).join('')
: '<tr><td colspan="7" class="dim">' + (disc.running ? 'No other coderai announcing on this link.' : '—') + '</td></tr>';
const usage = d.usage || {};
const byKey = usage.by_key || {}, byModel = usage.by_model || {};
const mins = s => (s >= 3600 ? (s/3600).toFixed(1) + ' h' : s >= 60 ? (s/60).toFixed(1) + ' min' : s.toFixed(1) + ' s');
const kv = o => Object.entries(o || {}).sort((a,b) => b[1]-a[1]).map(([k,v]) => esc(k) + ' ' + v).join(', ') || '—';
const keys = Object.entries(byKey).sort((a,b) => b[1].requests - a[1].requests);
document.getElementById('cl-usage-key').innerHTML = keys.length ? keys.map(([k, r]) => `<tr>
<td><b>${esc(k)}</b></td><td>${r.requests}</td><td>${r.errors ? '<span class="cl-bad">' + r.errors + '</span>' : 0}</td><td>${mins(r.seconds)}</td><td class="cl-dim">${kv(r.kinds)}</td><td class="cl-dim">${kv(r.engines)}</td></tr>`).join('')
: '<tr><td colspan="6" class="dim">No requests yet.</td></tr>';
const models = Object.entries(byModel).sort((a,b) => b[1].requests - a[1].requests);
document.getElementById('cl-usage-model').innerHTML = models.length ? models.map(([k, r]) => `<tr>
<td><b>${esc(k)}</b></td><td>${r.requests}</td><td>${r.errors ? '<span class="cl-bad">' + r.errors + '</span>' : 0}</td><td>${mins(r.seconds)}</td><td class="cl-dim">${kv(r.engines)}</td></tr>`).join('')
: '<tr><td colspan="5" class="dim">No requests yet.</td></tr>';
document.getElementById('cl-engines').innerHTML = engines.length ? engines.map(e => `<tr>
<td><b>${esc(e.name)}</b>${e.primary ? ' <span class="cl-dim">(primary)</span>' : ''}</td><td>${esc(e.backend)}</td>
......@@ -158,6 +223,20 @@ async function clRefresh(){
<td>${esc(p.state)}${p.healthy ? ' <span class="cl-ok">healthy</span>' : ''}</td><td>${p.inflight || 0}</td><td>${p.hourly_usd != null ? '$' + p.hourly_usd : '—'}</td></tr>`).join('')
: '<tr><td colspan="6" class="dim">None running.</td></tr>';
}
let _logNode = '';
async function clLogs(node){
_logNode = node || '';
const box = document.getElementById('cl-logbox'), pre = document.getElementById('cl-log');
box.style.display = '';
document.getElementById('cl-log-title').textContent = node ? 'Log of node ' + node : "This front's log";
pre.textContent = 'loading…';
try {
const r = await fetch(ROOT_PATH + '/admin/api/cluster/logs?lines=300' + (node ? '&node=' + encodeURIComponent(node) : ''));
const d = await r.json();
pre.textContent = (d.error ? '[' + d.error + ']\n' : '') + (d.lines || []).join('\n');
pre.scrollTop = pre.scrollHeight;
} catch (e) { pre.textContent = 'error: ' + e.message; }
}
clRefresh();
setInterval(clRefresh, 10000);
</script>
......
codai/admin/templates/models.html
View file @ 0d4a338b
......@@ -652,12 +652,18 @@ window.__DEFAULT_WHISPER_SERVER_PATH__ = {{ default_whisper_server_path|tojson }
<datalist id="cfg-rpc-known"></datalist>
<span class="form-hint" style="font-size:11px;line-height:1.5">Each <code>host:port</code> where an <code>rpc-server</code> listens becomes one more GPU in this model's layer split, <b>after this machine's cards</b> in that order (so a <b>Weight distribution</b> ratio reads: local cards…, then these). Known servers from the <a href="{{ root_path }}/admin/cluster">Cluster</a> page are suggested. Decode over a LAN is workable (activations per token are small); loading and prompt processing are wire-bound — 1 GbE hurts, 10 GbE is fine. The protocol is unauthenticated: LAN or WireGuard only. Tick <b>Split</b> above or leave Auto — a model naming RPC servers is split by definition.</span>
<div id="cfg-rpc-status" class="form-hint" style="font-size:11px;margin-top:.3rem"></div>
<div class="form-row" style="margin:.6rem 0 0"><label class="form-label">How the split is made</label>
<select id="cfg-split-mode" class="form-input">
<option value="">Layer split (default) — each card holds whole layers; only activations cross between cards</option>
<option value="row">Row split — tensor parallel: every layer over all cards at once (PCIe / 10 GbE+ / RDMA only)</option>
</select>
<span class="form-hint" style="font-size:11px">Applies to local cards and RPC servers alike. Row split trades one all-reduce per layer for using every card on every token: a win on one PCIe bus or a fast fabric, a loss on 1 GbE.</span></div>
</div>
<div class="form-row" id="cfg-vllm-row" style="margin-top:.75rem;display:none;border:1px solid var(--border,#333);border-radius:8px;padding:.6rem">
<label class="form-label" style="font-weight:600">vLLM <span class="muted" style="font-weight:400">— per-model parallelism and the machines it spans</span></label>
<span class="form-hint" style="font-size:11px;margin-bottom:.4rem">Blank = inherit the global vLLM settings. <b>Tensor parallel</b> splits every layer across GPUs (fast links only: one machine). <b>Pipeline parallel</b> gives each machine a slice of layers — that is the one to use across a network. With nodes listed (or pipeline &gt; 1) coderai starts a Ray head here, runs each node's command, waits until Ray sees tensor×pipeline GPUs, then launches vLLM on Ray.</span>
<span class="form-hint" style="font-size:11px;margin-bottom:.4rem">Blank = inherit the global vLLM settings. <b>Tensor parallel</b> splits every layer across GPUs; with nodes listed it may exceed this machine's cards and span the Ray cluster (an all-reduce per layer crosses the wire: 10 GbE is the floor, RDMA/InfiniBand the point). <b>Pipeline parallel</b> gives each machine a slice of layers — the one to use on ordinary Ethernet. Both can be combined (tensor × pipeline GPUs in total). With nodes listed (or pipeline &gt; 1) coderai starts a Ray head here, runs each node's command, waits until Ray sees tensor×pipeline GPUs, then launches vLLM on Ray.</span>
<div style="display:grid;grid-template-columns:1fr 1fr 1fr;gap:.6rem">
<div class="form-row" style="margin:0"><label class="form-label">Tensor parallel (GPUs per node)</label><input type="number" id="cfg-vllm-tp" class="form-input" min="1" placeholder="inherit"></div>
<div class="form-row" style="margin:0"><label class="form-label">Tensor parallel (GPUs, may span nodes)</label><input type="number" id="cfg-vllm-tp" class="form-input" min="1" placeholder="inherit"></div>
<div class="form-row" style="margin:0"><label class="form-label">Pipeline parallel (nodes)</label><input type="number" id="cfg-vllm-pp" class="form-input" min="1" placeholder="inherit / 1"></div>
<div class="form-row" style="margin:0"><label class="form-label">Executor</label>
<select id="cfg-vllm-executor" class="form-input"><option value="">auto (ray when nodes are listed)</option><option value="mp">mp — this machine only</option><option value="ray">ray</option></select></div>
......@@ -4364,6 +4370,7 @@ function _populateCluster(m, s){
_updateDistVisibility();
set('cfg-node-path', s.node_path || '');
set('cfg-rpc-servers', Array.isArray(s.rpc_servers) ? s.rpc_servers.join(', ') : (s.rpc_servers || ''));
set('cfg-split-mode', s.split_mode || '');
const dl = document.getElementById('cfg-rpc-known');
if (dl) { dl.innerHTML = ''; (_knownRpc || []).forEach(r => { const o = document.createElement('option'); o.value = r.endpoint; o.textContent = (r.node || '') + (r.device ? ' ' + r.device : '') + (r.reachable ? '' : ' (down)'); dl.appendChild(o); }); }
const st = document.getElementById('cfg-rpc-status');
......@@ -4389,6 +4396,7 @@ function _collectCluster(){
const out = {
node_path: val('cfg-node-path'),
rpc_servers: val('cfg-rpc-servers'),
split_mode: val('cfg-split-mode'),
components: { text_encoder: val('cfg-comp-te'), low_noise: val('cfg-comp-low'), vae: val('cfg-comp-vae') },
distribute: {
enabled: !!document.getElementById('cfg-dist-enabled').checked,
......
codai/admin/templates/settings.html
View file @ 0d4a338b
......@@ -47,6 +47,10 @@
<input type="text" id="s-cert" class="form-input" placeholder="/path/to/cert.pem">
</div>
</div>
<div class="form-row" style="margin-top:.8rem;margin-bottom:0">
<label class="toggle-row" style="display:flex;align-items:center;gap:.5rem;cursor:pointer"><input type="checkbox" id="s-metrics-public"><span>Serve <code>/metrics</code> (Prometheus) without a credential</span></label>
<span class="form-hint" style="font-size:11px">Off = a session, an API token or the cluster token is required. Exported: requests per key/model/kind/status/engine with wall time, engines and nodes up, VRAM, in-flight, RPC servers, discovery, RunPod spend.</span>
</div>
<div class="form-row" style="margin-top:1rem;margin-bottom:0">
<label class="form-label">Request queue max size</label>
<input type="number" id="s-queue-max" class="form-input" placeholder="6" min="1" max="1000" style="max-width:160px">
......@@ -885,8 +889,14 @@
<div class="form-group" style="margin:0"><label class="form-label">Node poll timeout (s)</label><input type="number" id="s-cluster-poll" class="form-input" min="1" step="0.5" placeholder="4"></div>
<div class="form-group" style="margin:0;align-self:end"><label class="toggle-row" style="display:flex;align-items:center;gap:.5rem;cursor:pointer"><input type="checkbox" id="s-cluster-serve"><span>Let other heads use this install as a node</span></label></div>
</div>
<div style="display:grid;grid-template-columns:1fr 1fr 1fr 1fr;gap:.6rem;margin-top:.6rem">
<div class="form-group" style="margin:0"><label class="form-label">Cluster token <span class="muted" id="s-cluster-token-state"></span></label><input type="password" id="s-cluster-token" class="form-input" placeholder="one secret shared by every box" autocomplete="new-password"><span class="form-hint" style="font-size:11px">Type the same on every install. Nodes accept it on <code>/cluster/*</code>; a head uses it for discovered nodes and for listed nodes with a blank token. Blank keeps the stored one, <code>-</code> clears it.</span></div>
<div class="form-group" style="margin:0;align-self:end"><label class="toggle-row" style="display:flex;align-items:center;gap:.5rem;cursor:pointer"><input type="checkbox" id="s-cluster-discovery"><span>Find nodes on the LAN (mDNS)</span></label><span class="form-hint" style="font-size:11px">Announces this install as <code>_coderai._tcp</code> and lists the others on the Cluster page. Link-local: same subnet, container on the host network.</span></div>
<div class="form-group" style="margin:0;align-self:end"><label class="toggle-row" style="display:flex;align-items:center;gap:.5rem;cursor:pointer"><input type="checkbox" id="s-cluster-autojoin"><span>Auto-join members as nodes</span></label><span class="form-hint" style="font-size:11px">Every discovered install with the same token becomes a node, no listing needed.</span></div>
<div class="form-group" style="margin:0;align-self:end"><label class="toggle-row" style="display:flex;align-items:center;gap:.5rem;cursor:pointer"><input type="checkbox" id="s-cluster-affinity"><span>Prefix-cache routing</span></label><input type="number" id="s-cluster-affinity-ttl" class="form-input" min="0" step="60" placeholder="1800" style="margin-top:.3rem" title="seconds a conversation stays bound to its engine"><span class="form-hint" style="font-size:11px">A conversation's next turn goes back to the engine or node that served it (warm KV cache) when the model is loaded in several places. TTL in seconds.</span></div>
</div>
<div id="cluster-fields" style="margin-top:.6rem">
<div style="font-weight:600;font-size:13px;margin:.4rem 0">Nodes</div>
<div style="font-weight:600;font-size:13px;margin:.4rem 0">Nodes <span class="muted" style="font-weight:400">— listed by hand; discovered members are added on their own</span></div>
<div id="s-cluster-nodes"></div>
<button type="button" class="btn btn-secondary btn-sm" onclick="_addClusterNode()">+ Add node</button>
<span class="form-hint" style="font-size:11px;margin-left:.6rem">Verify: <b>system</b> = public certificate, <b>pem</b> = paste the node's self-signed certificate, <b>off</b> = no check (plain http, or a LAN you trust). Capabilities blank = whatever the node reports.</span>
......@@ -1720,6 +1730,7 @@ async function loadSettings(){
document.getElementById('s-https').checked = !!d.server?.https;
document.getElementById('s-key').value = d.server?.https_key_path ?? '';
document.getElementById('s-cert').value = d.server?.https_cert_path ?? '';
document.getElementById('s-metrics-public').checked = !!d.server?.metrics_public;
document.getElementById('s-queue-max').value = d.server?.queue_max_size ?? 6;
document.getElementById('s-internal-port-base').value = d.server?.internal_port_base ?? 8780;
// Default engine — surfaced when 2+ engines are running. Sourced from the
......@@ -1994,6 +2005,7 @@ async function saveSettings(silent){
https: document.getElementById('s-https').checked,
https_key_path: strOrNull('s-key'),
https_cert_path: strOrNull('s-cert'),
metrics_public: document.getElementById('s-metrics-public').checked,
queue_max_size: parseInt(document.getElementById('s-queue-max').value) || 6,
internal_port_base: parseInt(document.getElementById('s-internal-port-base').value) || 8780,
max_parallel_requests: parseInt(document.getElementById('s-max-parallel').value) || 2,
......@@ -2451,6 +2463,12 @@ function _populateCluster(cl){
document.getElementById('s-cluster-poll').value = cl.poll_timeout_s ?? 4;
document.getElementById('s-cluster-serve').checked = cl.serve !== false;
document.getElementById('s-cluster-rpc-bin').value = cl.rpc_bin ?? '';
document.getElementById('s-cluster-token').value = '';
document.getElementById('s-cluster-token-state').textContent = cl.token_set ? '(set)' : '(not set)';
document.getElementById('s-cluster-discovery').checked = !!cl.discovery;
document.getElementById('s-cluster-autojoin').checked = cl.auto_join !== false;
document.getElementById('s-cluster-affinity').checked = cl.prefix_affinity !== false;
document.getElementById('s-cluster-affinity-ttl').value = cl.prefix_affinity_ttl_s ?? 1800;
document.getElementById('s-cluster-nodes').innerHTML = (cl.nodes || []).map(_clusterNodeRowHtml).join('');
document.getElementById('s-cluster-rpc').innerHTML = (cl.rpc_servers || []).map(_clusterRpcRowHtml).join('');
toggleClusterFields();
......@@ -2476,6 +2494,11 @@ function _collectCluster(){
poll_timeout_s: parseFloat(document.getElementById('s-cluster-poll').value) || 4,
serve: document.getElementById('s-cluster-serve').checked,
rpc_bin: document.getElementById('s-cluster-rpc-bin').value.trim(),
token: document.getElementById('s-cluster-token').value,
discovery: document.getElementById('s-cluster-discovery').checked,
auto_join: document.getElementById('s-cluster-autojoin').checked,
prefix_affinity: document.getElementById('s-cluster-affinity').checked,
prefix_affinity_ttl_s: parseFloat(document.getElementById('s-cluster-affinity-ttl').value) || 0,
nodes: nodes, rpc_servers: rpc,
};
}
......
codai/api/ratelimit.py
View file @ 0d4a338b
......@@ -59,6 +59,14 @@ def _unauthorized():
)
def _cluster_token() -> str:
try:
from codai.admin.routes import config_manager
return getattr(getattr(config_manager.config, "cluster", None), "token", "") or ""
except Exception:
return ""
class BearerAuthMiddleware(BaseHTTPMiddleware):
"""Reject /v1/ API requests that lack a valid Bearer token or active web session."""
......@@ -95,6 +103,13 @@ class BearerAuthMiddleware(BaseHTTPMiddleware):
if _env_token and auth_header.lower().startswith("bearer "):
if hmac.compare_digest(auth_header[7:].strip(), _env_token):
return await call_next(request)
# The shared cluster token (Settings → Cluster, or CODERAI_CLUSTER_TOKEN
# on a capability image): what a head that discovered this install
# presents on every request it forwards here.
if auth_header.lower().startswith("bearer "):
_ctok = os.environ.get("CODERAI_CLUSTER_TOKEN", "") or _cluster_token()
if _ctok and hmac.compare_digest(auth_header[7:].strip(), _ctok):
return await call_next(request)
from codai.admin import routes as _admin_routes
sm = _admin_routes.session_manager
......
codai/backends/vulkan.py
View file @ 0d4a338b
......@@ -1770,6 +1770,23 @@ class VulkanBackend(ModelBackend):
print(f" gpu split : tensor_split={_parsed} (multi-GPU pool)")
if _parsed and _llama_accepts('tensor_split'):
llama_kwargs['tensor_split'] = _parsed
# HOW the split is made across the devices (local cards and RPC
# servers alike):
# layer (llama.cpp default) — each device holds whole layers; per
# token only the activations between consecutive layers move,
# which is what an Ethernet link can carry;
# row — every matrix is split by rows over ALL devices (tensor
# parallelism): the devices work on each layer together, at the
# price of an all-reduce per layer. Faster when the devices sit
# on one PCIe bus; across a network it needs a fabric measured
# in tens of Gb/s (10 GbE is the floor, RDMA the point).
_sm = str(kwargs.get('split_mode', _raw_cfg.get('split_mode')) or 'layer').lower()
if _sm in ('row', 'tensor') and _llama_accepts('split_mode'):
_row = getattr(_llama_cpp, 'LLAMA_SPLIT_MODE_ROW', 2)
llama_kwargs['split_mode'] = _row
print(f" gpu split : split_mode=row (tensor parallel over "
f"{len(_parsed) if _parsed else 'all'} devices"
f"{', incl. RPC' if _rpc_eps else ''})")
elif _n_cuda > 0 and _n_vk > 0 and _llama_accepts('tensor_split'):
# NOT a split model, but the engine can see BOTH backends' cards (cross
# visibility was enabled for some other split model). Confine THIS model
......
codai/cluster/discovery.py 0 → 100644
View file @ 0d4a338b
# 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.
"""Finding the other coderai installs on the LAN — mDNS / DNS-SD.
Every install with ``cluster.discovery`` on announces itself as a
``_coderai._tcp.local.`` service (name, version, URL scheme, port, the
capabilities its engines offer) and browses for the others. Membership is
a **shared cluster token** (``cluster.token``): the announcement carries an
HMAC fingerprint of it — never the token — and only peers whose fingerprint
matches ours count as members. A head with ``cluster.auto_join`` on turns
every member it sees into a cluster node (``cluster.nodes`` entry in
memory, the shared token as its API key); a node accepts the shared token
on its ``/cluster/*`` endpoints. So one token typed on every box is the
whole setup: no URLs, no per-node API tokens.
mDNS is link-local multicast (224.0.0.251:5353): it crosses no router, and
a container sees it only with ``--network host`` (the launcher's default)
or a macvlan. Across subnets or VPNs, list the nodes by hand — both ways
coexist; a node named in ``cluster.nodes`` is never duplicated by discovery.
Needs the ``zeroconf`` package; without it discovery is simply off and the
Cluster page says so.
"""
import hmac
import hashlib
import socket
import threading
import time
import uuid
from typing import List, Optional
SERVICE = "_coderai._tcp.local."
_FP_MSG = b"coderai-cluster-membership-v1"
def token_fingerprint(token: str) -> str:
"""What an announcement carries instead of the token: HMAC-SHA256 keyed by
the token over a fixed message, truncated. Proves 'I hold the same token'
to anyone who also holds it, reveals nothing to anyone who doesn't."""
if not token:
return ""
return hmac.new(token.encode("utf-8"), _FP_MSG, hashlib.sha256).hexdigest()[:24]
def available() -> bool:
try:
import zeroconf # noqa: F401
return True
except Exception:
return False
def _txt(props: dict) -> dict:
return {str(k): str(v) for k, v in props.items() if v is not None}
class Discovery:
"""Announce this install and keep the list of the others seen.
``peers()`` is every coderai announcing on the link; ``members()`` is the
subset holding our token (and never ourselves)."""
def __init__(self, name: str, port: int, token: str = "", advertise_host: str = "",
scheme: str = "http", version: str = "", capabilities=None,
serve: bool = True, browse: bool = True, instance_id: str = ""):
self.name = (name or socket.gethostname()).strip()
self.port = int(port)
self.token = token or ""
self.advertise_host = advertise_host or ""
self.scheme = scheme or "http"
self.version = version or ""
self.capabilities = capabilities # callable or list; read at announce
self.serve = bool(serve)
self.browse = bool(browse)
self.instance_id = instance_id or uuid.uuid4().hex[:12]
self._zc = None
self._info = None
self._browser = None
self._peers: dict = {} # instance_id -> dict
self._lock = threading.Lock()
self.error = ""
self.started_at = 0.0
# ----------------------------------------------------------- lifecycle
def _address(self) -> str:
from codai.cluster.rpc import advertise_host
return advertise_host(self.advertise_host)
def _caps(self) -> List[str]:
c = self.capabilities
try:
c = c() if callable(c) else c
except Exception:
c = []
return sorted({str(x) for x in (c or [])})
def _service_info(self):
from zeroconf import ServiceInfo
addr = self._address()
props = _txt({
"name": self.name, "id": self.instance_id, "ver": self.version,
"scheme": self.scheme, "fp": token_fingerprint(self.token),
"caps": ",".join(self._caps()), "serve": "1" if self.serve else "0",
"path": "/cluster/state",
})
# The instance name must be unique on the link; the coderai node name
# plus a short instance id keeps two boxes with the same hostname apart.
inst = f"{self.name}-{self.instance_id}.{SERVICE}"
return ServiceInfo(SERVICE, inst, addresses=[socket.inet_aton(addr)],
port=self.port, properties=props, server=f"{self.name}.local.")
def start(self) -> bool:
try:
from zeroconf import Zeroconf, ServiceBrowser
except Exception as exc:
self.error = f"zeroconf not installed ({exc.__class__.__name__}); discovery off"
print(f"[discovery] {self.error}", flush=True)
return False
try:
self._zc = Zeroconf()
self._info = self._service_info()
self._zc.register_service(self._info)
if self.browse:
self._browser = ServiceBrowser(self._zc, SERVICE, handlers=[self._on_change])
self.started_at = time.time()
print(f"[discovery] announcing '{self.name}' as {SERVICE} on "
f"{self.scheme}://{self._address()}:{self.port}"
f"{' (token set)' if self.token else ' (NO cluster token: visible, joins nothing)'}",
flush=True)
return True
except Exception as exc:
self.error = f"mDNS not started: {exc}"
print(f"[discovery] {self.error}", flush=True)
self.stop()
return False
def refresh(self) -> None:
"""Re-announce (capabilities changed, e.g. an engine came up)."""
if self._zc is None or self._info is None:
return
try:
new = self._service_info()
self._zc.update_service(new)
self._info = new
except Exception as exc:
print(f"[discovery] re-announce failed: {exc}", flush=True)
def stop(self) -> None:
zc, self._zc = self._zc, None
if zc is None:
return
try:
if self._info is not None:
zc.unregister_service(self._info)
except Exception:
pass
try:
zc.close()
except Exception:
pass
# ------------------------------------------------------------- browsing
def _on_change(self, zeroconf, service_type, name, state_change):
try:
from zeroconf import ServiceStateChange
if state_change == ServiceStateChange.Removed:
with self._lock:
for k, p in list(self._peers.items()):
if p.get("service") == name:
self._peers.pop(k, None)
return
info = zeroconf.get_service_info(service_type, name, timeout=1500)
if info is None:
return
self._record(name, info)
except Exception as exc:
print(f"[discovery] browse event error: {exc}", flush=True)
def _record(self, service_name: str, info) -> None:
props = {}
for k, v in (info.properties or {}).items():
try:
props[k.decode() if isinstance(k, bytes) else str(k)] = \
v.decode() if isinstance(v, bytes) else ("" if v is None else str(v))
except Exception:
continue
iid = props.get("id") or service_name
addrs = []
try:
addrs = info.parsed_addresses()
except Exception:
try:
addrs = [socket.inet_ntoa(a) for a in info.addresses]
except Exception:
addrs = []
# Prefer an IPv4 address; mDNS may carry both.
addr = next((a for a in addrs if ":" not in a), addrs[0] if addrs else "")
if not addr:
return
scheme = props.get("scheme") or "http"
peer = {
"instance": iid, "service": service_name,
"name": props.get("name") or service_name.split(".")[0],
"url": f"{scheme}://{addr}:{info.port}",
"version": props.get("ver", ""), "fingerprint": props.get("fp", ""),
"capabilities": [c for c in (props.get("caps") or "").split(",") if c],
"serve": props.get("serve", "1") != "0",
"seen": time.time(),
}
with self._lock:
self._peers[iid] = peer
def peers(self, max_age_s: float = 0.0) -> List[dict]:
"""Everyone announcing (ourselves excluded)."""
now = time.time()
with self._lock:
out = [dict(p) for k, p in self._peers.items() if k != self.instance_id]
if max_age_s > 0:
out = [p for p in out if now - p["seen"] <= max_age_s]
for p in out:
p["member"] = bool(self.token) and p.get("fingerprint") == token_fingerprint(self.token)
return sorted(out, key=lambda p: p["name"].lower())
def members(self) -> List[dict]:
"""Peers holding our cluster token, that answer as a node."""
return [p for p in self.peers() if p["member"] and p.get("serve", True)]
def status(self) -> dict:
return {"available": available(), "running": self._zc is not None,
"error": self.error, "name": self.name, "instance": self.instance_id,
"token_set": bool(self.token), "port": self.port,
"peers": self.peers()}
def member_node_specs(members: List[dict], token: str, known_names, timeout_s: float = 4.0):
"""Discovered members as NodeSpecs for the supervisor, skipping any name
(or URL) that ``cluster.nodes`` already declares by hand."""
from codai.cluster.nodes import NodeSpec
known = {str(n).lower() for n in (known_names or [])}
out = []
for p in members:
name = p["name"]
if name.lower() in known or p["url"].lower() in known:
continue
# Two boxes announcing the same name: disambiguate with the instance id
# so both get an engine rather than one silently shadowing the other.
if any(o.name.lower() == name.lower() for o in out):
name = f"{name}-{p['instance'][:6]}"
out.append(NodeSpec(name=name, url=p["url"], api_key=token,
verify="off" if p["url"].startswith("https") else "system",
timeout_s=timeout_s))
return out
codai/cluster/overview.py
View file @ 0d4a338b
......@@ -156,4 +156,37 @@ def build_overview(front, local_rpc: list) -> dict:
"hosts": _host_targets(front),
"remotes": _remote_targets(front),
"pods": _pods(),
"discovery": _discovery(front, ccfg),
"usage": _usage(front),
}
def _discovery(front, ccfg) -> dict:
"""mDNS: what this install announces and who it sees (cluster/discovery.py)."""
from codai.cluster import discovery as _d
sup = getattr(front, "supervisor", None)
disc = getattr(sup, "discovery", None) if sup else None
want = bool(getattr(ccfg, "discovery", False)) if ccfg else False
out = {"enabled": want, "available": _d.available(), "running": False,
"token_set": bool(getattr(ccfg, "token", "")) if ccfg else False,
"auto_join": bool(getattr(ccfg, "auto_join", True)) if ccfg else True,
"error": "", "peers": []}
if disc is None:
if want and not out["available"]:
out["error"] = "the zeroconf package is not installed"
return out
st = disc.status()
out.update({"running": st["running"], "error": st["error"], "name": st["name"],
"instance": st["instance"], "port": st["port"]})
joined = {e.name.lower() for e in front.registry.remotes()}
for p in st["peers"]:
p["joined"] = p["name"].lower() in joined
out["peers"].append(p)
return out
def _usage(front) -> dict:
try:
return front.metrics.usage_tables()
except Exception:
return {"by_key": {}, "by_model": {}}
codai/config.py
View file @ 0d4a338b
......@@ -32,6 +32,9 @@ class ServerConfig:
https: bool = False
https_key_path: Optional[str] = None
https_cert_path: Optional[str] = None
# GET /metrics (Prometheus) without a credential. Off = a session cookie,
# an API token or the cluster token is required, as for the other telemetry.
metrics_public: bool = False
queue_max_size: int = 6
max_parallel_requests: int = 2
# Per-engine overrides for max_parallel_requests, keyed by engine name
......@@ -433,6 +436,23 @@ class ClusterConfig:
# advertises and the ray head it starts). Blank = best guess from the
# default route.
advertise_host: str = ""
# --- zero-config membership (codai/cluster/discovery.py) ---
# One secret shared by every box of the cluster. A node accepts it on its
# /cluster/* endpoints (besides its own API tokens); a head uses it as the
# API key of every node it discovers, and of a listed node whose api_key
# is blank. The mDNS announcement carries only an HMAC fingerprint of it.
token: str = ""
# Announce this install as _coderai._tcp on the LAN and listen for the
# others (needs the zeroconf package; link-local only).
discovery: bool = False
# Turn every discovered member (same token, serve on) into a cluster node.
auto_join: bool = True
# --- routing ---
# Prefer, for a follow-up turn of a conversation, the engine or node that
# served its earlier turns (its KV/prefix cache is warm there) when the
# model is resident in more than one place.
prefix_affinity: bool = True
prefix_affinity_ttl_s: float = 1800.0
@dataclass
......@@ -1038,6 +1058,17 @@ class ConfigManager:
)
else:
self.config = Config()
# A capability image has no config to edit: its cluster membership
# comes from the environment it was launched with (docker -e …).
_env = __import__("os").environ
if _env.get("CODERAI_CLUSTER_TOKEN"):
self.config.cluster.token = _env["CODERAI_CLUSTER_TOKEN"].strip()
if _env.get("CODERAI_DISCOVERY", "").strip().lower() in ("1", "true", "yes", "on"):
self.config.cluster.discovery = True
if _env.get("CODERAI_NODE_NAME"):
self.config.cluster.node_name = _env["CODERAI_NODE_NAME"].strip()
if _env.get("CODERAI_ADVERTISE_HOST"):
self.config.cluster.advertise_host = _env["CODERAI_ADVERTISE_HOST"].strip()
# Load models.json
if self.models_path.exists():
......@@ -1094,6 +1125,7 @@ class ConfigManager:
"https": self.config.server.https,
"https_key_path": self.config.server.https_key_path,
"https_cert_path": self.config.server.https_cert_path,
"metrics_public": self.config.server.metrics_public,
"queue_max_size": self.config.server.queue_max_size,
"max_parallel_requests": self.config.server.max_parallel_requests,
"max_parallel_requests_overrides": self.config.server.max_parallel_requests_overrides,
......
codai/frontproxy/affinity.py 0 → 100644
View file @ 0d4a338b
# 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.
"""Prefix-cache-aware routing across engines and nodes.
When a model is resident in more than one place — two local engines, a
local engine and a cluster node, replicas after a fan-out — the follow-up
turns of a conversation are worth sending where its earlier turns went:
that engine's KV / prefix cache (the multi-slot cache of the GGUF and HF
backends, vLLM's prefix caching) already holds the shared opening, and the
prompt is processed from the first new token instead of from scratch.
The key is the same one the engines use for their own slot affinity
(codai/api/text.py ``_conversation_session_key``): an explicit
``X-Session-Id`` header or OpenAI ``user`` field, else a hash of the
conversation's stable opening (system prompt + first user turn, or the
prompt head). The map is key → engine name with a TTL; entries are cheap
and bounded. A preference is only that: the router still requires the
engine to be alive, capable and to hold the model.
"""
import hashlib
import json
import threading
import time
from typing import Optional
_AFFINE_PATHS = ("/v1/chat/completions", "/v1/completions", "/v1/responses")
def conversation_key(path: str, body: Optional[bytes], headers=None) -> Optional[str]:
"""The affinity key of a request, or None when there is nothing stable to
hang it on (non-text paths, empty bodies)."""
p = (path or "").split("?", 1)[0].rstrip("/")
if p not in _AFFINE_PATHS or not body:
return None
try:
sid = ""
if headers is not None:
try:
sid = headers.get("x-session-id") or ""
except Exception:
sid = ""
if sid:
return f"sid:{sid}"
d = json.loads(body)
if not isinstance(d, dict):
return None
uid = d.get("user")
if uid:
return f"user:{uid}"
parts = []
msgs = d.get("messages")
if isinstance(msgs, list) and msgs:
for m in msgs:
if not isinstance(m, dict):
continue
role, content = m.get("role"), m.get("content")
if not isinstance(content, str):
content = json.dumps(content, sort_keys=True) if content is not None else ""
if role == "system":
parts.append("system:" + content)
elif role == "user":
parts.append("user:" + content)
break
else:
prompt = d.get("prompt") if "prompt" in d else d.get("input")
if isinstance(prompt, list):
prompt = prompt[0] if prompt else ""
if prompt:
parts.append(str(prompt)[:1024])
if not parts:
return None
return "hash:" + hashlib.sha256("\n".join(parts).encode("utf-8", "ignore")).hexdigest()[:16]
except Exception:
return None
class PrefixAffinity:
"""key → engine name, remembered for ``ttl_s`` after its last use."""
def __init__(self, ttl_s: float = 1800.0, max_entries: int = 20000):
self.ttl_s = float(ttl_s)
self.max_entries = int(max_entries)
self._m: dict = {}
self._lock = threading.Lock()
def get(self, key: Optional[str]) -> Optional[str]:
if not key:
return None
now = time.time()
with self._lock:
ent = self._m.get(key)
if ent is None:
return None
name, ts = ent
if self.ttl_s > 0 and now - ts > self.ttl_s:
self._m.pop(key, None)
return None
return name
def remember(self, key: Optional[str], engine_name: Optional[str]) -> None:
if not key or not engine_name:
return
now = time.time()
with self._lock:
self._m[key] = (engine_name, now)
if len(self._m) > self.max_entries:
# Drop the oldest tenth; a scan this rare is cheaper than an LRU list.
for k, _ in sorted(self._m.items(), key=lambda kv: kv[1][1])[: self.max_entries // 10]:
self._m.pop(k, None)
def forget_engine(self, engine_name: str) -> None:
with self._lock:
for k in [k for k, (n, _) in self._m.items() if n == engine_name]:
self._m.pop(k, None)
def __len__(self) -> int:
return len(self._m)
codai/frontproxy/app.py
View file @ 0d4a338b
......@@ -108,6 +108,16 @@ class FrontProxy:
# without asking the engine. Newest first; bounded.
import collections as _collections
self._recent_activity = _collections.deque(maxlen=25)
# Prometheus counters + per-key usage (codai/frontproxy/metrics.py), and
# the conversation → engine map for prefix-cache-aware routing.
from codai.frontproxy.metrics import Metrics
from codai.frontproxy.affinity import PrefixAffinity
self.metrics = Metrics()
_ccfg = getattr(config, "cluster", None)
self.affinity = PrefixAffinity(
ttl_s=float(getattr(_ccfg, "prefix_affinity_ttl_s", 1800.0) or 0.0))
self._key_names: dict = {}
self._key_names_at = 0.0
# Last-good per-model instance detail from an engine (synced only when the
# engine is idle), so model-loaded-status can serve loaded/running from the
# front's own state without ever hitting a busy engine.
......@@ -333,12 +343,18 @@ class FrontProxy:
model = None
_is_infer = _router.is_inference_path(path)
_akey = self._affinity_key(path, body, headers)
def _pick():
return _router.pick_engine(
e = _router.pick_engine(
self.registry, path, method, model,
required_cap=self._required_cap(path, model),
default_engine=self.default_engine, pinned=self._pin_for(model),
pin_fallback=bool(self._model_info(model).get("engine_fallback")))
pin_fallback=bool(self._model_info(model).get("engine_fallback")),
prefer=self.affinity.get(_akey))
if e is not None:
self.affinity.remember(_akey, e.name)
return e
# Brokered requests must not hard-fail in the startup/reload window where no
# engine is ready yet (e.g. an OOM-triggered evict+reload in progress). Wait
......@@ -435,7 +451,8 @@ class FrontProxy:
self._swap_release(_swap_tok)
self._rate_release(_rate_tok)
if _router.is_inference_path(path):
self._record_activity(model, self._task_kind(path), _status, _started)
self._record_activity(model, self._task_kind(path), _status, _started,
engine=engine, headers=headers)
# Surface the engine's actual reply so a brokered request that "doesn't get
# executed" (e.g. an instant small error body) is diagnosable from the log.
print("[front] broker route: %s %s -> engine#%s(%s) status=%s bytes=%d preview=%r"
......@@ -492,12 +509,18 @@ class FrontProxy:
"x_queue_info": {"status": "loading", "message": msg},
}) + "\n\n"
_akey = self._affinity_key(path, body, headers)
def _pick():
return _router.pick_engine(
e = _router.pick_engine(
self.registry, path, method, model,
required_cap=self._required_cap(path, model),
default_engine=self.default_engine, pinned=self._pin_for(model),
pin_fallback=bool(self._model_info(model).get("engine_fallback")))
pin_fallback=bool(self._model_info(model).get("engine_fallback")),
prefer=self.affinity.get(_akey))
if e is not None:
self.affinity.remember(_akey, e.name)
return e
engine = _pick()
if engine is None and _is_infer:
......@@ -613,7 +636,8 @@ class FrontProxy:
self._swap_release(_swap_tok)
self._rate_release(_rate_tok)
if _is_infer:
self._record_activity(model, self._task_kind(path), _status, _started)
self._record_activity(model, self._task_kind(path), _status, _started,
engine=engine, headers=headers)
# ------------------------------------------------------------------ helpers
@staticmethod
......@@ -691,6 +715,7 @@ class FrontProxy:
kind = _fan.kind_for(path)
if kind is None:
return None
_t0 = time.time()
ct = (request.headers.get("content-type") or "").lower()
if model is None and "multipart/form-data" in ct:
body_bytes = await request.body()
......@@ -730,7 +755,8 @@ class FrontProxy:
return None
if res is None:
return None
self._record_activity(model, self._task_kind(path), res.status, time.time())
self._record_activity(model, self._task_kind(path), res.status, _t0,
engine="fanout", headers=request.headers)
return Response(content=res.content, status_code=res.status, media_type=res.media_type)
def _load_pins(self) -> dict:
......@@ -786,19 +812,69 @@ class FrontProxy:
info[f[:-5]] = rec
return info
def _record_activity(self, model, kind, status, started_at):
def _record_activity(self, model, kind, status, started_at, engine=None, headers=None):
"""Append one completed inference to the recent-activity ring (newest first),
for the front-native Overview dashboard."""
for the front-native Overview dashboard — and count it for /metrics,
attributed to the API key (``headers``) and the engine that served it."""
try:
import time as _t
now = _t.time()
dur = now - (started_at or now)
self._recent_activity.appendleft({
"time": started_at or _t.time(),
"time": started_at or now,
"model": model or "", "type": kind or "text",
"status": int(status or 0),
"duration": round(_t.time() - (started_at or _t.time()), 1),
"duration": round(dur, 1),
})
ename = getattr(engine, "name", engine) if engine is not None else ""
self.metrics.observe(self._key_label(headers), model, kind, status,
ename or "", dur)
except Exception:
pass
def _key_label(self, headers) -> str:
"""Who made the request, for metering: the NAME of the API token (Tokens
page), ``session`` for a browser session, ``anonymous`` otherwise. Token
values never leave the process; the name map is refreshed every 30 s."""
try:
if headers is None:
return "anonymous"
get = headers.get if hasattr(headers, "get") else (lambda k, d="": d)
auth = get("authorization") or get("Authorization") or ""
if auth.lower().startswith("bearer "):
tok = auth[7:].strip()
import time as _t
if _t.time() - self._key_names_at > 30.0 and self._config_dir:
try:
from pathlib import Path
from codai.admin.auth import SessionManager
data = SessionManager(Path(self._config_dir))._load_auth_data()
self._key_names = {t.get("token", ""): (t.get("name") or f"token{t.get('id')}")
for t in data.get("tokens", []) if t.get("token")}
except Exception:
pass
self._key_names_at = _t.time()
if tok in self._key_names:
return self._key_names[tok]
ccfg = getattr(self.config, "cluster", None)
if ccfg is not None and getattr(ccfg, "token", "") and tok == ccfg.token:
return "cluster"
return "unknown-token"
cookie = get("cookie") or get("Cookie") or ""
if "session" in cookie:
return "session"
except Exception:
pass
return "anonymous"
def _affinity_key(self, path, body, headers=None):
ccfg = getattr(self.config, "cluster", None)
if ccfg is not None:
if not getattr(ccfg, "prefix_affinity", True):
return None
self.affinity.ttl_s = float(getattr(ccfg, "prefix_affinity_ttl_s", 1800.0) or 0.0)
from codai.frontproxy.affinity import conversation_key
return conversation_key(path, body, headers)
async def _spill_open(self, model, trigger, path, body_bytes):
"""Open a burst request against this model's RunPod spillover target.
......@@ -1614,6 +1690,7 @@ class FrontProxy:
"pid": pid,
"capabilities": sorted(e.capabilities or ()),
"remote": bool(getattr(e, "remote", False)),
"discovered": bool(getattr(e, "discovered", False)),
"url": e.url if getattr(e, "remote", False) else "",
"node_engines": list(getattr(e, "node_engines", []) or []),
"rpc_servers": list(getattr(e, "rpc_servers", []) or []),
......@@ -1998,7 +2075,8 @@ class FrontProxy:
_swap_acq.cancel()
self._swap_release(_swap_tok)
if _router.is_inference_path(path):
self._record_activity(model, self._task_kind(path), _status, _started)
self._record_activity(model, self._task_kind(path), _status, _started,
engine=engine, headers=request.headers)
return StreamingResponse(_gen(), status_code=200,
media_type="text/event-stream")
......@@ -2256,11 +2334,15 @@ class FrontProxy:
if fanned is not None:
return fanned
_akey = self._affinity_key(path, body_bytes, request.headers)
engine = _router.pick_engine(
self.registry, path, method, model,
required_cap=self._required_cap(path, model),
default_engine=self.default_engine, pinned=self._pin_for(model),
pin_fallback=bool(self._model_info(model).get("engine_fallback")))
pin_fallback=bool(self._model_info(model).get("engine_fallback")),
prefer=self.affinity.get(_akey))
if engine is not None:
self.affinity.remember(_akey, engine.name)
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.
......@@ -2399,7 +2481,8 @@ class FrontProxy:
self._rate_release(_rate_tok)
if _meta is not None:
self._record_activity(model, self._task_kind(path),
rp_resp.status_code, _started)
rp_resp.status_code, _started,
engine=engine, headers=request.headers)
# A node's JSON answer may name files by the node's own URL: rewrite
# them to this front (small bodies only; streams pass untouched).
......@@ -2736,7 +2819,15 @@ def build_app(config, config_dir=None) -> FastAPI:
if not auth.lower().startswith("bearer "):
return False
tok = auth[7:].strip()
if not tok or not config_dir:
if not tok:
return False
# The shared cluster token (Settings → Cluster) is enough: it is what a
# head that found this node on the LAN presents.
import hmac as _hmac
shared = (getattr(ccfg, "token", "") or "") if ccfg is not None else ""
if shared and _hmac.compare_digest(shared, tok):
return True
if not config_dir:
return False
try:
from pathlib import Path
......@@ -2878,6 +2969,62 @@ def build_app(config, config_dir=None) -> FastAPI:
None, build_overview, front, _rpc_endpoints())
return JSONResponse(data)
@app.get("/cluster/logs", include_in_schema=False)
async def _cluster_logs(request: Request):
"""This front's recent output (its own lines and every engine line it
re-emits), for a head's Cluster page. Node side."""
if not _cluster_token_ok(request):
return JSONResponse({"detail": "Unauthorized"}, status_code=401)
try:
n = max(1, min(2000, int(request.query_params.get("lines", "200"))))
except ValueError:
n = 200
return JSONResponse({"node": _node_name(), "lines": recent_log_lines(n)})
@app.get("/admin/api/cluster/logs", include_in_schema=False)
async def _cluster_node_logs(request: Request):
"""Head side: the log tail of one node (``?node=NAME``), or of this
front when ``node`` is blank."""
if not front._has_cred(request):
return JSONResponse({"detail": "Unauthorized"}, status_code=401)
name = (request.query_params.get("node") or "").strip()
try:
n = max(1, min(2000, int(request.query_params.get("lines", "200"))))
except ValueError:
n = 200
if not name:
return JSONResponse({"node": _node_name(), "lines": recent_log_lines(n)})
eng = next((e for e in front.registry.remotes() if e.name == name), None)
if eng is None:
return JSONResponse({"error": f"no node named {name!r}"}, status_code=404)
try:
r = await eng.http_short.get(eng.url + "/cluster/logs", params={"lines": n})
if r.status_code != 200:
return JSONResponse({"node": name, "error": f"HTTP {r.status_code}",
"lines": []}, status_code=502)
return JSONResponse(r.json())
except Exception as exc:
return JSONResponse({"node": name, "error": str(exc)[:200], "lines": []},
status_code=502)
@app.get("/admin/api/usage", include_in_schema=False)
async def _usage(request: Request):
"""Requests, errors and time per API key / model / kind since the front
started — the same numbers /metrics exports, as a table."""
if not front._has_cred(request):
return JSONResponse({"detail": "Unauthorized"}, status_code=401)
return JSONResponse(front.metrics.usage_tables())
@app.get("/metrics", include_in_schema=False)
async def _metrics(request: Request):
"""Prometheus exposition. Open when ``server.metrics_public`` is on,
otherwise the same light credential the telemetry endpoints take."""
pub = bool(getattr(config.server, "metrics_public", False))
if not pub and not front._has_cred(request) and not _cluster_token_ok(request):
return JSONResponse({"detail": "Unauthorized"}, status_code=401)
from codai.frontproxy.metrics import render
return Response(content=render(front), media_type="text/plain; version=0.0.4; charset=utf-8")
# Serve the admin / Studio UI pages from the FRONT (rendered here, sessions
# validated locally) so navigating the dashboard never waits on a GIL-busy
# engine mid-generation. The engine handles only generation; pages live here.
......@@ -2977,6 +3124,22 @@ class _TimestampedStdout:
def __init__(self, stream):
self._stream = stream
self._at_line_start = True
self._partial = ""
def _ring(self, text: str) -> None:
"""Keep the last lines (complete ones) for /cluster/logs."""
try:
self._partial += text
if '\n' not in self._partial:
return
*lines, self._partial = self._partial.split('\n')
for ln in lines:
if ln.startswith('\r'):
ln = ln.rsplit('\r', 1)[-1] # a progress bar: keep its last state
if ln.strip():
_LOG_RING.append(ln[:400])
except Exception:
pass
def write(self, s):
if not s:
......@@ -2996,12 +3159,23 @@ class _TimestampedStdout:
buf.append(ts)
self._at_line_start = False
buf.append(part)
return self._stream.write(''.join(buf))
out = ''.join(buf)
self._ring(out)
return self._stream.write(out)
def __getattr__(self, name):
return getattr(self._stream, name)
_LOG_RING = __import__("collections").deque(maxlen=2000)
def recent_log_lines(n: int = 200) -> list:
"""The front's last ``n`` output lines (what `docker logs` would show)."""
items = list(_LOG_RING)
return items[-n:] if n < len(items) else items
def run_front(config, args) -> None:
"""Build the front app, start engine supervision, and serve on the public port."""
# Timestamp every terminal line the front emits (raw prints, uvicorn logs, and
......
codai/frontproxy/engine_supervisor.py
View file @ 0d4a338b
......@@ -374,12 +374,40 @@ class EngineSupervisor:
_NODE_ID_BASE = 500
def _node_specs(self) -> list:
"""The nodes this head uses: ``cluster.nodes`` from config, plus — with
discovery + auto_join — every member found on the LAN holding the same
cluster token. A listed node with a blank api_key gets the shared
token, so one secret really is the whole setup."""
ccfg = getattr(self.config, "cluster", None)
try:
from codai.cluster.nodes import enabled_nodes
return enabled_nodes(getattr(self.config, "cluster", None))
specs = enabled_nodes(ccfg)
except Exception as exc:
print(f"[front] cluster nodes not read: {exc}", flush=True)
return []
specs = []
token = (getattr(ccfg, "token", "") or "") if ccfg else ""
if token:
for sp in specs:
if not sp.api_key:
sp.api_key = token
disc = getattr(self, "discovery", None)
if (disc is not None and token and ccfg is not None
and getattr(ccfg, "enabled", False) and getattr(ccfg, "auto_join", True)):
try:
from codai.cluster.discovery import member_node_specs
known = [sp.name for sp in specs] + [sp.url for sp in specs]
# Every listed node, enabled or not, blocks auto-join of the same name.
for n in (getattr(ccfg, "nodes", None) or []):
if isinstance(n, dict):
known += [str(n.get("name") or ""), str(n.get("url") or "").rstrip("/")]
found = member_node_specs(disc.members(), token, known,
timeout_s=float(getattr(ccfg, "poll_timeout_s", 4.0) or 4.0))
for sp in found:
sp.discovered = True
specs += found
except Exception as exc:
print(f"[front] discovered nodes not merged: {exc}", flush=True)
return specs
def _make_remote_engine(self, spec, eid: int) -> Engine:
from codai.cluster.nodes import async_clients, sync_client
......@@ -394,6 +422,7 @@ class EngineSupervisor:
e.http_short, e.http_long = async_clients(spec)
e.http_sync = sync_client(spec)
e.node_sig = self._node_sig(spec)
e.discovered = bool(getattr(spec, "discovered", False))
return e
def _build_remote_engines(self, next_id: int) -> list:
......@@ -758,6 +787,10 @@ class EngineSupervisor:
self.rpc_manager.start()
except Exception as exc:
print(f"[front] rpc servers not started: {exc}", flush=True)
# mDNS: announce this install, find the others (codai/cluster/discovery.py).
self.discovery = None
self._discovery_sig = None
self._sync_discovery()
self._poll_thread = threading.Thread(target=self._poll_loop, daemon=True)
self._poll_thread.start()
# Central thermal supervisor: monitor temps from the front (responsive even
......@@ -1118,6 +1151,18 @@ class EngineSupervisor:
except Exception as exc:
if self.debug:
print(f"[front] cluster node sync error: {exc}", flush=True)
try:
self._sync_discovery()
except Exception as exc:
print(f"[front] discovery sync error: {exc}", flush=True)
disc = getattr(self, "discovery", None)
if disc is not None and disc.started_at:
# Re-announce when what this machine offers changes (an engine
# came up with new capabilities), so heads see the right set.
caps = tuple(self._local_capabilities())
if caps != getattr(self, "_announced_caps", None):
self._announced_caps = caps
disc.refresh()
self._push_assignment_if_changed(client)
self._flush_pending_reloads(client) # deliver queued reloads to idle engines
for engine in self.registry.all():
......@@ -1187,10 +1232,63 @@ class EngineSupervisor:
self._stopped.wait(self.config.server.proxy_status_timeout)
client.close()
def _discovery_wanted(self):
"""(signature, kwargs) of the discovery the config asks for, or (None, None)."""
ccfg = getattr(self.config, "cluster", None)
if ccfg is None or not getattr(ccfg, "discovery", False):
return None, None
srv = self.config.server
kw = dict(name=(getattr(ccfg, "node_name", "") or "").strip(),
port=int(getattr(srv, "port", 8776) or 8776),
token=getattr(ccfg, "token", "") or "",
advertise_host=getattr(ccfg, "advertise_host", "") or "",
scheme="https" if getattr(srv, "https", False) else "http",
serve=bool(getattr(ccfg, "serve", True)))
return tuple(sorted(kw.items())), kw
def _sync_discovery(self) -> None:
"""Start, restart or stop discovery to match the (live-refreshed) config."""
sig, kw = self._discovery_wanted()
if sig == self._discovery_sig:
return
if self.discovery is not None:
try:
self.discovery.stop()
except Exception:
pass
self.discovery = None
self._discovery_sig = sig
if kw is None:
return
try:
from codai.cluster.discovery import Discovery
from codai import __version__
self.discovery = Discovery(version=__version__,
capabilities=self._local_capabilities, **kw)
if not self.discovery.start():
self.discovery.stop()
# keep the object: its .error tells the Cluster page why
except Exception as exc:
print(f"[front] discovery not started: {exc}", flush=True)
def _local_capabilities(self) -> list:
"""Union of the local engines' capabilities (what we announce on mDNS)."""
caps = set()
for e in self.registry.all():
if not e.remote and getattr(e, "role", "engine") == "engine":
caps |= set(e.capabilities or ())
return sorted(caps)
def _poll_remote(self, engine: Engine) -> None:
"""One state poll of a cluster node. Its front answers from its own
registry, so this is cheap and never waits on a busy GPU engine there."""
registry, so this is cheap and never waits on a busy GPU engine there.
Recovery: a node that comes back after being unreachable has very
likely restarted, and the model entries this head pushed live only in
its memory — so its assignment is queued for a fresh push, and the
models it was assigned load again there without anyone clicking."""
healthy = False
was_healthy = bool(engine.healthy)
try:
r = engine.http_sync.get(engine.url + engine.state_path)
if r.status_code == 200:
......@@ -1220,8 +1318,16 @@ class EngineSupervisor:
except Exception as exc:
self.registry.update_state(engine.id, healthy=False,
last_error=str(exc)[:200])
if self.debug and self._health.get(engine.id) != healthy:
if healthy and not was_healthy and self._health.get(engine.id) is False:
owned = sorted(engine.assigned_models)
if owned:
self._pending_reload[engine.name] = owned
self._node_recoveries = getattr(self, "_node_recoveries", 0) + 1
print(f"[front] node '{engine.name}' is back — re-pushing its "
f"{len(owned)} assigned model(s)", flush=True)
if self._health.get(engine.id) != healthy:
self._health[engine.id] = healthy
if self.debug or not healthy:
print(f"[front] node '{engine.name}' "
f"{'reachable' if healthy else 'not responding: ' + engine.last_error}",
flush=True)
......@@ -1351,6 +1457,11 @@ class EngineSupervisor:
self.rpc_manager.stop()
except Exception:
pass
try:
if getattr(self, "discovery", None) is not None:
self.discovery.stop()
except Exception:
pass
def _signal_group(proc, sig):
# Engines are started in their own session (setsid), so killing the
......
codai/frontproxy/metrics.py 0 → 100644
View file @ 0d4a338b
# 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.
"""Prometheus exposition for the front — ``GET /metrics`` — and the per-key
usage tables behind it.
No client library: the text format is a few lines per sample and writing it
by hand keeps the front's dependency list where it is. Counters live for
the life of the front process (Prometheus expects exactly that and takes
``rate()`` over them); everything else is read from the registry, the
supervisor and the RunPod ledger at scrape time.
What is exported, all prefixed ``coderai_``:
* ``requests_total{key,model,kind,status,engine}`` and
``request_seconds_sum/_count`` with the same labels — every inference the
front completed, attributed to the API key that made it (its name on the
Tokens page; ``session`` for a browser; ``anonymous`` when open);
* ``engine_up{engine,backend,remote}``, ``engine_inflight``,
``engine_vram_free_bytes`` / ``_total_bytes``, ``engine_loaded_models``;
* ``node_up{node}``, ``node_recoveries_total``;
* ``rpc_server_up{endpoint}``; ``discovery_peers`` / ``discovery_members``;
* ``runpod_spend_usd{model,period}`` (trailing hour/day/week/month), ``runpod_pods``;
* ``info{version}`` = 1.
"""
import threading
import time
from collections import defaultdict
def _esc(v) -> str:
return str(v).replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n")
def _labels(d: dict) -> str:
if not d:
return ""
return "{" + ",".join(f'{k}="{_esc(v)}"' for k, v in d.items()) + "}"
class Metrics:
"""Counters the front feeds from ``_record_activity``."""
def __init__(self):
self._lock = threading.Lock()
self._n = defaultdict(int) # (key, model, kind, status, engine) -> count
self._t = defaultdict(float) # same key -> seconds
self.started = time.time()
def observe(self, key: str, model: str, kind: str, status: int, engine: str,
seconds: float) -> None:
k = (key or "anonymous", model or "", kind or "text", int(status or 0), engine or "")
with self._lock:
self._n[k] += 1
self._t[k] += max(0.0, float(seconds or 0.0))
# ------------------------------------------------------------ tables
def usage_tables(self) -> dict:
"""Per key, per model and per kind — for the Cluster page."""
with self._lock:
items = [(k, self._n[k], self._t[k]) for k in self._n]
by_key, by_model = {}, {}
for (key, model, kind, status, engine), n, t in items:
for table, name in ((by_key, key), (by_model, model or "(none)")):
row = table.setdefault(name, {"requests": 0, "errors": 0, "seconds": 0.0,
"kinds": {}, "engines": {}})
row["requests"] += n
row["seconds"] += t
if status >= 400 or status == 0:
row["errors"] += n
row["kinds"][kind] = row["kinds"].get(kind, 0) + n
if engine:
row["engines"][engine] = row["engines"].get(engine, 0) + n
for table in (by_key, by_model):
for row in table.values():
row["seconds"] = round(row["seconds"], 1)
return {"since": self.started, "by_key": by_key, "by_model": by_model}
# ---------------------------------------------------------- exposition
def counter_lines(self) -> list:
with self._lock:
items = [(k, self._n[k], self._t[k]) for k in sorted(self._n)]
out = ["# HELP coderai_requests_total Inference requests completed by the front.",
"# TYPE coderai_requests_total counter"]
for (key, model, kind, status, engine), n, _ in items:
out.append("coderai_requests_total" + _labels(
{"key": key, "model": model, "kind": kind, "status": status, "engine": engine})
+ f" {n}")
out += ["# HELP coderai_request_seconds Wall time of completed inference requests.",
"# TYPE coderai_request_seconds summary"]
for (key, model, kind, status, engine), n, t in items:
lab = {"key": key, "model": model, "kind": kind, "status": status, "engine": engine}
out.append("coderai_request_seconds_sum" + _labels(lab) + f" {t:.3f}")
out.append("coderai_request_seconds_count" + _labels(lab) + f" {n}")
return out
def render(front) -> str:
"""The whole /metrics page."""
from codai import __version__
lines = ["# HELP coderai_info CoderAI version.", "# TYPE coderai_info gauge",
"coderai_info" + _labels({"version": __version__}) + " 1"]
m = getattr(front, "metrics", None)
if m is not None:
lines += m.counter_lines()
lines += ["# HELP coderai_front_start_time_seconds When this front started.",
"# TYPE coderai_front_start_time_seconds gauge",
f"coderai_front_start_time_seconds {m.started:.0f}"]
# Engines and nodes, from the registry.
up, infl, vfree, vtot, loaded, nup = [], [], [], [], [], []
for e in front.registry.all():
if getattr(e, "role", "engine") == "system":
continue
remote = bool(getattr(e, "remote", False))
lab = {"engine": e.name, "backend": e.backend or "", "remote": "1" if remote else "0"}
up.append("coderai_engine_up" + _labels(lab) + f" {1 if e.healthy else 0}")
infl.append("coderai_engine_inflight" + _labels(lab) + f" {int(getattr(e, 'inflight', 0) or 0)}")
v = e.vram or {}
if v:
vfree.append("coderai_engine_vram_free_bytes" + _labels(lab)
+ f" {float(v.get('free', 0) or 0) * (1 << 30):.0f}")
vtot.append("coderai_engine_vram_total_bytes" + _labels(lab)
+ f" {float(v.get('total', 0) or 0) * (1 << 30):.0f}")
loaded.append("coderai_engine_loaded_models" + _labels(lab) + f" {len(e.loaded_models)}")
if remote:
nup.append("coderai_node_up" + _labels({"node": e.name}) + f" {1 if e.healthy else 0}")
for help_, typ, rows in (
("Engine (or node) answering its health poll.", "gauge", up),
("Requests currently proxied to the engine.", "gauge", infl),
("Free VRAM the engine reports.", "gauge", vfree),
("Total VRAM the engine reports.", "gauge", vtot),
("Models resident on the engine.", "gauge", loaded),
("Cluster node reachable.", "gauge", nup)):
if rows:
name = rows[0].split("{", 1)[0]
lines += [f"# HELP {name} {help_}", f"# TYPE {name} {typ}"] + rows
sup = getattr(front, "supervisor", None)
lines += ["# HELP coderai_node_recoveries_total Nodes that came back and had their models re-pushed.",
"# TYPE coderai_node_recoveries_total counter",
f"coderai_node_recoveries_total {int(getattr(sup, '_node_recoveries', 0) or 0)}"]
# RPC servers this machine runs.
mgr = getattr(sup, "rpc_manager", None) if sup else None
if mgr is not None:
try:
rows = []
for ep in mgr.endpoints():
rows.append("coderai_rpc_server_up" + _labels({"endpoint": ep.get("endpoint", "")})
+ f" {1 if ep.get('listening') else 0}")
if rows:
lines += ["# HELP coderai_rpc_server_up llama.cpp rpc-server process running.",
"# TYPE coderai_rpc_server_up gauge"] + rows
except Exception:
pass
# Discovery.
disc = getattr(sup, "discovery", None) if sup else None
if disc is not None:
try:
peers = disc.peers()
lines += ["# HELP coderai_discovery_peers coderai installs seen on the LAN.",
"# TYPE coderai_discovery_peers gauge",
f"coderai_discovery_peers {len(peers)}",
"# HELP coderai_discovery_members Peers holding this cluster's token.",
"# TYPE coderai_discovery_members gauge",
f"coderai_discovery_members {sum(1 for p in peers if p.get('member'))}"]
except Exception:
pass
# RunPod spend.
try:
from codai.api import runpod_ledger, runpod_worker
tot = runpod_ledger.totals()
rows = []
for model, rec in (tot.get("per_model") or {}).items():
if not isinstance(rec, dict):
continue
for period, usd in rec.items():
try:
rows.append("coderai_runpod_spend_usd" + _labels({"model": model, "period": period})
+ f" {float(usd or 0):.4f}")
except (TypeError, ValueError):
continue
for period, usd in (tot.get("global") or {}).items():
try:
rows.append("coderai_runpod_spend_usd" + _labels({"model": "", "period": period})
+ f" {float(usd or 0):.4f}")
except (TypeError, ValueError):
continue
if rows:
lines += ["# HELP coderai_runpod_spend_usd Money spent on rented pods in the trailing period.",
"# TYPE coderai_runpod_spend_usd gauge"] + rows
try:
npods = len(runpod_worker.pods_status() or [])
except Exception:
npods = None
if npods is not None:
lines += ["# HELP coderai_runpod_pods Rented pods currently running.",
"# TYPE coderai_runpod_pods gauge", f"coderai_runpod_pods {npods}"]
except Exception:
pass
# Queue / active requests as the overview counts them.
try:
active = sum(int(getattr(e, "inflight", 0) or 0) for e in front.registry.all())
lines += ["# HELP coderai_requests_active Requests in flight through the front.",
"# TYPE coderai_requests_active gauge", f"coderai_requests_active {active}"]
except Exception:
pass
return "\n".join(lines) + "\n"
codai/frontproxy/registry.py
View file @ 0d4a338b
......@@ -272,24 +272,36 @@ class EngineRegistry:
elif not healthy:
e.cooling = None
@staticmethod
def engine_holds(e: Engine, model_key: str) -> bool:
"""Does this engine have the model resident (forgiving key match)?"""
if not model_key:
return False
short = _short_stem(model_key)
for k in e.loaded_models:
if k == model_key or _short_stem(k) == short \
or k.endswith(model_key) or model_key.endswith(k.split(":")[-1]):
return True
return False
def engine_for_model(self, model_key: str, required_cap: Optional[str] = None) -> Optional[Engine]:
"""Return a healthy, capability-compatible engine that already has the model
resident, if any.
resident, if any — the least busy one when several hold it (replicas
here and on nodes), local before remote on a tie.
Matching is forgiving: exact key, short-name, or type-prefixed variants —
the same fuzzy spirit the manager uses, but read-only over loaded keys."""
if not model_key:
return None
short = _short_stem(model_key)
with self._lock:
for e in self._engines.values():
if not e.healthy or not e.can_serve(required_cap):
continue
for k in e.loaded_models:
if k == model_key or _short_stem(k) == short \
or k.endswith(model_key) or model_key.endswith(k.split(":")[-1]):
return e
holders = [e for e in self._engines.values()
if e.healthy and e.can_serve(required_cap)
and self.engine_holds(e, model_key)]
if not holders:
return None
holders.sort(key=lambda e: (int(getattr(e, "inflight", 0) or 0),
1 if getattr(e, "remote", False) else 0, e.id))
return holders[0]
def engine_for_assigned(self, model_key: str) -> Optional[Engine]:
"""The engine the front ASSIGNED this model to (single owner), or None.
......
codai/frontproxy/router.py
View file @ 0d4a338b
......@@ -154,12 +154,15 @@ def pick_engine(registry: EngineRegistry, path: str, method: str,
model: Optional[str], required_cap: Optional[str] = None,
default_engine: Optional[str] = None,
pinned: Optional[str] = None,
pin_fallback: bool = False) -> Optional[Engine]:
pin_fallback: bool = False,
prefer: Optional[str] = None) -> Optional[Engine]:
"""Return the engine to proxy this request to, or None if none are ready.
Precedence for inference: per-model pin → engine already holding the model →
configured default engine → least-loaded compatible engine. Each candidate must
be capability-compatible (``required_cap``) and healthy. Works even when the
Precedence for inference: per-model pin → the engine that served this
conversation before (``prefer``, prefix-cache affinity) when it still
holds the model → engine already holding the model → configured default
engine → least-loaded compatible engine. Each candidate must be
capability-compatible (``required_cap``) and healthy. Works even when the
model id isn't known (e.g. a multipart transcription upload), routing purely by
capability.
"""
......@@ -204,6 +207,16 @@ def pick_engine(registry: EngineRegistry, path: str, method: str,
if not pin_fallback:
return None
# 0b. Prefix-cache affinity: the conversation's earlier turns ran on
# ``prefer`` — go back there while it is alive and still holds the
# model, so its warm KV/prefix cache is reused. Only matters when the
# model is resident in more than one place; never overrides a pin.
if prefer and model:
pe = registry.by_name(prefer)
if (pe is not None and pe.is_alive() and pe.can_serve(cap)
and registry.engine_holds(pe, model)):
return pe
# 1. The front's precomputed assignment is authoritative — it already folds
# in the default engine and balanced auto-selection, and keeps a model on
# exactly one engine. Honour it first when it's compatible.
......
docs/cluster.md
View file @ 0d4a338b
......@@ -44,6 +44,39 @@ the head holds. Optionally, in Settings → Cluster:
If the node serves HTTPS with a self-signed certificate, copy its PEM: the
head pastes it in the node row (verify = *pem*).
### Zero-config: a shared token and mDNS
The manual way above (a token per node, its URL on the head) still works
and is what you use across subnets or VPNs. On one LAN there is a shorter
way — the same on every box, head and nodes alike, Settings → Cluster:
1. **Cluster token** — one secret, typed on every install. A node accepts
it on its `/cluster/*` endpoints exactly like one of its own API tokens;
a head uses it as the API key of every node it discovers, and of any
listed node whose token field is blank.
2. **Find nodes on the LAN (mDNS)** — the install announces itself as a
`_coderai._tcp.local.` service (name, version, port, capabilities) and
browses for the others. The announcement carries an HMAC fingerprint of
the token, never the token: peers with the same fingerprint are
*members*, everyone else is merely *seen*. The Cluster page lists both.
3. **Auto-join members as nodes** (on by default) — every member that
serves becomes a cluster node of this head, with no row to fill in. It
leaves when its announcement stops; a node listed by hand under the same
name is never duplicated.
```json
"cluster": {"enabled": true, "token": "…", "discovery": true, "auto_join": true}
```
Every box can have all three on: each is then a head of the others and a
node of the others, and a model pinned to a name runs where that name is.
mDNS is link-local multicast: it does not cross routers, and a container
only sees it on the host's network — `coderai-docker --host-network` (the
server then listens on the host port directly; RPC and Ray ports need no
mapping either). The `zeroconf` package is in the images and in
`requirements.txt`; without it the Cluster page says discovery is off.
### On the head
Settings → Cluster → *Use the nodes below as engines*, one row per node:
......@@ -146,6 +179,17 @@ devices included (the VRAM and Speed strategies both see them).
"tensor_split": "0.4,0.3,0.3"}
```
**How the split is made** (same page, *Cards on other machines*):
*layer* (default) gives each device whole layers, and per token only the
activations between consecutive layers cross to the next device — what
Ethernet carries well. *Row* is llama.cpp's tensor parallelism: every
matrix is split by rows over all the devices, local cards and RPC servers
alike, so they work on every layer together — at the price of an
all-reduce per layer. On one PCIe bus it is faster; over a network it wants
a fabric measured in tens of Gb/s (10 GbE is the floor, RDMA the point),
and on 1 GbE it is slower than layer split. `"split_mode": "row"` in
models.json.
Under the hood (`codai/backends/ggml_rpc.py`): the RPC registration is
process-wide and permanent, so from the first registration on every load in
that engine gets an explicit device list — this machine's cards plus exactly
......@@ -164,10 +208,15 @@ vLLM and SGLang spread a model over machines on their own; coderai does the
choreography: starts the peers, tells them where rank 0 is, waits for them,
launches, and tears everything down with the service.
**Keep tensor parallel inside a machine** (an all-reduce every layer wants
NVLink/PCIe) and **make pipeline parallel the number of machines** (one
activation transfer per layer boundary). Every machine must run the same
build — the `coderai-vllm` image on all of them is the easy way.
**On ordinary Ethernet keep tensor parallel inside a machine** (an all-reduce
every layer wants NVLink/PCIe) and **make pipeline parallel the number of
machines** (one activation transfer per layer boundary). Tensor parallel
*may* span the nodes — set it larger than this machine's cards and Ray
places the ranks across the cluster, the two combining as tensor × pipeline
GPUs in total — which pays off on 10 GbE and up and is the right choice on
RDMA / InfiniBand (`NCCL_IB_*`, `NCCL_SOCKET_IFNAME` in the node's start
command environment). Every machine must run the same build — the
`coderai-vllm` image on all of them is the easy way.
### vLLM
......@@ -287,7 +336,27 @@ instead of one.
* Covers the in-process trainers: SD1.5, SDXL, Z-Image, the flow-DiT
family (LTX-2 …) and Wan, 4-bit QLoRA included.
## 7. Pools — several machines for one capability or one host model
## 7. Prefix-cache-aware routing
A model resident in more than one place — two local engines, a local engine
and a node, replicas after a fan-out — is served by the **least busy**
holder, local before remote on a tie. For a *conversation*, though, the
better place is where its earlier turns went: that engine's KV / prefix
cache (the multi-slot cache of the GGUF and HF backends, vLLM's prefix
caching) already holds the shared opening, and the prompt is processed
from the first new token rather than from scratch.
So the front keeps a small map *conversation → engine* and prefers that
engine while it is alive, capable and still holds the model. The key is
the one the engines use for their own slot affinity: an `X-Session-Id`
header or the OpenAI `user` field when the client sends one, else a hash
of the conversation's stable opening (system prompt + first user turn, or
the prompt head). A pin still wins; a preference never routes to an engine
that cannot serve. Settings → Cluster → *Prefix-cache routing* (on by
default, TTL 30 min): `cluster.prefix_affinity`,
`cluster.prefix_affinity_ttl_s`.
## 8. Pools — several machines for one capability or one host model
* **Remote capabilities** (Settings → Remote capabilities): a capability's
URL field takes several URLs, comma-separated. They form a pool: the
......@@ -299,21 +368,95 @@ instead of one.
one is started only when none is up, and one that dies mid-request hands
the request to another.
## 9. Operating it — metrics, usage, logs, recovery
* **`GET /metrics`** — Prometheus exposition, no client library. Per
completed request: `coderai_requests_total` and
`coderai_request_seconds_{sum,count}` labelled by **API key name** (the
name on the Tokens page; `session` for a browser; `anonymous`; `cluster`
for a head), model, kind, status and the engine or node that served it.
Gauges: `coderai_engine_up / _inflight / _vram_free_bytes / _vram_total_bytes
/ _loaded_models {engine,backend,remote}`, `coderai_node_up{node}`,
`coderai_node_recoveries_total`, `coderai_rpc_server_up{endpoint}`,
`coderai_discovery_peers / _members`, `coderai_runpod_spend_usd{model,period}`,
`coderai_runpod_pods`, `coderai_requests_active`, `coderai_info{version}`.
Guarded like the other telemetry (a session, an API token, or the cluster
token) unless Settings → Server → *Serve /metrics without a credential*
(`server.metrics_public`). A scrape config:
```yaml
scrape_configs:
- job_name: coderai
static_configs: [{targets: ["box1:8776", "box2:8776"]}]
authorization: {credentials: "<an API token or the cluster token>"}
```
* **Usage by key / by model** — the same counters as tables on the Cluster
page (`GET /admin/api/usage`): requests, errors, wall time, kinds, where
they ran. Since the front started; Prometheus is the durable store.
* **Node logs** — the Cluster page shows each node's recent output (its
front's lines and every engine line it re-emits) with the *log* button
on the node row, and this front's own with the button on the engines
card: `GET /cluster/logs?lines=N` on the node (token-guarded),
`GET /admin/api/cluster/logs?node=NAME` on the head.
* **Recovery** — a node that stops answering keeps its assignment; when it
answers again (it very likely restarted, and the entries the head pushed
lived only in its memory) the head pushes its assigned models again and
they load there without anyone clicking (`coderai_node_recoveries_total`).
Local engines have had this since the supervisor: a crashed engine is
respawned, a crash-looping one quarantined, and its models come back with
the next assignment push. A discovered node that vanishes from mDNS is
dropped and re-added when it reappears.
## 10. Apple Silicon, MLX and other accelerators — what fits
CoderAI's images are x86-64, CUDA + Vulkan; nothing of CoderAI runs
natively on macOS or on Ascend/Hygon/MThreads cards, and a container on
those has no GPU. What already works, and what would be needed:
* **A Mac as a text node today**: `mlx_lm.server` (or LM Studio, Ollama)
is OpenAI-compatible, so it is a *remote capability* for `text`
(Settings → Remote capabilities, its URL) or a `backend: host` model with
no start command. It is then one more place a text request can go — with
pools, failover and metering, but without CoderAI's model page, VRAM
accounting, or fan-out awareness. This is the cheapest useful step and
needs no code.
* **A real MLX node** would be a small Python package (`coderai-node-mlx`:
the front's `/cluster/state` + `/v1/chat/completions` + `/v1/embeddings`
over `mlx-lm`, later `mlx-audio` / diffusers-mlx) installed with pip on
the Mac and joining by token and mDNS like any node. Two to three days;
the node protocol is already the only contract a node must honour. That
is the point where "a stack of Macs" stops being exo's story alone.
* **Tensor parallel over Thunderbolt/RDMA** (exo's headline) is MLX-only
and macOS-only; nothing to port, and the same reasoning that keeps our
cross-machine split at layer/pipeline granularity applies on Ethernet.
* **Ascend / Hygon / MThreads / Iluvatar / MetaX** (GPUStack's list) each
come with a vLLM fork (vllm-ascend, …) or a llama.cpp backend (CANN,
MUSA). The cheapest route is the one we already have for anything that
speaks OpenAI: run the vendor's vLLM as a `service_url` / host model.
Native support means building their llama.cpp backends into the wheel
and a per-vendor image; do it when there is a card to test on, not
before.
## Reference
* Config: `cluster` section in `config.json` — `enabled`, `nodes`,
`node_name`, `serve`, `poll_timeout_s`, `rpc_servers`, `rpc_bin`,
`advertise_host`; `vllm.pipeline_parallel_size / distributed_executor_backend /
`advertise_host`, `token`, `discovery`, `auto_join`, `prefix_affinity`,
`prefix_affinity_ttl_s`; `server.metrics_public`; `vllm.pipeline_parallel_size / distributed_executor_backend /
ray_address / ray_port / nodes / nodes_ready_timeout_s`;
`ktransformers.nnodes / dist_init_addr / tp_size / nodes`.
* Per model (models.json): `engine` (a node name), `node_path` /
`node_paths`, `rpc_servers`, `vllm` block, `kt` block, `host.hosts`,
`node_paths`, `rpc_servers`, `split_mode`, `vllm` block, `kt` block, `host.hosts`,
`distribute` block, `components` block, `lora_train_nodes`.
* Node API (token of the node): `GET /cluster/state`, `GET /cluster/rpc-servers`,
`POST /cluster/reload-config`, `POST /cluster/model-load`, `POST /cluster/model-unload`.
* Head: `GET /admin/api/cluster` (what the Cluster page shows).
* Code: `codai/cluster/` (nodes, rpc, multinode, compat, overview, fanout,
components, ddp),
* Node API (a token of the node, or the cluster token): `GET /cluster/state`,
`GET /cluster/rpc-servers`, `GET /cluster/logs`, `POST /cluster/reload-config`,
`POST /cluster/model-load`, `POST /cluster/model-unload`.
* Head: `GET /admin/api/cluster` (what the Cluster page shows),
`GET /admin/api/cluster/logs?node=`, `GET /admin/api/usage`; `GET /metrics`.
* Code: `codai/cluster/` (nodes, discovery, rpc, multinode, compat, overview,
fanout, components, ddp), `codai/frontproxy/affinity.py`,
`codai/frontproxy/metrics.py`,
`codai/backends/ggml_rpc.py`, `codai/backends/overrides.py`,
`codai/frontproxy/engine_supervisor.py` (remote engines),
`codai/api/remote_gateway.EndpointPool`, `codai/api/host_worker.HostPool`.
docs/install-from-packages.md
View file @ 0d4a338b
......@@ -37,6 +37,22 @@ images) or 8776 (full image), and all of them can be a cluster node.
Every tag is also published with its version (`:0.2.20`); `:latest` moves
only after the image has completed a real request on real hardware.
**Every published image is signed** with [cosign](https://github.com/sigstore/cosign)
against the key pair in `packaging/cosign.pub` (the private half never
leaves the release machine). Check before you run:
```bash
cosign verify --key https://raw.githubusercontent.com/nextime/coderai/master/packaging/cosign.pub \
ghcr.io/nextime/coderai:latest
```
It prints the signed digest and the `version` / `git` annotations of the
build; a tag whose digest does not verify is not ours (or was re-pushed and
not yet re-signed — `packaging/sign-images.sh` is the release step that
signs, `… verify` the check across all images). Docker can enforce this:
a `policy-controller` / `cosign` admission rule on a cluster, or simply the
command above in whatever pulls.
For an offline machine, pull the image where there is a connection and
move it: `docker save ghcr.io/nextime/coderai:0.2.20 | gzip > coderai.tar.gz`
there, `docker load < coderai.tar.gz` here.
......@@ -60,7 +76,10 @@ the log; `./run_oci.sh --help` lists every option: `--vulkan`, `--all`,
`packaging/linux/run_oci.sh` installs itself as `coderai-docker`
(`~/.local/usr/bin`) on first use, so afterwards it is simply
`coderai-docker --nvidia -d` and `coderai-docker --upgrade`.
`coderai-docker --nvidia -d` and `coderai-docker --upgrade`. Add
`--host-network` on a box that should find, or be found by, the others over
mDNS (Settings → Cluster → discovery + a shared cluster token): link-local
multicast never crosses Docker's bridge.
Without the script, the plain `docker run` it builds is:
......@@ -114,7 +133,10 @@ docker run -d --name coderai-images --gpus all -p 8000:8000 \
then on the head: model → *Runs on: a machine of yours*, URL
`http://thatbox:8000`, token `some-token` — or Settings → Cluster → add it
as a node. `CODERAI_RPC_SERVER_PORT=50052` on the `coderai-llama` image also
as a node — or give both the same cluster token with discovery on and let
them find each other: on the capability container `--network host -e
CODERAI_CLUSTER_TOKEN=… -e CODERAI_DISCOVERY=1` (and `-e CODERAI_NODE_NAME=…`
to name it; `CODERAI_ADVERTISE_HOST` when it has several addresses). `CODERAI_RPC_SERVER_PORT=50052` on the `coderai-llama` image also
lends its card to a GGUF loaded elsewhere; `CODERAI_RAY_ADDRESS=…` turns the
`coderai-vllm` image into a Ray worker for a multi-node vLLM launch. See
[`cluster.md`](cluster.md) and [`remote-execution.md`](remote-execution.md).
packaging/cosign.pub 0 → 100644
View file @ 0d4a338b
-----BEGIN PUBLIC KEY-----
MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAEfxxda7vORqg1qr159ZE2ywAzTVXv
eUVcp92M6J+jbg1HKeedtnBvgiekjDOYaHU/lLFOCmIFTX2O7nLeRD5k7A==
-----END PUBLIC KEY-----
packaging/linux/Dockerfile.update
View file @ 0d4a338b
......@@ -50,6 +50,10 @@ RUN set -eux; \
# falls back to nvidia-smi when absent, hence `|| true` keeps offline builds OK.
RUN /opt/coderai/python/bin/python3 -m pip install --no-input --no-cache-dir "nvidia-ml-py>=12.0.0" || true
# mDNS cluster discovery (codai/cluster/discovery.py). Small; the code runs
# without it (discovery just stays off), hence `|| true`.
RUN /opt/coderai/python/bin/python3 -m pip install --no-input --no-cache-dir "zeroconf>=0.130" || true
# openssh-client so `coderai-docker --upgrade --ssh-key …` can fetch over SSH
# (git is present but the CUDA runtime base ships no ssh). Tiny; keep the overlay
# still fast. `|| true` keeps offline overlay builds working (HTTPS upgrade still
......
packaging/linux/run_oci.sh
View file @ 0d4a338b
......@@ -40,6 +40,10 @@ PORT="${CODERAI_PORT:-8776}"
# Host interface the published port binds to. Empty = Docker's default (all
# interfaces, 0.0.0.0). Set e.g. 127.0.0.1 to expose only on localhost.
HOST_BIND="${CODERAI_HOST_BIND:-}"
# --host-network: share the host's network namespace (no port mapping). Needed
# for mDNS cluster discovery (link-local multicast never crosses the docker
# bridge) and handy for RPC servers / Ray, whose ports then need no publishing.
HOST_NETWORK="${CODERAI_HOST_NETWORK:-0}"
# Extra CLI flags passed straight through to the coderai server inside the
# container (via CODERAI_EXTRA_ARGS, appended by the in-image coderai launcher).
# Built from --coderai-arg (repeatable, one token each) and --coderai-args "...".
......@@ -148,6 +152,9 @@ Upgrade (refresh the in-image code instead of running the server):
--host ADDR Host interface to bind the published port to (e.g.
127.0.0.1 for localhost-only, 0.0.0.0 for all interfaces).
Default: Docker's default (all interfaces).
--host-network Run on the host's network (docker --network host): the
server listens on PORT directly, and mDNS cluster
discovery, rpc-server and Ray ports work without mapping.
--coderai-arg ARG Pass one extra flag straight through to the coderai server
(e.g. --coderai-arg --some-flag). Repeatable; each ARG is a
single token (no embedded spaces).
......@@ -245,6 +252,8 @@ while [[ $# -gt 0 ]]; do
-p|--port)
[[ $# -ge 2 ]] || { echo "Error: $1 requires a port" >&2; exit 2; }
PORT="$2"; shift 2 ;;
--host-network)
HOST_NETWORK=1; shift ;;
--host)
[[ $# -ge 2 ]] || { echo "Error: --host requires an address" >&2; exit 2; }
HOST_BIND="$2"; shift 2 ;;
......@@ -458,7 +467,11 @@ if [[ -n "$HOST_BIND" ]]; then
else
PUBLISH="$PORT:8776"
fi
args=(run --rm --name "$NAME" --ipc=host -p "$PUBLISH" -e CODERAI_HOST=0.0.0.0 -e CODERAI_PORT=8776)
if [[ "$HOST_NETWORK" == "1" ]]; then
args=(run --rm --name "$NAME" --ipc=host --network host -e CODERAI_HOST="${HOST_BIND:-0.0.0.0}" -e CODERAI_PORT="$PORT")
else
args=(run --rm --name "$NAME" --ipc=host -p "$PUBLISH" -e CODERAI_HOST=0.0.0.0 -e CODERAI_PORT=8776)
fi
# Forward a HuggingFace token from the host env so the engines authenticate to the
# HF Hub (higher rate limits + gated models) instead of sending unauthenticated
# requests. huggingface_hub auto-reads these; no-op when unset.
......
packaging/runpod/profiles/core-light.txt
View file @ 0d4a338b
......@@ -57,3 +57,4 @@ websockets>=12.0
# (libnppicc.so.12). torch does not use NPP, so nothing else pulls it in, and
# torchcodec — which arrives as a transitive dependency of f5-tts — then fails
# to load with a message that blames FFmpeg and the torch version. It is neither.
zeroconf>=0.130
packaging/runpod/profiles/core.txt
View file @ 0d4a338b
......@@ -45,3 +45,4 @@ bitsandbytes>=0.43.0
# torchcodec — which arrives as a transitive dependency of f5-tts — then fails
# to load with a message that blames FFmpeg and the torch version. It is neither.
nvidia-npp-cu12
zeroconf>=0.130
packaging/runpod/publish_capability_images.sh
View file @ 0d4a338b
......@@ -157,6 +157,15 @@ else
say "${c_ok}all ${#built[@]} images pushed to ${NS}${c_off}"
fi
# ---- sign ------------------------------------------------------------------ #
# cosign signatures next to each pushed digest (packaging/sign-images.sh);
# users verify with packaging/cosign.pub. SIGN=0 skips it.
if [[ "${SIGN:-1}" == "1" && ${#push_failed[@]} -eq 0 ]]; then
sign_list=()
for t in "${built[@]}"; do sign_list+=("$t" "${t%:*}:latest"); done
IMAGES="${sign_list[*]}" NS="$NS" ./packaging/sign-images.sh || say "${c_warn}signing failed — run ./packaging/sign-images.sh all later${c_off}"
fi
# ---- make them public --------------------------------------------------- #
if [[ "$REGISTRY" == "ghcr.io" ]]; then
owner="${NS#*/}"
......
packaging/sign-images.sh 0 → 100755
View file @ 0d4a338b
#!/usr/bin/env bash
# Sign (or verify) the published CoderAI images with cosign.
#
# ./packaging/sign-images.sh # sign coderai:<version> + :latest
# ./packaging/sign-images.sh all # + the 19 capability images
# ./packaging/sign-images.sh verify [ref…] # check signatures (anyone can)
# IMAGES="ghcr.io/nextime/coderai:0.2.20" ./packaging/sign-images.sh
#
# Key-pair signing (works headless; keyless Sigstore needs a browser login):
# the private key lives OUTSIDE the repo at $COSIGN_KEY
# (default ~/.coderai-signing/cosign.key, generated on first use, encrypted
# with $COSIGN_PASSWORD — blank = an unencrypted key file, mode 600); the
# public key is committed as packaging/cosign.pub and is what users verify
# against. Signatures are attached to the image DIGEST in the registry
# (an OCI artifact next to the image), so a re-pushed tag with a different
# digest is unsigned until this script runs again.
#
# Verify (no account needed):
# cosign verify --key https://raw.githubusercontent.com/nextime/coderai/master/packaging/cosign.pub \
# ghcr.io/nextime/coderai:latest
set -euo pipefail
cd "$(dirname "$0")/.."
NS="${NS:-ghcr.io/nextime}"
VERSION="${VERSION:-$(python3 -c 'import re,pathlib; print(re.search(r"__version__ = \"([^\"]+)\"", pathlib.Path("codai/__init__.py").read_text()).group(1))')}"
COSIGN_KEY="${COSIGN_KEY:-$HOME/.coderai-signing/cosign.key}"
PUB="packaging/cosign.pub"
export COSIGN_PASSWORD="${COSIGN_PASSWORD:-}"
export COSIGN_YES=true # no interactive prompts
cosign_bin(){
if command -v cosign >/dev/null 2>&1; then echo cosign; return; fi
if [[ -x "$HOME/.local/bin/cosign" ]]; then echo "$HOME/.local/bin/cosign"; return; fi
echo "installing cosign into ~/.local/bin" >&2
mkdir -p "$HOME/.local/bin"
curl -fsSL -o "$HOME/.local/bin/cosign" \
"https://github.com/sigstore/cosign/releases/download/v3.1.3/cosign-linux-amd64"
chmod +x "$HOME/.local/bin/cosign"
echo "$HOME/.local/bin/cosign"
}
COSIGN="$(cosign_bin)"
capability_images(){
ls packaging/runpod/profiles/*.txt | xargs -n1 basename | sed 's/\.txt$//' \
| grep -v '^core' | grep -v '\.venv-' | sed "s#^#${NS}/coderai-#; s#\$#:latest#"
}
mode="${1:-sign}"
case "$mode" in
verify)
shift || true
refs=("$@")
[[ ${#refs[@]} -eq 0 ]] && refs=("${NS}/coderai:latest" "${NS}/coderai:${VERSION}")
rc=0
for ref in "${refs[@]}"; do
if "$COSIGN" verify --key "$PUB" "$ref" >/dev/null 2>&1; then
echo "OK $ref"
else
echo "UNSIGNED $ref"; rc=1
fi
done
exit $rc
;;
sign|all) ;;
*) echo "usage: $0 [sign|all|verify [ref…]]" >&2; exit 2 ;;
esac
# --- key ---------------------------------------------------------------
if [[ ! -f "$COSIGN_KEY" ]]; then
mkdir -p "$(dirname "$COSIGN_KEY")"; chmod 700 "$(dirname "$COSIGN_KEY")"
echo "generating a signing key pair at $COSIGN_KEY"
( cd "$(dirname "$COSIGN_KEY")" && "$COSIGN" generate-key-pair --output-key-prefix "$(basename "${COSIGN_KEY%.key}")" )
chmod 600 "$COSIGN_KEY"
fi
pubfile="${COSIGN_KEY%.key}.pub"
if [[ -f "$pubfile" ]] && ! cmp -s "$pubfile" "$PUB"; then
cp "$pubfile" "$PUB"
echo "public key written to $PUB — commit it"
fi
# --- images ------------------------------------------------------------
if [[ -n "${IMAGES:-}" ]]; then
mapfile -t images < <(tr ' ' '\n' <<<"$IMAGES" | sed '/^$/d')
else
images=("${NS}/coderai:${VERSION}" "${NS}/coderai:latest")
if [[ "$mode" == "all" ]]; then
mapfile -t caps < <(capability_images)
images+=("${caps[@]}")
fi
fi
# Sign by digest: a tag can move, a digest is the image.
for ref in "${images[@]}"; do
digest="$(docker buildx imagetools inspect "$ref" --format '{{json .Manifest.Digest}}' 2>/dev/null | tr -d '"' || true)"
if [[ -z "$digest" ]]; then
digest="$(docker manifest inspect -v "$ref" 2>/dev/null | python3 -c 'import sys,json
d=json.load(sys.stdin); d=d[0] if isinstance(d,list) else d; print(d.get("Descriptor",{}).get("digest",""))' || true)"
fi
if [[ -z "$digest" ]]; then
echo "SKIP $ref (not in the registry)"; continue
fi
name="${ref%%:*}"; name="${name%@*}"
if "$COSIGN" verify --key "$PUB" "${name}@${digest}" >/dev/null 2>&1; then
echo "OK $ref already signed (${digest:7:12})"; continue
fi
echo "SIGN $ref ${digest:7:12}"
"$COSIGN" sign --key "$COSIGN_KEY" \
-a "version=${VERSION}" -a "git=$(git rev-parse --short HEAD 2>/dev/null || echo unknown)" \
"${name}@${digest}"
done
echo "done. verify with: cosign verify --key $PUB <image>"
requirements.txt
View file @ 0d4a338b
......@@ -6,6 +6,7 @@ pydantic>=2.5.0
# CLI dependencies
requests>=2.31.0 # for the coder CLI tool
python-multipart>=0.0.6 # for multipart form data parsing
zeroconf>=0.130 # cluster discovery: mDNS/DNS-SD announce + browse (codai/cluster/discovery.py)
# PyTorch - Uncomment the appropriate version for your system.
# IMPORTANT: Use quotes around version specifiers to prevent shell interpretation!
......
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