feat(TALE-7): reimplement integration system

app/integration/docker.ts

@@ -5,123 +5,143 @@ import { Client } from 'undici'
 
 import { HeadscaleError, pull } from '~/utils/headscale'
 
-import type { Integration } from '.'
+import { createIntegration } from './integration'
 
-// Integration name
+interface Context {
-const name = 'Docker'
+	client: Client | undefined
+	container: string | undefined
+	maxAttempts: number
+}
 
-let url: URL | undefined
+export default createIntegration<Context>({
-let container: string | undefined
+	name: 'Docker',
+	context: {
+		client: undefined,
+		container: undefined,
+		maxAttempts: 10,
+	},
+	isAvailable: async ({ client, container }) => {
+		// Check for the HEADSCALE_CONTAINER environment variable first
+		// to avoid unnecessary fetching of the Docker socket
+		container = process.env.HEADSCALE_CONTAINER
+			?.trim()
+			.toLowerCase()
 
-async function preflight() {
+		if (!container || container.length === 0) {
-	const path = process.env.DOCKER_SOCK ?? 'unix:///var/run/docker.sock'
+			return false
+		}
 
-	try {
+		const path = process.env.DOCKER_SOCK ?? 'unix:///var/run/docker.sock'
-		url = new URL(path)
+		let url: URL | undefined
-	} catch {
-		return false
-	}
 
-	// The API is available as an HTTP endpoint
-	if (url.protocol === 'tcp:') {
-		url.protocol = 'http:'
-	}
-
-	// Check if the socket is accessible
-	if (url.protocol === 'unix:') {
 		try {
-			await access(path, constants.R_OK)
+			url = new URL(path)
 		} catch {
 			return false
 		}
-	}
 
-	if (url.protocol === 'http:') {
+		if (url.protocol !== 'tcp:' && url.protocol !== 'unix:') {
-		try {
-			await fetch(new URL('/v1.30/version', url).href)
-		} catch {
 			return false
 		}
-	}
 
-	if (url.protocol !== 'http:' && url.protocol !== 'unix:') {
+		// The API is available as an HTTP endpoint and this
-		return false
+		// will simplify the fetching logic in undici
-	}
+		if (url.protocol === 'tcp:') {
+			url.protocol = 'http:'
+			try {
+				await fetch(new URL('/v1.30/version', url).href)
+			} catch {
+				return false
+			}
 
-	container = process.env.HEADSCALE_CONTAINER
+			client = new Client(url.href)
-		?.trim()
+		}
-		.toLowerCase()
 
-	if (!container || container.length === 0) {
+		// Check if the socket is accessible
-		return false
+		if (url.protocol === 'unix:') {
-	}
+			try {
+				await access(path, constants.R_OK)
+			} catch {
+				return false
+			}
 
-	return true
+			client = new Client('http://localhost', {
-}
+				socketPath: path,
+			})
+		}
 
-async function sighup() {
+		return client === undefined
-	if (!url || !container) {
+	},
-		return
-	}
 
-	// Supports the DOCKER_SOCK environment variable
+	onAclChange: async ({ client, container, maxAttempts }) => {
-	const client = url.protocol === 'unix:'
+		if (!client || !container) {
-		? new Client('http://localhost', {
-			socketPath: url.href,
-		})
-		: new Client(url.href)
-
-	const response = await client.request({
-		method: 'POST',
-		path: `/v1.30/containers/${container}/kill?signal=SIGHUP`,
-	})
-
-	if (!response.statusCode || response.statusCode !== 204) {
-		throw new Error('Failed to send SIGHUP to Headscale')
-	}
-}
-
-async function restart() {
-	if (!url || !container) {
-		return
-	}
-
-	// Supports the DOCKER_SOCK environment variable
-	const client = url.protocol === 'unix:'
-		? new Client('http://localhost', {
-			socketPath: url.href,
-		})
-		: new Client(url.href)
-
-	const response = await client.request({
-		method: 'POST',
-		path: `/v1.30/containers/${container}/restart`,
-	})
-
-	if (!response.statusCode || response.statusCode !== 204) {
-		throw new Error('Failed to restart Headscale')
-	}
-
-	// Wait for Headscale to restart before continuing
-	let attempts = 0
-	// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition, no-constant-condition
-	while (true) {
-		try {
-			await pull('v1', '')
 			return
-		} catch (error) {
-			if (error instanceof HeadscaleError && error.status === 401) {
-				break
-			}
-
-			if (attempts > 10) {
-				throw new Error('Headscale did not restart in time')
-			}
-
-			attempts++
-			await setTimeout(1000)
 		}
-	}
-}
 
-export default { name, preflight, sighup, restart } satisfies Integration
+		let attempts = 0
+		while (attempts <= maxAttempts) {
+			const response = await client.request({
+				method: 'POST',
+				path: `/v1.30/containers/${container}/kill?signal=SIGHUP`,
+			})
+
+			if (response.statusCode !== 204) {
+				if (attempts < 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}`)
+			}
+		}
+	},
+
+	onConfigChange: async ({ client, container, maxAttempts }) => {
+		if (!client || !container) {
+			return
+		}
+
+		let attempts = 0
+		while (attempts <= maxAttempts) {
+			const response = await client.request({
+				method: 'POST',
+				path: `/v1.30/containers/${container}/restart`,
+			})
+
+			if (response.statusCode !== 204) {
+				if (attempts < 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}`)
+			}
+		}
+
+		attempts = 0
+		while (attempts <= maxAttempts) {
+			try {
+				await pull('v1', '')
+				return
+			} catch (error) {
+				if (error instanceof HeadscaleError && error.status === 401) {
+					break
+				}
+
+				if (attempts < maxAttempts) {
+					attempts++
+					await setTimeout(1000)
+					continue
+				}
+
+				throw new Error(`Missed restart deadline for ${container}`)
+			}
+		}
+	},
+})

app/integration/index.ts

@@ -1,28 +1,18 @@
-import docker from './docker'
+import dockerIntegration from './docker'
-import kubernetes from './kubernetes'
+import kubernetesIntegration from './kubernetes'
-import proc from './proc'
+import procIntegration from './proc'
 
-export interface Integration {
+export * from './integration'
-	name: string
-	preflight: () => Promise<boolean>
-	sighup?: () => Promise<void>
-	restart?: () => Promise<void>
-}
 
-// Because we previously supported the Docker integration by
+export function loadIntegration() {
-// checking for the HEADSCALE_CONTAINER variable, we need to
-// check for it here as well.
-//
-// This ensures that when people upgrade from older versions
-// of Headplane, they don't explicitly need to define the new
-// HEADSCALE_INTEGRATION variable that is needed to configure
-// an integration.
-export async function checkIntegration() {
 	let integration = process.env.HEADSCALE_INTEGRATION
 		?.trim()
 		.toLowerCase()
 
 	// Old HEADSCALE_CONTAINER variable upgrade path
+	// This ensures that when people upgrade from older versions of Headplane
+	// they don't explicitly need to define the new HEADSCALE_INTEGRATION
+	// variable that is needed to configure docker
 	if (!integration && process.env.HEADSCALE_CONTAINER) {
 		integration = 'docker'
 	}
@@ -32,32 +22,22 @@ export async function checkIntegration() {
 		return
 	}
 
-	let module: Integration | undefined
+	switch (integration.toLowerCase().trim()) {
-	try {
-		module = getIntegration(integration)
-		await module.preflight()
-	} catch (error) {
-		console.error('Failed to load integration', error)
-		return
-	}
-
-	return module
-}
-
-function getIntegration(name: string) {
-	switch (name) {
 		case 'docker': {
-			return docker
+			return dockerIntegration
 		}
-		case 'proc': {
+
-			return proc
+		case 'proc':
+		case 'native':
+		case 'linux': {
+			return procIntegration
 		}
+
 		case 'kubernetes':
 		case 'k8s': {
-			return kubernetes
+			return kubernetesIntegration
-		}
-		default: {
-			throw new Error(`Unknown integration: ${name}`)
 		}
 	}
+
+	console.error('Unknown integration:', integration)
 }

app/integration/integration.ts

@@ -0,0 +1,14 @@
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+export interface IntegrationFactory<T = any> {
+	name: string
+	context: T
+	isAvailable: (context: T) => Promise<boolean> | boolean
+	onAclChange?: (context: T) => Promise<void> | void
+	onConfigChange?: (context: T) => Promise<void> | void
+}
+
+export function createIntegration<T>(
+	options: IntegrationFactory<T>,
+) {
+	return options
+}

app/integration/kubernetes.ts

@@ -1,189 +1,162 @@
-import { access, constants, readdir, readFile } from 'node:fs/promises'
+import { readdir, readFile } from 'node:fs/promises'
 import { platform } from 'node:os'
 import { join, resolve } from 'node:path'
 import { kill } from 'node:process'
 
 import { Config, CoreV1Api, KubeConfig } from '@kubernetes/client-node'
 
-import type { Integration } from '.'
+import { createIntegration } from './integration'
 
-// Integration name
+interface Context {
-const name = 'Kubernetes (k8s)'
+	pid: number | undefined
-
-// Check if we have a proper service account and /proc
-// This is because the Kubernetes integration is basically
-// the /proc integration plus some extra steps.
-async function preflight() {
-	if (platform() !== 'linux') {
-		console.error('Not running on k8s Linux')
-		return false
-	}
-
-	const dir = resolve('/proc')
-	try {
-		await access(dir, constants.R_OK)
-	} catch (error) {
-		console.error('Failed to access /proc', error)
-		return false
-	}
-
-	const secretsDir = resolve(Config.SERVICEACCOUNT_ROOT)
-	try {
-		const files = await readdir(secretsDir)
-		if (files.length === 0) {
-			console.error('No Kubernetes service account found')
-			return false
-		}
-
-		const mappedFiles = new Set(files.map(file => join(secretsDir, file)))
-		const expectedFiles = [
-			Config.SERVICEACCOUNT_CA_PATH,
-			Config.SERVICEACCOUNT_TOKEN_PATH,
-			Config.SERVICEACCOUNT_NAMESPACE_PATH,
-		]
-
-		if (!expectedFiles.every(file => mappedFiles.has(file))) {
-			console.error('Kubernetes service account is incomplete')
-			return false
-		}
-	} catch (error) {
-		console.error('Failed to access Kubernetes service account', error)
-		return false
-	}
-
-	const namespace = await readFile(Config.SERVICEACCOUNT_NAMESPACE_PATH, 'utf8')
-	if (namespace.trim().length === 0) {
-		console.error('Kubernetes namespace is empty')
-		return false
-	}
-
-	const skip = process.env.HEADSCALE_INTEGRATION_UNSTRICT
-	if (skip === 'true' || skip === '1') {
-		console.warn('Skipping strict Kubernetes integration check')
-		return true
-	}
-
-	// Some very ugly nesting but it's necessary
-	const pod = process.env.POD_NAME
-	if (!pod) {
-		console.error('No pod name found (POD_NAME)')
-		return false
-	}
-
-	const result = await checkPod(pod, namespace)
-	if (!result) {
-		return false
-	}
-
-	return true
 }
 
-async function checkPod(pod: string, namespace: string) {
+export default createIntegration<Context>({
-	if (pod.trim().length === 0) {
+	name: 'Kubernetes (k8s)',
-		console.error('Pod name is empty')
+	context: {
-		return false
+		pid: undefined,
-	}
+	},
+	isAvailable: async ({ pid }) => {
+		if (platform() !== 'linux') {
+			return false
+		}
 
-	try {
+		const svcRoot = Config.SERVICEACCOUNT_ROOT
-		const kc = new KubeConfig()
+		try {
-		kc.loadFromCluster()
+			const files = await readdir(svcRoot)
+			if (files.length === 0) {
+				console.error('No Kubernetes service account found')
+				return false
+			}
 
-		const kCoreV1Api = kc.makeApiClient(CoreV1Api)
+			const mappedFiles = new Set(files.map(file => join(svcRoot, file)))
-		const { response, body } = await kCoreV1Api.readNamespacedPod(
+			const expectedFiles = [
-			pod,
+				Config.SERVICEACCOUNT_CA_PATH,
-			namespace,
+				Config.SERVICEACCOUNT_TOKEN_PATH,
+				Config.SERVICEACCOUNT_NAMESPACE_PATH,
+			]
+
+			if (!expectedFiles.every(file => mappedFiles.has(file))) {
+				console.error('Kubernetes service account is incomplete')
+				return false
+			}
+		} catch (error) {
+			console.error('Failed to access Kubernetes service account', error)
+			return false
+		}
+
+		const namespace = await readFile(
+			Config.SERVICEACCOUNT_NAMESPACE_PATH,
+			'utf8',
 		)
 
-		if (response.statusCode !== 200) {
+		// Some very ugly nesting but it's necessary
-			console.error('Failed to read pod', response.statusCode)
+		if (process.env.HEADSCALE_INTEGRATION_UNSTRICT === 'true') {
-			return false
+			console.warn('Skipping strict Kubernetes integration check')
+		} else {
+			const pod = process.env.POD_NAME
+			if (!pod) {
+				console.error('No pod name found (POD_NAME)')
+				return false
+			}
+
+			if (pod.trim().length === 0) {
+				console.error('Pod name is empty')
+				return false
+			}
+
+			try {
+				const kc = new KubeConfig()
+				kc.loadFromCluster()
+
+				const kCoreV1Api = kc.makeApiClient(CoreV1Api)
+				const { response, body } = await kCoreV1Api.readNamespacedPod(
+					pod,
+					namespace,
+				)
+
+				if (response.statusCode !== 200) {
+					console.error('Failed to read pod', response.statusCode)
+					return false
+				}
+
+				const shared = body.spec?.shareProcessNamespace
+				if (shared === undefined) {
+					console.error('Pod does not have shareProcessNamespace set')
+					return false
+				}
+
+				if (!shared) {
+					console.error('Pod has disabled shareProcessNamespace')
+					return false
+				}
+			} catch (error) {
+				console.error('Failed to check pod', error)
+				return false
+			}
 		}
 
-		const shared = body.spec?.shareProcessNamespace
+		const dir = resolve('/proc')
-		if (shared === undefined) {
+		try {
-			console.error('Pod does not have shareProcessNamespace set')
+			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 {
+					const data = await readFile(path, 'utf8')
+					if (data.includes('headscale')) {
+						return pid
+					}
+				} catch {}
+			})
+
+			const results = await Promise.allSettled(promises)
+			const pids = []
+
+			for (const result of results) {
+				if (result.status === 'fulfilled' && result.value) {
+					pids.push(result.value)
+				}
+			}
+
+			if (pids.length > 1) {
+				console.warn('Found multiple Headscale processes', pids)
+				console.log('Disabling the /proc integration')
+				return false
+			}
+
+			if (pids.length === 0) {
+				console.warn('Could not find Headscale process')
+				console.log('Disabling the /proc integration')
+				return false
+			}
+
+			pid = pids[0]
+			console.log('Found Headscale process', pid)
+			return true
+		} catch {
 			return false
 		}
+	},
 
-		if (!shared) {
+	onAclChange: ({ pid }) => {
-			console.error('Pod has disabled shareProcessNamespace')
+		if (!pid) {
-			return false
-		}
-	} catch (error) {
-		console.error('Failed to check pod', error)
-		return false
-	}
-
-	return true
-}
-
-async function findPid() {
-	const dirs = await readdir('/proc')
-
-	const promises = dirs.map(async (dir) => {
-		const pid = Number.parseInt(dir, 10)
-
-		if (Number.isNaN(pid)) {
 			return
 		}
 
-		const path = join('/proc', dir, 'cmdline')
-		try {
-			const data = await readFile(path, 'utf8')
-			if (data.includes('headscale')) {
-				return pid
-			}
-		} catch {}
-	})
-
-	const results = await Promise.allSettled(promises)
-	const pids = []
-
-	for (const result of results) {
-		if (result.status === 'fulfilled' && result.value) {
-			pids.push(result.value)
-		}
-	}
-
-	if (pids.length > 1) {
-		console.warn('Found multiple Headscale processes', pids)
-		console.log('Disabling the k8s integration')
-		return
-	}
-
-	if (pids.length === 0) {
-		console.warn('Could not find Headscale process')
-		console.log('Disabling the k8s integration')
-		return
-	}
-
-	return pids[0]
-}
-
-async function sighup() {
-	const pid = await findPid()
-	if (!pid) {
-		return
-	}
-
-	try {
 		kill(pid, 'SIGHUP')
-	} catch (error) {
+	},
-		console.error('Failed to send SIGHUP to Headscale', error)
-	}
-}
 
-async function restart() {
+	onConfigChange: ({ pid }) => {
-	const pid = await findPid()
+		if (!pid) {
-	if (!pid) {
+			return
-		return
+		}
-	}
 
-	try {
 		kill(pid, 'SIGTERM')
-	} catch (error) {
+	},
-		console.error('Failed to send SIGTERM to Headscale', error)
+})
-	}
-}
-
-export default { name, preflight, sighup, restart } satisfies Integration

app/integration/proc.ts

@@ -1,83 +1,77 @@
-import { access, constants, readdir, readFile } from 'node:fs/promises'
+import { readdir, readFile } from 'node:fs/promises'
 import { platform } from 'node:os'
 import { join, resolve } from 'node:path'
 import { kill } from 'node:process'
 
-import type { Integration } from '.'
+import { createIntegration } from './integration'
 
-// Integration name
+interface Context {
-const name = 'Native Linux (/proc)'
+	pid: number | undefined
-
-// Check if we have a /proc and if it's readable
-async function preflight() {
-	if (platform() !== 'linux') {
-		return false
-	}
-
-	const dir = resolve('/proc')
-	try {
-		await access(dir, constants.R_OK)
-		return true
-	} catch (error) {
-		console.error('Failed to access /proc', error)
-		return false
-	}
 }
 
-async function findPid() {
+export default createIntegration<Context>({
-	const dirs = await readdir('/proc')
+	name: 'Native Linux (/proc)',
+	context: {
+		pid: undefined,
+	},
+	isAvailable: async ({ pid }) => {
+		if (platform() !== 'linux') {
+			return false
+		}
 
-	const promises = dirs.map(async (dir) => {
+		const dir = resolve('/proc')
-		const pid = Number.parseInt(dir, 10)
+		try {
+			const subdirs = await readdir(dir)
+			const promises = subdirs.map(async (dir) => {
+				const pid = Number.parseInt(dir, 10)
 
-		if (Number.isNaN(pid)) {
+				if (Number.isNaN(pid)) {
+					return
+				}
+
+				const path = join('/proc', dir, 'cmdline')
+				try {
+					const data = await readFile(path, 'utf8')
+					if (data.includes('headscale')) {
+						return pid
+					}
+				} catch {}
+			})
+
+			const results = await Promise.allSettled(promises)
+			const pids = []
+
+			for (const result of results) {
+				if (result.status === 'fulfilled' && result.value) {
+					pids.push(result.value)
+				}
+			}
+
+			if (pids.length > 1) {
+				console.warn('Found multiple Headscale processes', pids)
+				console.log('Disabling the /proc integration')
+				return false
+			}
+
+			if (pids.length === 0) {
+				console.warn('Could not find Headscale process')
+				console.log('Disabling the /proc integration')
+				return false
+			}
+
+			pid = pids[0]
+			console.log('Found Headscale process', pid)
+			return true
+		} catch {
+			return false
+		}
+	},
+
+	onAclChange: ({ pid }) => {
+		if (!pid) {
 			return
 		}
 
-		const path = join('/proc', dir, 'cmdline')
-		try {
-			const data = await readFile(path, 'utf8')
-			if (data.includes('headscale')) {
-				return pid
-			}
-		} catch {}
-	})
-
-	const results = await Promise.allSettled(promises)
-	const pids = []
-
-	for (const result of results) {
-		if (result.status === 'fulfilled' && result.value) {
-			pids.push(result.value)
-		}
-	}
-
-	if (pids.length > 1) {
-		console.warn('Found multiple Headscale processes', pids)
-		console.log('Disabling the /proc integration')
-		return
-	}
-
-	if (pids.length === 0) {
-		console.warn('Could not find Headscale process')
-		console.log('Disabling the /proc integration')
-		return
-	}
-
-	return pids[0]
-}
-
-async function sighup() {
-	const pid = await findPid()
-	if (!pid) {
-		return
-	}
-
-	try {
 		kill(pid, 'SIGHUP')
-	} catch (error) {
+	},
-		console.error('Failed to send SIGHUP to Headscale', error)
+})
-	}
-}
-
-export default { name, preflight, sighup } satisfies Integration

app/utils/config/headplane.ts

@@ -8,14 +8,14 @@ import { resolve } from 'node:path'
 
 import { parse } from 'yaml'
 
-import { checkIntegration, Integration } from '~/integration'
+import { IntegrationFactory, loadIntegration } from '~/integration'
 
 import { HeadscaleConfig, loadConfig } from './headscale'
 
 export interface HeadplaneContext {
 	headscaleUrl: string
 	cookieSecret: string
-	integration: Integration | undefined
+	integration: IntegrationFactory | undefined
 
 	config: {
 		read: boolean
@@ -67,7 +67,7 @@ export async function loadContext(): Promise<HeadplaneContext> {
 	context = {
 		headscaleUrl,
 		cookieSecret,
-		integration: await checkIntegration(),
+		integration: loadIntegration(),
 		config: contextData,
 		acl: await checkAcl(config),
 		oidc: await checkOidc(config),