Add fleet control: service configs, replica floors, monitor, pause/restart
Some checks are pending
offline / test (push) Waiting to run
Some checks are pending
offline / test (push) Waiting to run
Central fleet.json sets min/max copies. Tree-node keepFloor respawns until three healthy unpaused replicas remain. CLI and loopback UI pause, resume, stop, and restart instances that fail health checks.
This commit is contained in:
parent
f3dc0e6eee
commit
dd99ce1c64
33 changed files with 1368 additions and 0 deletions
129
packages/verae-fleet/src/cli.js
Normal file
129
packages/verae-fleet/src/cli.js
Normal file
|
|
@ -0,0 +1,129 @@
|
|||
#!/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 <service> turn on, spawn up to min
|
||||
stop <service|instance> turn off (service disable) or stop one replica
|
||||
pause <service|instance> pause (does not count toward tree-node floor)
|
||||
resume <service|instance>
|
||||
restart <instance> kill + respawn
|
||||
reconcile force floor check
|
||||
|
||||
Central spec: fleet.json
|
||||
Per service: services/<id>.json
|
||||
`);
|
||||
}
|
||||
|
||||
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}`);
|
||||
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;
|
||||
});
|
||||
90
packages/verae-fleet/src/health.js
Normal file
90
packages/verae-fleet/src/health.js
Normal file
|
|
@ -0,0 +1,90 @@
|
|||
/**
|
||||
* 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';
|
||||
|
||||
export function httpProbe(port, pathname = '/health', timeoutMs = 800) {
|
||||
return new Promise((resolve) => {
|
||||
const req = http.get(
|
||||
{ host: '127.0.0.1', port, path: pathname, timeout: timeoutMs },
|
||||
(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'));
|
||||
});
|
||||
}
|
||||
|
||||
export function postControl(port, pathname) {
|
||||
return new Promise((resolve, reject) => {
|
||||
const req = http.request(
|
||||
{ host: '127.0.0.1', port, path: pathname, method: 'POST', timeout: 800 },
|
||||
(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);
|
||||
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 };
|
||||
}
|
||||
}
|
||||
87
packages/verae-fleet/src/load.js
Normal file
87
packages/verae-fleet/src/load.js
Normal file
|
|
@ -0,0 +1,87 @@
|
|||
/**
|
||||
* 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';
|
||||
|
||||
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) {
|
||||
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,
|
||||
};
|
||||
}
|
||||
return {
|
||||
root,
|
||||
fleetPath: path.relative(root, fleetPath) || 'fleet.json',
|
||||
control: fleet.control,
|
||||
nats: fleet.nats,
|
||||
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;
|
||||
}
|
||||
38
packages/verae-fleet/src/monitor.js
Normal file
38
packages/verae-fleet/src/monitor.js
Normal file
|
|
@ -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;
|
||||
}
|
||||
}
|
||||
85
packages/verae-fleet/src/server.js
Normal file
85
packages/verae-fleet/src/server.js
Normal file
|
|
@ -0,0 +1,85 @@
|
|||
/**
|
||||
* 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 === '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;
|
||||
}
|
||||
323
packages/verae-fleet/src/supervisor.js
Normal file
323
packages/verae-fleet/src/supervisor.js
Normal file
|
|
@ -0,0 +1,323 @@
|
|||
/**
|
||||
* 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';
|
||||
|
||||
const WORKER = path.join(path.dirname(fileURLToPath(import.meta.url)), 'worker.js');
|
||||
|
||||
export class Supervisor {
|
||||
/**
|
||||
* @param {object} [opts]
|
||||
* @param {ReturnType<typeof loadFleet>} [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<string, object>} */
|
||||
this.instances = new Map();
|
||||
this.events = [];
|
||||
this._seq = 0;
|
||||
}
|
||||
|
||||
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 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 rec = {
|
||||
id: instance,
|
||||
service: serviceId,
|
||||
index,
|
||||
pid: child.pid,
|
||||
healthPort,
|
||||
stateDir,
|
||||
child,
|
||||
paused: false,
|
||||
startedAt: Date.now(),
|
||||
restarts: 0,
|
||||
lastError: null,
|
||||
};
|
||||
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);
|
||||
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.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 postControl(rec.healthPort, '/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 postControl(rec.healthPort, '/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 } = rec;
|
||||
rec.child?.kill('SIGTERM');
|
||||
await waitExit(rec.child, 2000);
|
||||
this.instances.delete(instanceId);
|
||||
const next = await this.startOne(service, { index });
|
||||
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 postControl(rec.healthPort, '/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);
|
||||
rec.paused = Boolean(r.paused);
|
||||
rec.lastProbe = r;
|
||||
rec.healthy = r.ok;
|
||||
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,
|
||||
healthPort: i.healthPort,
|
||||
paused: i.paused,
|
||||
healthy: i.healthy,
|
||||
restarts: i.restarts,
|
||||
})),
|
||||
};
|
||||
}
|
||||
return {
|
||||
control: this.loaded.control,
|
||||
nats: this.loaded.nats,
|
||||
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(port, ms) {
|
||||
const deadline = Date.now() + ms;
|
||||
while (Date.now() < deadline) {
|
||||
const r = await httpProbe(port, '/health', 300);
|
||||
if (r.ok || r.statusCode === 503) return r;
|
||||
await new Promise((r) => setTimeout(r, 50));
|
||||
}
|
||||
return { ok: false, error: 'start-timeout' };
|
||||
}
|
||||
113
packages/verae-fleet/src/worker.js
Normal file
113
packages/verae-fleet/src/worker.js
Normal file
|
|
@ -0,0 +1,113 @@
|
|||
#!/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 { fileURLToPath } from 'node:url';
|
||||
|
||||
const role = process.env.FLEET_ROLE || 'unknown';
|
||||
const service = process.env.FLEET_SERVICE || role;
|
||||
const instance = process.env.FLEET_INSTANCE || `${service}-0`;
|
||||
const port = Number(process.env.FLEET_HEALTH_PORT || 0);
|
||||
const stateDir = process.env.FLEET_STATE_DIR || path.join(process.cwd(), 'data', instance);
|
||||
const startedAt = new Date().toISOString();
|
||||
|
||||
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 snapshot() {
|
||||
return {
|
||||
ok: !paused && !forceUnhealthy,
|
||||
paused,
|
||||
forceUnhealthy,
|
||||
role,
|
||||
service,
|
||||
instance,
|
||||
port,
|
||||
pid: process.pid,
|
||||
startedAt,
|
||||
puts,
|
||||
archiveId: archive?.archiveId || null,
|
||||
};
|
||||
}
|
||||
|
||||
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());
|
||||
}
|
||||
json(404, { error: 'not found' });
|
||||
});
|
||||
|
||||
server.listen(port, '127.0.0.1', () => {
|
||||
beat();
|
||||
process.stdout.write(`fleet-worker ${instance} health http://127.0.0.1:${port}/health\n`);
|
||||
});
|
||||
|
||||
const iv = setInterval(beat, 400);
|
||||
function shutdown() {
|
||||
clearInterval(iv);
|
||||
try {
|
||||
server.close();
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
process.exit(0);
|
||||
}
|
||||
process.on('SIGTERM', shutdown);
|
||||
process.on('SIGINT', shutdown);
|
||||
Loading…
Add table
Add a link
Reference in a new issue