gsadmin/headplane
1
0
Fork 0
mirror of https://github.com/tale/headplane.git synced 2026-08-20 18:02:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #477 from drifterza/fix/issue-427-hostinfo-update-frequency

Browse Source
This commit is contained in:
Aarnav Tale
2026-03-07 14:51:15 -05:00
committed by GitHub
parent 432cd29cd0 779103147d
commit cb97b73abd
2 changed files with 382 additions and 349 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/server/hp-agent.ts
+321 -349
View File
@@ -1,179 +1,155 @@
import { ChildProcessWithoutNullStreams, spawn } from 'node:child_process'; import { inArray } from "drizzle-orm";
import EventEmitter from 'node:events'; import { LibSQLDatabase } from "drizzle-orm/libsql/driver-core";
import { access, constants, mkdir, open } from 'node:fs/promises'; import { ChildProcessWithoutNullStreams, spawn } from "node:child_process";
import { getegid, geteuid } from 'node:process'; import EventEmitter from "node:events";
import { createInterface, Interface } from 'node:readline'; import { access, constants, mkdir, open } from "node:fs/promises";
import { inArray } from 'drizzle-orm'; import { getegid, geteuid } from "node:process";
import { LibSQLDatabase } from 'drizzle-orm/libsql/driver-core'; import { createInterface, Interface } from "node:readline";
import { HostInfo } from '~/types';
import log from '~/utils/log'; import { HostInfo } from "~/types";
import { HeadplaneConfig } from './config/config-schema'; import log from "~/utils/log";
import { hostInfo } from './db/schema';
import { HeadplaneConfig } from "./config/config-schema";
import { hostInfo } from "./db/schema";
export async function createHeadplaneAgent( export async function createHeadplaneAgent(
config: NonNullable<HeadplaneConfig['integration']>['agent'] | undefined, config: NonNullable<HeadplaneConfig["integration"]>["agent"] | undefined,
headscaleUrl: string, headscaleUrl: string,
db: LibSQLDatabase, db: LibSQLDatabase,
) { ) {
if (!config?.enabled) { if (!config?.enabled) {
return; return;
} }
if (!config.pre_authkey) { if (!config.pre_authkey) {
log.error('agent', 'Agent `pre_authkey` is not set'); log.error("agent", "Agent `pre_authkey` is not set");
log.warn('agent', 'The agent will not run until resolved'); log.warn("agent", "The agent will not run until resolved");
return; return;
} }
try { try {
await access(config.work_dir, constants.R_OK | constants.W_OK); await access(config.work_dir, constants.R_OK | constants.W_OK);
log.debug('config', 'Using agent work dir at %s', config.work_dir); log.debug("config", "Using agent work dir at %s", config.work_dir);
} catch (error) { } catch (error) {
// Try to create the directory just in case // Try to create the directory just in case
try { try {
await mkdir(config.work_dir, { recursive: true }); await mkdir(config.work_dir, { recursive: true });
log.debug('config', 'Created agent work dir at %s', config.work_dir); log.debug("config", "Created agent work dir at %s", config.work_dir);
log.info( log.info("config", "Created missing agent work dir at %s", config.work_dir);
'config',
'Created missing agent work dir at %s',
config.work_dir,
);
return; return;
} catch (innerError) { } catch (innerError) {
log.error( log.error("config", "Failed to create agent work dir at %s", config.work_dir);
'config', log.info("config", "Agent work dir not accessible at %s", config.work_dir);
'Failed to create agent work dir at %s', log.debug("config", "Error details: %s", error);
config.work_dir, log.debug("config", "Create error details: %s", innerError);
); return;
log.info( }
'config', }
'Agent work dir not accessible at %s',
config.work_dir,
);
log.debug('config', 'Error details: %s', error);
log.debug('config', 'Create error details: %s', innerError);
return;
}
}
try { try {
const handle = await open(config.cache_path, 'a+'); const handle = await open(config.cache_path, "a+");
log.info('agent', 'Using agent cache file at %s', config.cache_path); log.info("agent", "Using agent cache file at %s", config.cache_path);
await handle.close(); await handle.close();
} catch (error) { } catch (error) {
log.info( log.info("agent", "Agent cache file not accessible at %s", config.cache_path);
'agent', log.debug("agent", "Error details: %s", error);
'Agent cache file not accessible at %s', return;
config.cache_path, }
);
log.debug('agent', 'Error details: %s', error);
return;
}
const agent = new HeadplaneAgent({ const agent = new HeadplaneAgent({
...config, ...config,
headscaleUrl, headscaleUrl,
}); });
agent.on('spawn', () => { agent.on("spawn", () => {
log.info('agent', 'Headplane agent started'); log.info("agent", "Headplane agent started");
}); });
agent.on('ready', () => { agent.on("ready", () => {
log.info('agent', 'Headplane agent is ready and serving queries'); log.info("agent", "Headplane agent is ready and serving queries");
}); });
agent.on('error', (err) => { agent.on("error", (err) => {
log.warn('agent', 'Headplane agent experienced an error: %s', err.message); log.warn("agent", "Headplane agent experienced an error: %s", err.message);
log.debug('agent', 'Error details: %o', err); log.debug("agent", "Error details: %o", err);
}); });
agent.on('exit', ({ code, signal }) => { agent.on("exit", ({ code, signal }) => {
log.warn( log.warn("agent", "Headplane agent exited with code %s and signal %s", code, signal);
'agent', });
'Headplane agent exited with code %s and signal %s',
code,
signal,
);
});
agent.on('restart', ({ delay, attempt }) => { agent.on("restart", ({ delay, attempt }) => {
log.warn( log.warn(
'agent', "agent",
'Headplane agent will restart in %f seconds (attempt %d)', "Headplane agent will restart in %f seconds (attempt %d)",
delay / 1000, delay / 1000,
attempt, attempt,
); );
}); });
agent.on('stderr', (data) => { agent.on("stderr", (data) => {
log.error('agent', 'Headplane agent stderr:', data); log.error("agent", "Headplane agent stderr:", data);
}); });
agent.on('info', async ({ id, info }) => { agent.on("info", async ({ id, info }) => {
log.debug('agent', 'Received HostInfo for %s', id); log.debug("agent", "Received HostInfo for %s", id);
try { try {
const parsedInfo = JSON.parse(info) as HostInfo; const parsedInfo = JSON.parse(info) as HostInfo;
await db await db
.insert(hostInfo) .insert(hostInfo)
.values({ .values({
host_id: id, host_id: id,
payload: parsedInfo, payload: parsedInfo,
updated_at: new Date(), updated_at: new Date(),
}) })
.onConflictDoUpdate({ .onConflictDoUpdate({
target: hostInfo.host_id, target: hostInfo.host_id,
set: { set: {
payload: parsedInfo, payload: parsedInfo,
updated_at: new Date(), updated_at: new Date(),
}, },
}); });
} catch (error) { } catch (error) {
log.error( log.error(
'agent', "agent",
'Failed to parse HostInfo for %s: %s', "Failed to parse HostInfo for %s: %s",
id, id,
error instanceof Error ? error.message : String(error), error instanceof Error ? error.message : String(error),
); );
return; return;
} }
}); });
agent.start(); agent.start();
process.on('SIGTERM', () => agent.shutdown()); process.on("SIGTERM", () => agent.shutdown());
process.on('SIGINT', () => agent.shutdown()); process.on("SIGINT", () => agent.shutdown());
return { return {
agentID: () => agent.agentID(), agentID: () => agent.agentID(),
lookup: async (nodes: string[]) => { lookup: async (nodes: string[]) => {
const results = await db const results = await db.select().from(hostInfo).where(inArray(hostInfo.host_id, nodes));
.select()
.from(hostInfo)
.where(inArray(hostInfo.host_id, nodes));
return Object.fromEntries( return Object.fromEntries(
results.filter((r) => r.payload).map((r) => [r.host_id, r.payload]), results.filter((r) => r.payload).map((r) => [r.host_id, r.payload]),
) as Record<string, HostInfo>; ) as Record<string, HostInfo>;
}, },
}; };
} }
type AgentOptions = NonNullable< type AgentOptions = NonNullable<NonNullable<HeadplaneConfig["integration"]>["agent"]> & {
NonNullable<HeadplaneConfig['integration']>['agent'] headscaleUrl: string;
> & {
headscaleUrl: string;
}; };
interface AgentEvents { interface AgentEvents {
ready: []; ready: [];
spawn: []; spawn: [];
error: [Error]; error: [Error];
exit: [{ code?: number; signal?: NodeJS.Signals }]; exit: [{ code?: number; signal?: NodeJS.Signals }];
restart: [{ delay: number; attempt: number }]; restart: [{ delay: number; attempt: number }];
stderr: [string]; stderr: [string];
info: [{ id: string; info: string }]; info: [{ id: string; info: string }];
} }
/** /**
@@ -184,237 +160,233 @@ interface AgentEvents {
* - The agent is restarted on a backoff strategy * - The agent is restarted on a backoff strategy
*/ */
class HeadplaneAgent extends EventEmitter<AgentEvents> { class HeadplaneAgent extends EventEmitter<AgentEvents> {
private child?: ChildProcessWithoutNullStreams; private child?: ChildProcessWithoutNullStreams;
private readline?: Interface; private readline?: Interface;
private options: AgentOptions; private options: AgentOptions;
private hbInterval?: NodeJS.Timeout; private hbInterval?: NodeJS.Timeout;
private hbDeadline?: NodeJS.Timeout; private hbDeadline?: NodeJS.Timeout;
private restartTimer?: NodeJS.Timeout; private restartTimer?: NodeJS.Timeout;
private isWaitingForAck = false; private refreshInterval?: NodeJS.Timeout;
private isShuttingDown = false; private isWaitingForAck = false;
private backoffAttempt = 0; private isShuttingDown = false;
private agentId?: string; private backoffAttempt = 0;
private agentId?: string;
private BASE_BACKOFF_MS = 1.5 * 1000; // 1.5 seconds private BASE_BACKOFF_MS = 1.5 * 1000; // 1.5 seconds
private MAX_BACKOFF_MS = 30 * 1000; // 30 seconds private MAX_BACKOFF_MS = 30 * 1000; // 30 seconds
private PROBE_COOLDOWN_MS = 5 * 60_000; // 5 minutes private PROBE_COOLDOWN_MS = 5 * 60_000; // 5 minutes
private PROBE_ATTEMPT_INTERVAL = 10; // Every 10th attempt private PROBE_ATTEMPT_INTERVAL = 10; // Every 10th attempt
private HEARTBEAT_INTERVAL_MS = 5 * 1000; // 5 seconds private HEARTBEAT_INTERVAL_MS = 5 * 1000; // 5 seconds
private HEARTBEAT_TIMEOUT_MS = 3 * 1000; // 3 seconds private HEARTBEAT_TIMEOUT_MS = 3 * 1000; // 3 seconds
constructor(options: AgentOptions) { constructor(options: AgentOptions) {
super(); super();
this.options = options; this.options = options;
} }
agentID() { agentID() {
return this.agentId; return this.agentId;
} }
start() { start() {
this.isShuttingDown = false; this.isShuttingDown = false;
this.spawnInternalChild(); this.spawnInternalChild();
} }
shutdown() { shutdown() {
this.isShuttingDown = true; this.isShuttingDown = true;
this.agentId = undefined; this.agentId = undefined;
clearTimeout(this.restartTimer); clearTimeout(this.restartTimer);
clearInterval(this.hbInterval); clearInterval(this.hbInterval);
clearTimeout(this.hbDeadline); clearInterval(this.refreshInterval);
this.isWaitingForAck = false; clearTimeout(this.hbDeadline);
this.isWaitingForAck = false;
this.send('SHUTDOWN'); this.send("SHUTDOWN");
this.child?.kill('SIGTERM'); this.child?.kill("SIGTERM");
this.readline?.close(); this.readline?.close();
} }
private spawnInternalChild() { private spawnInternalChild() {
this.child = spawn(this.options.executable_path, { this.child = spawn(this.options.executable_path, {
stdio: ['pipe', 'pipe', 'pipe'], stdio: ["pipe", "pipe", "pipe"],
uid: geteuid?.() ?? undefined, uid: geteuid?.() ?? undefined,
gid: getegid?.() ?? undefined, gid: getegid?.() ?? undefined,
env: { env: {
HOME: process.env.HOME, HOME: process.env.HOME,
HEADPLANE_AGENT_WORK_DIR: this.options.work_dir, HEADPLANE_AGENT_WORK_DIR: this.options.work_dir,
HEADPLANE_AGENT_DEBUG: log.debugEnabled ? 'true' : 'false', HEADPLANE_AGENT_DEBUG: log.debugEnabled ? "true" : "false",
HEADPLANE_AGENT_HOSTNAME: this.options.host_name, HEADPLANE_AGENT_HOSTNAME: this.options.host_name,
HEADPLANE_AGENT_TS_SERVER: this.options.headscaleUrl, HEADPLANE_AGENT_TS_SERVER: this.options.headscaleUrl,
HEADPLANE_AGENT_TS_AUTHKEY: this.options.pre_authkey, HEADPLANE_AGENT_TS_AUTHKEY: this.options.pre_authkey,
}, },
}); });
this.emit('spawn'); this.emit("spawn");
this.child.on('error', (err) => this.emit('error', err)); this.child.on("error", (err) => this.emit("error", err));
this.child.stderr.on('data', (data) => this.child.stderr.on("data", (data) => this.emit("stderr", data.toString()));
this.emit('stderr', data.toString()),
);
this.child.on('exit', (code, signal) => { this.child.on("exit", (code, signal) => {
this.agentId = undefined; this.agentId = undefined;
this.emit('exit', { this.emit("exit", {
code: code ?? undefined, code: code ?? undefined,
signal: signal ?? undefined, signal: signal ?? undefined,
}); });
this.readline?.close(); this.readline?.close();
clearInterval(this.hbInterval); clearInterval(this.hbInterval);
clearTimeout(this.hbDeadline); clearInterval(this.refreshInterval);
this.isWaitingForAck = false; clearTimeout(this.hbDeadline);
this.isWaitingForAck = false;
if (this.isShuttingDown) { if (this.isShuttingDown) {
log.info('agent', 'Child process exited gracefully'); log.info("agent", "Child process exited gracefully");
return; return;
} }
this.backoffAttempt++; this.backoffAttempt++;
const delay = this.calculateBackoff(); const delay = this.calculateBackoff();
this.emit('restart', { delay, attempt: this.backoffAttempt }); this.emit("restart", { delay, attempt: this.backoffAttempt });
this.restartTimer = setTimeout(() => this.spawnInternalChild(), delay); this.restartTimer = setTimeout(() => this.spawnInternalChild(), delay);
}); });
this.readline = createInterface({ input: this.child.stdout }); this.readline = createInterface({ input: this.child.stdout });
this.readline.on('line', (line) => this.readlineHandler(line)); this.readline.on("line", (line) => this.readlineHandler(line));
this.send('START'); this.send("START");
// Start the heartbeat loop with our custom interval // Start the heartbeat loop with our custom interval
this.hbInterval = setInterval(() => { this.hbInterval = setInterval(() => {
if (!this.child || this.child.killed) return; if (!this.child || this.child.killed) return;
// If we get here, we missed the last PONG response and can die // If we get here, we missed the last PONG response and can die
if (this.isWaitingForAck) { if (this.isWaitingForAck) {
this.agentId = undefined; this.agentId = undefined;
this.emit('error', new Error('Agent heartbeat missed')); this.emit("error", new Error("Agent heartbeat missed"));
this.child.kill('SIGTERM'); this.child.kill("SIGTERM");
return; return;
} }
this.isWaitingForAck = true; this.isWaitingForAck = true;
this.send('PING'); this.send("PING");
clearTimeout(this.hbDeadline); clearTimeout(this.hbDeadline);
this.hbDeadline = setTimeout(() => { this.hbDeadline = setTimeout(() => {
if (this.isWaitingForAck) { if (this.isWaitingForAck) {
this.agentId = undefined; this.agentId = undefined;
this.emit('error', new Error('Agent heartbeat timeout')); this.emit("error", new Error("Agent heartbeat timeout"));
this.child?.kill('SIGTERM'); this.child?.kill("SIGTERM");
} }
}, this.HEARTBEAT_TIMEOUT_MS); }, this.HEARTBEAT_TIMEOUT_MS);
}, this.HEARTBEAT_INTERVAL_MS); }, this.HEARTBEAT_INTERVAL_MS);
} }
private send(s: string) { private send(s: string) {
if (!this.child || this.child.killed) return; if (!this.child || this.child.killed) return;
const ok = this.child.stdin.write(`${s}\n`); const ok = this.child.stdin.write(`${s}\n`);
if (!ok) this.child.stdin.once('drain', () => {}); if (!ok) this.child.stdin.once("drain", () => {});
} }
/** /**
* Calculates a backoff time based on the current attempt. * Calculates a backoff time based on the current attempt.
* Supports a randomized jitter to avoid thundering herd problems. * Supports a randomized jitter to avoid thundering herd problems.
* *
* @param min The minimum backoff time in milliseconds. * @param min The minimum backoff time in milliseconds.
* @param max The maximum backoff time in milliseconds. * @param max The maximum backoff time in milliseconds.
* @returns The calculated backoff time in milliseconds. * @returns The calculated backoff time in milliseconds.
*/ */
private calculateBackoff() { private calculateBackoff() {
const attempt = this.backoffAttempt; const attempt = this.backoffAttempt;
if (attempt > 0 && attempt % this.PROBE_ATTEMPT_INTERVAL === 0) { if (attempt > 0 && attempt % this.PROBE_ATTEMPT_INTERVAL === 0) {
const jitter = Math.floor(Math.random() * (this.MAX_BACKOFF_MS + 1)); const jitter = Math.floor(Math.random() * (this.MAX_BACKOFF_MS + 1));
const sign = Math.random() < 0.5 ? -1 : 1; const sign = Math.random() < 0.5 ? -1 : 1;
return Math.max(0, this.PROBE_COOLDOWN_MS + jitter * sign); return Math.max(0, this.PROBE_COOLDOWN_MS + jitter * sign);
} }
const cap = Math.min( const cap = Math.min(this.MAX_BACKOFF_MS, this.BASE_BACKOFF_MS * 2 ** attempt);
this.MAX_BACKOFF_MS,
this.BASE_BACKOFF_MS * 2 ** attempt,
);
return Math.floor(Math.random() * (cap + 1)); return Math.floor(Math.random() * (cap + 1));
} }
/** /**
* Processes and dispatches the appropriate response based on the message. * Processes and dispatches the appropriate response based on the message.
* @param line The message to process (piped straight from readline) * @param line The message to process (piped straight from readline)
*/ */
private readlineHandler(line: string) { private readlineHandler(line: string) {
// When we are ready we force a refresh so that the UI has the most // When we are ready we force a refresh so that the UI has the most
// up-to-date information and will gracefully handle new info being sent // up-to-date information and will gracefully handle new info being sent
if (line.startsWith('READY')) { if (line.startsWith("READY")) {
this.backoffAttempt = 0; this.backoffAttempt = 0;
this.send('REFRESH'); this.send("REFRESH");
this.emit('ready'); this.emit("ready");
const agentId = line.slice(5).trim(); // Start periodic refresh using cache_ttl
if (this.agentId && this.agentId !== agentId) { clearInterval(this.refreshInterval);
log.warn( this.refreshInterval = setInterval(() => {
'agent', if (!this.child || this.child.killed) return;
'Agent ID changed from %s to %s', log.debug("agent", "Sending periodic REFRESH");
this.agentId, this.send("REFRESH");
agentId, }, this.options.cache_ttl);
);
}
this.agentId = agentId; const agentId = line.slice(5).trim();
return; if (this.agentId && this.agentId !== agentId) {
} log.warn("agent", "Agent ID changed from %s to %s", this.agentId, agentId);
}
if (line.startsWith('PONG')) { this.agentId = agentId;
this.isWaitingForAck = false; return;
clearTimeout(this.hbDeadline); }
const agentId = line.slice(5).trim(); if (line.startsWith("PONG")) {
if (this.agentId && this.agentId !== agentId) { this.isWaitingForAck = false;
log.warn( clearTimeout(this.hbDeadline);
'agent',
'Agent ID changed from %s to %s',
this.agentId,
agentId,
);
}
this.agentId = agentId; const agentId = line.slice(5).trim();
return; if (this.agentId && this.agentId !== agentId) {
} log.warn("agent", "Agent ID changed from %s to %s", this.agentId, agentId);
}
if (line.startsWith('HOSTINFO')) { this.agentId = agentId;
const data = line.slice(9).trim(); return;
const [id, ...infoParts] = data.split(' '); }
const info = infoParts.join(' ');
this.emit('info', { id, info });
return;
}
if (line.startsWith('ERROR')) { if (line.startsWith("HOSTINFO")) {
const error = line.slice(6).trim(); const data = line.slice(9).trim();
this.emit('error', new Error(error)); const [id, ...infoParts] = data.split(" ");
return; const info = infoParts.join(" ");
} this.emit("info", { id, info });
return;
}
if (line.startsWith('LOG')) { if (line.startsWith("ERROR")) {
const logSnippet = line.slice(4).trim(); const error = line.slice(6).trim();
const [level, ...messageParts] = logSnippet.split(' '); this.emit("error", new Error(error));
const message = messageParts.join(' '); return;
switch (level) { }
case 'INFO':
log.info('agent', message);
break;
case 'WARN':
log.warn('agent', message);
break;
case 'ERROR':
log.error('agent', message);
break;
default:
log.debug('agent', message);
}
return; if (line.startsWith("LOG")) {
} const logSnippet = line.slice(4).trim();
} const [level, ...messageParts] = logSnippet.split(" ");
const message = messageParts.join(" ");
switch (level) {
case "INFO":
log.info("agent", message);
break;
case "WARN":
log.warn("agent", message);
break;
case "ERROR":
log.error("agent", message);
break;
default:
log.debug("agent", message);
}
return;
}
}
} }
tests/unit/agent/refresh-interval.test.ts
+61
View File
@@ -0,0 +1,61 @@
import { describe, expect, test, vi, beforeEach, afterEach } from "vitest";
describe("Agent refresh interval", () => {
beforeEach(() => {
vi.useFakeTimers();
});
afterEach(() => {
vi.useRealTimers();
});
test("setInterval is called with cache_ttl value", () => {
const setIntervalSpy = vi.spyOn(global, "setInterval");
const cache_ttl = 180000; // 3 minutes
// Simulate what the agent does when starting refresh
const refreshInterval = setInterval(() => {
// refresh logic
}, cache_ttl);
expect(setIntervalSpy).toHaveBeenCalledWith(expect.any(Function), cache_ttl);
clearInterval(refreshInterval);
setIntervalSpy.mockRestore();
});
test("interval triggers callback at expected times", () => {
const callback = vi.fn();
const cache_ttl = 60000; // 1 minute for test
const interval = setInterval(callback, cache_ttl);
expect(callback).not.toHaveBeenCalled();
vi.advanceTimersByTime(cache_ttl);
expect(callback).toHaveBeenCalledTimes(1);
vi.advanceTimersByTime(cache_ttl);
expect(callback).toHaveBeenCalledTimes(2);
vi.advanceTimersByTime(cache_ttl);
expect(callback).toHaveBeenCalledTimes(3);
clearInterval(interval);
});
test("clearInterval stops the refresh", () => {
const callback = vi.fn();
const cache_ttl = 60000;
const interval = setInterval(callback, cache_ttl);
vi.advanceTimersByTime(cache_ttl);
expect(callback).toHaveBeenCalledTimes(1);
clearInterval(interval);
vi.advanceTimersByTime(cache_ttl * 5);
expect(callback).toHaveBeenCalledTimes(1); // still 1, not called again
});
});