fix: fix integrations not loading

app/server/README.md

@@ -8,6 +8,12 @@ many side-effects (in this case, importing a module may run code).
 server
 ├── index.ts: Loads everything and starts the web server.
 ├── config/
+│   ├── integration/
+│   │   ├── abstract.ts: Defines the abstract class for integrations.
+│   │   ├── docker.ts: Contains the Docker integration.
+│   │   ├── index.ts: Determines the correct integration to use (if any).
+│   │   ├── kubernetes.ts: Contains the Kubernetes integration.
+│   │   ├── proc.ts: Contains the Proc integration.
 │   ├── env.ts: Checks the environment variables for custom overrides.
 │   ├── loader.ts: Checks the configuration file and coalesces with ENV.
 │   ├── schema.ts: Defines the schema for the Headplane configuration.

app/server/config/integration/abstract.ts

@@ -0,0 +1,16 @@
+import type { ApiClient } from '~/server/headscale/api-client';
+
+export abstract class Integration<T> {
+	protected context: NonNullable<T>;
+	constructor(context: T) {
+		if (!context) {
+			throw new Error('Missing integration context');
+		}
+
+		this.context = context;
+	}
+
+	abstract isAvailable(): Promise<boolean> | boolean;
+	abstract onConfigChange(client: ApiClient): Promise<void> | void;
+	abstract get name(): string;
+}

app/server/config/integration/docker.ts

@@ -0,0 +1,142 @@
+import { constants, access } from 'node:fs/promises';
+import { setTimeout } from 'node:timers/promises';
+import { Client } from 'undici';
+import { ApiClient } from '~/server/headscale/api-client';
+import log from '~/utils/log';
+import type { HeadplaneConfig } from '../schema';
+import { Integration } from './abstract';
+
+type T = NonNullable<HeadplaneConfig['integration']>['docker'];
+export default class DockerIntegration extends Integration<T> {
+	private maxAttempts = 10;
+	private client: Client | undefined;
+
+	get name() {
+		return 'Docker';
+	}
+
+	async isAvailable() {
+		if (this.context.container_name.length === 0) {
+			log.error('config', 'Docker container name is empty');
+			return false;
+		}
+
+		log.info('config', 'Using container: %s', this.context.container_name);
+		let url: URL | undefined;
+		try {
+			url = new URL(this.context.socket);
+		} catch {
+			log.error(
+				'config',
+				'Invalid Docker socket path: %s',
+				this.context.socket,
+			);
+			return false;
+		}
+
+		if (url.protocol !== 'tcp:' && url.protocol !== 'unix:') {
+			log.error('config', 'Invalid Docker socket protocol: %s', url.protocol);
+			return false;
+		}
+
+		// The API is available as an HTTP endpoint and this
+		// will simplify the fetching logic in undici
+		if (url.protocol === 'tcp:') {
+			// Apparently setting url.protocol doesn't work anymore?
+			const fetchU = url.href.replace(url.protocol, 'http:');
+
+			try {
+				log.info('config', 'Checking API: %s', fetchU);
+				await fetch(new URL('/v1.30/version', fetchU).href);
+			} catch (error) {
+				log.error('config', 'Failed to connect to Docker API: %s', error);
+				log.debug('config', 'Connection error: %o', error);
+				return false;
+			}
+
+			this.client = new Client(fetchU);
+		}
+
+		// Check if the socket is accessible
+		if (url.protocol === 'unix:') {
+			try {
+				log.info('config', 'Checking socket: %s', url.pathname);
+				await access(url.pathname, constants.R_OK);
+			} catch (error) {
+				log.error('config', 'Failed to access Docker socket: %s', url.pathname);
+				log.debug('config', 'Access error: %o', error);
+				return false;
+			}
+
+			this.client = new Client('http://localhost', {
+				socketPath: url.pathname,
+			});
+		}
+
+		return this.client !== undefined;
+	}
+
+	async onConfigChange(client: ApiClient) {
+		if (!this.client) {
+			return;
+		}
+
+		log.info('config', 'Restarting Headscale via Docker');
+
+		let attempts = 0;
+		while (attempts <= this.maxAttempts) {
+			log.debug(
+				'config',
+				'Restarting container: %s (attempt %d)',
+				this.context.container_name,
+				attempts,
+			);
+
+			const response = await this.client.request({
+				method: 'POST',
+				path: `/v1.30/containers/${this.context.container_name}/restart`,
+			});
+
+			if (response.statusCode !== 204) {
+				if (attempts < this.maxAttempts) {
+					attempts++;
+					await setTimeout(1000);
+					continue;
+				}
+
+				const stringCode = response.statusCode.toString();
+				const body = await response.body.text();
+				throw new Error(`API request failed: ${stringCode} ${body}`);
+			}
+
+			break;
+		}
+
+		attempts = 0;
+		while (attempts <= this.maxAttempts) {
+			try {
+				log.debug('config', 'Checking Headscale status (attempt %d)', attempts);
+				const status = await client.healthcheck();
+				if (status === false) {
+					throw new Error('Headscale is not running');
+				}
+
+				log.info('config', 'Headscale is up and running');
+				return;
+			} catch (error) {
+				if (attempts < this.maxAttempts) {
+					attempts++;
+					await setTimeout(1000);
+					continue;
+				}
+
+				log.error(
+					'config',
+					'Missed restart deadline for %s',
+					this.context.container_name,
+				);
+				return;
+			}
+		}
+	}
+}

