factorio-clusterio/packages/host/host.ts

#!/usr/bin/env node

/**
 * Clusterio host
 *
 * Connects to the controller and hosts Factorio servers that can
 * communicate with the cluster.  It is remotely controlled by {@link
 * module:controller/controller}.
 *
 * @module host/host
 * @author Danielv123, Hornwitser
 * @example
 * npx clusteriohost run
 */
import fs from "node:fs/promises";
import path from "path";
import yargs from "yargs";
import packageConfig from "@clusterio/host/package.json" with { type: "json" };
import winston from "winston";
import "winston-daily-rotate-file";

// internal libraries
// Reduce startup time by lazy compiling schemas.
import "./src/set_lazy_schema_compilation.js";
import * as lib from "@clusterio/lib";
import { ConsoleTransport, levels, logger } from "@clusterio/lib";

import type Host from "./src/Host.js";
import { createScenarioCommand, handleCreateScenarioCommand } from "./src/create_scenario.js";
import { PairingError, ensurePaired } from "./src/pairing.js";

let host: Host | undefined;

void new lib.Gauge(
	"clusterio_host_pending_requests",
	"Count of pending link requests currently waiting in memory on the host.",
	{
		labels: ["host_id"],
		callback: (gauge) => {
			if (!host) {
				return;
			}

			let count = host.pendingRequestCount;
			for (const connection of host.instanceConnections.values()) {
				count += connection.pendingRequestCount;
				count += connection.instance.pendingRequestCount;
			}
			gauge.labels(String(host.config.get("host.id"))).set(count);
		},
	}
);

export class HostConnector extends lib.WebSocketClientConnector {
	constructor(
		public hostConfig: lib.HostConfig,
		public pluginInfos: lib.PluginNodeEnvInfo[]
	) {
		super(
			hostConfig.get("host.controller_url"),
			hostConfig.get("host.max_reconnect_delay"),
		);
	}

	register() {
		logger.info("Connector | registering host");
		let plugins: Record<string, string> = {};
		for (let pluginInfo of this.pluginInfos) {
			plugins[pluginInfo.name] = pluginInfo.version;
		}

		this.sendHandshake(
			new lib.MessageRegisterHost(
				new lib.RegisterHostData(
					this.hostConfig.get("host.controller_token"),
					packageConfig.version,
					this.hostConfig.get("host.id"),
					plugins,
				)
			)
		);
	}
}


