From 0a7edcf9993c029ed9b0e366bb4d5ef7de55cd9c Mon Sep 17 00:00:00 2001 From: George Lambert Date: Fri, 11 Sep 2026 13:37:59 -0400 Subject: [PATCH] Initial import of verae-fleet from zapier monorepo --- .gitignore | 3 + NATS.md | 15 + README.md | 42 +++ SERVICES.md | 24 ++ SUMMARY.md | 9 + fleet.json | 26 ++ machines.json | 47 +++ machines.secrets.json.example | 13 + package.json | 13 + public/index.html | 148 +++++++++ services/archive-aggregator.json | 16 + services/archive-worm.json | 16 + services/job-poller.json | 16 + services/middleware-http.json | 17 ++ services/nats.json | 11 + services/tree-node.json | 16 + services/unmanaged.json | 37 +++ services/webhook-deliver.json | 16 + services/zapier-simulator.json | 14 + services/zappier-edge.json | 14 + src/agent.js | 86 ++++++ src/classify.js | 29 ++ src/cli.js | 145 +++++++++ src/health.js | 118 +++++++ src/load.js | 94 ++++++ src/machines.js | 107 +++++++ src/monitor.js | 38 +++ src/rtt.js | 57 ++++ src/server.js | 105 +++++++ src/ssh.js | 170 +++++++++++ src/supervisor.js | 510 +++++++++++++++++++++++++++++++ src/worker.js | 153 ++++++++++ test/classify.test.js | 36 +++ test/fleet.test.js | 141 +++++++++ test/rtt.test.js | 42 +++ test/ssh.test.js | 56 ++++ 36 files changed, 2400 insertions(+) create mode 100644 .gitignore create mode 100644 NATS.md create mode 100644 README.md create mode 100644 SERVICES.md create mode 100644 SUMMARY.md create mode 100644 fleet.json create mode 100644 machines.json create mode 100644 machines.secrets.json.example create mode 100644 package.json create mode 100644 public/index.html create mode 100644 services/archive-aggregator.json create mode 100644 services/archive-worm.json create mode 100644 services/job-poller.json create mode 100644 services/middleware-http.json create mode 100644 services/nats.json create mode 100644 services/tree-node.json create mode 100644 services/unmanaged.json create mode 100644 services/webhook-deliver.json create mode 100644 services/zapier-simulator.json create mode 100644 services/zappier-edge.json create mode 100644 src/agent.js create mode 100644 src/classify.js create mode 100644 src/cli.js create mode 100644 src/health.js create mode 100644 src/load.js create mode 100644 src/machines.js create mode 100644 src/monitor.js create mode 100644 src/rtt.js create mode 100644 src/server.js create mode 100644 src/ssh.js create mode 100644 src/supervisor.js create mode 100644 src/worker.js create mode 100644 test/classify.test.js create mode 100644 test/fleet.test.js create mode 100644 test/rtt.test.js create mode 100644 test/ssh.test.js diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..e094d00 --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +data/ +machines.secrets.json + diff --git a/NATS.md b/NATS.md new file mode 100644 index 0000000..73525c7 --- /dev/null +++ b/NATS.md @@ -0,0 +1,15 @@ +# NATS — verae-fleet + +Fleet control **does not subscribe** to JetStream. It only records which workers should. + +| Service | NATS in | NATS out | +|---------|---------|----------| +| nats (external) | — | — (the broker) | +| middleware-http | jobs.events | jobs.watch | +| job-poller | jobs.watch | jobs.events | +| webhook-deliver | webhooks.deliver, jobs.events | HTTPS to Zapier | +| archive-aggregator | archive.reply.* | archive.query (broadcast) | +| archive-worm | archive.put, archive.query | archive.reply.* on bloom hit | +| tree-node | archive.put, archive.query | archive.reply.* on bloom hit | + +NS1 nats-server stays on `127.0.0.1:4222`. `fleet.json` `nats.publicBind` is false. diff --git a/README.md b/README.md new file mode 100644 index 0000000..8a0d2f9 --- /dev/null +++ b/README.md @@ -0,0 +1,42 @@ +# verae-fleet + +Operator control plane for Verae Time × Zapier **runtime** services: + +- a **list** of every service and its config file +- a **central replica spec** (`fleet.json`) — how many copies must be up +- a **monitor** of active / paused / unhealthy replicas +- **restart** when a copy is offline or fails `/health` +- **on / off / pause / resume** per service or per instance +- **keepFloor** on tree nodes so at least `min` copies stay available (paused copies do not count) + +```bash +cd packages/verae-fleet +npm test +node src/cli.js list +node src/cli.js serve # http://127.0.0.1:3850/ +``` + +Against a running daemon: + +```bash +node src/cli.js status +node src/cli.js pause tree-node-0 # floor starts another tree-node +node src/cli.js resume tree-node-0 +node src/cli.js restart tree-node-1 +node src/cli.js stop webhook-deliver # disable that service +node src/cli.js start webhook-deliver +``` + +Add capacity in `machines.json` or the monitor **Add machine** form (`kind=ssh`, user, host, identity file path). New replicas land on the least-loaded eligible host. + +```bash +node src/cli.js ssh-check ns1 # marchon@70.88.205.138 with ~/.ssh/id_ed25519 +``` + +Private keys stay on disk (`~/.ssh/id_ed25519`); git stores only the path. Optional overrides: `machines.secrets.json` (gitignored). + +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/SERVICES.md b/SERVICES.md new file mode 100644 index 0000000..b2f9215 --- /dev/null +++ b/SERVICES.md @@ -0,0 +1,24 @@ +# Service list + +Central replica spec: [`fleet.json`](fleet.json). +Per-service files: [`services/`](services/). + +| Service | Config | min | max | keepFloor | Spawned by fleet | +|---------|--------|-----|-----|-----------|------------------| +| nats | `services/nats.json` | 1 | 1 | no | no (external monitor) | +| zappier-edge | `services/zappier-edge.json` | 1 | 1 | yes | yes | +| middleware-http | `services/middleware-http.json` | 1 | 1 | yes | yes | +| job-poller | `services/job-poller.json` | 1 | 2 | yes | yes | +| webhook-deliver | `services/webhook-deliver.json` | 1 | 2 | yes | yes | +| archive-aggregator | `services/archive-aggregator.json` | 1 | 2 | yes | yes | +| archive-worm | `services/archive-worm.json` | 3 | 6 | yes | yes | +| **tree-node** | `services/tree-node.json` | **3** | 9 | **yes** | yes | +| zapier-simulator | `services/zapier-simulator.json` | 0 | 1 | no | optional | +| zapier-platform-app | `services/unmanaged.json` | — | — | — | no (Zapier cloud) | +| verae-activate | `services/unmanaged.json` | — | — | — | no | +| request-splitter | `services/unmanaged.json` | — | — | — | no (library) | +| 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/SUMMARY.md b/SUMMARY.md new file mode 100644 index 0000000..162fd11 --- /dev/null +++ b/SUMMARY.md @@ -0,0 +1,9 @@ +# verae-fleet + +**Job:** Catalog services, hold per-service configs, monitor health, restart bad replicas, pause/resume/on/off, and keep `fleet.json` replica floors — especially tree-node `min`. + +**Expects:** operator CLI/HTTP on `127.0.0.1:3850`. Spawns `src/worker.js` children with `/health`. + +**Sends:** nothing to Zapier or public NATS. + +**Test:** `npm test` — list configs; start 3 tree-nodes; pause one still ≥3 available; unhealthy restart; stop+replace. diff --git a/fleet.json b/fleet.json new file mode 100644 index 0000000..afc9c94 --- /dev/null +++ b/fleet.json @@ -0,0 +1,26 @@ +{ + "apiVersion": "verae.fleet/v1", + "control": { + "bind": "127.0.0.1", + "port": 3850, + "probeIntervalMs": 750, + "unhealthyAfterMs": 2500, + "restartBackoffMs": [200, 800, 2000], + "maxRestartsPerHour": 40 + }, + "nats": { + "url": "nats://127.0.0.1:4222", + "publicBind": false + }, + "services": { + "nats": { "min": 1, "max": 1, "keepFloor": false, "enabled": true, "managed": false }, + "zappier-edge": { "min": 1, "max": 1, "keepFloor": true, "enabled": true }, + "middleware-http": { "min": 1, "max": 1, "keepFloor": true, "enabled": true }, + "job-poller": { "min": 1, "max": 2, "keepFloor": true, "enabled": true }, + "webhook-deliver": { "min": 1, "max": 2, "keepFloor": true, "enabled": true }, + "archive-aggregator": { "min": 1, "max": 2, "keepFloor": true, "enabled": true }, + "archive-worm": { "min": 3, "max": 6, "keepFloor": true, "enabled": true }, + "tree-node": { "min": 3, "max": 9, "keepFloor": true, "enabled": true }, + "zapier-simulator": { "min": 0, "max": 1, "keepFloor": false, "enabled": true } + } +} diff --git a/machines.json b/machines.json new file mode 100644 index 0000000..dd47bcc --- /dev/null +++ b/machines.json @@ -0,0 +1,47 @@ +{ + "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": "ssh", + "host": "70.88.205.138", + "user": "marchon", + "sshPort": 22, + "identityFile": "~/.ssh/id_ed25519", + "remoteDir": "~/verae-fleet-runtime", + "enabled": true, + "capacity": 24, + "roles": [ + "tree-node", + "archive-worm", + "archive-aggregator", + "job-poller", + "webhook-deliver" + ], + "notes": "SSH as marchon with ~/.ssh/id_ed25519. Workers bind 127.0.0.1 on NS1; control plane probes via ssh+curl. Private key is never stored in git." + }, + { + "id": "lan-134", + "title": "70.88.205.134", + "kind": "ssh", + "host": "70.88.205.134", + "user": "marchon", + "sshPort": 22, + "identityFile": "~/.ssh/id_ed25519", + "remoteDir": "~/verae-fleet-runtime", + "enabled": false, + "capacity": 12, + "roles": ["tree-node", "archive-worm"], + "notes": "Same user/key as NS1. Enable after confirming node is on PATH." + } + ] +} diff --git a/machines.secrets.json.example b/machines.secrets.json.example new file mode 100644 index 0000000..474231d --- /dev/null +++ b/machines.secrets.json.example @@ -0,0 +1,13 @@ +{ + "ns1": { + "user": "marchon", + "host": "70.88.205.138", + "sshPort": 22, + "identityFile": "~/.ssh/id_ed25519" + }, + "lan-134": { + "user": "marchon", + "host": "70.88.205.134", + "identityFile": "~/.ssh/id_ed25519" + } +} diff --git a/package.json b/package.json new file mode 100644 index 0000000..6e77001 --- /dev/null +++ b/package.json @@ -0,0 +1,13 @@ +{ + "name": "verae-fleet", + "version": "0.1.0", + "type": "module", + "description": "Service catalog, replica floors, monitor, restart, pause/resume for Verae Zapier workers and tree nodes", + "bin": { "verae-fleet": "./src/cli.js" }, + "scripts": { + "start": "node src/cli.js serve", + "list": "node src/cli.js list", + "test": "node --test test/*.test.js" + }, + "engines": { "node": ">=20" } +} diff --git a/public/index.html b/public/index.html new file mode 100644 index 0000000..29f0fc4 --- /dev/null +++ b/public/index.html @@ -0,0 +1,148 @@ + + + + + + Verae fleet monitor + + + +
+