app/server/config/integration/index.ts

@@ -0,0 +1,62 @@
+import { HeadplaneConfig } from '~/server/config/schema';
+import log from '~/utils/log';
+import dockerIntegration from './docker';
+import kubernetesIntegration from './kubernetes';
+import procIntegration from './proc';
+
+export async function loadIntegration(context: HeadplaneConfig['integration']) {
+	const integration = getIntegration(context);
+	if (!integration) {
+		return;
+	}
+
+	try {
+		const res = await integration.isAvailable();
+		if (!res) {
+			log.error('config', 'Integration %s is not available', integration);
+			return;
+		}
+	} catch (error) {
+		log.error(
+			'config',
+			'Failed to load integration %s: %s',
+			integration,
+			error,
+		);
+		log.debug('config', 'Loading error: %o', error);
+		return;
+	}
+
+	return integration;
+}
+
+function getIntegration(integration: HeadplaneConfig['integration']) {
+	const docker = integration?.docker;
+	const k8s = integration?.kubernetes;
+	const proc = integration?.proc;
+
+	if (!docker?.enabled && !k8s?.enabled && !proc?.enabled) {
+		log.debug('config', 'No integrations enabled');
+		return;
+	}
+
+	if (docker?.enabled && k8s?.enabled && proc?.enabled) {
+		log.error('config', 'Multiple integrations enabled, please pick one only');
+		return;
+	}
+
+	if (docker?.enabled) {
+		log.info('config', 'Using Docker integration');
+		return new dockerIntegration(integration?.docker);
+	}
+
+	if (k8s?.enabled) {
+		log.info('config', 'Using Kubernetes integration');
+		return new kubernetesIntegration(integration?.kubernetes);
+	}
+
+	if (proc?.enabled) {
+		log.info('config', 'Using Proc integration');
+		return new procIntegration(integration?.proc);
+	}
+}

app/server/config/integration/kubernetes.ts