async function startHost() {
	// argument parsing
	// eslint-disable-next-line node/no-sync
	const args = yargs(process.argv.slice(2))
		.scriptName("host")
		.usage("$0 <command> [options]")
		.option("log-level", {
			nargs: 1,
			describe: "Log level to print to stdout",
			default: "info",
			choices: ["none"].concat(Object.keys(levels)),
			type: "string",
		})
		.option("log-directory", {
			nargs: 1,
			describe: "Directory to place logs in",
			default: "logs",
			type: "string",
		})
		.option("config", {
			nargs: 1,
			describe: "host config file to use",
			default: "config-host.json",
			type: "string",
		})
		.option("create-config", {
			nargs: 0,
			describe: "Create config file if it does not exist",
			default: false,
			type: "boolean",
		})
		.option("plugin-list", {
			nargs: 1,
			describe: "File containing list of plugins available with their install path",
			default: "plugin-list.json",
			type: "string",
		})
		.option("bypass-lock-file", { hidden: true, type: "boolean", nargs: 0, default: false })
		.command("plugin", "Manage available plugins", lib.pluginCommand)
		.command("config", "Manage Host config", lib.configCommand)
		.command(
			"create-scenario <output>",
			"Create a scenario patched with modules for use outside Clusterio",
			createScenarioCommand,
		)
		.command("run", "Run host", yargs => {
			yargs.option("can-restart", {
				type: "boolean", nargs: 0, default: false,
				describe: "Indicate that a process monitor will restart this host on failure",
			});
			yargs.option("recovery", {
				type: "boolean", nargs: 0, default: false,
				describe: "Start the host in recovery mode with all plugins disabled and controller disconnected",
			});
			yargs.option("user-token", {
				type: "string", nargs: 1,
				describe: "Token of a user allowed to approve hosts, approves this host without waiting. Not saved",
			});
		})
		.demandCommand(1, "You need to specify a command to run")
		.strict()
		.parseSync()
	;

	logger.add(new winston.transports.DailyRotateFile({
		format: winston.format.json(),
		filename: "host-%DATE%.log",
		utc: true,
		dirname: path.join(args.logDirectory, "host"),
	}));
	if (args.logLevel !== "none") {
		logger.add(new ConsoleTransport({
			level: args.logLevel,
			format: new lib.TerminalFormat(),
			filter: (info: any) => info.instance_id === undefined,
		}));
	}
	lib.handleUnhandledErrors();

	let command = args._[0];
	if (command === "run") {
		logger.info(`Starting Clusterio host ${packageConfig.version}`);
		if (args.recovery) {
			logger.warn("Host recovery mode enabled. Some features will be disabled.");
		}
	}

	logger.info(`Loading available plugins from ${args.pluginList}`);
	const skippedNpmPlugins = new Map<string, string>();
	let pluginList = await lib.loadPluginList(args.pluginList, true, skippedNpmPlugins);

	// If the command is plugin management we don't try to load plugins
	if (command === "plugin") {
		await lib.handlePluginCommand(args, pluginList, args.pluginList);
		return;
	}

	logger.info("Loading Plugin info");
	let pluginInfos = await lib.loadPluginInfos(pluginList, skippedNpmPlugins);
	lib.registerPluginMessages(pluginInfos);
	lib.registerPluginPermissions(pluginInfos);
	lib.addPluginConfigFields(pluginInfos);
	if (!args.recovery) {
		await lib.loadPluginMigrations(pluginInfos);
	}

	const [hostConfig, hostConfigLock] = await lib.loadConfigFromArgs(
		args,
		lib.HostConfig,
		"host",
	);

	hostConfig.set("host.version", packageConfig.version); // Allows tracking last loaded version

	if (command === "config") {
		await lib.handleConfigCommand(args, hostConfig, hostConfigLock);
		return;
	}

	if (command === "create-scenario") {
		await handleCreateScenarioCommand(args, hostConfig, pluginInfos);
		return;
	}

	// From this point on the host should run
	await hostConfigLock.acquire();

	// If we get here the command was run
	await fs.mkdir(hostConfig.get("host.instances_directory"), { recursive: true });
	await fs.mkdir(hostConfig.get("host.mods_directory"), { recursive: true });
	await fs.mkdir("modules", { recursive: true });

	// Set the process title, shows up as the title of the CMD window on windows
	// and as the process name in ps/top on linux.
	process.title = "clusteriohost";

	// make sure url ends with /
	if (!hostConfig.get("host.controller_url").endsWith("/")) {
		logger.fatal("ERROR invalid config!");
		logger.fatal("host.controller_url must end with '/'");
		process.exitCode = 1;
		return;
	}

	if (!args.recovery) {
		const abort = new AbortController();
		const onSignal = () => { abort.abort(); };
		process.on("SIGINT", onSignal);
		process.on("SIGTERM", onSignal);
		process.on("SIGHUP", onSignal);
		try {
			await ensurePaired(hostConfig, {
				version: packageConfig.version,
				userToken: args.userToken as string | undefined,
				signal: abort.signal,
			});
		} catch (err: any) {
			if (abort.signal.aborted) {
				logger.info("Stopped waiting for approval");
				return;
			}
			if (err instanceof PairingError) {
				logger.fatal(`Unable to join the cluster: ${err.message}`);
				// Exit code 8 stops the run scripts from restarting the host.
				process.exitCode = err.status === 403 ? 8 : 1;
				return;
			}
			throw err;
		} finally {
			process.off("SIGINT", onSignal);
			process.off("SIGTERM", onSignal);
			process.off("SIGHUP", onSignal);
		}
	}

	// Loaded here as the server is not needed by the other commands
	const { default: Host } = await import("./src/Host.js");
	let hostConnector = new HostConnector(hostConfig, pluginInfos);
	host = await Host.fromConfig(
		hostConfig,
		pluginInfos,
		hostConnector,
		Boolean(args.canRestart),
		Boolean(args.recovery),
	);

	// Handle interrupts
	process.on("SIGINT", lib.createShutdownGuard(logger, "interrupt", host.shutdown.bind(host)));
	process.on("SIGTERM", lib.createShutdownGuard(logger, "termination", host.shutdown.bind(host)));
	process.on("SIGHUP", () => {
		logger.info("Terminal closed, shutting down");
		host!.shutdown();
	});

	try {
		await host.loadPlugins();
	} catch (err) {
		await host.shutdown();
		throw err;
	}

	hostConnector.once("connect", () => {
		logger.add(new lib.LinkTransport({ link: host! }));
	});

	await hostConnector.connect();
	host.startedAtMs = Date.now();
	logger.info("Started host");
}

export function bootstrap() {
	startHost().catch(err => {
		if (err instanceof lib.WebSocketError) {
			// Already handled by connector.on("error") within Host

		} else if (err instanceof lib.StartupError || err.code === "InstallationError") {
			logger.fatal(`
+----------------------------------+
| Unable to to start clusteriohost |
+----------------------------------+
${err.stack}`
			);

		} else if (err instanceof lib.PluginError) {
			logger.fatal(`
${err.pluginName} plugin threw an unexpected error
during startup, please report it to the plugin author.
------------------------------------------------------
${err.original.stack}`
			);

		} else {
			logger.fatal(`
+------------------------------------------------------------+
| Unexpected error occured while starting host, please       |
| report it to https://github.com/clusterio/clusterio/issues |
+------------------------------------------------------------+
${err.stack}`
			);
		}

		if (err instanceof lib.AuthenticationFailed) {
			process.exitCode = 8;
		} else {
			process.exitCode = 1;
		}
	});
}

if (import.meta.main) {
	bootstrap();
}