From b325f6f697a41f4c613d23fad560d3e0f786ce8f Mon Sep 17 00:00:00 2001 From: George Lambert Date: Fri, 11 Sep 2026 13:25:05 -0400 Subject: [PATCH] Spread fleet replicas across extra machines and show message RTT percentiles machines.json plus Add-machine UI place copies on the least-loaded host. Monitor samples processing RTT and displays min, avg, p50, and p90 for capacity planning. Remote hosts run src/agent.js. --- docs/02-architecture/fleet.md | 13 ++ .../modules/verae-fleet/SUMMARY.md | 2 +- packages/verae-fleet/README.md | 4 + packages/verae-fleet/SERVICES.md | 2 + packages/verae-fleet/machines.json | 30 ++++ packages/verae-fleet/public/index.html | 78 ++++++-- packages/verae-fleet/src/agent.js | 86 +++++++++ packages/verae-fleet/src/cli.js | 2 + packages/verae-fleet/src/health.js | 36 +++- packages/verae-fleet/src/load.js | 9 +- packages/verae-fleet/src/machines.js | 92 ++++++++++ packages/verae-fleet/src/rtt.js | 57 ++++++ packages/verae-fleet/src/server.js | 16 ++ packages/verae-fleet/src/supervisor.js | 167 ++++++++++++++---- packages/verae-fleet/src/worker.js | 44 ++++- packages/verae-fleet/test/fleet.test.js | 46 ++++- packages/verae-fleet/test/rtt.test.js | 42 +++++ 17 files changed, 673 insertions(+), 53 deletions(-) create mode 100644 packages/verae-fleet/machines.json create mode 100644 packages/verae-fleet/src/agent.js create mode 100644 packages/verae-fleet/src/machines.js create mode 100644 packages/verae-fleet/src/rtt.js create mode 100644 packages/verae-fleet/test/rtt.test.js diff --git a/docs/02-architecture/fleet.md b/docs/02-architecture/fleet.md index 1207b3b..9b9c055 100644 --- a/docs/02-architecture/fleet.md +++ b/docs/02-architecture/fleet.md @@ -18,4 +18,17 @@ operator --HTTP 127.0.0.1:3850--> fleet control - **pause / off** — pause does not count toward `min`; keepFloor starts another tree-node. `stop` on a service disables it. - **tree-node floor** — `fleet.json` `tree-node.min` (default 3). Do not drop this without changing the spec; bulk-summary leaf queries need several bloom-filtered nodes. +## Extra machines + +Define hosts in `packages/verae-fleet/machines.json` (or **Add machine** on the monitor). Each machine has `capacity` and `roles` it may run. The supervisor places new replicas on the **least-loaded** eligible host. + +- `kind: local` — spawn on the control plane. +- `kind: agent` — HTTP to `http://host:agentPort` (`node src/agent.js` on that box). Do not publish NATS. + +Raising `tree-node.max` and adding machines increases bulk-summary lookup capacity. + +## Message RTT (planning) + +The monitor samples `POST /message` on each healthy replica and shows **min / avg / p50 / p90** milliseconds of message processing (service row, instance, and machine). Use p90 when sizing extra tree nodes. + Zapier cloud is not spawned. NATS is monitored, not bound publicly. diff --git a/packages/docs-master/modules/verae-fleet/SUMMARY.md b/packages/docs-master/modules/verae-fleet/SUMMARY.md index 0fb54b9..7c9900c 100644 --- a/packages/docs-master/modules/verae-fleet/SUMMARY.md +++ b/packages/docs-master/modules/verae-fleet/SUMMARY.md @@ -2,4 +2,4 @@ **Job:** Service catalog, per-service configs, monitor, restart, pause/on/off, keep tree-node replica floor. -**Config:** `fleet.json` (min/max) + `services/.json`. +**Config:** `fleet.json` (min/max) + `machines.json` (hosts) + `services/.json`. Monitor shows message RTT min/avg/p50/p90. diff --git a/packages/verae-fleet/README.md b/packages/verae-fleet/README.md index b60502e..b056789 100644 --- a/packages/verae-fleet/README.md +++ b/packages/verae-fleet/README.md @@ -27,6 +27,10 @@ node src/cli.js stop webhook-deliver # disable that service node src/cli.js start webhook-deliver ``` +Add capacity by editing `machines.json` or the monitor **Add machine** form. Remote boxes run `node src/agent.js` (`FLEET_AGENT_PORT=3851`). New replicas land on the least-loaded host that allows the role. + +The UI shows message-processing **min / avg / p50 / p90** RTT per service, instance, and machine. + Zapier cloud apps are listed but **not spawned**. NATS on NS1 is **monitored only** (loopback `:4222`, never a public bind). Clone: `ssh://git@git.georgelambert.org:2223/marchon/verae-fleet.git` diff --git a/packages/verae-fleet/SERVICES.md b/packages/verae-fleet/SERVICES.md index 3ee2c11..b2f9215 100644 --- a/packages/verae-fleet/SERVICES.md +++ b/packages/verae-fleet/SERVICES.md @@ -20,3 +20,5 @@ Per-service files: [`services/`](services/). | verae-chain-client | `services/unmanaged.json` | — | — | — | no (library) | `keepFloor`: monitor starts replacements when **available** (running, healthy, not paused) drops below `min`. That is how tree nodes stay at three live copies if one is paused, crashed, or not responding. + +Hosts: [`machines.json`](machines.json). Add machines in the monitor UI or POST `/api/machines`. Placement is least-loaded eligible host. Remote: `node src/agent.js`. diff --git a/packages/verae-fleet/machines.json b/packages/verae-fleet/machines.json new file mode 100644 index 0000000..b132df1 --- /dev/null +++ b/packages/verae-fleet/machines.json @@ -0,0 +1,30 @@ +{ + "machines": [ + { + "id": "local", + "title": "Control plane (this host)", + "kind": "local", + "host": "127.0.0.1", + "enabled": true, + "capacity": 32, + "roles": ["*"] + }, + { + "id": "ns1", + "title": "NS1.GEORGELAMBERT.ORG", + "kind": "agent", + "host": "70.88.205.138", + "agentPort": 3851, + "enabled": false, + "capacity": 24, + "roles": [ + "tree-node", + "archive-worm", + "archive-aggregator", + "job-poller", + "webhook-deliver" + ], + "notes": "Run `node src/agent.js` on NS1, then enable this machine to spread tree-node / WORM capacity. Agent port is loopback-plus-SSH-tunnel or private LAN — do not publish NATS." + } + ] +} diff --git a/packages/verae-fleet/public/index.html b/packages/verae-fleet/public/index.html index 263cd1d..eea4777 100644 --- a/packages/verae-fleet/public/index.html +++ b/packages/verae-fleet/public/index.html @@ -11,9 +11,10 @@ header { background:var(--ink); color:#f6f3ee; padding:.9rem 1.1rem; } header h1 { margin:0; font-size:1.1rem; } header p { margin:.35rem 0 0; color:#c5d0d8; font-size:13px; } - main { padding:1rem; max-width:1100px; margin:0 auto; } + main { padding:1rem; max-width:1280px; margin:0 auto; } + h2 { font-size:13px; letter-spacing:.08em; text-transform:uppercase; color:var(--muted); margin:1.2rem 0 .4rem; } table { width:100%; border-collapse:collapse; background:#fff; } - th, td { text-align:left; padding:.4rem .5rem; border-bottom:1px solid var(--line); font-size:13px; } + th, td { text-align:left; padding:.4rem .5rem; border-bottom:1px solid var(--line); font-size:13px; vertical-align:top; } th { font-size:11px; letter-spacing:.06em; text-transform:uppercase; color:var(--muted); } tr.state-operational { background:#e3f6e8; } tr.state-degraded { background:#fff3bf; } @@ -25,28 +26,65 @@ button { margin-right:.25rem; font:650 12px system-ui; border:1px solid var(--line); background:#fff; border-radius:5px; padding:.25rem .45rem; cursor:pointer; } code { font:12px ui-monospace,Menlo,monospace; } .inst { font:12px ui-monospace,Menlo,monospace; margin:.15rem 0; } + .rtt { font:12px ui-monospace,Menlo,monospace; white-space:nowrap; } + form.add { display:flex; flex-wrap:wrap; gap:.4rem; margin:.5rem 0 1rem; align-items:end; } + form.add label { font-size:11px; color:var(--muted); display:flex; flex-direction:column; gap:.15rem; } + form.add input, form.add select { padding:.35rem .45rem; border:1px solid var(--line); border-radius:5px; font:13px ui-monospace,Menlo,monospace; }