@@ -0,0 +1,248 @@
+import { readFile, readdir } from 'node:fs/promises';
+import { platform } from 'node:os';
+import { join, resolve } from 'node:path';
+import { kill } from 'node:process';
+import { setTimeout } from 'node:timers/promises';
+import { Config, CoreV1Api, KubeConfig } from '@kubernetes/client-node';
+import { ApiClient } from '~/server/headscale/api-client';
+import log from '~/utils/log';
+import { HeadplaneConfig } from '../schema';
+import { Integration } from './abstract';
+
+// TODO: Upgrade to the new CoreV1Api from @kubernetes/client-node
+type T = NonNullable<HeadplaneConfig['integration']>['kubernetes'];
+export default class KubernetesIntegration extends Integration<T> {
+	private pid: number | undefined;
+	private maxAttempts = 10;
+
+	get name() {
+		return 'Kubernetes (k8s)';
+	}
+
+	async isAvailable() {
+		if (platform() !== 'linux') {
+			log.error('config', 'Kubernetes is only available on Linux');
+			return false;
+		}
+
+		const svcRoot = Config.SERVICEACCOUNT_ROOT;
+		try {
+			log.debug('config', 'Checking Kubernetes service account at %s', svcRoot);
+			const files = await readdir(svcRoot);
+			if (files.length === 0) {
+				log.error('config', 'Kubernetes service account not found');
+				return false;
+			}
+
+			const mappedFiles = new Set(files.map((file) => join(svcRoot, file)));
+			const expectedFiles = [
+				Config.SERVICEACCOUNT_CA_PATH,
+				Config.SERVICEACCOUNT_TOKEN_PATH,
+				Config.SERVICEACCOUNT_NAMESPACE_PATH,
+			];
+
+			log.debug('config', 'Looking for %s', expectedFiles.join(', '));
+			if (!expectedFiles.every((file) => mappedFiles.has(file))) {
+				log.error('config', 'Malformed Kubernetes service account');
+				return false;
+			}
+		} catch (error) {
+			log.error('config', 'Failed to access %s: %s', svcRoot, error);
+			return false;
+		}
+
+		log.debug('config', 'Reading Kubernetes service account at %s', svcRoot);
+		const namespace = await readFile(
+			Config.SERVICEACCOUNT_NAMESPACE_PATH,
+			'utf8',
+		);
+
+		// Some very ugly nesting but it's necessary
+		if (this.context.validate_manifest === false) {
+			log.warn('config', 'Skipping strict Pod status check');
+		} else {
+			const pod = this.context.pod_name;
+			if (!pod) {
+				log.error('config', 'Missing POD_NAME variable');
+				return false;
+			}
+
+			if (pod.trim().length === 0) {
+				log.error('config', 'Pod name is empty');
+				return false;
+			}
+
+			log.debug(
+				'config',
+				'Checking Kubernetes pod %s in namespace %s',
+				pod,
+				namespace,
+			);
+
+			try {
+				log.debug('config', 'Attempgin to get cluster KubeConfig');
+				const kc = new KubeConfig();
+				kc.loadFromCluster();
+
+				const cluster = kc.getCurrentCluster();
+				if (!cluster) {
+					log.error('config', 'Malformed kubeconfig');
+					return false;
+				}
+
+				log.info(
+					'config',
+					'Service account connected to %s (%s)',
+					cluster.name,
+					cluster.server,
+				);
+
+				const kCoreV1Api = kc.makeApiClient(CoreV1Api);
+
+				log.info(
+					'config',
+					'Checking pod %s in namespace %s (%s)',
+					pod,
+					namespace,
+					kCoreV1Api.basePath,
+				);
+
+				log.debug('config', 'Reading pod info for %s', pod);
+				const { response, body } = await kCoreV1Api.readNamespacedPod(
+					pod,
+					namespace,
+				);
+
+				if (response.statusCode !== 200) {
+					log.error(
+						'config',
+						'Failed to read pod info: http %d',
+						response.statusCode,
+					);
+					return false;
+				}
+
+				log.debug('config', 'Got pod info: %o', body.spec);
+				const shared = body.spec?.shareProcessNamespace;
+				if (shared === undefined) {
+					log.error(
+						'config',
+						'Pod does not have spec.shareProcessNamespace set',
+					);
+					return false;
+				}
+
+				if (!shared) {
+					log.error(
+						'config',
+						'Pod has set but disabled spec.shareProcessNamespace',
+					);
+					return false;
+				}
+
+				log.info('config', 'Pod %s enabled shared processes', pod);
+			} catch (error) {
+				log.error('config', 'Failed to read pod info: %s', error);
+				return false;
+			}
+		}
+
+		log.debug('config', 'Looking for namespaced process in /proc');
+		const dir = resolve('/proc');
+		try {
+			const subdirs = await readdir(dir);
+			const promises = subdirs.map(async (dir) => {
+				const pid = Number.parseInt(dir, 10);
+
+				if (Number.isNaN(pid)) {
+					return;
+				}
+
+				const path = join('/proc', dir, 'cmdline');
+				try {
+					log.debug('config', 'Reading %s', path);
+					const data = await readFile(path, 'utf8');
+					if (data.includes('headscale')) {
+						return pid;
+					}
+				} catch (error) {
+					log.debug('config', 'Failed to read %s: %s', path, error);
+				}
+			});
+
+			const results = await Promise.allSettled(promises);
+			const pids = [];
+
+			for (const result of results) {
+				if (result.status === 'fulfilled' && result.value) {
+					pids.push(result.value);
+				}
+			}
+
+			log.debug('config', 'Found Headscale processes: %o', pids);
+			if (pids.length > 1) {
+				log.error(
+					'config',
+					'Found %d Headscale processes: %s',
+					pids.length,
+					pids.join(', '),
+				);
+				return false;
+			}
+
+			if (pids.length === 0) {
+				log.error('config', 'Could not find Headscale process');
+				return false;
+			}
+
+			this.pid = pids[0];
+			log.info('config', 'Found Headscale process with PID: %d', this.pid);
+			return true;
+		} catch {
+			log.error('config', 'Failed to read /proc');
+			return false;
+		}
+	}
+
+	async onConfigChange(client: ApiClient) {
+		if (!this.pid) {
+			return;
+		}
+
+		try {
+			log.info('config', 'Sending SIGTERM to Headscale');
+			kill(this.pid, 'SIGTERM');
+		} catch (error) {
+			log.error('config', 'Failed to send SIGTERM to Headscale: %s', error);
+			log.debug('config', 'kill(1) error: %o', error);
+		}
+
+		await setTimeout(1000);
+		let attempts = 0;
+		while (attempts <= this.maxAttempts) {
+			try {
+				log.debug('config', 'Checking Headscale status (attempt %d)', attempts);
+				const status = await client.healthcheck();
+				if (status === false) {
+					throw new Error('Headscale is not running');
+				}
+
+				log.info('config', 'Headscale is up and running');
+				return;
+			} catch (error) {
+				if (attempts < this.maxAttempts) {
+					attempts++;
+					await setTimeout(1000);
+					continue;
+				}
+
+				log.error(
+					'config',
+					'Missed restart deadline for Headscale (pid %d)',
+					this.pid,
+				);
+				return;
+			}
+		}
+	}
+}