Verae fleet monitor

+

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

+
+ + + + + + + + + +
+ + + +
MachineSSH / hostKindCapacityRunningRTT ms min / avg / p50 / p90
+

Services

+ + + +
ServiceStatemin/maxavailableRTT ms min / avg / p50 / p90instancesactions
+
+ + + diff --git a/services/archive-aggregator.json b/services/archive-aggregator.json new file mode 100644 index 0000000..2c024d2 --- /dev/null +++ b/services/archive-aggregator.json @@ -0,0 +1,16 @@ +{ + "id": "archive-aggregator", + "title": "Archive reply aggregator", + "kind": "nats-worker", + "package": "verae-archive-aggregator", + "role": "archive-aggregator", + "managed": true, + "runtime": "query fan-out + merge", + "health": { "type": "http", "path": "/health", "timeoutMs": 800 }, + "ports": { "healthBase": 13400 }, + "nats": { + "in": ["verae.archive.reply.*"], + "out": ["verae.archive.query"] + }, + "notes": "Broadcast archive.query (no queue group). kinds may include tree." +} diff --git a/services/archive-worm.json b/services/archive-worm.json new file mode 100644 index 0000000..dd3360c --- /dev/null +++ b/services/archive-worm.json @@ -0,0 +1,16 @@ +{ + "id": "archive-worm", + "title": "Bloom-filtered WORM archive", + "kind": "nats-archive", + "package": "verae-archive-worm", + "role": "archive-worm", + "managed": true, + "runtime": "N copies, bloom miss = silence", + "health": { "type": "http", "path": "/health", "timeoutMs": 800 }, + "ports": { "healthBase": 13500 }, + "nats": { + "in": ["verae.archive.put", "verae.archive.query"], + "out": ["verae.archive.reply."] + }, + "notes": "Subscribe to query without a shared queue group." +} diff --git a/services/job-poller.json b/services/job-poller.json new file mode 100644 index 0000000..0e78df1 --- /dev/null +++ b/services/job-poller.json @@ -0,0 +1,16 @@ +{ + "id": "job-poller", + "title": "Job poller worker", + "kind": "nats-worker", + "package": "verae-zapier-middleware", + "role": "job-poller", + "managed": true, + "runtime": "JetStream consumer", + "health": { "type": "http", "path": "/health", "timeoutMs": 800 }, + "ports": { "healthBase": 13200 }, + "nats": { + "in": ["verae.zapier.jobs.watch"], + "out": ["verae.zapier.jobs.events"] + }, + "notes": "GET chain status; emit terminal events. Queue group job-poller." +} diff --git a/services/middleware-http.json b/services/middleware-http.json new file mode 100644 index 0000000..5b819d2 --- /dev/null +++ b/services/middleware-http.json @@ -0,0 +1,17 @@ +{ + "id": "middleware-http", + "title": "Verae Zapier HTTP middleware", + "kind": "http", + "package": "verae-zapier-middleware", + "role": "middleware-http", + "managed": true, + "runtime": "HTTPS :3100 /zapier/v1", + "health": { "type": "http", "path": "/health", "timeoutMs": 800 }, + "ports": { "healthBase": 13100 }, + "env": { "PORT": "3100", "NATS_URL": "nats://127.0.0.1:4222" }, + "nats": { + "in": ["verae.zapier.jobs.events"], + "out": ["verae.zapier.jobs.watch", "verae.zapier.webhooks.deliver"] + }, + "notes": "Wait path subscribes jobs.events. Never expose NATS to Zapier." +} diff --git a/services/nats.json b/services/nats.json new file mode 100644 index 0000000..60dcb37 --- /dev/null +++ b/services/nats.json @@ -0,0 +1,11 @@ +{ + "id": "nats", + "title": "NATS JetStream (NS1 loopback)", + "kind": "external", + "package": null, + "managed": false, + "runtime": "nats-server -js on 127.0.0.1:4222", + "health": { "type": "tcp", "host": "127.0.0.1", "port": 4222, "timeoutMs": 400 }, + "nats": { "in": [], "out": [] }, + "notes": "Do not bind 0.0.0.0. Zapier never connects here. Fleet monitors only; does not spawn nats-server." +} diff --git a/services/tree-node.json b/services/tree-node.json new file mode 100644 index 0000000..0c85964 --- /dev/null +++ b/services/tree-node.json @@ -0,0 +1,16 @@ +{ + "id": "tree-node", + "title": "Merkle tree-node archive", + "kind": "nats-archive", + "package": "verae-tree-node", + "role": "tree-node", + "managed": true, + "runtime": "N copies holding leaf proofs for bulk summaries", + "health": { "type": "http", "path": "/health", "timeoutMs": 800 }, + "ports": { "healthBase": 13600 }, + "nats": { + "in": ["verae.archive.put", "verae.archive.query"], + "out": ["verae.archive.reply."] + }, + "notes": "keepFloor in fleet.json must stay >= 3 so a leaf query still has a node that might hold the shard. Bloom miss = no packet." +} diff --git a/services/unmanaged.json b/services/unmanaged.json new file mode 100644 index 0000000..df7e1bf --- /dev/null +++ b/services/unmanaged.json @@ -0,0 +1,37 @@ +{ + "id": "_unmanaged", + "title": "Not fleet-spawned (listed for operators)", + "entries": [ + { + "id": "zapier-platform-app", + "package": "verae-zapier", + "runtime": "Zapier cloud", + "notes": "HTTPS only to zappier-edge. Never NATS." + }, + { + "id": "verae-activate", + "package": "verae-activate", + "runtime": "Zapier cloud (activate-now app)" + }, + { + "id": "request-splitter", + "package": "verae-request-splitter", + "runtime": "library in middleware" + }, + { + "id": "verae-chain-client", + "package": "verae-zapier-middleware", + "runtime": "library → api.veraetime.net or MOCK" + }, + { + "id": "zapier-user-docs", + "package": "zapier-user-docs", + "runtime": "static" + }, + { + "id": "docs-master", + "package": "docs-master", + "runtime": "static" + } + ] +} diff --git a/services/webhook-deliver.json b/services/webhook-deliver.json new file mode 100644 index 0000000..6c058e3 --- /dev/null +++ b/services/webhook-deliver.json @@ -0,0 +1,16 @@ +{ + "id": "webhook-deliver", + "title": "Zapier REST Hook deliverer", + "kind": "nats-worker", + "package": "verae-zapier-middleware", + "role": "webhook-deliver", + "managed": true, + "runtime": "JetStream consumer + HTTPS POST", + "health": { "type": "http", "path": "/health", "timeoutMs": 800 }, + "ports": { "healthBase": 13300 }, + "nats": { + "in": ["verae.zapier.webhooks.deliver", "verae.zapier.jobs.events"], + "out": [] + }, + "notes": "HTTPS POST to Zapier hook URL. Not a NATS client in Zapier cloud." +} diff --git a/services/zapier-simulator.json b/services/zapier-simulator.json new file mode 100644 index 0000000..e454710 --- /dev/null +++ b/services/zapier-simulator.json @@ -0,0 +1,14 @@ +{ + "id": "zapier-simulator", + "title": "Zapier interface simulator + trace console", + "kind": "http", + "package": "verae-zapier-simulator", + "role": "zapier-simulator", + "managed": true, + "runtime": "HTTP 127.0.0.1:3847", + "health": { "type": "http", "path": "/health", "timeoutMs": 800 }, + "ports": { "healthBase": 13847 }, + "env": { "SIM_PORT": "3847" }, + "nats": { "in": [], "out": [] }, + "notes": "Optional. min 0 in central fleet.json. Does not talk to live NATS." +} diff --git a/services/zappier-edge.json b/services/zappier-edge.json new file mode 100644 index 0000000..a43ca91 --- /dev/null +++ b/services/zappier-edge.json @@ -0,0 +1,14 @@ +{ + "id": "zappier-edge", + "title": "Zappier commercial HTTPS edge", + "kind": "http", + "package": "zappier", + "role": "zappier-edge", + "managed": true, + "runtime": "HTTPS :3000 (portal, admin, meter)", + "health": { "type": "http", "path": "/health", "timeoutMs": 800 }, + "ports": { "healthBase": 13000 }, + "env": { "PORT": "3000" }, + "nats": { "in": [], "out": [] }, + "notes": "Zapier x-api-key lands here. HTTPS to middleware only." +} diff --git a/src/agent.js b/src/agent.js new file mode 100644 index 0000000..05a3d9d --- /dev/null +++ b/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/src/classify.js b/src/classify.js new file mode 100644 index 0000000..3313b7a --- /dev/null +++ b/src/classify.js @@ -0,0 +1,29 @@ +/** + * Row health: operational | degraded | down + * @module classify + */ + +/** + * @param {{ + * min?: number, + * available?: number, + * running?: number, + * paused?: number, + * managed?: boolean, + * instances?: Array<{ pid?: number|null, paused?: boolean, healthy?: boolean }> + * }} sv + * @returns {'operational'|'degraded'|'down'} + */ +export function classifyService(sv) { + const min = sv.min ?? 0; + const available = sv.available ?? 0; + const running = sv.running ?? 0; + const paused = sv.paused ?? 0; + const inst = sv.instances || []; + const unhealthy = inst.filter((i) => i.pid && i.healthy === false && !i.paused).length; + + if (min === 0 && running === 0) return 'operational'; + if (available === 0 && min > 0) return 'down'; + if (available >= min && paused === 0 && unhealthy === 0) return 'operational'; + return 'degraded'; +} diff --git a/src/cli.js b/src/cli.js new file mode 100644 index 0000000..fdc2951 --- /dev/null +++ b/src/cli.js @@ -0,0 +1,145 @@ +#!/usr/bin/env node +/** + * verae-fleet — catalog, monitor, restart, pause. + * + * Local (no daemon): list + * Daemon: serve | monitor + * Against daemon: status | start | stop | pause | resume | restart + */ +import http from 'node:http'; +import { loadFleet, listServices } from './load.js'; +import { Supervisor } from './supervisor.js'; +import { Monitor } from './monitor.js'; +import { startControlServer } from './server.js'; + +const loaded = loadFleet(); +const [cmd = 'help', target] = process.argv.slice(2); +const BASE = process.env.FLEET_URL || `http://127.0.0.1:${loaded.control.port}`; + +function print(obj) { + process.stdout.write(`${typeof obj === 'string' ? obj : JSON.stringify(obj, null, 2)}\n`); +} + +function help() { + print(`verae-fleet + + list services, config files, replica min/max + serve start floors + monitor + UI ${BASE}/ + monitor same without printing extra + status GET daemon status + start turn on, spawn up to min + stop turn off (service disable) or stop one replica + pause pause (does not count toward tree-node floor) + resume + restart kill + respawn + reconcile force floor check + +Central spec: fleet.json +Machines: machines.json (add hosts to spread replicas) +Per service: services/.json +Remote SSH: machines.json user + host + identityFile (path only) + ssh-check [id] test SSH login (default ns1) +`); +} + +function api(method, pathname) { + return new Promise((resolve, reject) => { + const u = new URL(pathname, BASE); + const req = http.request( + { hostname: u.hostname, port: u.port, path: u.pathname, method, timeout: 8000 }, + (res) => { + const chunks = []; + res.on('data', (c) => chunks.push(c)); + res.on('end', () => { + const raw = Buffer.concat(chunks).toString('utf8'); + try { + resolve({ status: res.statusCode, body: JSON.parse(raw) }); + } catch { + resolve({ status: res.statusCode, body: raw }); + } + }); + }, + ); + req.on('error', reject); + req.end(); + }); +} + +async function main() { + if (cmd === 'help' || cmd === '-h' || cmd === '--help') return help(); + if (cmd === 'list') { + for (const r of listServices(loaded)) { + print( + `${r.id.padEnd(22)} min=${String(r.min).padStart(2)} max=${String(r.max).padStart(2)} floor=${r.keepFloor ? 'yes' : 'no '} managed=${r.managed ? 'yes' : 'no '} ${r.configPath}`, + ); + } + print(`\ncentral replica spec: ${loaded.fleetPath}`); + print(`tree-node floor: min=${loaded.services['tree-node'].min} keepFloor=${loaded.services['tree-node'].keepFloor}`); + print('\nmachines:'); + for (const m of loaded.machines) { + const ssh = m.kind === 'ssh' ? `${m.user}@${m.host} key=${m.identityFile}` : m.host; + print(` ${m.id.padEnd(12)} ${m.kind.padEnd(6)} ${m.enabled ? 'on ' : 'off'} cap=${m.capacity} ${ssh}`); + } + return; + } + if (cmd === 'ssh-check') { + const { sshCheck } = await import('./ssh.js'); + const id = target || 'ns1'; + const m = loaded.machines.find((x) => x.id === id); + if (!m) throw new Error(`unknown machine ${id}`); + print(await sshCheck(m)); + return; + } + if (cmd === 'serve' || cmd === 'monitor') { + const sup = new Supervisor({ loaded }); + for (const id of Object.keys(loaded.services)) { + const s = loaded.services[id]; + if (s.managed && s.enabled && s.min > 0) await sup.startService(id); + } + const mon = new Monitor(sup); + await mon.tick(); + mon.start(); + startControlServer(sup, mon); + print(`fleet ${cmd} pid=${process.pid}\nUI ${BASE}/\nSIGINT stops all managed replicas`); + process.on('SIGINT', async () => { + mon.stop(); + await sup.stopAll(); + process.exit(0); + }); + process.on('SIGTERM', async () => { + mon.stop(); + await sup.stopAll(); + process.exit(0); + }); + return; + } + + const instance = target && /-\d+$/.test(target); + const map = { + status: ['GET', '/api/status'], + reconcile: ['POST', '/api/reconcile'], + }; + if (map[cmd] && !target) { + const [m, p] = map[cmd]; + const r = await api(m, p); + print(r.body); + if (r.status >= 400) process.exitCode = 1; + return; + } + if (!target) { + help(); + process.exitCode = 1; + return; + } + const pathName = instance + ? `/api/instances/${target}/${cmd}` + : `/api/services/${target}/${cmd}`; + const r = await api('POST', pathName); + print(r.body); + if (r.status >= 400) process.exitCode = 1; +} + +main().catch((err) => { + print({ error: err.message, hint: 'Is `verae-fleet serve` running?' }); + process.exitCode = 1; +}); diff --git a/src/health.js b/src/health.js new file mode 100644 index 0000000..50ec05f --- /dev/null +++ b/src/health.js @@ -0,0 +1,118 @@ +/** + * Probe a replica: HTTP /health or TCP connect. + * @module health + */ + +import http from 'node:http'; +import net from 'node:net'; +import fs from 'node:fs'; +import path from 'node:path'; + +/** + * @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, port, path: pathName, timeout: tmo }, + (res) => { + const chunks = []; + res.on('data', (c) => chunks.push(c)); + res.on('end', () => { + let body = {}; + try { + body = JSON.parse(Buffer.concat(chunks).toString('utf8') || '{}'); + } catch { + body = {}; + } + resolve({ + ok: res.statusCode >= 200 && res.statusCode < 300 && body.ok !== false, + statusCode: res.statusCode, + paused: Boolean(body.paused), + body, + }); + }); + }, + ); + req.on('error', (err) => resolve({ ok: false, error: err.message, paused: false })); + req.on('timeout', () => { + req.destroy(); + resolve({ ok: false, error: 'timeout', paused: false }); + }); + }); +} + +export function tcpProbe(host, port, timeoutMs = 400) { + return new Promise((resolve) => { + const sock = net.connect({ host, port }); + const done = (ok, error) => { + try { + sock.destroy(); + } catch { + /* ignore */ + } + resolve({ ok, error, paused: false }); + }; + sock.setTimeout(timeoutMs); + sock.on('connect', () => done(true)); + sock.on('error', (err) => done(false, err.message)); + sock.on('timeout', () => done(false, 'timeout')); + }); +} + +/** + * @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, + 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)); + res.on('end', () => { + try { + resolve(JSON.parse(Buffer.concat(chunks).toString('utf8') || '{}')); + } catch { + resolve({}); + } + }); + }, + ); + req.on('error', reject); + if (body) req.write(body); + req.end(); + }); +} + +export function heartbeatFresh(stateDir, maxAgeMs) { + const file = path.join(stateDir, 'heartbeat.json'); + try { + const st = fs.statSync(file); + const body = JSON.parse(fs.readFileSync(file, 'utf8')); + return { ok: Date.now() - st.mtimeMs < maxAgeMs && body.ok !== false, body, ageMs: Date.now() - st.mtimeMs }; + } catch (err) { + return { ok: false, error: err.message }; + } +} diff --git a/src/load.js b/src/load.js new file mode 100644 index 0000000..96d9662 --- /dev/null +++ b/src/load.js @@ -0,0 +1,94 @@ +/** + * Load central fleet.json + per-service JSON configs. + * @module load + */ + +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, '..'); + +export function readJson(file) { + return JSON.parse(fs.readFileSync(file, 'utf8')); +} + +/** + * @param {string} [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'); + const services = {}; + const unmanaged = []; + for (const name of fs.readdirSync(dir).sort()) { + if (!name.endsWith('.json')) continue; + const spec = readJson(path.join(dir, name)); + const configPath = path.join('services', name); + if (spec.id === '_unmanaged') { + for (const e of spec.entries || []) { + unmanaged.push({ ...e, managed: false, configPath }); + } + continue; + } + const central = fleet.services?.[spec.id] || {}; + services[spec.id] = { + ...spec, + configPath, + min: central.min ?? 0, + max: central.max ?? 1, + keepFloor: central.keepFloor !== false, + enabled: central.enabled !== false, + 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, + }; +} + +export function listServices(loaded) { + const rows = Object.values(loaded.services).map((s) => ({ + id: s.id, + title: s.title, + kind: s.kind, + managed: s.managed, + enabled: s.enabled, + min: s.min, + max: s.max, + keepFloor: s.keepFloor, + configPath: s.configPath, + package: s.package, + runtime: s.runtime, + })); + for (const u of loaded.unmanaged) { + rows.push({ + id: u.id, + title: u.id, + kind: 'unmanaged', + managed: false, + enabled: false, + min: 0, + max: 0, + keepFloor: false, + configPath: u.configPath, + package: u.package, + runtime: u.runtime, + }); + } + return rows; +} diff --git a/src/machines.js b/src/machines.js new file mode 100644 index 0000000..551bd55 --- /dev/null +++ b/src/machines.js @@ -0,0 +1,107 @@ +/** + * 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); + } + const secretsPath = path.join(root, 'machines.secrets.json'); + if (fs.existsSync(secretsPath)) { + const secrets = JSON.parse(fs.readFileSync(secretsPath, 'utf8')); + const byId = secrets.machines || secrets; + for (const m of list) { + const extra = byId[m.id]; + if (extra && typeof extra === 'object') Object.assign(m, extra); + } + } + return list.map(normalizeMachine); +} + +export function normalizeMachine(m) { + const kind = m.kind === 'agent' || m.kind === 'ssh' ? m.kind : 'local'; + return { + id: String(m.id || '').trim(), + title: m.title || m.id, + kind, + host: m.host || '127.0.0.1', + user: m.user || (kind === 'ssh' ? 'marchon' : ''), + sshPort: Number(m.sshPort || 22), + identityFile: m.identityFile || (kind === 'ssh' ? '~/.ssh/id_ed25519' : ''), + remoteDir: m.remoteDir || '~/verae-fleet-runtime', + 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, exclude = []) { + const skip = new Set(exclude); + const eligible = machines.filter((m) => canHost(m, serviceId, role) && !skip.has(m.id)); + 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/src/monitor.js b/src/monitor.js new file mode 100644 index 0000000..6cdb2e1 --- /dev/null +++ b/src/monitor.js @@ -0,0 +1,38 @@ +/** + * Periodic probe + reconcile so replica floors (especially tree-node) stay met. + * @module monitor + */ + +export class Monitor { + /** + * @param {import('./supervisor.js').Supervisor} supervisor + * @param {{ intervalMs?: number }} [opts] + */ + constructor(supervisor, opts = {}) { + this.supervisor = supervisor; + this.intervalMs = opts.intervalMs ?? supervisor.loaded.control?.probeIntervalMs ?? 750; + this.timer = null; + this.last = null; + } + + async tick() { + const actions = await this.supervisor.reconcile(); + this.last = { t: new Date().toISOString(), actions, status: this.supervisor.status() }; + return this.last; + } + + start() { + if (this.timer) return; + this.timer = setInterval(() => { + this.tick().catch((err) => { + this.supervisor.log('monitor-error', { error: err.message }); + }); + }, this.intervalMs); + this.timer.unref?.(); + } + + stop() { + if (this.timer) clearInterval(this.timer); + this.timer = null; + } +} diff --git a/src/rtt.js b/src/rtt.js new file mode 100644 index 0000000..09ec7bb --- /dev/null +++ b/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/src/server.js b/src/server.js new file mode 100644 index 0000000..7baaf07 --- /dev/null +++ b/src/server.js @@ -0,0 +1,105 @@ +/** + * Operator HTTP: list, status, pause/resume/stop/start/restart. + * Binds 127.0.0.1 only. + * @module server + */ + +import http from 'node:http'; +import fs from 'node:fs'; +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { listServices } from './load.js'; + +const PUBLIC = path.join(path.dirname(fileURLToPath(import.meta.url)), '..', 'public'); + +function json(res, code, obj) { + res.writeHead(code, { 'content-type': 'application/json', 'cache-control': 'no-store' }); + res.end(JSON.stringify(obj, null, 2)); +} + +/** + * @param {import('./supervisor.js').Supervisor} sup + * @param {import('./monitor.js').Monitor} [mon] + */ +export function startControlServer(sup, mon) { + const bind = sup.loaded.control?.bind || '127.0.0.1'; + const port = Number(sup.loaded.control?.port || 3850); + + 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-control' }); + } + if (req.method === 'GET' && url.pathname === '/api/services') { + return json(res, 200, { services: listServices(sup.loaded), fleetPath: sup.loaded.fleetPath }); + } + 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|check)$/); + if (mach && req.method === 'POST') { + const [, id, op] = mach; + if (op === 'check') { + const check = await sup.checkMachine(id); + return json(res, check.ok ? 200 : 400, { check, machines: sup.status().machines }); + } + 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() }); + } + const svc = url.pathname.match(/^\/api\/services\/([^/]+)\/(start|stop|pause|resume)$/); + if (svc && req.method === 'POST') { + const [, id, op] = svc; + const fn = { + start: () => sup.startService(id), + stop: () => sup.stopService(id), + pause: () => sup.pauseService(id), + resume: () => sup.resumeService(id), + }[op]; + const result = await fn(); + return json(res, 200, { op, id, result, status: sup.status().services[id] }); + } + const inst = url.pathname.match(/^\/api\/instances\/([^/]+)\/(pause|resume|restart|stop|unhealthy)$/); + if (inst && req.method === 'POST') { + const [, id, op] = inst; + const fn = { + pause: () => sup.pauseInstance(id), + resume: () => sup.resumeInstance(id), + restart: () => sup.restartInstance(id), + stop: () => sup.stopInstance(id, { replace: true }), + unhealthy: () => sup.markUnhealthy(id), + }[op]; + const result = await fn(); + return json(res, 200, { op, id, result: result?.id || result, status: sup.status() }); + } + if (req.method === 'GET' && (url.pathname === '/' || url.pathname === '/index.html')) { + const dest = path.join(PUBLIC, 'index.html'); + res.writeHead(200, { 'content-type': 'text/html; charset=utf-8' }); + fs.createReadStream(dest).pipe(res); + return; + } + json(res, 404, { error: 'not found' }); + } catch (err) { + json(res, 400, { error: err.message }); + } + }); + + server.listen(port, bind, () => { + process.stdout.write(`fleet control http://${bind}:${port}/\n`); + }); + return server; +} diff --git a/src/ssh.js b/src/ssh.js new file mode 100644 index 0000000..e8a86c7 --- /dev/null +++ b/src/ssh.js @@ -0,0 +1,170 @@ +/** + * SSH control of remote fleet workers. Uses username + host + identity file path. + * Never logs or stores private-key material. + * @module ssh + */ + +import { spawn } from 'node:child_process'; +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; + +const HERE = path.dirname(fileURLToPath(import.meta.url)); + +export function expandHome(p) { + if (!p) return p; + if (p === '~') return os.homedir(); + if (p.startsWith('~/')) return path.join(os.homedir(), p.slice(2)); + return p; +} + +export function shellQuote(s) { + return `'${String(s).replace(/'/g, `'\"'\"'`)}'`; +} + +export function sshTarget(machine) { + return `${machine.user || 'marchon'}@${machine.host}`; +} + +export function sshBaseArgs(machine) { + const ident = expandHome(machine.identityFile || ''); + const args = [ + '-n', + '-o', 'BatchMode=yes', + '-o', 'StrictHostKeyChecking=accept-new', + '-o', 'IdentitiesOnly=yes', + '-o', 'ConnectTimeout=8', + '-p', String(machine.sshPort || 22), + ]; + if (ident) { + if (!fs.existsSync(ident)) throw new Error(`identityFile not found: ${ident}`); + args.push('-i', ident); + } + return args; +} + +export function scpBaseArgs(machine) { + const ident = expandHome(machine.identityFile || ''); + const args = [ + '-o', 'BatchMode=yes', + '-o', 'StrictHostKeyChecking=accept-new', + '-o', 'IdentitiesOnly=yes', + '-o', 'ConnectTimeout=8', + '-P', String(machine.sshPort || 22), + ]; + if (ident) args.push('-i', ident); + return args; +} + +export function sshExec(machine, remoteCommand, { timeoutMs = 20000 } = {}) { + return new Promise((resolve, reject) => { + const args = [...sshBaseArgs(machine), sshTarget(machine), remoteCommand]; + const child = spawn('ssh', args, { stdio: ['ignore', 'pipe', 'pipe'] }); + const out = []; + const err = []; + child.stdout.on('data', (c) => out.push(c)); + child.stderr.on('data', (c) => err.push(c)); + const t = setTimeout(() => { + child.kill('SIGKILL'); + reject(new Error(`ssh timeout: ${machine.id || machine.host}`)); + }, timeoutMs); + child.on('error', (e) => { + clearTimeout(t); + reject(e); + }); + child.on('close', (code) => { + clearTimeout(t); + const stdout = Buffer.concat(out).toString('utf8').trim(); + const stderr = Buffer.concat(err).toString('utf8').trim(); + if (code !== 0) { + reject(new Error(stderr || stdout || `ssh exit ${code}`)); + return; + } + resolve({ stdout, stderr, code }); + }); + }); +} + +export function scpTo(machine, localFiles, remoteDir) { + return new Promise((resolve, reject) => { + const dest = `${sshTarget(machine)}:${remoteDir}/`; + const args = [...scpBaseArgs(machine), ...localFiles, dest]; + const child = spawn('scp', args, { stdio: ['ignore', 'pipe', 'pipe'] }); + const err = []; + child.stderr.on('data', (c) => err.push(c)); + const t = setTimeout(() => { + child.kill('SIGKILL'); + reject(new Error('scp timeout')); + }, 30000); + child.on('close', (code) => { + clearTimeout(t); + if (code !== 0) reject(new Error(Buffer.concat(err).toString('utf8') || `scp exit ${code}`)); + else resolve(); + }); + child.on('error', (e) => { + clearTimeout(t); + reject(e); + }); + }); +} + +const synced = new Set(); + +export async function ensureRemoteRuntime(machine) { + const remoteDir = machine.remoteDir || '~/verae-fleet-runtime'; + const key = `${machine.id}:${remoteDir}`; + if (synced.has(key)) return remoteDir; + await sshExec(machine, `mkdir -p ${remoteDir}/src ${remoteDir}/data`); + await scpTo(machine, [path.join(HERE, 'worker.js'), path.join(HERE, 'rtt.js')], `${remoteDir}/src`); + synced.add(key); + return remoteDir; +} + +export async function sshSpawnWorker(machine, { instance, env }) { + const remoteDir = await ensureRemoteRuntime(machine); + const fleetEnv = Object.entries(env) + .filter(([k]) => k.startsWith('FLEET_')) + .map(([k, v]) => `${k}=${shellQuote(v)}`) + .join(' '); + const inner = `node src/worker.js data/${instance}/worker.log 2>&1 & echo $!`; + const cmd = `mkdir -p ${remoteDir}/data/${instance} && cd ${remoteDir} && ${fleetEnv} bash -c ${shellQuote(inner)}`; + const { stdout } = await sshExec(machine, cmd); + const pid = Number(String(stdout).split('\n').pop()); + if (!Number.isFinite(pid)) throw new Error(`remote spawn produced no pid: ${stdout}`); + return { pid, remoteDir }; +} + +export async function sshKillWorker(machine, pid, instance) { + const cmd = pid + ? `kill -TERM ${Number(pid)} 2>/dev/null || true` + : `pkill -f ${shellQuote(`FLEET_INSTANCE=${instance}`)} 2>/dev/null || true`; + await sshExec(machine, cmd); +} + +export async function sshHttp(machine, method, urlPath, { port, body } = {}) { + const url = `http://127.0.0.1:${port}${urlPath}`; + let cmd = `curl -sS -m 3 -X ${method}`; + if (body) { + cmd += ` -H 'content-type: application/json' -d ${shellQuote(JSON.stringify(body))}`; + } + cmd += ` ${shellQuote(url)}`; + const { stdout } = await sshExec(machine, cmd, { timeoutMs: 8000 }); + try { + return JSON.parse(stdout || '{}'); + } catch { + return { raw: stdout }; + } +} + +export async function sshCheck(machine) { + const { stdout } = await sshExec(machine, 'echo OK; whoami; hostname; command -v node; node -v'); + const lines = stdout.split('\n'); + return { + ok: lines[0] === 'OK', + user: lines[1], + hostname: lines[2], + node: lines[3] ? `${lines[3]} ${lines[4] || ''}`.trim() : null, + identityFile: expandHome(machine.identityFile || ''), + }; +} diff --git a/src/supervisor.js b/src/supervisor.js new file mode 100644 index 0000000..5bb543e --- /dev/null +++ b/src/supervisor.js @@ -0,0 +1,510 @@ +/** + * Spawn, pause, resume, stop, restart replicas. Keep replica floor (tree-node min). + * @module supervisor + */ + +import { spawn } from 'node:child_process'; +import fs from 'node:fs'; +import path from 'node:path'; +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'; +import { sshSpawnWorker, sshKillWorker, sshHttp, sshCheck } from './ssh.js'; + +const WORKER = path.join(path.dirname(fileURLToPath(import.meta.url)), 'worker.js'); + +export class Supervisor { + /** + * @param {object} [opts] + * @param {ReturnType} [opts.loaded] + * @param {string} [opts.stateRoot] + */ + constructor(opts = {}) { + this.loaded = opts.loaded || loadFleet(); + this.stateRoot = opts.stateRoot || path.join(this.loaded.root, 'data', 'runtime'); + fs.mkdirSync(this.stateRoot, { recursive: true }); + /** @type {Map} */ + 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; + } + + async checkMachine(id) { + const m = this.machines().find((x) => x.id === id); + if (!m) throw new Error(`unknown machine ${id}`); + if (m.kind === 'ssh') return sshCheck(m); + if (m.kind === 'agent') { + const r = await httpProbe({ host: m.host, port: m.agentPort, path: '/health', timeoutMs: 2000 }); + return { ok: r.ok, kind: 'agent', body: r.body }; + } + return { ok: true, kind: 'local', host: m.host }; + } + + machineOf(rec) { + return this.machines().find((x) => x.id === rec.machine); + } + + async controlInstance(rec, pathname, body) { + if (rec.kind === 'ssh') { + const m = this.machineOf(rec); + if (!m) throw new Error(`machine ${rec.machine} missing`); + return sshHttp(m, 'POST', pathname, { port: rec.healthPort, body }); + } + return postControl({ + host: rec.host || '127.0.0.1', + port: rec.healthPort, + path: pathname, + body, + }); + } + + log(type, detail) { + const ev = { t: new Date().toISOString(), type, ...detail }; + this.events.push(ev); + if (this.events.length > 500) this.events.splice(0, this.events.length - 400); + return ev; + } + + spec(serviceId) { + const s = this.loaded.services[serviceId]; + if (!s) throw new Error(`unknown service ${serviceId}`); + return s; + } + + nextIndex(serviceId) { + const used = new Set( + [...this.instances.values()].filter((i) => i.service === serviceId).map((i) => i.index), + ); + const max = this.spec(serviceId).max ?? 8; + for (let i = 0; i < max; i += 1) if (!used.has(i)) return i; + return null; + } + + /** + * @param {string} serviceId + * @param {{ index?: number }} [opts] + */ + async startOne(serviceId, opts = {}) { + const spec = this.spec(serviceId); + if (!spec.managed) { + this.log('skip', { service: serviceId, reason: 'unmanaged' }); + return null; + } + if (!spec.enabled) { + this.log('skip', { service: serviceId, reason: 'disabled' }); + return null; + } + const index = opts.index ?? this.nextIndex(serviceId); + if (index == null) { + this.log('skip', { service: serviceId, reason: 'at-max' }); + return null; + } + 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, + opts.excludeMachines || [], + ); + 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 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' : '127.0.0.1', + FLEET_STATE_DIR: machine.kind === 'ssh' ? `${machine.remoteDir || '~/verae-fleet-runtime'}/data/${instance}` : stateDir, + FLEET_MESSAGE_HZ: process.env.FLEET_MESSAGE_HZ || '8', + }; + let child = null; + let pid = null; + if (machine.kind === 'ssh') { + try { + const spawned = await sshSpawnWorker(machine, { instance, env }); + pid = spawned.pid; + } catch (err) { + this.log('skip', { service: serviceId, reason: 'ssh-spawn-failed', machine: machine.id, error: err.message }); + const exclude = [...(opts.excludeMachines || []), machine.id]; + if (exclude.length < this.machines().length) { + return this.startOne(serviceId, { ...opts, excludeMachines: exclude }); + } + return null; + } + } else 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, + machine: machine.id, + host: machine.host, + pid, + healthPort, + stateDir, + child, + kind: machine.kind, + _machine: machine, + paused: false, + startedAt: Date.now(), + restarts: 0, + lastError: null, + rttSamples: [], + }; + 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, machine: machine.id, pid: rec.pid, healthPort }); + const h = await waitForHealth(rec, 8000); + rec.healthy = h.ok; + rec.lastProbe = h; + return rec; + } + + async stopInstance(instanceId, { replace = true } = {}) { + const rec = this.instances.get(instanceId); + if (!rec) return null; + rec.stopping = true; + if (rec.kind === 'ssh') { + const m = this.machines().find((x) => x.id === rec.machine); + if (m) { + try { + await sshKillWorker(m, rec.pid, instanceId); + } catch { + /* ignore */ + } + } + } else 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); + } + this.instances.delete(instanceId); + this.log('stop', { instance: instanceId, service: rec.service }); + if (replace) await this.reconcile(rec.service); + return rec; + } + + async pauseInstance(instanceId, { replace = true } = {}) { + const rec = this.instances.get(instanceId); + if (!rec) throw new Error(`no instance ${instanceId}`); + await this.controlInstance(rec, '/pause'); + rec.paused = true; + this.log('pause', { instance: instanceId, service: rec.service }); + if (replace) await this.reconcile(rec.service); + return rec; + } + + async resumeInstance(instanceId) { + const rec = this.instances.get(instanceId); + if (!rec) throw new Error(`no instance ${instanceId}`); + await this.controlInstance(rec, '/resume'); + rec.paused = false; + this.log('resume', { instance: instanceId, service: rec.service }); + return rec; + } + + async restartInstance(instanceId) { + const rec = this.instances.get(instanceId); + if (!rec) throw new Error(`no instance ${instanceId}`); + 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; + } + + async markUnhealthy(instanceId) { + const rec = this.instances.get(instanceId); + if (!rec) throw new Error(`no instance ${instanceId}`); + await this.controlInstance(rec, '/unhealthy'); + return rec; + } + + async probe(rec) { + const spec = this.spec(rec.service); + const pathName = spec.health?.path || '/health'; + let r; + if (rec.kind === 'ssh') { + const m = this.machineOf(rec); + try { + const body = await sshHttp(m, 'GET', pathName, { port: rec.healthPort }); + r = { ok: body.ok !== false, paused: Boolean(body.paused), body, statusCode: body.ok === false ? 503 : 200 }; + } catch (err) { + r = { ok: false, paused: false, error: err.message }; + } + } else { + r = await httpProbe({ + host: rec.host || '127.0.0.1', + port: rec.healthPort, + path: pathName, + 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 this.controlInstance(rec, '/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; + } + + available(serviceId) { + return [...this.instances.values()].filter( + (i) => i.service === serviceId && i.pid && !i.paused && i.healthy !== false, + ); + } + + /** + * Ensure running healthy unpaused copies >= min when keepFloor. + * @param {string} [only] + */ + async reconcile(only) { + const actions = []; + const ids = only ? [only] : Object.keys(this.loaded.services); + for (const serviceId of ids) { + const spec = this.loaded.services[serviceId]; + if (!spec?.managed || !spec.enabled) continue; + for (const rec of [...this.instances.values()].filter((i) => i.service === serviceId)) { + if (!rec.pid && !rec.stopping) { + this.instances.delete(rec.id); + continue; + } + if (rec.pid) await this.probe(rec); + if (rec.pid && rec.healthy === false && !rec.paused) { + this.log('unhealthy', { instance: rec.id, service: serviceId }); + await this.restartInstance(rec.id); + actions.push({ op: 'restart', instance: rec.id }); + } + } + const avail = this.available(serviceId).length; + const want = spec.keepFloor ? spec.min : Math.min(spec.min, spec.max); + if (avail < want) { + const need = want - avail; + for (let n = 0; n < need; n += 1) { + const started = await this.startOne(serviceId); + if (started) actions.push({ op: 'start', instance: started.id }); + } + } + } + return actions; + } + + async startService(serviceId) { + const spec = this.spec(serviceId); + spec.enabled = true; + const have = [...this.instances.values()].filter((i) => i.service === serviceId && i.pid).length; + const want = spec.min || 1; + const started = []; + for (let i = have; i < want; i += 1) { + const rec = await this.startOne(serviceId); + if (rec) started.push(rec.id); + } + return started; + } + + async stopService(serviceId, { holdFloor = false } = {}) { + const spec = this.spec(serviceId); + if (!holdFloor) spec.enabled = false; + const ids = [...this.instances.values()].filter((i) => i.service === serviceId).map((i) => i.id); + for (const id of ids) await this.stopInstance(id, { replace: false }); + if (holdFloor) await this.reconcile(serviceId); + return ids; + } + + async pauseService(serviceId) { + const ids = [...this.instances.values()].filter((i) => i.service === serviceId && i.pid).map((i) => i.id); + for (const id of ids) await this.pauseInstance(id, { replace: false }); + await this.reconcile(serviceId); + return ids; + } + + async resumeService(serviceId) { + const ids = [...this.instances.values()].filter((i) => i.service === serviceId).map((i) => i.id); + for (const id of ids) await this.resumeInstance(id); + return ids; + } + + async stopAll() { + for (const id of [...this.instances.keys()]) { + await this.stopInstance(id, { replace: false }); + } + } + + status() { + const services = {}; + for (const spec of Object.values(this.loaded.services)) { + const inst = [...this.instances.values()].filter((i) => i.service === spec.id); + const available = inst.filter((i) => i.pid && !i.paused && i.healthy !== false).length; + services[spec.id] = { + id: spec.id, + title: spec.title, + configPath: spec.configPath, + managed: spec.managed, + enabled: spec.enabled, + min: spec.min, + max: spec.max, + keepFloor: spec.keepFloor, + running: inst.filter((i) => i.pid).length, + paused: inst.filter((i) => i.paused).length, + available, + belowFloor: spec.keepFloor && spec.enabled && spec.managed && available < spec.min, + 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), + }; + } +} + +function waitExit(child, ms) { + return new Promise((resolve) => { + if (!child || child.exitCode != null) return resolve(); + const t = setTimeout(() => { + try { + child.kill('SIGKILL'); + } catch { + /* ignore */ + } + resolve(); + }, ms); + child.once('exit', () => { + clearTimeout(t); + resolve(); + }); + }); +} + +async function waitForHealth(rec, ms) { + const deadline = Date.now() + ms; + while (Date.now() < deadline) { + let r; + if (rec.kind === 'ssh' && rec._machine) { + try { + const body = await sshHttp(rec._machine, 'GET', '/health', { port: rec.healthPort }); + r = { ok: body.ok !== false, statusCode: 200, paused: Boolean(body.paused), body }; + } catch { + r = { ok: false }; + } + } else { + r = await httpProbe({ + host: rec.host || '127.0.0.1', + port: rec.healthPort, + path: '/health', + timeoutMs: 400, + }); + } + if (r.ok || r.statusCode === 503) return r; + await new Promise((res) => setTimeout(res, 80)); + } + return { ok: false, error: 'start-timeout' }; +} diff --git a/src/worker.js b/src/worker.js new file mode 100644 index 0000000..bc010a2 --- /dev/null +++ b/src/worker.js @@ -0,0 +1,153 @@ +#!/usr/bin/env node +/** + * Fleet-managed replica: health HTTP + pause + heartbeat. + * Roles tree-node / archive-worm hold an in-process WormArchive. + */ +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 }); + +let paused = false; +let forceUnhealthy = false; +let archive = null; +let puts = 0; + +if (role === 'tree-node' || role === 'archive-worm') { + try { + const wormPath = path.resolve( + path.dirname(fileURLToPath(import.meta.url)), + '../../verae-archive-worm/src/archive.js', + ); + const { WormArchive } = await import(wormPath); + archive = new WormArchive(instance); + } catch { + archive = { + archiveId: instance, + put() { + puts += 1; + }, + query() { + return null; + }, + }; + } +} + +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, + paused, + forceUnhealthy, + role, + service, + instance, + machine, + port, + pid: process.pid, + startedAt, + puts, + archiveId: archive?.archiveId || null, + rtt: rtt.stats(), + }; +} + +function beat() { + const body = { ...snapshot(), ts: Date.now() }; + fs.writeFileSync(path.join(stateDir, 'heartbeat.json'), JSON.stringify(body)); +} + +const server = http.createServer((req, res) => { + const url = new URL(req.url, 'http://127.0.0.1'); + const json = (code, obj) => { + res.writeHead(code, { 'content-type': 'application/json' }); + res.end(JSON.stringify(obj)); + }; + if (url.pathname === '/health' && req.method === 'GET') { + const snap = snapshot(); + return json(snap.ok ? 200 : 503, snap); + } + if (url.pathname === '/info' && req.method === 'GET') return json(200, snapshot()); + if (url.pathname === '/pause' && req.method === 'POST') { + paused = true; + beat(); + return json(200, snapshot()); + } + if (url.pathname === '/resume' && req.method === 'POST') { + paused = false; + forceUnhealthy = false; + beat(); + return json(200, snapshot()); + } + if (url.pathname === '/unhealthy' && req.method === 'POST') { + forceUnhealthy = true; + 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, bind, () => { + beat(); + 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 { + /* ignore */ + } + process.exit(0); +} +process.on('SIGTERM', shutdown); +process.on('SIGINT', shutdown); diff --git a/test/classify.test.js b/test/classify.test.js new file mode 100644 index 0000000..7b5b083 --- /dev/null +++ b/test/classify.test.js @@ -0,0 +1,36 @@ +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { classifyService } from '../src/classify.js'; + +describe('classifyService', () => { + it('operational when available meets min and none paused', () => { + assert.equal( + classifyService({ min: 3, available: 3, running: 3, paused: 0, instances: [{ pid: 1, healthy: true }] }), + 'operational', + ); + }); + + it('degraded when a replica is paused even if floor is still met', () => { + assert.equal( + classifyService({ + min: 3, + available: 3, + running: 4, + paused: 1, + instances: [ + { pid: 1, paused: true, healthy: false }, + { pid: 2, paused: false, healthy: true }, + ], + }), + 'degraded', + ); + }); + + it('not operational when min>0 and nothing available', () => { + assert.equal(classifyService({ min: 1, available: 0, running: 0, paused: 0, instances: [] }), 'down'); + }); + + it('optional idle (min 0, not running) is operational', () => { + assert.equal(classifyService({ min: 0, available: 0, running: 0, paused: 0 }), 'operational'); + }); +}); diff --git a/test/fleet.test.js b/test/fleet.test.js new file mode 100644 index 0000000..c680cc0 --- /dev/null +++ b/test/fleet.test.js @@ -0,0 +1,141 @@ +import { describe, it, after } from 'node:test'; +import assert from 'node:assert/strict'; +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(FLEET_ROOT, { overlay: false }); + for (const s of Object.values(loaded.services)) { + s.enabled = s.id === 'tree-node'; + if (s.id === 'tree-node') { + s.min = 3; + s.max = 6; + s.keepFloor = true; + 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; +} + +describe('verae-fleet', () => { + /** @type {Supervisor[]} */ + const supervisors = []; + after(async () => { + for (const s of supervisors) await s.stopAll(); + }); + + it('lists every service with a config file and central replica spec', () => { + const loaded = loadFleet(); + const rows = listServices(loaded); + const ids = rows.map((r) => r.id); + for (const need of [ + 'nats', + 'zappier-edge', + 'middleware-http', + 'job-poller', + 'webhook-deliver', + 'archive-aggregator', + 'archive-worm', + 'tree-node', + 'zapier-simulator', + 'zapier-platform-app', + ]) { + assert.ok(ids.includes(need), `missing ${need}`); + } + const tree = loaded.services['tree-node']; + assert.equal(tree.min, 3); + assert.equal(tree.keepFloor, true); + assert.equal(tree.configPath, 'services/tree-node.json'); + assert.equal(loaded.fleetPath, 'fleet.json'); + }); + + it('starts the tree-node floor (3 available copies)', async () => { + const sup = new Supervisor({ loaded: treeOnly(14600) }); + supervisors.push(sup); + await sup.startService('tree-node'); + const st = sup.status().services['tree-node']; + assert.equal(st.available, 3); + assert.equal(st.belowFloor, false); + assert.equal(st.instances.length, 3); + }); + + it('pauses one tree-node and respawns so min available stays 3', async () => { + const sup = new Supervisor({ loaded: treeOnly(14700) }); + supervisors.push(sup); + await sup.startService('tree-node'); + await sup.pauseInstance('tree-node-0'); + const st = sup.status().services['tree-node']; + assert.ok(st.available >= 3, `available ${st.available}`); + assert.ok(st.paused >= 1); + assert.ok(st.running >= 4); + assert.equal(st.belowFloor, false); + }); + + it('restarts an unhealthy replica and keeps the floor', async () => { + const sup = new Supervisor({ loaded: treeOnly(14800) }); + supervisors.push(sup); + await sup.startService('tree-node'); + const before = sup.instances.get('tree-node-1').pid; + await sup.markUnhealthy('tree-node-1'); + const mon = new Monitor(sup, { intervalMs: 50 }); + await mon.tick(); + const rec = sup.instances.get('tree-node-1'); + assert.ok(rec.pid); + assert.notEqual(rec.pid, before); + assert.equal(rec.healthy, true); + assert.equal(sup.status().services['tree-node'].available, 3); + }); + + it('stop+replace of a replica keeps min tree-nodes available', async () => { + const sup = new Supervisor({ loaded: treeOnly(14900) }); + supervisors.push(sup); + await sup.startService('tree-node'); + await sup.stopInstance('tree-node-2', { replace: true }); + const st = sup.status().services['tree-node']; + 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/test/rtt.test.js b/test/rtt.test.js new file mode 100644 index 0000000..def37cf --- /dev/null +++ b/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'); + }); +}); diff --git a/test/ssh.test.js b/test/ssh.test.js new file mode 100644 index 0000000..624bd74 --- /dev/null +++ b/test/ssh.test.js @@ -0,0 +1,56 @@ +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import { expandHome, shellQuote, sshBaseArgs, sshTarget } from '../src/ssh.js'; +import { normalizeMachine, pickMachine } from '../src/machines.js'; + +describe('SSH remote machine config', () => { + it('expands ~ in identityFile and quotes remote commands', () => { + assert.equal(expandHome('~/foo'), `${os.homedir()}/foo`); + assert.equal(shellQuote("a'b"), `'a'"'"'b'`); + }); + + it('builds ssh argv with user, host, port, and key path (not key bytes)', () => { + const key = path.join(os.tmpdir(), 'verae-fleet-test-id'); + fs.writeFileSync(key, 'not-a-real-key\n', { mode: 0o600 }); + const m = normalizeMachine({ + id: 'ns1', + kind: 'ssh', + host: '70.88.205.138', + user: 'marchon', + sshPort: 22, + identityFile: key, + }); + assert.equal(m.kind, 'ssh'); + assert.equal(sshTarget(m), 'marchon@70.88.205.138'); + const args = sshBaseArgs(m); + assert.ok(args.includes('-i')); + assert.ok(args.includes(key)); + assert.ok(!args.some((a) => a.includes('BEGIN'))); + assert.equal(m.user, 'marchon'); + fs.unlinkSync(key); + }); + + it('places load on ssh hosts when they have free capacity', () => { + const machines = [ + normalizeMachine({ id: 'local', kind: 'local', host: '127.0.0.1', capacity: 2, roles: ['*'] }), + normalizeMachine({ + id: 'ns1', + kind: 'ssh', + host: '70.88.205.138', + user: 'marchon', + identityFile: '~/.ssh/id_ed25519', + capacity: 8, + roles: ['tree-node'], + }), + ]; + const instances = [ + { machine: 'local', pid: 1 }, + { machine: 'local', pid: 2 }, + ]; + const pick = pickMachine(machines, instances, 'tree-node', 'tree-node'); + assert.equal(pick.id, 'ns1'); + }); +});