Verae fleet monitor

-

Active replicas, replica floors, pause / restart. Tree-node keepFloor respawns until min copies are healthy and unpaused. Loopback only.

+

Spread replicas across machines. Track message-processing RTT (min / avg / p50 / p90) for capacity planning. Paused copies do not count toward the tree-node floor.

+

Machines

+
+ + + + + + +
- + + +
ServiceStateConfigmin/maxavailableinstancesactions
MachineHostKindCapacityRunningRTT ms min / avg / p50 / p90
+

Services

+ +
ServiceStatemin/maxavailableRTT ms min / avg / p50 / p90instancesactions
diff --git a/packages/verae-fleet/src/agent.js b/packages/verae-fleet/src/agent.js new file mode 100644 index 0000000..05a3d9d --- /dev/null +++ b/packages/verae-fleet/src/agent.js @@ -0,0 +1,86 @@ +#!/usr/bin/env node +/** + * Remote host agent: spawn/kill fleet workers so the control plane can spread replicas. + * + * FLEET_AGENT_PORT=3851 node src/agent.js + * + * Bind loopback or a private interface. Do not publish NATS. + */ +import http from 'node:http'; +import { spawn } from 'node:child_process'; +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; + +const WORKER = path.join(path.dirname(fileURLToPath(import.meta.url)), 'worker.js'); +const PORT = Number(process.env.FLEET_AGENT_PORT || 3851); +const BIND = process.env.FLEET_AGENT_BIND || '127.0.0.1'; +const children = new Map(); + +function json(res, code, obj) { + res.writeHead(code, { 'content-type': 'application/json' }); + res.end(JSON.stringify(obj)); +} + +async function readBody(req) { + const chunks = []; + for await (const c of req) chunks.push(c); + const raw = Buffer.concat(chunks).toString('utf8'); + return raw ? JSON.parse(raw) : {}; +} + +export function startAgent({ port = PORT, bind = BIND } = {}) { + const server = http.createServer(async (req, res) => { + const url = new URL(req.url, `http://${bind}:${port}`); + try { + if (req.method === 'GET' && url.pathname === '/health') { + return json(res, 200, { + ok: true, + role: 'fleet-agent', + workers: [...children.keys()], + }); + } + if (req.method === 'POST' && url.pathname === '/spawn') { + const body = await readBody(req); + const instance = body.instance; + if (!instance) return json(res, 400, { error: 'instance required' }); + if (children.has(instance)) { + return json(res, 200, { pid: children.get(instance).pid, instance, existing: true }); + } + const child = spawn(process.execPath, [WORKER], { + env: { ...process.env, ...(body.env || {}) }, + stdio: ['ignore', 'pipe', 'pipe'], + }); + children.set(instance, child); + child.on('exit', () => children.delete(instance)); + return json(res, 200, { pid: child.pid, instance }); + } + if (req.method === 'POST' && url.pathname === '/kill') { + const body = await readBody(req); + const child = children.get(body.instance); + if (child) { + child.kill('SIGTERM'); + children.delete(body.instance); + } + return json(res, 200, { killed: Boolean(child), instance: body.instance }); + } + json(res, 404, { error: 'not found' }); + } catch (err) { + json(res, 400, { error: err.message }); + } + }); + server.listen(port, bind, () => { + process.stdout.write(`fleet-agent http://${bind}:${port}/\n`); + }); + return { + server, + async close() { + for (const c of children.values()) c.kill('SIGTERM'); + children.clear(); + await new Promise((r) => server.close(r)); + }, + }; +} + +if (process.argv[1] && path.basename(process.argv[1]) === 'agent.js') { + startAgent(); +} diff --git a/packages/verae-fleet/src/cli.js b/packages/verae-fleet/src/cli.js index a8c4127..0265744 100644 --- a/packages/verae-fleet/src/cli.js +++ b/packages/verae-fleet/src/cli.js @@ -35,7 +35,9 @@ function help() { reconcile force floor check Central spec: fleet.json +Machines: machines.json (add hosts to spread replicas) Per service: services/.json +Remote host: FLEET_AGENT_PORT=3851 node src/agent.js `); } diff --git a/packages/verae-fleet/src/health.js b/packages/verae-fleet/src/health.js index 38755cd..50ec05f 100644 --- a/packages/verae-fleet/src/health.js +++ b/packages/verae-fleet/src/health.js @@ -8,10 +8,21 @@ import net from 'node:net'; import fs from 'node:fs'; import path from 'node:path'; -export function httpProbe(port, pathname = '/health', timeoutMs = 800) { +/** + * @param {number|{ host?: string, port: number, path?: string, timeoutMs?: number }} portOrOpts + */ +export function httpProbe(portOrOpts, pathname = '/health', timeoutMs = 800) { + const opts = + typeof portOrOpts === 'object' + ? portOrOpts + : { port: portOrOpts, path: pathname, timeoutMs }; + const host = opts.host || '127.0.0.1'; + const port = opts.port; + const pathName = opts.path || pathname; + const tmo = opts.timeoutMs || timeoutMs; return new Promise((resolve) => { const req = http.get( - { host: '127.0.0.1', port, path: pathname, timeout: timeoutMs }, + { host, port, path: pathName, timeout: tmo }, (res) => { const chunks = []; res.on('data', (c) => chunks.push(c)); @@ -57,10 +68,26 @@ export function tcpProbe(host, port, timeoutMs = 400) { }); } -export function postControl(port, pathname) { +/** + * @param {number|{ host?: string, port: number, path?: string, body?: object }} portOrOpts + * @param {string} [pathname] + */ +export function postControl(portOrOpts, pathname = '/') { + const opts = typeof portOrOpts === 'object' ? portOrOpts : { port: portOrOpts, path: pathname }; + const host = opts.host || '127.0.0.1'; + const port = opts.port; + const pathName = opts.path || pathname; + const body = opts.body ? JSON.stringify(opts.body) : ''; return new Promise((resolve, reject) => { const req = http.request( - { host: '127.0.0.1', port, path: pathname, method: 'POST', timeout: 800 }, + { + host, + port, + path: pathName, + method: 'POST', + timeout: opts.timeoutMs || 2000, + headers: body ? { 'content-type': 'application/json', 'content-length': Buffer.byteLength(body) } : {}, + }, (res) => { const chunks = []; res.on('data', (c) => chunks.push(c)); @@ -74,6 +101,7 @@ export function postControl(port, pathname) { }, ); req.on('error', reject); + if (body) req.write(body); req.end(); }); } diff --git a/packages/verae-fleet/src/load.js b/packages/verae-fleet/src/load.js index d8ed9da..96d9662 100644 --- a/packages/verae-fleet/src/load.js +++ b/packages/verae-fleet/src/load.js @@ -6,6 +6,7 @@ import fs from 'node:fs'; import path from 'node:path'; import { fileURLToPath } from 'node:url'; +import { loadMachines } from './machines.js'; const HERE = path.dirname(fileURLToPath(import.meta.url)); export const FLEET_ROOT = path.resolve(HERE, '..'); @@ -17,7 +18,7 @@ export function readJson(file) { /** * @param {string} [root] */ -export function loadFleet(root = FLEET_ROOT) { +export function loadFleet(root = FLEET_ROOT, opts = {}) { const fleetPath = path.join(root, 'fleet.json'); const fleet = readJson(fleetPath); const dir = path.join(root, 'services'); @@ -44,11 +45,17 @@ export function loadFleet(root = FLEET_ROOT) { managed: central.managed === false ? false : spec.managed !== false, }; } + const overlay = + opts.overlay === false ? null : path.join(root, 'data', 'runtime', 'machines-overlay.json'); + const machines = loadMachines(root, overlay); return { root, fleetPath: path.relative(root, fleetPath) || 'fleet.json', + machinesPath: 'machines.json', + machinesOverlay: overlay, control: fleet.control, nats: fleet.nats, + machines, services, unmanaged, }; diff --git a/packages/verae-fleet/src/machines.js b/packages/verae-fleet/src/machines.js new file mode 100644 index 0000000..8ca5faa --- /dev/null +++ b/packages/verae-fleet/src/machines.js @@ -0,0 +1,92 @@ +/** + * Host catalog: where replicas may run. Least-loaded placement. + * @module machines + */ + +import fs from 'node:fs'; +import path from 'node:path'; + +export function defaultMachines() { + return [ + { + id: 'local', + title: 'Control plane (this host)', + kind: 'local', + host: '127.0.0.1', + enabled: true, + capacity: 32, + roles: ['*'], + }, + ]; +} + +export function loadMachines(root, overlayPath) { + const file = path.join(root, 'machines.json'); + let list = defaultMachines(); + if (fs.existsSync(file)) { + const raw = JSON.parse(fs.readFileSync(file, 'utf8')); + if (Array.isArray(raw.machines) && raw.machines.length) list = raw.machines; + } + if (overlayPath && fs.existsSync(overlayPath)) { + const extra = JSON.parse(fs.readFileSync(overlayPath, 'utf8')); + const incoming = Array.isArray(extra) ? extra : extra.machines || []; + for (const m of incoming) upsertMachine(list, m); + } + return list.map(normalizeMachine); +} + +export function normalizeMachine(m) { + return { + id: String(m.id || '').trim(), + title: m.title || m.id, + kind: m.kind === 'agent' ? 'agent' : 'local', + host: m.host || '127.0.0.1', + agentPort: Number(m.agentPort || 3851), + enabled: m.enabled !== false, + capacity: Number(m.capacity || 8), + roles: Array.isArray(m.roles) && m.roles.length ? m.roles : ['*'], + notes: m.notes || '', + }; +} + +export function upsertMachine(list, spec) { + const m = normalizeMachine(spec); + if (!m.id) throw new Error('machine id required'); + const i = list.findIndex((x) => x.id === m.id); + if (i >= 0) list[i] = { ...list[i], ...m }; + else list.push(m); + return m; +} + +export function canHost(machine, serviceId, role) { + if (!machine?.enabled) return false; + const roles = machine.roles || ['*']; + if (roles.includes('*')) return true; + return roles.includes(serviceId) || (role && roles.includes(role)); +} + +/** + * Pick the enabled machine with the most free capacity (fewest running / capacity). + * @param {object[]} machines + * @param {Array<{ machine?: string }>} instances + * @param {string} serviceId + * @param {string} [role] + */ +export function pickMachine(machines, instances, serviceId, role) { + const eligible = machines.filter((m) => canHost(m, serviceId, role)); + if (!eligible.length) return null; + const scored = eligible.map((m) => { + const running = instances.filter((i) => i.machine === m.id && i.pid).length; + const free = m.capacity - running; + const load = m.capacity <= 0 ? 1 : running / m.capacity; + return { m, running, free, load }; + }); + scored.sort((a, b) => a.load - b.load || a.running - b.running || a.m.id.localeCompare(b.m.id)); + const best = scored.find((s) => s.free > 0); + return best ? best.m : null; +} + +export function saveOverlay(overlayPath, machines) { + fs.mkdirSync(path.dirname(overlayPath), { recursive: true }); + fs.writeFileSync(overlayPath, JSON.stringify({ machines }, null, 2)); +} diff --git a/packages/verae-fleet/src/rtt.js b/packages/verae-fleet/src/rtt.js new file mode 100644 index 0000000..09ec7bb --- /dev/null +++ b/packages/verae-fleet/src/rtt.js @@ -0,0 +1,57 @@ +/** + * Ring buffer of message-processing RTTs (ms) for planning percentiles. + * @module rtt + */ + +export class RttWindow { + /** + * @param {number} [cap=400] + */ + constructor(cap = 400) { + this.cap = cap; + /** @type {number[]} */ + this.samples = []; + } + + add(ms) { + const n = Number(ms); + if (!Number.isFinite(n) || n < 0) return; + this.samples.push(n); + if (this.samples.length > this.cap) this.samples.splice(0, this.samples.length - this.cap); + } + + stats() { + return summarizeRtt(this.samples); + } +} + +/** + * @param {number[]} samples + */ +export function summarizeRtt(samples) { + const xs = samples.filter((n) => Number.isFinite(n) && n >= 0).slice().sort((a, b) => a - b); + if (!xs.length) { + return { count: 0, minMs: null, avgMs: null, p50Ms: null, p90Ms: null }; + } + const sum = xs.reduce((a, b) => a + b, 0); + return { + count: xs.length, + minMs: round1(xs[0]), + avgMs: round1(sum / xs.length), + p50Ms: round1(percentile(xs, 0.5)), + p90Ms: round1(percentile(xs, 0.9)), + }; +} + +function percentile(sorted, p) { + if (sorted.length === 1) return sorted[0]; + const idx = (sorted.length - 1) * p; + const lo = Math.floor(idx); + const hi = Math.ceil(idx); + if (lo === hi) return sorted[lo]; + return sorted[lo] + (sorted[hi] - sorted[lo]) * (idx - lo); +} + +function round1(n) { + return Math.round(n * 10) / 10; +} diff --git a/packages/verae-fleet/src/server.js b/packages/verae-fleet/src/server.js index f909f01..1ac15f7 100644 --- a/packages/verae-fleet/src/server.js +++ b/packages/verae-fleet/src/server.js @@ -37,6 +37,22 @@ export function startControlServer(sup, mon) { if (req.method === 'GET' && url.pathname === '/api/status') { return json(res, 200, { ...sup.status(), monitor: mon?.last || null }); } + if (req.method === 'GET' && url.pathname === '/api/machines') { + return json(res, 200, { machines: sup.status().machines }); + } + if (req.method === 'POST' && url.pathname === '/api/machines') { + const chunks = []; + for await (const c of req) chunks.push(c); + const spec = JSON.parse(Buffer.concat(chunks).toString('utf8') || '{}'); + const m = sup.addMachine(spec); + return json(res, 200, { machine: m, machines: sup.status().machines }); + } + const mach = url.pathname.match(/^\/api\/machines\/([^/]+)\/(enable|disable)$/); + if (mach && req.method === 'POST') { + const [, id, op] = mach; + const m = sup.setMachineEnabled(id, op === 'enable'); + return json(res, 200, { machine: m, machines: sup.status().machines }); + } if (req.method === 'POST' && url.pathname === '/api/reconcile') { const actions = await sup.reconcile(); return json(res, 200, { actions, status: sup.status() }); diff --git a/packages/verae-fleet/src/supervisor.js b/packages/verae-fleet/src/supervisor.js index 81f6d00..e78a3a6 100644 --- a/packages/verae-fleet/src/supervisor.js +++ b/packages/verae-fleet/src/supervisor.js @@ -10,6 +10,8 @@ import { fileURLToPath } from 'node:url'; import { loadFleet } from './load.js'; import { httpProbe, postControl } from './health.js'; import { classifyService } from './classify.js'; +import { pickMachine, upsertMachine, saveOverlay } from './machines.js'; +import { summarizeRtt } from './rtt.js'; const WORKER = path.join(path.dirname(fileURLToPath(import.meta.url)), 'worker.js'); @@ -27,6 +29,30 @@ export class Supervisor { this.instances = new Map(); this.events = []; this._seq = 0; + if (!Array.isArray(this.loaded.machines) || !this.loaded.machines.length) { + this.loaded.machines = [ + { id: 'local', title: 'local', kind: 'local', host: '127.0.0.1', enabled: true, capacity: 32, roles: ['*'] }, + ]; + } + } + + machines() { + return this.loaded.machines; + } + + addMachine(spec) { + const m = upsertMachine(this.loaded.machines, spec); + if (this.loaded.machinesOverlay) saveOverlay(this.loaded.machinesOverlay, this.loaded.machines); + this.log('machine-add', { id: m.id, host: m.host, kind: m.kind }); + return m; + } + + setMachineEnabled(id, enabled) { + const m = this.loaded.machines.find((x) => x.id === id); + if (!m) throw new Error(`unknown machine ${id}`); + m.enabled = Boolean(enabled); + if (this.loaded.machinesOverlay) saveOverlay(this.loaded.machinesOverlay, this.loaded.machines); + return m; } log(type, detail) { @@ -72,43 +98,76 @@ export class Supervisor { } const instance = `${serviceId}-${index}`; if (this.instances.has(instance)) return this.instances.get(instance); + const machine = + opts.machine || + pickMachine(this.machines(), [...this.instances.values()], serviceId, spec.role || serviceId); + if (!machine) { + this.log('skip', { service: serviceId, reason: 'no-machine-capacity' }); + return null; + } const healthPort = (spec.ports?.healthBase || 19000) + index; const stateDir = path.join(this.stateRoot, instance); fs.mkdirSync(stateDir, { recursive: true }); - const child = spawn(process.execPath, [WORKER], { - env: { - ...process.env, - FLEET_ROLE: spec.role || serviceId, - FLEET_SERVICE: serviceId, - FLEET_INSTANCE: instance, - FLEET_HEALTH_PORT: String(healthPort), - FLEET_STATE_DIR: stateDir, - }, - stdio: ['ignore', 'pipe', 'pipe'], - }); + const env = { + ...process.env, + FLEET_ROLE: spec.role || serviceId, + FLEET_SERVICE: serviceId, + FLEET_INSTANCE: instance, + FLEET_MACHINE: machine.id, + FLEET_HEALTH_PORT: String(healthPort), + FLEET_HEALTH_BIND: machine.kind === 'local' ? '127.0.0.1' : '0.0.0.0', + FLEET_STATE_DIR: stateDir, + FLEET_MESSAGE_HZ: process.env.FLEET_MESSAGE_HZ || '8', + }; + let child = null; + let pid = null; + if (machine.kind === 'agent') { + try { + const spawned = await postControl({ + host: machine.host, + port: machine.agentPort, + path: '/spawn', + body: { env, instance }, + timeoutMs: 4000, + }); + pid = spawned.pid || null; + } catch (err) { + this.log('skip', { service: serviceId, reason: 'agent-spawn-failed', machine: machine.id, error: err.message }); + return null; + } + } else { + child = spawn(process.execPath, [WORKER], { env, stdio: ['ignore', 'pipe', 'pipe'] }); + pid = child.pid; + } const rec = { id: instance, service: serviceId, index, - pid: child.pid, + machine: machine.id, + host: machine.host, + pid, healthPort, stateDir, child, + kind: machine.kind, paused: false, startedAt: Date.now(), restarts: 0, lastError: null, + rttSamples: [], }; - child.stderr?.on('data', () => {}); - child.on('exit', (code, signal) => { - rec.exitCode = code; - rec.signal = signal; - rec.pid = null; - this.log('exit', { instance, service: serviceId, code, signal }); - }); + if (child) { + child.stderr?.on('data', () => {}); + child.on('exit', (code, signal) => { + rec.exitCode = code; + rec.signal = signal; + rec.pid = null; + this.log('exit', { instance, service: serviceId, code, signal }); + }); + } this.instances.set(instance, rec); - this.log('start', { instance, service: serviceId, pid: rec.pid, healthPort }); - const h = await waitForHealth(healthPort, 4000); + this.log('start', { instance, service: serviceId, machine: machine.id, pid: rec.pid, healthPort }); + const h = await waitForHealth(rec.host, healthPort, 4000); rec.healthy = h.ok; rec.lastProbe = h; return rec; @@ -118,7 +177,16 @@ export class Supervisor { const rec = this.instances.get(instanceId); if (!rec) return null; rec.stopping = true; - if (rec.child && rec.pid) { + if (rec.kind === 'agent') { + const m = this.machines().find((x) => x.id === rec.machine); + if (m) { + try { + await postControl({ host: m.host, port: m.agentPort, path: '/kill', body: { instance: instanceId } }); + } catch { + /* ignore */ + } + } + } else if (rec.child && rec.pid) { rec.child.kill('SIGTERM'); await waitExit(rec.child, 2000); } @@ -131,7 +199,7 @@ export class Supervisor { async pauseInstance(instanceId, { replace = true } = {}) { const rec = this.instances.get(instanceId); if (!rec) throw new Error(`no instance ${instanceId}`); - await postControl(rec.healthPort, '/pause'); + await postControl({ host: rec.host || '127.0.0.1', port: rec.healthPort, path: '/pause' }); rec.paused = true; this.log('pause', { instance: instanceId, service: rec.service }); if (replace) await this.reconcile(rec.service); @@ -141,7 +209,7 @@ export class Supervisor { async resumeInstance(instanceId) { const rec = this.instances.get(instanceId); if (!rec) throw new Error(`no instance ${instanceId}`); - await postControl(rec.healthPort, '/resume'); + await postControl({ host: rec.host || '127.0.0.1', port: rec.healthPort, path: '/resume' }); rec.paused = false; this.log('resume', { instance: instanceId, service: rec.service }); return rec; @@ -150,11 +218,9 @@ export class Supervisor { async restartInstance(instanceId) { const rec = this.instances.get(instanceId); if (!rec) throw new Error(`no instance ${instanceId}`); - const { service, index } = rec; - rec.child?.kill('SIGTERM'); - await waitExit(rec.child, 2000); - this.instances.delete(instanceId); - const next = await this.startOne(service, { index }); + const { service, index, machine } = rec; + await this.stopInstance(instanceId, { replace: false }); + const next = await this.startOne(service, { index, machine: this.machines().find((m) => m.id === machine) }); if (next) next.restarts = (rec.restarts || 0) + 1; this.log('restart', { instance: instanceId, service }); return next; @@ -163,16 +229,35 @@ export class Supervisor { async markUnhealthy(instanceId) { const rec = this.instances.get(instanceId); if (!rec) throw new Error(`no instance ${instanceId}`); - await postControl(rec.healthPort, '/unhealthy'); + await postControl({ host: rec.host || '127.0.0.1', port: rec.healthPort, path: '/unhealthy' }); return rec; } async probe(rec) { const spec = this.spec(rec.service); - const r = await httpProbe(rec.healthPort, spec.health?.path || '/health', spec.health?.timeoutMs || 800); + const host = rec.host || '127.0.0.1'; + const r = await httpProbe({ + host, + port: rec.healthPort, + path: spec.health?.path || '/health', + timeoutMs: spec.health?.timeoutMs || 800, + }); rec.paused = Boolean(r.paused); rec.lastProbe = r; rec.healthy = r.ok; + if (r.ok && !rec.paused) { + try { + const msg = await postControl({ host, port: rec.healthPort, path: '/message' }); + if (Number.isFinite(msg.rttMs)) { + rec.rttSamples = rec.rttSamples || []; + rec.rttSamples.push(msg.rttMs); + if (rec.rttSamples.length > 200) rec.rttSamples.splice(0, rec.rttSamples.length - 200); + } + rec.rtt = msg.rtt || r.body?.rtt || summarizeRtt(rec.rttSamples); + } catch { + rec.rtt = r.body?.rtt || summarizeRtt(rec.rttSamples || []); + } + } return r; } @@ -279,17 +364,33 @@ export class Supervisor { instances: inst.map((i) => ({ id: i.id, pid: i.pid, + machine: i.machine || 'local', + host: i.host || '127.0.0.1', healthPort: i.healthPort, paused: i.paused, healthy: i.healthy, restarts: i.restarts, + rtt: i.rtt || summarizeRtt(i.rttSamples || []), })), }; + const allSamples = inst.flatMap((i) => i.rttSamples || []); + services[spec.id].rtt = summarizeRtt(allSamples); services[spec.id].health = classifyService(services[spec.id]); } + const machines = this.machines().map((m) => { + const inst = [...this.instances.values()].filter((i) => i.machine === m.id); + const samples = inst.flatMap((i) => i.rttSamples || []); + return { + ...m, + running: inst.filter((i) => i.pid).length, + available: inst.filter((i) => i.pid && !i.paused && i.healthy !== false).length, + rtt: summarizeRtt(samples), + }; + }); return { control: this.loaded.control, nats: this.loaded.nats, + machines, services, events: this.events.slice(-50), }; @@ -314,10 +415,10 @@ function waitExit(child, ms) { }); } -async function waitForHealth(port, ms) { +async function waitForHealth(host, port, ms) { const deadline = Date.now() + ms; while (Date.now() < deadline) { - const r = await httpProbe(port, '/health', 300); + const r = await httpProbe({ host: host || '127.0.0.1', port, path: '/health', timeoutMs: 300 }); if (r.ok || r.statusCode === 503) return r; await new Promise((r) => setTimeout(r, 50)); } diff --git a/packages/verae-fleet/src/worker.js b/packages/verae-fleet/src/worker.js index 5b3fe56..bc010a2 100644 --- a/packages/verae-fleet/src/worker.js +++ b/packages/verae-fleet/src/worker.js @@ -6,14 +6,20 @@ import http from 'node:http'; import fs from 'node:fs'; import path from 'node:path'; +import { createHash, randomBytes } from 'node:crypto'; import { fileURLToPath } from 'node:url'; +import { RttWindow } from './rtt.js'; const role = process.env.FLEET_ROLE || 'unknown'; const service = process.env.FLEET_SERVICE || role; const instance = process.env.FLEET_INSTANCE || `${service}-0`; +const machine = process.env.FLEET_MACHINE || 'local'; const port = Number(process.env.FLEET_HEALTH_PORT || 0); +const bind = process.env.FLEET_HEALTH_BIND || '127.0.0.1'; const stateDir = process.env.FLEET_STATE_DIR || path.join(process.cwd(), 'data', instance); const startedAt = new Date().toISOString(); +const rtt = new RttWindow(400); +const hz = Number(process.env.FLEET_MESSAGE_HZ || 6); fs.mkdirSync(stateDir, { recursive: true }); @@ -43,6 +49,24 @@ if (role === 'tree-node' || role === 'archive-worm') { } } +function processMessage() { + const t0 = process.hrtime.bigint(); + if (paused) return null; + const payload = randomBytes(32); + if (archive) { + const sha = createHash('sha256').update(payload).digest('hex'); + archive.put({ sha256: sha, kind: role === 'tree-node' ? 'tree' : 'publicMeta', record: { n: puts } }); + archive.query(sha); + puts += 1; + } else { + createHash('sha256').update(payload).digest('hex'); + puts += 1; + } + const ms = Number(process.hrtime.bigint() - t0) / 1e6; + rtt.add(ms); + return ms; +} + function snapshot() { return { ok: !paused && !forceUnhealthy, @@ -51,11 +75,13 @@ function snapshot() { role, service, instance, + machine, port, pid: process.pid, startedAt, puts, archiveId: archive?.archiveId || null, + rtt: rtt.stats(), }; } @@ -91,17 +117,31 @@ const server = http.createServer((req, res) => { beat(); return json(200, snapshot()); } + if (url.pathname === '/message' && req.method === 'POST') { + const ms = processMessage(); + return json(paused ? 503 : 200, { rttMs: ms, ...snapshot() }); + } + if (url.pathname === '/metrics' && req.method === 'GET') { + return json(200, { rtt: rtt.stats(), samples: rtt.samples.slice(-80) }); + } json(404, { error: 'not found' }); }); -server.listen(port, '127.0.0.1', () => { +server.listen(port, bind, () => { beat(); - process.stdout.write(`fleet-worker ${instance} health http://127.0.0.1:${port}/health\n`); + process.stdout.write(`fleet-worker ${instance}@${machine} health http://${bind}:${port}/health\n`); }); const iv = setInterval(beat, 400); +const msgIv = + hz > 0 + ? setInterval(() => { + if (!paused && !forceUnhealthy) processMessage(); + }, Math.max(50, Math.round(1000 / hz))) + : null; function shutdown() { clearInterval(iv); + if (msgIv) clearInterval(msgIv); try { server.close(); } catch { diff --git a/packages/verae-fleet/test/fleet.test.js b/packages/verae-fleet/test/fleet.test.js index 90e7a29..c680cc0 100644 --- a/packages/verae-fleet/test/fleet.test.js +++ b/packages/verae-fleet/test/fleet.test.js @@ -1,11 +1,11 @@ import { describe, it, after } from 'node:test'; import assert from 'node:assert/strict'; -import { loadFleet, listServices } from '../src/load.js'; +import { loadFleet, listServices, FLEET_ROOT } from '../src/load.js'; import { Supervisor } from '../src/supervisor.js'; import { Monitor } from '../src/monitor.js'; function treeOnly(healthBase) { - const loaded = loadFleet(); + const loaded = loadFleet(FLEET_ROOT, { overlay: false }); for (const s of Object.values(loaded.services)) { s.enabled = s.id === 'tree-node'; if (s.id === 'tree-node') { @@ -15,6 +15,13 @@ function treeOnly(healthBase) { s.ports = { healthBase }; } } + loaded.machinesOverlay = null; + loaded.machines = loaded.machines.filter((m) => m.kind === 'local' && m.enabled); + if (!loaded.machines.length) { + loaded.machines = [ + { id: 'local', title: 'local', kind: 'local', host: '127.0.0.1', enabled: true, capacity: 32, roles: ['*'] }, + ]; + } return loaded; } @@ -96,4 +103,39 @@ describe('verae-fleet', () => { assert.equal(st.available, 3); assert.ok(st.instances.some((i) => i.id === 'tree-node-2' || i.id === 'tree-node-0')); }); + + it('spreads replicas across two defined machines', async () => { + const loaded = treeOnly(15100); + loaded.services['tree-node'].min = 4; + loaded.services['tree-node'].max = 6; + loaded.machines = [ + { id: 'rack-a', title: 'A', kind: 'local', host: '127.0.0.1', enabled: true, capacity: 8, roles: ['*'] }, + { id: 'rack-b', title: 'B', kind: 'local', host: '127.0.0.1', enabled: true, capacity: 8, roles: ['*'] }, + ]; + const sup = new Supervisor({ loaded }); + supervisors.push(sup); + await sup.startService('tree-node'); + const hosts = new Set(sup.status().services['tree-node'].instances.map((i) => i.machine)); + assert.ok(hosts.has('rack-a') && hosts.has('rack-b'), `hosts ${[...hosts]}`); + const machines = sup.status().machines; + assert.equal(machines.length, 2); + assert.ok(machines.every((m) => m.running >= 1)); + }); + + it('records message-processing RTT percentiles', async () => { + const loaded = treeOnly(15200); + loaded.services['tree-node'].min = 1; + loaded.services['tree-node'].max = 2; + const sup = new Supervisor({ loaded }); + supervisors.push(sup); + await sup.startService('tree-node'); + const rec = [...sup.instances.values()][0]; + for (let i = 0; i < 8; i += 1) await sup.probe(rec); + const rtt = rec.rtt; + assert.ok(rtt.count >= 8, `count ${rtt.count}`); + assert.ok(rtt.minMs <= rtt.p50Ms && rtt.p50Ms <= rtt.p90Ms); + const svc = sup.status().services['tree-node'].rtt; + assert.ok(svc.count >= 8); + assert.ok(svc.minMs != null && svc.avgMs != null); + }); }); diff --git a/packages/verae-fleet/test/rtt.test.js b/packages/verae-fleet/test/rtt.test.js new file mode 100644 index 0000000..def37cf --- /dev/null +++ b/packages/verae-fleet/test/rtt.test.js @@ -0,0 +1,42 @@ +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { summarizeRtt, RttWindow } from '../src/rtt.js'; +import { pickMachine } from '../src/machines.js'; + +describe('RTT planning stats', () => { + it('computes min, avg, p50, p90 in order', () => { + const s = summarizeRtt([10, 20, 30, 40, 50, 60, 70, 80, 90, 100]); + assert.equal(s.count, 10); + assert.equal(s.minMs, 10); + assert.equal(s.avgMs, 55); + assert.ok(s.p50Ms >= s.minMs && s.p50Ms <= s.p90Ms); + assert.ok(s.p90Ms <= 100); + }); + + it('empty window is nulls', () => { + const s = new RttWindow().stats(); + assert.equal(s.count, 0); + assert.equal(s.minMs, null); + }); +}); + +describe('machine placement', () => { + it('spreads onto the least-loaded eligible host', () => { + const machines = [ + { id: 'a', enabled: true, capacity: 4, roles: ['*'] }, + { id: 'b', enabled: true, capacity: 4, roles: ['*'] }, + ]; + const instances = [{ machine: 'a', pid: 1 }]; + const pick = pickMachine(machines, instances, 'tree-node', 'tree-node'); + assert.equal(pick.id, 'b'); + }); + + it('skips machines that do not allow the role', () => { + const machines = [ + { id: 'edge', enabled: true, capacity: 8, roles: ['zappier-edge'] }, + { id: 'trees', enabled: true, capacity: 8, roles: ['tree-node'] }, + ]; + const pick = pickMachine(machines, [], 'tree-node', 'tree-node'); + assert.equal(pick.id, 'trees'); + }); +});