app/server/config/integration/proc.ts

@@ -0,0 +1,125 @@
+import { readFile, readdir } from 'node:fs/promises';
+import { platform } from 'node:os';
+import { join, resolve } from 'node:path';
+import { kill } from 'node:process';
+import { setTimeout } from 'node:timers/promises';
+import { ApiClient } from '~/server/headscale/api-client';
+import log from '~/utils/log';
+import { HeadplaneConfig } from '../schema';
+import { Integration } from './abstract';
+
+type T = NonNullable<HeadplaneConfig['integration']>['proc'];
+export default class ProcIntegration extends Integration<T> {
+	private pid: number | undefined;
+	private maxAttempts = 10;
+
+	get name() {
+		return 'Native Linux (/proc)';
+	}
+
+	async isAvailable() {
+		if (platform() !== 'linux') {
+			log.error('config', '/proc is only available on Linux');
+			return false;
+		}
+
+		log.debug('config', 'Checking /proc for Headscale process');
+		const dir = resolve('/proc');
+		try {
+			const subdirs = await readdir(dir);
+			const promises = subdirs.map(async (dir) => {
+				const pid = Number.parseInt(dir, 10);
+
+				if (Number.isNaN(pid)) {
+					return;
+				}
+
+				const path = join('/proc', dir, 'cmdline');
+				try {
+					log.debug('config', 'Reading %s', path);
+					const data = await readFile(path, 'utf8');
+					if (data.includes('headscale')) {
+						return pid;
+					}
+				} catch (error) {
+					log.error('config', 'Failed to read %s: %s', path, error);
+				}
+			});
+
+			const results = await Promise.allSettled(promises);
+			const pids = [];
+
+			for (const result of results) {
+				if (result.status === 'fulfilled' && result.value) {
+					pids.push(result.value);
+				}
+			}
+
+			log.debug('config', 'Found Headscale processes: %o', pids);
+			if (pids.length > 1) {
+				log.error(
+					'config',
+					'Found %d Headscale processes: %s',
+					pids.length,
+					pids.join(', '),
+				);
+				return false;
+			}
+
+			if (pids.length === 0) {
+				log.error('config', 'Could not find Headscale process');
+				return false;
+			}
+
+			this.pid = pids[0];
+			log.info('config', 'Found Headscale process with PID: %d', this.pid);
+			return true;
+		} catch {
+			log.error('config', 'Failed to read /proc');
+			return false;
+		}
+	}
+
+	async onConfigChange(client: ApiClient) {
+		if (!this.pid) {
+			return;
+		}
+
+		try {
+			log.info('config', 'Sending SIGTERM to Headscale');
+			kill(this.pid, 'SIGTERM');
+		} catch (error) {
+			log.error('config', 'Failed to send SIGTERM to Headscale: %s', error);
+			log.debug('config', 'kill(1) error: %o', error);
+		}
+
+		await setTimeout(1000);
+		let attempts = 0;
+		while (attempts <= this.maxAttempts) {
+			try {
+				log.debug('config', 'Checking Headscale status (attempt %d)', attempts);
+				const status = await client.healthcheck();
+				if (status === false) {
+					log.error('config', 'Headscale is not running');
+					return;
+				}
+
+				log.info('config', 'Headscale is up and running');
+				return;
+			} catch (error) {
+				if (attempts < this.maxAttempts) {
+					attempts++;
+					await setTimeout(1000);
+					continue;
+				}
+
+				log.error(
+					'config',
+					'Missed restart deadline for Headscale (pid %d)',
+					this.pid,
+				);
+				return;
+			}
+		}
+	}
+}

app/server/headscale/api-client.ts

@@ -34,10 +34,7 @@ export class ResponseError extends Error {
 	}
 }
 
-// Represents an error that occurred during a request
-// class RequestError extends Error {
-
-class ApiClient {
+export class ApiClient {
 	private agent: Agent;
 	private base: string;
 

app/server/index.ts

@@ -4,6 +4,7 @@ import { createHonoServer } from 'react-router-hono-server/node';
 import type { WebSocket } from 'ws';
 import log from '~/utils/log';
 import { configureConfig, configureLogger, envVariables } from './config/env';
+import { loadIntegration } from './config/integration';
 import { loadConfig } from './config/loader';
 import { createApiClient } from './headscale/api-client';
 import { loadHeadscaleConfig } from './headscale/config-loader';
@@ -61,6 +62,7 @@ const appLoadContext = {
 		config.server.agent.ttl,
 	),
 
+	integration: await loadIntegration(config.integration),
 	oidc: config.oidc ? await createOidcClient(config.oidc) : undefined,
 };