diff options
Diffstat (limited to 'src/server/Session/session.ts')
| -rw-r--r-- | src/server/Session/session.ts | 338 |
1 files changed, 338 insertions, 0 deletions
diff --git a/src/server/Session/session.ts b/src/server/Session/session.ts new file mode 100644 index 000000000..cf2231b1f --- /dev/null +++ b/src/server/Session/session.ts @@ -0,0 +1,338 @@ +import { red, cyan, green, yellow, magenta, blue } from "colors"; +import { on, fork, setupMaster, Worker } from "cluster"; +import { execSync } from "child_process"; +import { get } from "request-promise"; +import { Utils } from "../../Utils"; +import Repl, { ReplAction } from "../repl"; +import { readFileSync } from "fs"; +import { validate, ValidationError } from "jsonschema"; +import { configurationSchema } from "./session_config_schema"; + +const onWindows = process.platform === "win32"; + +/** + * This namespace relies on NodeJS's cluster module, which allows a parent (master) process to share + * code with its children (workers). A simple `isMaster` flag indicates who is trying to access + * the code, and thus determines the functionality that actually gets invoked (checked by the caller, not internally). + * + * Think of the master thread as a factory, and the workers as the helpers that actually run the server. + * + * So, when we run `npm start`, given the appropriate check, initializeMaster() is called in the parent process + * This will spawn off its own child process (by default, mirrors the execution path of its parent), + * in which initializeWorker() is invoked. + */ +export namespace Session { + + interface Configuration { + showServerOutput: boolean; + masterIdentifier: string; + workerIdentifier: string; + ports: { [description: string]: number }; + pollingRoute: string; + pollingIntervalSeconds: number; + [key: string]: any; + } + + const defaultConfiguration: Configuration = { + showServerOutput: false, + masterIdentifier: yellow("__monitor__:"), + workerIdentifier: magenta("__server__:"), + ports: { server: 3000 }, + pollingRoute: "/", + pollingIntervalSeconds: 30 + }; + + interface MasterExtensions { + addReplCommand: (basename: string, argPatterns: (RegExp | string)[], action: ReplAction) => void; + addChildMessageHandler: (message: string, handler: ActionHandler) => void; + } + + export interface NotifierHooks { + key?: (key: string) => boolean | Promise<boolean>; + crash?: (error: Error) => boolean | Promise<boolean>; + } + + export interface SessionAction { + message: string; + args: any; + } + + export type ExitHandler = (error: Error) => void | Promise<void>; + export type ActionHandler = (action: SessionAction) => void | Promise<void>; + export interface EmailTemplate { + subject: string; + body: string; + } + + function loadAndValidateConfiguration(): any { + try { + const configuration: Configuration = JSON.parse(readFileSync('./session.config.json', 'utf8')); + const options = { + throwError: true, + allowUnknownAttributes: false + }; + // ensure all necessary and no excess information is specified by the configuration file + validate(configuration, configurationSchema, options); + let formatMaster = true; + let formatWorker = true; + Object.keys(defaultConfiguration).forEach(property => { + if (!configuration[property]) { + if (property === "masterIdentifier") { + formatMaster = false; + } else if (property === "workerIdentifier") { + formatWorker = false; + } + configuration[property] = defaultConfiguration[property]; + } + }); + if (formatMaster) { + configuration.masterIdentifier = yellow(configuration.masterIdentifier + ":"); + } + if (formatWorker) { + configuration.workerIdentifier = magenta(configuration.workerIdentifier + ":"); + } + return configuration; + } catch (error) { + if (error instanceof ValidationError) { + console.log(red("\nSession configuration failed.")); + console.log("The given session.config.json configuration file is invalid."); + console.log(`${error.instance}: ${error.stack}`); + process.exit(0); + } else if (error.code === "ENOENT" && error.path === "./session.config.json") { + console.log(cyan("Loading default session parameters...")); + console.log("Consider including a session.config.json configuration file in your project root for customization."); + return defaultConfiguration; + } else { + console.log(red("\nSession configuration failed.")); + console.log("The following unknown error occurred during configuration."); + console.log(error.stack); + process.exit(0); + } + } + } + + function timestamp() { + return blue(`[${new Date().toUTCString()}]`); + } + + /** + * Validates and reads the configuration file, accordingly builds a child process factory + * and spawns off an initial process that will respawn as predecessors die. + */ + export async function initializeMonitorThread(notifiers?: NotifierHooks): Promise<MasterExtensions> { + let activeWorker: Worker; + const childMessageHandlers: { [message: string]: (action: SessionAction, args: any) => void } = {}; + + // read in configuration .json file only once, in the master thread + // pass down any variables the pertinent to the child processes as environment variables + const { + masterIdentifier, + workerIdentifier, + ports, + pollingRoute, + showServerOutput, + pollingIntervalSeconds + } = loadAndValidateConfiguration(); + + const masterLog = (...optionalParams: any[]) => console.log(timestamp(), masterIdentifier, ...optionalParams); + + // this sends a pseudorandomly generated guid to the configuration's recipients, allowing them alone + // to kill the server via the /kill/:key route + let key: string | undefined; + if (notifiers && notifiers.key) { + key = Utils.GenerateGuid(); + const success = await notifiers.key(key); + const statement = success ? green("distributed session key to recipients") : red("distribution of session key failed"); + masterLog(statement); + } + + // handle exceptions in the master thread - there shouldn't be many of these + // the IPC (inter process communication) channel closed exception can't seem + // to be caught in a try catch, and is inconsequential, so it is ignored + process.on("uncaughtException", ({ message, stack }) => { + if (message !== "Channel closed") { + masterLog(red(message)); + if (stack) { + masterLog(`uncaught exception\n${red(stack)}`); + } + } + }); + + // determines whether or not we see the compilation / initialization / runtime output of each child server process + setupMaster({ silent: !showServerOutput }); + + // attempts to kills the active worker ungracefully + const tryKillActiveWorker = (graceful = false): boolean => { + if (activeWorker && !activeWorker.isDead()) { + if (graceful) { + activeWorker.kill(); + } else { + activeWorker.process.kill(); + } + return true; + } + return false; + }; + + const restart = () => { + // indicate to the worker that we are 'expecting' this restart + activeWorker.send({ setResponsiveness: false }); + tryKillActiveWorker(); + }; + + const setPort = (port: string, value: number, immediateRestart: boolean) => { + if (value > 1023 && value < 65536) { + ports[port] = value; + if (immediateRestart) { + restart(); + } + } else { + masterLog(red(`${port} is an invalid port number`)); + } + }; + + // kills the current active worker and proceeds to spawn a new worker, + // feeding in configuration information as environment variables + const spawn = (): void => { + tryKillActiveWorker(); + activeWorker = fork({ + pollingRoute, + serverPort: ports.server, + socketPort: ports.socket, + pollingIntervalSeconds, + session_key: key + }); + masterLog(`spawned new server worker with process id ${activeWorker.process.pid}`); + // an IPC message handler that executes actions on the master thread when prompted by the active worker + activeWorker.on("message", async ({ lifecycle, action }) => { + if (action) { + const { message, args } = action as SessionAction; + console.log(timestamp(), `${workerIdentifier} action requested (${cyan(message)})`); + switch (message) { + case "kill": + masterLog(red("an authorized user has manually ended the server session")); + tryKillActiveWorker(true); + process.exit(0); + case "notify_crash": + if (notifiers && notifiers.crash) { + const { error } = args; + const success = await notifiers.crash(error); + const statement = success ? green("distributed crash notification to recipients") : red("distribution of crash notification failed"); + masterLog(statement); + } + case "set_port": + const { port, value, immediateRestart } = args; + setPort(port, value, immediateRestart); + default: + const handler = childMessageHandlers[message]; + if (handler) { + handler(action, args); + } + } + } else if (lifecycle) { + console.log(timestamp(), `${workerIdentifier} lifecycle phase (${lifecycle})`); + } + }); + }; + + // a helpful cluster event called on the master thread each time a child process exits + on("exit", ({ process: { pid } }, code, signal) => { + const prompt = `server worker with process id ${pid} has exited with code ${code}${signal === null ? "" : `, having encountered signal ${signal}`}.`; + masterLog(cyan(prompt)); + // to make this a robust, continuous session, every time a child process dies, we immediately spawn a new one + spawn(); + }); + + // builds the repl that allows the following commands to be typed into stdin of the master thread + const repl = new Repl({ identifier: () => `${timestamp()} ${masterIdentifier}` }); + repl.registerCommand("exit", [], () => execSync(onWindows ? "taskkill /f /im node.exe" : "killall -9 node")); + repl.registerCommand("restart", [], restart); + repl.registerCommand("set", [/[a-zA-Z]+/, "port", /\d+/, /true|false/], args => setPort(args[0], Number(args[2]), args[3] === "true")); + // finally, set things in motion by spawning off the first child (server) process + spawn(); + + // returned to allow the caller to add custom commands + return { + addReplCommand: repl.registerCommand, + addChildMessageHandler: (message: string, handler: ActionHandler) => { childMessageHandlers[message] = handler; } + }; + } + + /** + * Effectively, each worker repairs the connection to the server by reintroducing a consistent state + * if its predecessor has died. It itself also polls the server heartbeat, and exits with a notification + * email if the server encounters an uncaught exception or if the server cannot be reached. + * @param work the function specifying the work to be done by each worker thread + */ + export async function initializeWorkerThread(work: Function): Promise<(handler: ExitHandler) => void> { + let shouldServerBeResponsive = false; + const exitHandlers: ExitHandler[] = []; + + // notify master thread (which will log update in the console) of initialization via IPC + process.send?.({ lifecycle: green("compiling and initializing...") }); + + // updates the local value of listening to the value sent from master + process.on("message", ({ setResponsiveness }) => shouldServerBeResponsive = setResponsiveness); + + // called whenever the process has a reason to terminate, either through an uncaught exception + // in the process (potentially inconsistent state) or the server cannot be reached + const activeExit = async (error: Error): Promise<void> => { + if (!shouldServerBeResponsive) { + return; + } + shouldServerBeResponsive = false; + // communicates via IPC to the master thread that it should dispatch a crash notification email + process.send?.({ + action: { + message: "notify_crash", + args: { error } + } + }); + await Promise.all(exitHandlers.map(handler => handler(error))); + // notify master thread (which will log update in the console) of crash event via IPC + process.send?.({ lifecycle: red(`crash event detected @ ${new Date().toUTCString()}`) }); + process.send?.({ lifecycle: red(error.message) }); + process.exit(1); + }; + + // one reason to exit, as the process might be in an inconsistent state after such an exception + process.on('uncaughtException', activeExit); + + const { + pollingIntervalSeconds, + pollingRoute, + serverPort + } = process.env; + // this monitors the health of the server by submitting a get request to whatever port / route specified + // by the configuration every n seconds, where n is also given by the configuration. + const pollTarget = `http://localhost:${serverPort}${pollingRoute}`; + const pollServer = async (): Promise<void> => { + await new Promise<void>(resolve => { + setTimeout(async () => { + try { + await get(pollTarget); + if (!shouldServerBeResponsive) { + // notify master thread (which will log update in the console) via IPC that the server is up and running + process.send?.({ lifecycle: green(`listening on ${serverPort}...`) }); + } + shouldServerBeResponsive = true; + resolve(); + } catch (error) { + // if we expect the server to be unavailable, i.e. during compilation, + // the listening variable is false, activeExit will return early and the child + // process will continue + activeExit(error); + } + }, 1000 * Number(pollingIntervalSeconds)); + }); + // controlled, asynchronous infinite recursion achieves a persistent poll that does not submit a new request until the previous has completed + pollServer(); + }; + + work(); + pollServer(); // begin polling + + return (handler: ExitHandler) => exitHandlers.push(handler); + } + +}
\ No newline at end of file |