mirror of
https://github.com/tale/headplane.git
synced 2026-07-26 07:48:14 +00:00
feat: make api calls more resilient and stuff
This commit is contained in:
+1
-1
@@ -45,7 +45,7 @@ export async function loader({ request, context }: Route.LoaderArgs) {
|
||||
: { name: principal.displayName, subject: "api_key" };
|
||||
|
||||
// MARK: The session should stay valid if Headscale isn't healthy
|
||||
const isHealthy = await api.isHealthy();
|
||||
const isHealthy = await context.headscale.health();
|
||||
if (isHealthy) {
|
||||
try {
|
||||
await api.apiKeys.list();
|
||||
|
||||
@@ -33,6 +33,8 @@ export async function loginAction({ request, context }: Route.LoaderArgs) {
|
||||
};
|
||||
}
|
||||
|
||||
// Build a client with the candidate API key the user just submitted, so the
|
||||
// GET /api/v1/apikey call below validates the key against Headscale itself.
|
||||
const api = context.headscale.client(apiKey);
|
||||
try {
|
||||
const apiKeys = await api.apiKeys.list();
|
||||
|
||||
@@ -56,6 +56,9 @@ export async function loader({ request, context }: Route.LoaderArgs) {
|
||||
});
|
||||
|
||||
try {
|
||||
// Looks up the Headscale user that matches this OIDC identity. We use
|
||||
// the configured admin API key here — not a per-request one — because
|
||||
// there is no per-request key yet (the session is being created).
|
||||
const hsApi = context.headscale.client(context.headscaleApiKey!);
|
||||
const hsUsers = await hsApi.users.list();
|
||||
const hsUser = findHeadscaleUserBySubject(hsUsers, identity.subject, identity.email);
|
||||
|
||||
@@ -16,9 +16,6 @@ export async function dnsAction({ request, context }: Route.ActionArgs) {
|
||||
return data({ success: false }, 403);
|
||||
}
|
||||
|
||||
// We only need it for health checks which don't require auth
|
||||
const api = context.headscale.client("fake-api-key");
|
||||
|
||||
const formData = await request.formData();
|
||||
const action = formData.get("action_id")?.toString();
|
||||
if (!action) {
|
||||
@@ -39,7 +36,7 @@ export async function dnsAction({ request, context }: Route.ActionArgs) {
|
||||
},
|
||||
]);
|
||||
|
||||
await context.integration?.onConfigChange(api);
|
||||
await context.integration?.onConfigChange(context.headscale);
|
||||
return { message: "Tailnet renamed successfully" };
|
||||
}
|
||||
case "toggle_magic": {
|
||||
@@ -55,7 +52,7 @@ export async function dnsAction({ request, context }: Route.ActionArgs) {
|
||||
},
|
||||
]);
|
||||
|
||||
await context.integration?.onConfigChange(api);
|
||||
await context.integration?.onConfigChange(context.headscale);
|
||||
return { message: "Magic DNS state updated successfully" };
|
||||
}
|
||||
case "remove_ns": {
|
||||
@@ -88,7 +85,7 @@ export async function dnsAction({ request, context }: Route.ActionArgs) {
|
||||
]);
|
||||
}
|
||||
|
||||
await context.integration?.onConfigChange(api);
|
||||
await context.integration?.onConfigChange(context.headscale);
|
||||
return { message: "Nameserver removed successfully" };
|
||||
}
|
||||
case "add_ns": {
|
||||
@@ -123,7 +120,7 @@ export async function dnsAction({ request, context }: Route.ActionArgs) {
|
||||
]);
|
||||
}
|
||||
|
||||
await context.integration?.onConfigChange(api);
|
||||
await context.integration?.onConfigChange(context.headscale);
|
||||
return { message: "Nameserver added successfully" };
|
||||
}
|
||||
case "remove_domain": {
|
||||
@@ -141,7 +138,7 @@ export async function dnsAction({ request, context }: Route.ActionArgs) {
|
||||
},
|
||||
]);
|
||||
|
||||
await context.integration?.onConfigChange(api);
|
||||
await context.integration?.onConfigChange(context.headscale);
|
||||
return { message: "Domain removed successfully" };
|
||||
}
|
||||
case "add_domain": {
|
||||
@@ -161,7 +158,7 @@ export async function dnsAction({ request, context }: Route.ActionArgs) {
|
||||
},
|
||||
]);
|
||||
|
||||
await context.integration?.onConfigChange(api);
|
||||
await context.integration?.onConfigChange(context.headscale);
|
||||
return { message: "Domain added successfully" };
|
||||
}
|
||||
case "remove_record": {
|
||||
@@ -183,7 +180,7 @@ export async function dnsAction({ request, context }: Route.ActionArgs) {
|
||||
return;
|
||||
}
|
||||
|
||||
await context.integration?.onConfigChange(api);
|
||||
await context.integration?.onConfigChange(context.headscale);
|
||||
return { message: "DNS record removed successfully" };
|
||||
}
|
||||
case "add_record": {
|
||||
@@ -205,7 +202,7 @@ export async function dnsAction({ request, context }: Route.ActionArgs) {
|
||||
return;
|
||||
}
|
||||
|
||||
await context.integration?.onConfigChange(api);
|
||||
await context.integration?.onConfigChange(context.headscale);
|
||||
return { message: "DNS record added successfully" };
|
||||
}
|
||||
case "override_dns": {
|
||||
@@ -222,7 +219,7 @@ export async function dnsAction({ request, context }: Route.ActionArgs) {
|
||||
},
|
||||
]);
|
||||
|
||||
await context.integration?.onConfigChange(api);
|
||||
await context.integration?.onConfigChange(context.headscale);
|
||||
return { message: "DNS override updated successfully" };
|
||||
}
|
||||
default:
|
||||
|
||||
@@ -28,8 +28,6 @@ export async function restrictionAction({ request, context }: Route.ActionArgs)
|
||||
});
|
||||
}
|
||||
|
||||
// We only need healthchecks which don't rely on an API key
|
||||
const api = context.headscale.client("fake-api-key");
|
||||
switch (action) {
|
||||
case "add_domain": {
|
||||
const domain = formData.get("domain")?.toString()?.trim();
|
||||
@@ -48,7 +46,7 @@ export async function restrictionAction({ request, context }: Route.ActionArgs)
|
||||
},
|
||||
]);
|
||||
|
||||
context.integration?.onConfigChange(api);
|
||||
context.integration?.onConfigChange(context.headscale);
|
||||
return data("Domain added successfully.");
|
||||
}
|
||||
|
||||
@@ -76,7 +74,7 @@ export async function restrictionAction({ request, context }: Route.ActionArgs)
|
||||
value: domains,
|
||||
},
|
||||
]);
|
||||
context.integration?.onConfigChange(api);
|
||||
context.integration?.onConfigChange(context.headscale);
|
||||
return data("Domain removed successfully.");
|
||||
}
|
||||
|
||||
@@ -97,7 +95,7 @@ export async function restrictionAction({ request, context }: Route.ActionArgs)
|
||||
},
|
||||
]);
|
||||
|
||||
context.integration?.onConfigChange(api);
|
||||
context.integration?.onConfigChange(context.headscale);
|
||||
return data("Group added successfully.");
|
||||
}
|
||||
|
||||
@@ -126,7 +124,7 @@ export async function restrictionAction({ request, context }: Route.ActionArgs)
|
||||
},
|
||||
]);
|
||||
|
||||
context.integration?.onConfigChange(api);
|
||||
context.integration?.onConfigChange(context.headscale);
|
||||
return data("Group removed successfully.");
|
||||
}
|
||||
|
||||
@@ -147,7 +145,7 @@ export async function restrictionAction({ request, context }: Route.ActionArgs)
|
||||
},
|
||||
]);
|
||||
|
||||
context.integration?.onConfigChange(api);
|
||||
context.integration?.onConfigChange(context.headscale);
|
||||
return data("User added successfully.");
|
||||
}
|
||||
|
||||
@@ -176,7 +174,7 @@ export async function restrictionAction({ request, context }: Route.ActionArgs)
|
||||
},
|
||||
]);
|
||||
|
||||
context.integration?.onConfigChange(api);
|
||||
context.integration?.onConfigChange(context.headscale);
|
||||
return data("User removed successfully.");
|
||||
}
|
||||
|
||||
|
||||
@@ -1,9 +1,7 @@
|
||||
import type { Route } from "./+types/healthz";
|
||||
|
||||
export async function loader({ context }: Route.LoaderArgs) {
|
||||
// Use a fake API key for healthcheck
|
||||
const api = context.headscale.client("fake-api-key");
|
||||
const healthy = await api.isHealthy();
|
||||
const healthy = await context.headscale.health();
|
||||
|
||||
return new Response(JSON.stringify({ status: healthy ? "OK" : "ERROR" }), {
|
||||
status: healthy ? 200 : 500,
|
||||
|
||||
@@ -34,9 +34,7 @@ export async function loader({ request, context }: Route.LoaderArgs) {
|
||||
);
|
||||
}
|
||||
|
||||
// Use a fake API key for healthcheck
|
||||
const api = context.headscale.client("fake-api-key");
|
||||
const healthy = await api.isHealthy();
|
||||
const healthy = await context.headscale.health();
|
||||
|
||||
const body = {
|
||||
status: healthy ? "healthy" : "unhealthy",
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import type { HeadscaleClient } from "~/server/headscale/api";
|
||||
import type { Headscale } from "~/server/headscale/api";
|
||||
|
||||
export abstract class Integration<T> {
|
||||
protected context: NonNullable<T>;
|
||||
@@ -11,6 +11,6 @@ export abstract class Integration<T> {
|
||||
}
|
||||
|
||||
abstract isAvailable(): Promise<boolean> | boolean;
|
||||
abstract onConfigChange(client: HeadscaleClient): Promise<void> | void;
|
||||
abstract onConfigChange(headscale: Headscale): Promise<void> | void;
|
||||
abstract get name(): string;
|
||||
}
|
||||
|
||||
@@ -4,7 +4,7 @@ import { setTimeout } from "node:timers/promises";
|
||||
import { type } from "arktype";
|
||||
import { Client } from "undici";
|
||||
|
||||
import type { HeadscaleClient } from "~/server/headscale/api";
|
||||
import type { Headscale } from "~/server/headscale/api";
|
||||
import log from "~/utils/log";
|
||||
|
||||
import { Integration } from "./abstract";
|
||||
@@ -255,7 +255,7 @@ export default class DockerIntegration extends Integration<typeof configSchema.f
|
||||
return this.client !== undefined && this.containerId !== undefined;
|
||||
}
|
||||
|
||||
async onConfigChange(client: HeadscaleClient) {
|
||||
async onConfigChange(headscale: Headscale) {
|
||||
if (!this.client) {
|
||||
return;
|
||||
}
|
||||
@@ -290,7 +290,7 @@ export default class DockerIntegration extends Integration<typeof configSchema.f
|
||||
while (attempts <= this.maxAttempts) {
|
||||
try {
|
||||
log.debug("config", "Checking Headscale status (attempt %d)", attempts);
|
||||
const status = await client.isHealthy();
|
||||
const status = await headscale.health();
|
||||
if (status === false) {
|
||||
throw new Error("Headscale is not running");
|
||||
}
|
||||
|
||||
@@ -5,7 +5,7 @@ import { join } from "node:path";
|
||||
import { CoreV1Api, KubeConfig } from "@kubernetes/client-node";
|
||||
import { type } from "arktype";
|
||||
|
||||
import type { HeadscaleClient } from "~/server/headscale/api";
|
||||
import type { Headscale } from "~/server/headscale/api";
|
||||
import log from "~/utils/log";
|
||||
|
||||
import { Integration } from "./abstract";
|
||||
@@ -154,12 +154,12 @@ export default class KubernetesIntegration extends Integration<typeof configSche
|
||||
}
|
||||
}
|
||||
|
||||
async onConfigChange(client: HeadscaleClient) {
|
||||
async onConfigChange(headscale: Headscale) {
|
||||
if (!this.pid) {
|
||||
return;
|
||||
}
|
||||
|
||||
await signalAndWaitHealthy(client, {
|
||||
await signalAndWaitHealthy(headscale, {
|
||||
pid: this.pid,
|
||||
signal: "SIGHUP",
|
||||
});
|
||||
|
||||
@@ -3,7 +3,7 @@ import { join } from "node:path";
|
||||
import { kill } from "node:process";
|
||||
import { setTimeout } from "node:timers/promises";
|
||||
|
||||
import type { HeadscaleClient } from "~/server/headscale/api";
|
||||
import type { Headscale } from "~/server/headscale/api";
|
||||
import log from "~/utils/log";
|
||||
|
||||
/**
|
||||
@@ -75,12 +75,12 @@ export interface SignalHeadscaleOptions {
|
||||
|
||||
/**
|
||||
* Sends a signal to the headscale process and waits for it to become healthy.
|
||||
* @param client The HeadscaleClient to check health
|
||||
* @param headscale The Headscale instance to health-check
|
||||
* @param options Options for signaling and waiting
|
||||
* @returns True if headscale became healthy, false otherwise
|
||||
*/
|
||||
export async function signalAndWaitHealthy(
|
||||
client: HeadscaleClient,
|
||||
headscale: Headscale,
|
||||
options: SignalHeadscaleOptions,
|
||||
): Promise<boolean> {
|
||||
const { pid, signal = "SIGHUP", maxAttempts = 10, retryDelayMs = 1000 } = options;
|
||||
@@ -96,7 +96,7 @@ export async function signalAndWaitHealthy(
|
||||
await setTimeout(retryDelayMs);
|
||||
for (let attempt = 1; attempt <= maxAttempts; attempt++) {
|
||||
try {
|
||||
const healthy = await client.isHealthy();
|
||||
const healthy = await headscale.health();
|
||||
if (healthy) {
|
||||
log.info("config", "Headscale is healthy after restart");
|
||||
return true;
|
||||
|
||||
@@ -2,7 +2,7 @@ import { platform } from "node:os";
|
||||
|
||||
import { type } from "arktype";
|
||||
|
||||
import type { HeadscaleClient } from "~/server/headscale/api";
|
||||
import type { Headscale } from "~/server/headscale/api";
|
||||
import log from "~/utils/log";
|
||||
|
||||
import { Integration } from "./abstract";
|
||||
@@ -51,12 +51,12 @@ export default class ProcIntegration extends Integration<typeof configSchema.ful
|
||||
}
|
||||
}
|
||||
|
||||
async onConfigChange(client: HeadscaleClient) {
|
||||
async onConfigChange(headscale: Headscale) {
|
||||
if (!this.pid) {
|
||||
return;
|
||||
}
|
||||
|
||||
await signalAndWaitHealthy(client, {
|
||||
await signalAndWaitHealthy(headscale, {
|
||||
pid: this.pid,
|
||||
signal: "SIGHUP",
|
||||
});
|
||||
|
||||
@@ -15,7 +15,7 @@ import { makePolicyApi, type PolicyApi } from "./resources/policy";
|
||||
import { makePreAuthKeyApi, type PreAuthKeyApi } from "./resources/pre-auth-keys";
|
||||
import { makeUserApi, type UserApi } from "./resources/users";
|
||||
import { formatServerVersion, parseServerVersion, type ServerVersion } from "./server-version";
|
||||
import { createTransport, type Transport } from "./transport";
|
||||
import { createTransport } from "./transport";
|
||||
|
||||
export interface Headscale {
|
||||
readonly version: ServerVersion;
|
||||
@@ -34,48 +34,84 @@ export interface HeadscaleClient {
|
||||
policy: PolicyApi;
|
||||
preAuthKeys: PreAuthKeyApi;
|
||||
apiKeys: ApiKeyApi;
|
||||
/**
|
||||
* Convenience passthrough to `Headscale.health()`. Headscale's
|
||||
* `/health` endpoint is unauthenticated so callers that already
|
||||
* have a client (loaders/actions) don't need to also reach for
|
||||
* the top-level `Headscale` value.
|
||||
*/
|
||||
isHealthy(): Promise<boolean>;
|
||||
}
|
||||
|
||||
export interface CreateHeadscaleOptions {
|
||||
url: string;
|
||||
certPath?: string;
|
||||
/**
|
||||
* How often to retry `/version` while Headscale is unreachable.
|
||||
* Defaults to 30 seconds. Exposed for tests.
|
||||
*/
|
||||
retryIntervalMs?: number;
|
||||
}
|
||||
|
||||
const DEFAULT_RETRY_INTERVAL_MS = 30_000;
|
||||
|
||||
export async function createHeadscale(opts: CreateHeadscaleOptions): Promise<Headscale> {
|
||||
const transport = await createTransport({ url: opts.url, certPath: opts.certPath });
|
||||
const retryIntervalMs = opts.retryIntervalMs ?? DEFAULT_RETRY_INTERVAL_MS;
|
||||
|
||||
const versionInfo = await transport.getPublic<{ version: string }>("/version");
|
||||
const version = parseServerVersion(versionInfo.version);
|
||||
const capabilities = capabilitiesFor(version);
|
||||
let version: ServerVersion = parseServerVersion("unreachable");
|
||||
let capabilities: Capabilities = capabilitiesFor(version);
|
||||
let detected = false;
|
||||
let retryTimer: ReturnType<typeof setTimeout> | undefined;
|
||||
let disposed = false;
|
||||
|
||||
if (version.unknown) {
|
||||
log.warn(
|
||||
"api",
|
||||
"Could not parse Headscale version %s, assuming newest known capabilities",
|
||||
versionInfo.version,
|
||||
);
|
||||
} else {
|
||||
log.info("api", "Connected to Headscale %s", formatServerVersion(version));
|
||||
async function detectOnce(): Promise<boolean> {
|
||||
try {
|
||||
const { version: raw } = await transport.getPublic<{ version: string }>("/version");
|
||||
const parsed = parseServerVersion(raw);
|
||||
version = parsed;
|
||||
capabilities = capabilitiesFor(parsed);
|
||||
detected = true;
|
||||
if (parsed.unknown) {
|
||||
log.warn(
|
||||
"api",
|
||||
"Could not parse Headscale version %s, assuming newest known capabilities",
|
||||
raw,
|
||||
);
|
||||
} else {
|
||||
log.info("api", "Connected to Headscale %s", formatServerVersion(parsed));
|
||||
}
|
||||
return true;
|
||||
} catch (error) {
|
||||
log.debug("api", "Headscale /version probe failed: %s", String(error));
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
return makeHeadscale(transport, version, capabilities);
|
||||
}
|
||||
function scheduleRetry() {
|
||||
if (disposed || detected) return;
|
||||
retryTimer = setTimeout(async () => {
|
||||
retryTimer = undefined;
|
||||
if (disposed) return;
|
||||
if (await detectOnce()) return;
|
||||
scheduleRetry();
|
||||
}, retryIntervalMs);
|
||||
// Don't keep the event loop alive on this timer alone — Headplane
|
||||
// should still shut down cleanly while we're waiting to retry.
|
||||
retryTimer.unref?.();
|
||||
}
|
||||
|
||||
if (!(await detectOnce())) {
|
||||
log.warn(
|
||||
"api",
|
||||
"Headscale unreachable at boot; defaulting to newest-known capabilities and retrying every %dms",
|
||||
retryIntervalMs,
|
||||
);
|
||||
scheduleRetry();
|
||||
}
|
||||
|
||||
function makeHeadscale(
|
||||
transport: Transport,
|
||||
version: ServerVersion,
|
||||
capabilities: Capabilities,
|
||||
): Headscale {
|
||||
return {
|
||||
version,
|
||||
capabilities,
|
||||
// Getters so callers always observe the latest detected values
|
||||
// without having to know about the retry loop.
|
||||
get version() {
|
||||
return version;
|
||||
},
|
||||
get capabilities() {
|
||||
return capabilities;
|
||||
},
|
||||
health: () => transport.health(),
|
||||
client(apiKey) {
|
||||
return {
|
||||
@@ -84,9 +120,15 @@ function makeHeadscale(
|
||||
policy: makePolicyApi(transport, capabilities, apiKey),
|
||||
preAuthKeys: makePreAuthKeyApi(transport, capabilities, apiKey),
|
||||
apiKeys: makeApiKeyApi(transport, capabilities, apiKey),
|
||||
isHealthy: () => transport.health(),
|
||||
};
|
||||
},
|
||||
dispose: () => transport.dispose(),
|
||||
async dispose() {
|
||||
disposed = true;
|
||||
if (retryTimer) {
|
||||
clearTimeout(retryTimer);
|
||||
retryTimer = undefined;
|
||||
}
|
||||
await transport.dispose();
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
@@ -208,6 +208,7 @@ export function startHttpServer(opts: StartOptions): Server {
|
||||
}
|
||||
process.exit(0);
|
||||
});
|
||||
|
||||
// Force exit if connections don't drain in time.
|
||||
setTimeout(() => process.exit(0), 5_000).unref();
|
||||
};
|
||||
|
||||
@@ -2,7 +2,7 @@ import { describe, expect, test } from "vitest";
|
||||
|
||||
import { gte } from "~/server/headscale/api/server-version";
|
||||
|
||||
import { getBootstrapClient, getRuntimeClient, HS_VERSIONS, Version } from "../setup/env";
|
||||
import { getBootstrapClient, HS_VERSIONS, Version } from "../setup/env";
|
||||
|
||||
describe.for(HS_VERSIONS)("Headscale %s: Runtime Client", (version) => {
|
||||
test("the runtime client is usable", async () => {
|
||||
@@ -34,8 +34,8 @@ describe.for(HS_VERSIONS)("Headscale %s: Runtime Client", (version) => {
|
||||
});
|
||||
|
||||
test("the health check endpoint works", async () => {
|
||||
const client = await getRuntimeClient(version);
|
||||
const health = await client.isHealthy();
|
||||
const bootstrapper = await getBootstrapClient(version);
|
||||
const health = await bootstrapper.health();
|
||||
expect(health).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -81,8 +81,8 @@ describe("DockerIntegration", () => {
|
||||
socketPath: "/var/run/docker.sock",
|
||||
});
|
||||
|
||||
const mockClient = {
|
||||
isHealthy: async () => {
|
||||
const mockHeadscale = {
|
||||
health: async () => {
|
||||
try {
|
||||
const res = await dockerClient.request({
|
||||
method: "GET",
|
||||
@@ -96,9 +96,9 @@ describe("DockerIntegration", () => {
|
||||
},
|
||||
} as any;
|
||||
|
||||
await integration.onConfigChange(mockClient);
|
||||
await integration.onConfigChange(mockHeadscale);
|
||||
|
||||
const healthy = await mockClient.isHealthy();
|
||||
const healthy = await mockHeadscale.health();
|
||||
expect(healthy).toBe(true);
|
||||
}, 60_000);
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user