From 1bc1ed8c5c31a0d3126d27434c690b8d744971ba Mon Sep 17 00:00:00 2001 From: Sam Wilkins Date: Wed, 8 Jan 2020 02:42:42 -0500 Subject: final session changes --- src/server/DashSession.ts | 61 +++++++++++++++++++++ src/server/Session/session.ts | 125 ++++++++++++++++++++++++++++++------------ src/server/index.ts | 57 ++----------------- 3 files changed, 157 insertions(+), 86 deletions(-) create mode 100644 src/server/DashSession.ts (limited to 'src') diff --git a/src/server/DashSession.ts b/src/server/DashSession.ts new file mode 100644 index 000000000..9c36fa17f --- /dev/null +++ b/src/server/DashSession.ts @@ -0,0 +1,61 @@ +import { Session } from "./Session/session"; +import { Email } from "./ActionUtilities"; +import { red, yellow } from "colors"; +import { SolrManager } from "./ApiManagers/SearchManager"; +import { execSync } from "child_process"; +import { isMaster } from "cluster"; +import { Utils } from "../Utils"; +import { WebSocket } from "./Websocket/Websocket"; +import { MessageStore } from "./Message"; +import { launchServer } from "."; + +const notificationRecipients = ["samuel_wilkins@brown.edu"]; +const signature = "-Dash Server Session Manager"; + +const monitorHooks: Session.MonitorNotifierHooks = { + key: async (key, masterLog) => { + const content = `The key for this session (started @ ${new Date().toUTCString()}) is ${key}.\n\n${signature}`; + const failures = await Email.dispatchAll(notificationRecipients, "Server Termination Key", content); + if (failures) { + failures.map(({ recipient, error: { message } }) => masterLog(red(`dispatch failure @ ${recipient} (${yellow(message)})`))); + return false; + } + return true; + }, + crash: async ({ name, message, stack }, masterLog) => { + const body = [ + "You, as a Dash Administrator, are being notified of a server crash event. Here's what we know:", + `name:\n${name}`, + `message:\n${message}`, + `stack:\n${stack}`, + "The server is already restarting itself, but if you're concerned, use the Remote Desktop Connection to monitor progress.", + ].join("\n\n"); + const content = `${body}\n\n${signature}`; + const failures = await Email.dispatchAll(notificationRecipients, "Dash Web Server Crash", content); + if (failures) { + failures.map(({ recipient, error: { message } }) => masterLog(red(`dispatch failure @ ${recipient} (${yellow(message)})`))); + return false; + } + return true; + } +}; + +export class DashSessionAgent extends Session.AppliedSessionAgent { + + /** + * If we're the monitor (master) thread, we should launch the monitor logic for the session. + * Otherwise, we must be on a worker thread that was spawned *by* the monitor (master) thread, and thus + * our job should be to run the server. + */ + protected async launchImplementation() { + if (isMaster) { + this.sessionMonitor = await Session.initializeMonitorThread(monitorHooks); + this.sessionMonitor.addReplCommand("pull", [], () => execSync("git pull", { stdio: ["ignore", "inherit", "inherit"] })); + this.sessionMonitor.addReplCommand("solr", [/start|stop/g], args => SolrManager.SetRunning(args[0] === "start")); + } else { + this.serverWorker = await Session.initializeWorkerThread(launchServer); // server initialization delegated to worker + this.serverWorker.addExitHandler(() => Utils.Emit(WebSocket._socket, MessageStore.ConnectionTerminated, "Manual")); + } + } + +} \ No newline at end of file diff --git a/src/server/Session/session.ts b/src/server/Session/session.ts index 2a483fbab..b22b6404d 100644 --- a/src/server/Session/session.ts +++ b/src/server/Session/session.ts @@ -1,5 +1,5 @@ import { red, cyan, green, yellow, magenta, blue } from "colors"; -import { on, fork, setupMaster, Worker } from "cluster"; +import { on, fork, setupMaster, Worker, isMaster } from "cluster"; import { get } from "request-promise"; import { Utils } from "../../Utils"; import Repl, { ReplAction } from "../repl"; @@ -20,6 +20,51 @@ import { configurationSchema } from "./session_config_schema"; */ export namespace Session { + export abstract class AppliedSessionAgent { + + private launched = false; + + protected sessionMonitorRef: Session.Monitor | undefined; + public get sessionMonitor(): Session.Monitor { + if (!isMaster) { + throw new Error("Cannot access the session monitor directly from the server worker thread"); + } + return this.sessionMonitorRef!; + } + public set sessionMonitor(monitor: Session.Monitor) { + if (!isMaster) { + throw new Error("Cannot set the session monitor directly from the server worker thread"); + } + this.sessionMonitorRef = monitor; + } + + protected serverWorkerRef: Session.ServerWorker | undefined; + public get serverWorker(): Session.ServerWorker { + if (isMaster) { + throw new Error("Cannot access the server worker directly from the session monitor thread"); + } + return this.serverWorkerRef!; + } + public set serverWorker(worker: Session.ServerWorker) { + if (isMaster) { + throw new Error("Cannot set the server worker directly from the session monitor thread"); + } + this.serverWorkerRef = worker; + } + + public async launch(): Promise { + if (!this.launched) { + this.launched = true; + await this.launchImplementation(); + } else { + throw new Error("Cannot launch a session thread more than once per process."); + } + } + + protected abstract async launchImplementation(): Promise; + + } + interface Configuration { showServerOutput: boolean; masterIdentifier: string; @@ -41,12 +86,21 @@ export namespace Session { pollingFailureTolerance: 1 }; - interface MasterExtensions { + export interface Monitor { + log: (...optionalParams: any[]) => void; + restartServer: () => void; + setPort: (port: "server" | "socket" | string, value: number, immediateRestart: boolean) => void; + killSession: (graceful?: boolean) => never; addReplCommand: (basename: string, argPatterns: (RegExp | string)[], action: ReplAction) => void; addChildMessageHandler: (message: string, handler: ActionHandler) => void; } - export interface NotifierHooks { + export interface ServerWorker { + killSession: () => void; + addExitHandler: (handler: ExitHandler) => void; + } + + export interface MonitorNotifierHooks { key?: (key: string, masterLog: (...optionalParams: any[]) => void) => boolean | Promise; crash?: (error: Error, masterLog: (...optionalParams: any[]) => void) => boolean | Promise; } @@ -56,14 +110,8 @@ export namespace Session { args: any; } - export interface SessionHooks { - masterLog: (...optionalParams: any[]) => void; - killSession: (graceful?: boolean) => never; - restartServer: () => void; - } - - export type ExitHandler = (error: Error) => void | Promise; - export type ActionHandler = (action: SessionAction, hooks: SessionHooks) => void | Promise; + export type ExitHandler = (reason: Error | null) => void | Promise; + export type ActionHandler = (action: SessionAction) => void | Promise; export interface EmailTemplate { subject: string; body: string; @@ -125,7 +173,7 @@ export namespace Session { * 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 { + export async function initializeMonitorThread(notifiers?: MonitorNotifierHooks): Promise { console.log(timestamp(), cyan("initializing session...")); let activeWorker: Worker; const childMessageHandlers: { [message: string]: ActionHandler } = {}; @@ -143,26 +191,26 @@ export namespace Session { } = configuration; let { pollingIntervalSeconds } = configuration; - const masterLog = (...optionalParams: any[]) => console.log(timestamp(), masterIdentifier, ...optionalParams); + const log = (...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, masterLog); + const success = await notifiers.key(key, log); const statement = success ? green("distributed session key to recipients") : red("distribution of session key failed"); - masterLog(statement); + log(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 }) => { + process.on("uncaughtException", ({ message, stack }): void => { if (message !== "Channel closed") { - masterLog(red(message)); + log(red(message)); if (stack) { - masterLog(`uncaught exception\n${red(stack)}`); + log(`uncaught exception\n${red(stack)}`); } } }); @@ -183,26 +231,26 @@ export namespace Session { return false; }; - const restartServer = () => { + const restartServer = (): void => { // indicate to the worker that we are 'expecting' this restart activeWorker.send({ setResponsiveness: false }); tryKillActiveWorker(true); }; - const killSession = (graceful = true) => { - masterLog(cyan(`exiting session ${graceful ? "clean" : "immediate"}ly`)); + const killSession = (graceful = true): never => { + log(cyan(`exiting session ${graceful ? "clean" : "immediate"}ly`)); tryKillActiveWorker(graceful); process.exit(0); }; - const setPort = (port: string, value: number, immediateRestart: boolean) => { + const setPort = (port: "server" | "socket" | string, value: number, immediateRestart: boolean): void => { if (value > 1023 && value < 65536) { ports[port] = value; if (immediateRestart) { restartServer(); } } else { - masterLog(red(`${port} is an invalid port number`)); + log(red(`${port} is an invalid port number`)); } }; @@ -218,7 +266,7 @@ export namespace Session { pollingIntervalSeconds, session_key: key }); - masterLog(cyan(`spawned new server worker with process id ${activeWorker.process.pid}`)); + log(cyan(`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) { @@ -226,14 +274,14 @@ export namespace Session { console.log(timestamp(), `${workerIdentifier} action requested (${cyan(message)})`); switch (message) { case "kill": - masterLog(red("an authorized user has manually ended the server session")); + log(red("an authorized user has manually ended the server session")); killSession(); case "notify_crash": if (notifiers && notifiers.crash) { const { error } = args; - const success = await notifiers.crash(error, masterLog); + const success = await notifiers.crash(error, log); const statement = success ? green("distributed crash notification to recipients") : red("distribution of crash notification failed"); - masterLog(statement); + log(statement); } case "set_port": const { port, value, immediateRestart } = args; @@ -241,7 +289,7 @@ export namespace Session { default: const handler = childMessageHandlers[message]; if (handler) { - handler({ message, args }, { restartServer, killSession, masterLog }); + handler({ message, args }); } } } else if (lifecycle) { @@ -253,7 +301,7 @@ export namespace Session { // 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)); + log(cyan(prompt)); // to make this a robust, continuous session, every time a child process dies, we immediately spawn a new one spawn(); }); @@ -269,7 +317,7 @@ export namespace Session { repl.registerCommand("set", [/polling/, number, boolean], args => { const newPollingIntervalSeconds = Math.floor(Number(args[2])); if (newPollingIntervalSeconds < 0) { - masterLog(red("the polling interval must be a non-negative integer")); + log(red("the polling interval must be a non-negative integer")); } else { if (newPollingIntervalSeconds !== pollingIntervalSeconds) { pollingIntervalSeconds = newPollingIntervalSeconds; @@ -285,7 +333,11 @@ export namespace Session { // returned to allow the caller to add custom commands return { addReplCommand: repl.registerCommand, - addChildMessageHandler: (message: string, handler: ActionHandler) => { childMessageHandlers[message] = handler; } + addChildMessageHandler: (message: string, handler: ActionHandler) => { childMessageHandlers[message] = handler; }, + restartServer, + killSession, + setPort, + log }; } @@ -295,7 +347,7 @@ export namespace Session { * 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> { + export async function initializeWorkerThread(work: Function): Promise { let shouldServerBeResponsive = false; const exitHandlers: ExitHandler[] = []; let pollingFailureCount = 0; @@ -315,6 +367,8 @@ export namespace Session { } }); + const executeExitHandlers = async (reason: Error | null) => Promise.all(exitHandlers.map(handler => handler(reason))); + // 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 => { @@ -326,7 +380,7 @@ export namespace Session { args: { error } } }); - await Promise.all(exitHandlers.map(handler => handler(error))); + await executeExitHandlers(error); // notify master thread (which will log update in the console) of crash event via IPC lifecycleNotification(red(`crash event detected @ ${new Date().toUTCString()}`)); lifecycleNotification(red(error.message)); @@ -375,7 +429,10 @@ export namespace Session { work(); pollServer(); // begin polling - return (handler: ExitHandler) => exitHandlers.push(handler); + return { + addExitHandler: (handler: ExitHandler) => exitHandlers.push(handler), + killSession: () => process.send!({ action: { message: "kill" } }) + }; } } \ No newline at end of file diff --git a/src/server/index.ts b/src/server/index.ts index 9bc3c7617..27e9a677b 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -3,7 +3,6 @@ import { GoogleApiServerUtils } from "./apis/google/GoogleApiServerUtils"; import * as mobileDetect from 'mobile-detect'; import * as path from 'path'; import { Database } from './database'; -const serverPort = 4321; import { DashUploadUtils } from './DashUploadUtils'; import RouteSubscriber from './RouteSubscriber'; import initializeServer from './server_initialization'; @@ -24,10 +23,7 @@ import GooglePhotosManager from "./ApiManagers/GooglePhotosManager"; import { Logger } from "./ProcessFactory"; import { yellow, red } from "colors"; import { Session } from "./Session/session"; -import { isMaster } from "cluster"; -import { execSync } from "child_process"; -import { Utils } from "../Utils"; -import { MessageStore } from "./Message"; +import { DashSessionAgent } from "./DashSession"; export const publicDirectory = path.resolve(__dirname, "public"); export const filesDirectory = path.resolve(publicDirectory, "files"); @@ -96,7 +92,7 @@ function routeSetter({ isRelease, addSupervisedRoute, logRegistrationOutcome }: secureHandler: ({ req, res }) => { if (req.params.key === process.env.session_key) { res.send(""); - process.send!({ action: { message: "kill" } }); + sessionAgent.serverWorker.killSession(); } else { res.redirect("/home"); } @@ -136,7 +132,7 @@ function routeSetter({ isRelease, addSupervisedRoute, logRegistrationOutcome }: * however, this becomes the logic invoked by a single worker thread spawned by * the main monitor (master) thread. */ -async function launchServer() { +export async function launchServer() { await log_execution({ startMessage: "\nstarting execution of preliminary functions", endMessage: "completed preliminary functions\n", @@ -145,50 +141,7 @@ async function launchServer() { await initializeServer(routeSetter); } -/** - * If we're the monitor (master) thread, we should launch the monitor logic for the session. - * Otherwise, we must be on a worker thread that was spawned *by* the monitor (master) thread, and thus - * our job should be to run the server. - */ -async function launchMonitoredSession() { - if (isMaster) { - const notificationRecipients = ["samuel_wilkins@brown.edu"]; - const signature = "-Dash Server Session Manager"; - const extensions = await Session.initializeMonitorThread({ - key: async (key, masterLog) => { - const content = `The key for this session (started @ ${new Date().toUTCString()}) is ${key}.\n\n${signature}`; - const failures = await Email.dispatchAll(notificationRecipients, "Server Termination Key", content); - if (failures) { - failures.map(({ recipient, error: { message } }) => masterLog(red(`dispatch failure @ ${recipient} (${yellow(message)})`))); - return false; - } - return true; - }, - crash: async ({ name, message, stack }, masterLog) => { - const body = [ - "You, as a Dash Administrator, are being notified of a server crash event. Here's what we know:", - `name:\n${name}`, - `message:\n${message}`, - `stack:\n${stack}`, - "The server is already restarting itself, but if you're concerned, use the Remote Desktop Connection to monitor progress.", - ].join("\n\n"); - const content = `${body}\n\n${signature}`; - const failures = await Email.dispatchAll(notificationRecipients, "Dash Web Server Crash", content); - if (failures) { - failures.map(({ recipient, error: { message } }) => masterLog(red(`dispatch failure @ ${recipient} (${yellow(message)})`))); - return false; - } - return true; - } - }); - extensions.addReplCommand("pull", [], () => execSync("git pull", { stdio: ["ignore", "inherit", "inherit"] })); - extensions.addReplCommand("solr", [/start|stop/g], args => SolrManager.SetRunning(args[0] === "start")); - } else { - const addExitHandler = await Session.initializeWorkerThread(launchServer); // server initialization delegated to worker - addExitHandler(() => Utils.Emit(WebSocket._socket, MessageStore.ConnectionTerminated, "Manual")); - } -} - +export const sessionAgent: Session.AppliedSessionAgent = new DashSessionAgent(); /** * If you're in development mode, you won't need to run a session. * The session spawns off new server processes each time an error is encountered, and doesn't @@ -196,7 +149,7 @@ async function launchMonitoredSession() { * So, the 'else' clause is exactly what we've always run when executing npm start. */ if (process.env.RELEASE) { - launchMonitoredSession(); + sessionAgent.launch(); } else { launchServer(); } \ No newline at end of file -- cgit v1.2.3-70-g09d2 From da50d941f597fb3ec52bc3653d94f8c30affe552 Mon Sep 17 00:00:00 2001 From: Sam Wilkins Date: Wed, 8 Jan 2020 02:43:26 -0500 Subject: small change --- src/server/index.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'src') diff --git a/src/server/index.ts b/src/server/index.ts index 27e9a677b..f18e9abb6 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -141,7 +141,7 @@ export async function launchServer() { await initializeServer(routeSetter); } -export const sessionAgent: Session.AppliedSessionAgent = new DashSessionAgent(); +export const sessionAgent = new DashSessionAgent(); /** * If you're in development mode, you won't need to run a session. * The session spawns off new server processes each time an error is encountered, and doesn't -- cgit v1.2.3-70-g09d2 From 7378b5d063d9da34d485c8384efa71ba83272a61 Mon Sep 17 00:00:00 2001 From: Sam Wilkins Date: Wed, 8 Jan 2020 03:42:14 -0500 Subject: session cleanup --- src/client/DocServer.ts | 4 +- src/server/DashSession.ts | 89 +++++++++++++++++++++---------------------- src/server/Session/session.ts | 69 +++++++++++++++++++++------------ src/server/index.ts | 2 +- 4 files changed, 92 insertions(+), 72 deletions(-) (limited to 'src') diff --git a/src/client/DocServer.ts b/src/client/DocServer.ts index 47c63bfb7..ed7fbd7ba 100644 --- a/src/client/DocServer.ts +++ b/src/client/DocServer.ts @@ -82,7 +82,9 @@ export namespace DocServer { Utils.AddServerHandler(_socket, MessageStore.UpdateField, respondToUpdate); Utils.AddServerHandler(_socket, MessageStore.DeleteField, respondToDelete); Utils.AddServerHandler(_socket, MessageStore.DeleteFields, respondToDelete); - Utils.AddServerHandler(_socket, MessageStore.ConnectionTerminated, () => alert("Your connection to the server has been terminated.")); + Utils.AddServerHandler(_socket, MessageStore.ConnectionTerminated, () => { + alert("Your connection to the server has been terminated."); + }); } function errorFunc(): never { diff --git a/src/server/DashSession.ts b/src/server/DashSession.ts index 9c36fa17f..c0ebc9687 100644 --- a/src/server/DashSession.ts +++ b/src/server/DashSession.ts @@ -3,59 +3,58 @@ import { Email } from "./ActionUtilities"; import { red, yellow } from "colors"; import { SolrManager } from "./ApiManagers/SearchManager"; import { execSync } from "child_process"; -import { isMaster } from "cluster"; import { Utils } from "../Utils"; import { WebSocket } from "./Websocket/Websocket"; import { MessageStore } from "./Message"; import { launchServer } from "."; -const notificationRecipients = ["samuel_wilkins@brown.edu"]; -const signature = "-Dash Server Session Manager"; +/** +* If we're the monitor (master) thread, we should launch the monitor logic for the session. +* Otherwise, we must be on a worker thread that was spawned *by* the monitor (master) thread, and thus +* our job should be to run the server. +*/ +export class DashSessionAgent extends Session.AppliedSessionAgent { -const monitorHooks: Session.MonitorNotifierHooks = { - key: async (key, masterLog) => { - const content = `The key for this session (started @ ${new Date().toUTCString()}) is ${key}.\n\n${signature}`; - const failures = await Email.dispatchAll(notificationRecipients, "Server Termination Key", content); - if (failures) { - failures.map(({ recipient, error: { message } }) => masterLog(red(`dispatch failure @ ${recipient} (${yellow(message)})`))); - return false; - } - return true; - }, - crash: async ({ name, message, stack }, masterLog) => { - const body = [ - "You, as a Dash Administrator, are being notified of a server crash event. Here's what we know:", - `name:\n${name}`, - `message:\n${message}`, - `stack:\n${stack}`, - "The server is already restarting itself, but if you're concerned, use the Remote Desktop Connection to monitor progress.", - ].join("\n\n"); - const content = `${body}\n\n${signature}`; - const failures = await Email.dispatchAll(notificationRecipients, "Dash Web Server Crash", content); - if (failures) { - failures.map(({ recipient, error: { message } }) => masterLog(red(`dispatch failure @ ${recipient} (${yellow(message)})`))); - return false; - } - return true; - } -}; + private readonly notificationRecipients = ["samuel_wilkins@brown.edu"]; + private readonly signature = "-Dash Server Session Manager"; -export class DashSessionAgent extends Session.AppliedSessionAgent { + protected async launchMonitor() { + const monitor = await Session.initializeMonitorThread({ + key: async (key, masterLog) => { + const content = `The key for this session (started @ ${new Date().toUTCString()}) is ${key}.\n\n${this.signature}`; + const failures = await Email.dispatchAll(this.notificationRecipients, "Server Termination Key", content); + if (failures) { + failures.map(({ recipient, error: { message } }) => masterLog(red(`dispatch failure @ ${recipient} (${yellow(message)})`))); + return false; + } + return true; + }, + crash: async ({ name, message, stack }, masterLog) => { + const body = [ + "You, as a Dash Administrator, are being notified of a server crash event. Here's what we know:", + `name:\n${name}`, + `message:\n${message}`, + `stack:\n${stack}`, + "The server is already restarting itself, but if you're concerned, use the Remote Desktop Connection to monitor progress.", + ].join("\n\n"); + const content = `${body}\n\n${this.signature}`; + const failures = await Email.dispatchAll(this.notificationRecipients, "Dash Web Server Crash", content); + if (failures) { + failures.map(({ recipient, error: { message } }) => masterLog(red(`dispatch failure @ ${recipient} (${yellow(message)})`))); + return false; + } + return true; + } + }); + monitor.addReplCommand("pull", [], () => execSync("git pull", { stdio: ["ignore", "inherit", "inherit"] })); + monitor.addReplCommand("solr", [/start|stop/g], args => SolrManager.SetRunning(args[0] === "start")); + return monitor; + } - /** - * If we're the monitor (master) thread, we should launch the monitor logic for the session. - * Otherwise, we must be on a worker thread that was spawned *by* the monitor (master) thread, and thus - * our job should be to run the server. - */ - protected async launchImplementation() { - if (isMaster) { - this.sessionMonitor = await Session.initializeMonitorThread(monitorHooks); - this.sessionMonitor.addReplCommand("pull", [], () => execSync("git pull", { stdio: ["ignore", "inherit", "inherit"] })); - this.sessionMonitor.addReplCommand("solr", [/start|stop/g], args => SolrManager.SetRunning(args[0] === "start")); - } else { - this.serverWorker = await Session.initializeWorkerThread(launchServer); // server initialization delegated to worker - this.serverWorker.addExitHandler(() => Utils.Emit(WebSocket._socket, MessageStore.ConnectionTerminated, "Manual")); - } + protected async launchServerWorker() { + const worker = await Session.initializeWorkerThread(launchServer); // server initialization delegated to worker + worker.addExitHandler(() => Utils.Emit(WebSocket._socket, MessageStore.ConnectionTerminated, "Manual")); + return worker; } } \ No newline at end of file diff --git a/src/server/Session/session.ts b/src/server/Session/session.ts index b22b6404d..cc9e7dd1a 100644 --- a/src/server/Session/session.ts +++ b/src/server/Session/session.ts @@ -22,46 +22,44 @@ export namespace Session { export abstract class AppliedSessionAgent { + public killSession(graceful = true) { + const target = isMaster ? this.sessionMonitor : this.serverWorker; + target.killSession(graceful); + } + private launched = false; - protected sessionMonitorRef: Session.Monitor | undefined; + private sessionMonitorRef: Session.Monitor | undefined; public get sessionMonitor(): Session.Monitor { if (!isMaster) { throw new Error("Cannot access the session monitor directly from the server worker thread"); } return this.sessionMonitorRef!; } - public set sessionMonitor(monitor: Session.Monitor) { - if (!isMaster) { - throw new Error("Cannot set the session monitor directly from the server worker thread"); - } - this.sessionMonitorRef = monitor; - } - protected serverWorkerRef: Session.ServerWorker | undefined; + private serverWorkerRef: Session.ServerWorker | undefined; public get serverWorker(): Session.ServerWorker { if (isMaster) { throw new Error("Cannot access the server worker directly from the session monitor thread"); } return this.serverWorkerRef!; } - public set serverWorker(worker: Session.ServerWorker) { - if (isMaster) { - throw new Error("Cannot set the server worker directly from the session monitor thread"); - } - this.serverWorkerRef = worker; - } public async launch(): Promise { if (!this.launched) { this.launched = true; - await this.launchImplementation(); + if (isMaster) { + this.sessionMonitorRef = await this.launchMonitor(); + } else { + this.serverWorkerRef = await this.launchServerWorker(); + } } else { throw new Error("Cannot launch a session thread more than once per process."); } } - protected abstract async launchImplementation(): Promise; + protected abstract async launchMonitor(): Promise; + protected abstract async launchServerWorker(): Promise; } @@ -92,11 +90,14 @@ export namespace Session { setPort: (port: "server" | "socket" | string, value: number, immediateRestart: boolean) => void; killSession: (graceful?: boolean) => never; addReplCommand: (basename: string, argPatterns: (RegExp | string)[], action: ReplAction) => void; - addChildMessageHandler: (message: string, handler: ActionHandler) => void; + addServerMessageListener: (message: string, handler: ActionHandler) => void; + removeServerMessageListener: (message: string, handler: ActionHandler) => void; + clearServerMessageListeners: (message: string) => void; } export interface ServerWorker { - killSession: () => void; + killSession: (graceful?: boolean) => void; + sendSessionAction: (message: string, args?: any) => void; addExitHandler: (handler: ExitHandler) => void; } @@ -176,7 +177,7 @@ export namespace Session { export async function initializeMonitorThread(notifiers?: MonitorNotifierHooks): Promise { console.log(timestamp(), cyan("initializing session...")); let activeWorker: Worker; - const childMessageHandlers: { [message: string]: ActionHandler } = {}; + const onMessage: { [message: string]: ActionHandler[] | undefined } = {}; // read in configuration .json file only once, in the master thread // pass down any variables the pertinent to the child processes as environment variables @@ -275,7 +276,7 @@ export namespace Session { switch (message) { case "kill": log(red("an authorized user has manually ended the server session")); - killSession(); + killSession(args.graceful); case "notify_crash": if (notifiers && notifiers.crash) { const { error } = args; @@ -287,9 +288,9 @@ export namespace Session { const { port, value, immediateRestart } = args; setPort(port, value, immediateRestart); default: - const handler = childMessageHandlers[message]; - if (handler) { - handler({ message, args }); + const handlers = onMessage[message]; + if (handlers) { + handlers.forEach(handler => handler({ message, args })); } } } else if (lifecycle) { @@ -333,7 +334,24 @@ export namespace Session { // returned to allow the caller to add custom commands return { addReplCommand: repl.registerCommand, - addChildMessageHandler: (message: string, handler: ActionHandler) => { childMessageHandlers[message] = handler; }, + addServerMessageListener: (message: string, handler: ActionHandler) => { + const handlers = onMessage[message]; + if (handlers) { + handlers.push(handler); + } else { + onMessage[message] = [handler]; + } + }, + removeServerMessageListener: (message: string, handler: ActionHandler) => { + const handlers = onMessage[message]; + if (handlers) { + const index = handlers.indexOf(handler); + if (index > -1) { + handlers.splice(index, 1); + } + } + }, + clearServerMessageListeners: (message: string) => onMessage[message] = undefined, restartServer, killSession, setPort, @@ -431,7 +449,8 @@ export namespace Session { return { addExitHandler: (handler: ExitHandler) => exitHandlers.push(handler), - killSession: () => process.send!({ action: { message: "kill" } }) + killSession: (graceful = true) => process.send!({ action: { message: "kill", args: { graceful } } }), + sendSessionAction: (message: string, args?: any) => process.send!({ action: { message, args } }) }; } diff --git a/src/server/index.ts b/src/server/index.ts index f18e9abb6..ffab0f380 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -92,7 +92,7 @@ function routeSetter({ isRelease, addSupervisedRoute, logRegistrationOutcome }: secureHandler: ({ req, res }) => { if (req.params.key === process.env.session_key) { res.send(""); - sessionAgent.serverWorker.killSession(); + sessionAgent.killSession(); } else { res.redirect("/home"); } -- cgit v1.2.3-70-g09d2 From 78bedabbbe0682d089c343ad94d90d0311bdfe0e Mon Sep 17 00:00:00 2001 From: Sam Wilkins Date: Wed, 8 Jan 2020 06:30:51 -0500 Subject: graceful exiting --- src/server/DashSession.ts | 14 +- src/server/Session/session.ts | 696 ++++++++++++++++++++++++------------------ src/server/index.ts | 4 +- 3 files changed, 402 insertions(+), 312 deletions(-) (limited to 'src') diff --git a/src/server/DashSession.ts b/src/server/DashSession.ts index c0ebc9687..7c2cfaf8d 100644 --- a/src/server/DashSession.ts +++ b/src/server/DashSession.ts @@ -19,17 +19,19 @@ export class DashSessionAgent extends Session.AppliedSessionAgent { private readonly signature = "-Dash Server Session Manager"; protected async launchMonitor() { - const monitor = await Session.initializeMonitorThread({ - key: async (key, masterLog) => { + const monitor = new Session.Monitor({ + key: async key => { + // this sends a pseudorandomly generated guid to the configuration's recipients, allowing them alone + // to kill the server via the /kill/:key route const content = `The key for this session (started @ ${new Date().toUTCString()}) is ${key}.\n\n${this.signature}`; const failures = await Email.dispatchAll(this.notificationRecipients, "Server Termination Key", content); if (failures) { - failures.map(({ recipient, error: { message } }) => masterLog(red(`dispatch failure @ ${recipient} (${yellow(message)})`))); + failures.map(({ recipient, error: { message } }) => monitor.log(red(`dispatch failure @ ${recipient} (${yellow(message)})`))); return false; } return true; }, - crash: async ({ name, message, stack }, masterLog) => { + crash: async ({ name, message, stack }) => { const body = [ "You, as a Dash Administrator, are being notified of a server crash event. Here's what we know:", `name:\n${name}`, @@ -40,7 +42,7 @@ export class DashSessionAgent extends Session.AppliedSessionAgent { const content = `${body}\n\n${this.signature}`; const failures = await Email.dispatchAll(this.notificationRecipients, "Dash Web Server Crash", content); if (failures) { - failures.map(({ recipient, error: { message } }) => masterLog(red(`dispatch failure @ ${recipient} (${yellow(message)})`))); + failures.map(({ recipient, error: { message } }) => monitor.log(red(`dispatch failure @ ${recipient} (${yellow(message)})`))); return false; } return true; @@ -52,7 +54,7 @@ export class DashSessionAgent extends Session.AppliedSessionAgent { } protected async launchServerWorker() { - const worker = await Session.initializeWorkerThread(launchServer); // server initialization delegated to worker + const worker = new Session.ServerWorker(launchServer); // server initialization delegated to worker worker.addExitHandler(() => Utils.Emit(WebSocket._socket, MessageStore.ConnectionTerminated, "Manual")); return worker; } diff --git a/src/server/Session/session.ts b/src/server/Session/session.ts index cc9e7dd1a..8bee99f41 100644 --- a/src/server/Session/session.ts +++ b/src/server/Session/session.ts @@ -1,4 +1,4 @@ -import { red, cyan, green, yellow, magenta, blue } from "colors"; +import { red, cyan, green, yellow, magenta, blue, white } from "colors"; import { on, fork, setupMaster, Worker, isMaster } from "cluster"; import { get } from "request-promise"; import { Utils } from "../../Utils"; @@ -8,26 +8,32 @@ import { validate, ValidationError } from "jsonschema"; import { configurationSchema } from "./session_config_schema"; /** - * 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. - */ + * 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 { export abstract class AppliedSessionAgent { + // the following two methods allow the developer to create a custom + // session and use the built in customization options for each thread + protected abstract async launchMonitor(): Promise; + protected abstract async launchServerWorker(): Promise; + + private launched = false; + public killSession(graceful = true) { const target = isMaster ? this.sessionMonitor : this.serverWorker; target.killSession(graceful); } - private launched = false; private sessionMonitorRef: Session.Monitor | undefined; public get sessionMonitor(): Session.Monitor { @@ -58,9 +64,6 @@ export namespace Session { } } - protected abstract async launchMonitor(): Promise; - protected abstract async launchServerWorker(): Promise; - } interface Configuration { @@ -81,377 +84,462 @@ export namespace Session { ports: { server: 3000 }, pollingRoute: "/", pollingIntervalSeconds: 30, - pollingFailureTolerance: 1 + pollingFailureTolerance: 0 }; - export interface Monitor { - log: (...optionalParams: any[]) => void; - restartServer: () => void; - setPort: (port: "server" | "socket" | string, value: number, immediateRestart: boolean) => void; - killSession: (graceful?: boolean) => never; - addReplCommand: (basename: string, argPatterns: (RegExp | string)[], action: ReplAction) => void; - addServerMessageListener: (message: string, handler: ActionHandler) => void; - removeServerMessageListener: (message: string, handler: ActionHandler) => void; - clearServerMessageListeners: (message: string) => void; - } + export type ExitHandler = (reason: Error | null) => void | Promise; - export interface ServerWorker { - killSession: (graceful?: boolean) => void; - sendSessionAction: (message: string, args?: any) => void; - addExitHandler: (handler: ExitHandler) => void; - } + export namespace Monitor { - export interface MonitorNotifierHooks { - key?: (key: string, masterLog: (...optionalParams: any[]) => void) => boolean | Promise; - crash?: (error: Error, masterLog: (...optionalParams: any[]) => void) => boolean | Promise; - } + export interface NotifierHooks { + key?: (key: string) => (boolean | Promise); + crash?: (error: Error) => (boolean | Promise); + } - export interface SessionAction { - message: string; - args: any; - } + export interface Action { + message: string; + args: any; + } + + export type ServerMessageHandler = (action: Action) => void | Promise; - export type ExitHandler = (reason: Error | null) => void | Promise; - export type ActionHandler = (action: SessionAction) => void | Promise; - export interface EmailTemplate { - subject: string; - body: string; } - function loadAndValidateConfiguration(): Configuration { - try { - console.log(timestamp(), cyan("validating configuration...")); - 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; + /** + * 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 class Monitor { + + private exitHandlers: ExitHandler[] = []; + private readonly notifiers: Monitor.NotifierHooks | undefined; + private readonly configuration: Configuration; + private onMessage: { [message: string]: Monitor.ServerMessageHandler[] | undefined } = {}; + private activeWorker: Worker | undefined; + private key: string | undefined; + private repl: Repl; + + /** + * Kill this session and its active child + * server process, either gracefully (may wait + * indefinitely, but at least allows active networking + * requests to complete) or immediately. + */ + public killSession = async (graceful = true): Promise => { + this.log(cyan(`exiting session ${graceful ? "clean" : "immediate"}ly`)); + this.tryKillActiveWorker(graceful); + process.exit(0); + } + + /** + * Execute the list of functions registered to be called + * whenever the process exits. + */ + public addExitHandler = (handler: ExitHandler) => this.exitHandlers.push(handler); + + /** + * Extend the default repl by adding in custom commands + * that can invoke application logic external to this module + */ + public addReplCommand = (basename: string, argPatterns: (RegExp | string)[], action: ReplAction) => { + this.repl.registerCommand(basename, argPatterns, action); + } + + /** + * Add a listener at this message. When the monitor process + * receives a message, it will invoke all registered functions. + */ + public addServerMessageListener = (message: string, handler: Monitor.ServerMessageHandler) => { + const handlers = this.onMessage[message]; + if (handlers) { + handlers.push(handler); + } else { + this.onMessage[message] = [handler]; + } + } + + /** + * Unregister a given listener at this message. + */ + public removeServerMessageListener = (message: string, handler: Monitor.ServerMessageHandler) => { + const handlers = this.onMessage[message]; + if (handlers) { + const index = handlers.indexOf(handler); + if (index > -1) { + handlers.splice(index, 1); + } + } + } + + /** + * Unregister all listeners at this message. + */ + public clearServerMessageListeners = (message: string) => this.onMessage[message] = undefined; + + constructor(notifiers?: Monitor.NotifierHooks) { + this.notifiers = notifiers; + + console.log(this.timestamp(), cyan("initializing session...")); + + this.configuration = this.loadAndValidateConfiguration(); + this.initializeSessionKey(); + // determines whether or not we see the compilation / initialization / runtime output of each child server process + setupMaster({ silent: !this.configuration.showServerOutput }); + + // 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 }): void => { + if (message !== "Channel closed") { + this.log(red(message)); + if (stack) { + this.log(`uncaught exception\n${red(stack)}`); } - 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); - } + + // 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}`}.`; + this.log(cyan(prompt)); + // to make this a robust, continuous session, every time a child process dies, we immediately spawn a new one + this.spawn(); + }); + + this.repl = this.initializeRepl(); + this.spawn(); } - } - 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?: MonitorNotifierHooks): Promise { - console.log(timestamp(), cyan("initializing session...")); - let activeWorker: Worker; - const onMessage: { [message: string]: ActionHandler[] | undefined } = {}; - - // 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 configuration = loadAndValidateConfiguration(); - const { - masterIdentifier, - workerIdentifier, - ports, - pollingRoute, - showServerOutput, - pollingFailureTolerance - } = configuration; - let { pollingIntervalSeconds } = configuration; - - const log = (...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, log); - const statement = success ? green("distributed session key to recipients") : red("distribution of session key failed"); - log(statement); + /** + * Generates a blue UTC string associated with the time + * of invocation. + */ + private timestamp = () => blue(`[${new Date().toUTCString()}]`); + + /** + * A formatted, identified and timestamped log in color + */ + public log = (...optionalParams: any[]) => { + console.log(this.timestamp(), this.configuration.masterIdentifier, ...optionalParams); + } + + /** + * If the caller has indicated an interest + * in being notified of this feature, creates + * a GUID for this session that can, for example, + * be used as authentication for killing the server + * (checked externally). + */ + private initializeSessionKey = async (): Promise => { + if (this.notifiers?.key) { + this.key = Utils.GenerateGuid(); + const success = await this.notifiers.key(this.key); + const statement = success ? green("distributed session key to recipients") : red("distribution of session key failed"); + this.log(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 }): void => { - if (message !== "Channel closed") { - log(red(message)); - if (stack) { - log(`uncaught exception\n${red(stack)}`); + /** + * Builds the repl that allows the following commands to be typed into stdin of the master thread. + */ + private initializeRepl = (): Repl => { + const repl = new Repl({ identifier: () => `${this.timestamp()} ${this.configuration.masterIdentifier}` }); + const boolean = /true|false/; + const number = /\d+/; + const letters = /[a-zA-Z]+/; + repl.registerCommand("exit", [/clean|force/], args => this.killSession(args[0] === "clean")); + repl.registerCommand("restart", [/clean|force/], args => this.tryKillActiveWorker(args[0] === "clean")); + repl.registerCommand("set", [letters, "port", number, boolean], args => this.setPort(args[0], Number(args[2]), args[3] === "true")); + repl.registerCommand("set", [/polling/, number, boolean], args => { + const newPollingIntervalSeconds = Math.floor(Number(args[2])); + if (newPollingIntervalSeconds < 0) { + this.log(red("the polling interval must be a non-negative integer")); + } else { + if (newPollingIntervalSeconds !== this.configuration.pollingIntervalSeconds) { + this.configuration.pollingIntervalSeconds = newPollingIntervalSeconds; + if (args[3] === "true") { + this.activeWorker?.send({ newPollingIntervalSeconds }); + } + } + } + }); + return repl; + } + + /** + * Reads in configuration .json file only once, in the master thread + * and pass down any variables the pertinent to the child processes as environment variables. + */ + private loadAndValidateConfiguration = (): Configuration => { + try { + console.log(this.timestamp(), cyan("validating configuration...")); + 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); } } - }); + } - // 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, unless otherwise specified - const tryKillActiveWorker = (graceful = false): boolean => { - if (activeWorker && !activeWorker.isDead()) { + private executeExitHandlers = async (reason: Error | null) => Promise.all(this.exitHandlers.map(handler => handler(reason))); + + /** + * Attempts to kill the active worker gracefully, unless otherwise specified. + */ + private tryKillActiveWorker = (graceful = true): boolean => { + if (!this.activeWorker?.isDead()) { if (graceful) { - activeWorker.kill(); + this.activeWorker?.send({ manualExit: true }); } else { - activeWorker.process.kill(); + this.activeWorker?.process.kill(); } return true; } return false; - }; - - const restartServer = (): void => { - // indicate to the worker that we are 'expecting' this restart - activeWorker.send({ setResponsiveness: false }); - tryKillActiveWorker(true); - }; - - const killSession = (graceful = true): never => { - log(cyan(`exiting session ${graceful ? "clean" : "immediate"}ly`)); - tryKillActiveWorker(graceful); - process.exit(0); - }; + } - const setPort = (port: "server" | "socket" | string, value: number, immediateRestart: boolean): void => { + /** + * Allows the caller to set the port at which the target (be it the server, + * the websocket, some other custom port) is listening. If an immediate restart + * is specified, this monitor will kill the active child and re-launch the server + * at the port. Otherwise, the updated port won't be used until / unless the child + * dies on its own and triggers a restart. + */ + private setPort = (port: "server" | "socket" | string, value: number, immediateRestart: boolean): void => { if (value > 1023 && value < 65536) { - ports[port] = value; + this.configuration.ports[port] = value; if (immediateRestart) { - restartServer(); + this.tryKillActiveWorker(); } } else { - log(red(`${port} is an invalid port number`)); + this.log(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({ + /** + * Kills the current active worker and proceeds to spawn a new worker, + * feeding in configuration information as environment variables. + */ + private spawn = (): void => { + const { + pollingRoute, + pollingFailureTolerance, + pollingIntervalSeconds, + ports + } = this.configuration; + this.tryKillActiveWorker(); + this.activeWorker = fork({ pollingRoute, pollingFailureTolerance, serverPort: ports.server, socketPort: ports.socket, pollingIntervalSeconds, - session_key: key + session_key: this.key }); - log(cyan(`spawned new server worker with process id ${activeWorker.process.pid}`)); + this.log(cyan(`spawned new server worker with process id ${this.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 }) => { + this.activeWorker.on("message", async ({ lifecycle, action }) => { if (action) { - const { message, args } = action as SessionAction; - console.log(timestamp(), `${workerIdentifier} action requested (${cyan(message)})`); + const { message, args } = action as Monitor.Action; + console.log(this.timestamp(), `${this.configuration.workerIdentifier} action requested (${cyan(message)})`); switch (message) { case "kill": - log(red("an authorized user has manually ended the server session")); - killSession(args.graceful); + this.log(red("an authorized user has manually ended the server session")); + this.killSession(args.graceful); case "notify_crash": - if (notifiers && notifiers.crash) { + if (this.notifiers?.crash) { const { error } = args; - const success = await notifiers.crash(error, log); + const success = await this.notifiers.crash(error); const statement = success ? green("distributed crash notification to recipients") : red("distribution of crash notification failed"); - log(statement); + this.log(statement); } case "set_port": const { port, value, immediateRestart } = args; - setPort(port, value, immediateRestart); - default: - const handlers = onMessage[message]; - if (handlers) { - handlers.forEach(handler => handler({ message, args })); - } + this.setPort(port, value, immediateRestart); + } + const handlers = this.onMessage[message]; + if (handlers) { + handlers.forEach(handler => handler({ message, args })); } } else if (lifecycle) { - console.log(timestamp(), `${workerIdentifier} lifecycle phase (${lifecycle})`); + console.log(this.timestamp(), `${this.configuration.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}`}.`; - log(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}` }); - const boolean = /true|false/; - const number = /\d+/; - const letters = /[a-zA-Z]+/; - repl.registerCommand("exit", [/clean|force/], args => killSession(args[0] === "clean")); - repl.registerCommand("restart", [], restartServer); - repl.registerCommand("set", [letters, "port", number, boolean], args => setPort(args[0], Number(args[2]), args[3] === "true")); - repl.registerCommand("set", [/polling/, number, boolean], args => { - const newPollingIntervalSeconds = Math.floor(Number(args[2])); - if (newPollingIntervalSeconds < 0) { - log(red("the polling interval must be a non-negative integer")); - } else { - if (newPollingIntervalSeconds !== pollingIntervalSeconds) { - pollingIntervalSeconds = newPollingIntervalSeconds; - if (args[3] === "true") { - activeWorker.send({ newPollingIntervalSeconds }); - } - } - } - }); - // 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, - addServerMessageListener: (message: string, handler: ActionHandler) => { - const handlers = onMessage[message]; - if (handlers) { - handlers.push(handler); - } else { - onMessage[message] = [handler]; - } - }, - removeServerMessageListener: (message: string, handler: ActionHandler) => { - const handlers = onMessage[message]; - if (handlers) { - const index = handlers.indexOf(handler); - if (index > -1) { - handlers.splice(index, 1); - } - } - }, - clearServerMessageListeners: (message: string) => onMessage[message] = undefined, - restartServer, - killSession, - setPort, - log - }; + } + } + + /** * 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 { - let shouldServerBeResponsive = false; - const exitHandlers: ExitHandler[] = []; - let pollingFailureCount = 0; - - const lifecycleNotification = (lifecycle: string) => process.send?.({ lifecycle }); - - // notify master thread (which will log update in the console) of initialization via IPC - lifecycleNotification(green("compiling and initializing...")); + export class ServerWorker { + + private shouldServerBeResponsive = false; + private exitHandlers: ExitHandler[] = []; + private pollingFailureCount = 0; + private pollingIntervalSeconds: number; + private pollingFailureTolerance: number; + private pollTarget: string; + private serverPort: number; + + /** + * Allows developers to invoke application specific logic + * by hooking into the exiting of the server process. + */ + public addExitHandler = (handler: ExitHandler) => this.exitHandlers.push(handler); + + /** + * Kill the session monitor (parent process) from this + * server worker (child process). This will also kill + * this process (child process). + */ + public killSession = (graceful = true) => this.sendMonitorAction("kill", { graceful }); + + /** + * A convenience wrapper to tell the session monitor (parent process) + * to carry out the action with the specified message and arguments. + */ + public sendMonitorAction = (message: string, args?: any) => process.send!({ action: { message, args } }); + + constructor(work: Function) { + this.lifecycleNotification(green(`initializing process... (${white(`${process.execPath} ${process.execArgv.join(" ")}`)})`)); + + const { pollingRoute, serverPort, pollingIntervalSeconds, pollingFailureTolerance } = process.env; + this.serverPort = Number(serverPort); + this.pollingIntervalSeconds = Number(pollingIntervalSeconds); + this.pollingFailureTolerance = Number(pollingFailureTolerance); + this.pollTarget = `http://localhost:${serverPort}${pollingRoute}`; + + this.configureProcess(); + work(); + this.pollServer(); + } - // updates the local value of listening to the value sent from master - process.on("message", ({ setResponsiveness, newPollingIntervalSeconds }) => { - if (setResponsiveness) { - shouldServerBeResponsive = setResponsiveness; - } - if (newPollingIntervalSeconds) { - pollingIntervalSeconds = newPollingIntervalSeconds; - } - }); + /** + * Set up message and uncaught exception handlers for this + * server process. + */ + private configureProcess = () => { + // updates the local values of variables to the those sent from master + process.on("message", async ({ setResponsiveness, newPollingIntervalSeconds, manualExit }) => { + if (setResponsiveness !== undefined) { + this.shouldServerBeResponsive = setResponsiveness; + } + if (newPollingIntervalSeconds !== undefined) { + this.pollingIntervalSeconds = newPollingIntervalSeconds; + } + if (manualExit !== undefined) { + await this.executeExitHandlers(null); + process.exit(0); + } + }); - const executeExitHandlers = async (reason: Error | null) => Promise.all(exitHandlers.map(handler => handler(reason))); + // one reason to exit, as the process might be in an inconsistent state after such an exception + process.on('uncaughtException', this.proactiveUnplannedExit); + } - // 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 => { - shouldServerBeResponsive = false; + /** + * Execute the list of functions registered to be called + * whenever the process exits. + */ + private executeExitHandlers = async (reason: Error | null) => Promise.all(this.exitHandlers.map(handler => handler(reason))); + + /** + * Notify master thread (which will log update in the console) of initialization via IPC. + */ + private lifecycleNotification = (event: string) => process.send?.({ lifecycle: event }); + + /** + * 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. + */ + private proactiveUnplannedExit = async (error: Error): Promise => { + this.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 executeExitHandlers(error); + this.sendMonitorAction("notify_crash", { error }); + await this.executeExitHandlers(error); // notify master thread (which will log update in the console) of crash event via IPC - lifecycleNotification(red(`crash event detected @ ${new Date().toUTCString()}`)); - lifecycleNotification(red(error.message)); + this.lifecycleNotification(red(`crash event detected @ ${new Date().toUTCString()}`)); + this.lifecycleNotification(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 { env } = process; - const { pollingRoute, serverPort } = env; - let pollingIntervalSeconds = Number(env.pollingIntervalSeconds); - const pollingFailureTolerance = Number(env.pollingFailureTolerance); - // 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 => { + } + + /** + * 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. + */ + private pollServer = async (): Promise => { await new Promise(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}...`) }); + await get(this.pollTarget); + if (!this.shouldServerBeResponsive) { + // notify monitor thread that the server is up and running + this.lifecycleNotification(green(`listening on ${this.serverPort}...`)); } - shouldServerBeResponsive = true; + this.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 - if (shouldServerBeResponsive) { - if (++pollingFailureCount > pollingFailureTolerance) { - activeExit(error); + if (this.shouldServerBeResponsive) { + if (++this.pollingFailureCount > this.pollingFailureTolerance) { + this.proactiveUnplannedExit(error); } else { - lifecycleNotification(yellow(`the server has encountered ${pollingFailureCount} of ${pollingFailureTolerance} tolerable failures`)); + this.lifecycleNotification(yellow(`the server has encountered ${this.pollingFailureCount} of ${this.pollingFailureTolerance} tolerable failures`)); } } } - }, 1000 * pollingIntervalSeconds); + }, 1000 * this.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 + this.pollServer(); + } - return { - addExitHandler: (handler: ExitHandler) => exitHandlers.push(handler), - killSession: (graceful = true) => process.send!({ action: { message: "kill", args: { graceful } } }), - sendSessionAction: (message: string, args?: any) => process.send!({ action: { message, args } }) - }; } } \ No newline at end of file diff --git a/src/server/index.ts b/src/server/index.ts index ffab0f380..8e0ddc206 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -25,6 +25,7 @@ import { yellow, red } from "colors"; import { Session } from "./Session/session"; import { DashSessionAgent } from "./DashSession"; +export let sessionAgent: Session.AppliedSessionAgent; export const publicDirectory = path.resolve(__dirname, "public"); export const filesDirectory = path.resolve(publicDirectory, "files"); @@ -141,7 +142,6 @@ export async function launchServer() { await initializeServer(routeSetter); } -export const sessionAgent = new DashSessionAgent(); /** * If you're in development mode, you won't need to run a session. * The session spawns off new server processes each time an error is encountered, and doesn't @@ -149,7 +149,7 @@ export const sessionAgent = new DashSessionAgent(); * So, the 'else' clause is exactly what we've always run when executing npm start. */ if (process.env.RELEASE) { - sessionAgent.launch(); + (sessionAgent = new DashSessionAgent()).launch(); } else { launchServer(); } \ No newline at end of file -- cgit v1.2.3-70-g09d2 From 643e8a295c954c5b2fd40ed17269a239a9653c00 Mon Sep 17 00:00:00 2001 From: Sam Wilkins Date: Wed, 8 Jan 2020 06:37:04 -0500 Subject: clean up --- src/server/Session/session.ts | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) (limited to 'src') diff --git a/src/server/Session/session.ts b/src/server/Session/session.ts index 8bee99f41..3b2be9d6a 100644 --- a/src/server/Session/session.ts +++ b/src/server/Session/session.ts @@ -464,10 +464,7 @@ export namespace Session { */ private configureProcess = () => { // updates the local values of variables to the those sent from master - process.on("message", async ({ setResponsiveness, newPollingIntervalSeconds, manualExit }) => { - if (setResponsiveness !== undefined) { - this.shouldServerBeResponsive = setResponsiveness; - } + process.on("message", async ({ newPollingIntervalSeconds, manualExit }) => { if (newPollingIntervalSeconds !== undefined) { this.pollingIntervalSeconds = newPollingIntervalSeconds; } -- cgit v1.2.3-70-g09d2 From 252a8ea3483852b29171b459c65217aa593c25f1 Mon Sep 17 00:00:00 2001 From: Sam Wilkins Date: Wed, 8 Jan 2020 07:14:00 -0500 Subject: factory --- src/server/DashSession.ts | 4 ++-- src/server/Session/session.ts | 30 +++++++++++++++++++++++++----- src/server/repl.ts | 4 ++-- 3 files changed, 29 insertions(+), 9 deletions(-) (limited to 'src') diff --git a/src/server/DashSession.ts b/src/server/DashSession.ts index 7c2cfaf8d..22bcbadc9 100644 --- a/src/server/DashSession.ts +++ b/src/server/DashSession.ts @@ -19,7 +19,7 @@ export class DashSessionAgent extends Session.AppliedSessionAgent { private readonly signature = "-Dash Server Session Manager"; protected async launchMonitor() { - const monitor = new Session.Monitor({ + const monitor = Session.Monitor.Create({ key: async key => { // this sends a pseudorandomly generated guid to the configuration's recipients, allowing them alone // to kill the server via the /kill/:key route @@ -54,7 +54,7 @@ export class DashSessionAgent extends Session.AppliedSessionAgent { } protected async launchServerWorker() { - const worker = new Session.ServerWorker(launchServer); // server initialization delegated to worker + const worker = Session.ServerWorker.Create(launchServer); // server initialization delegated to worker worker.addExitHandler(() => Utils.Emit(WebSocket._socket, MessageStore.ConnectionTerminated, "Manual")); return worker; } diff --git a/src/server/Session/session.ts b/src/server/Session/session.ts index 3b2be9d6a..144d50c52 100644 --- a/src/server/Session/session.ts +++ b/src/server/Session/session.ts @@ -111,6 +111,7 @@ export namespace Session { */ export class Monitor { + private static count = 0; private exitHandlers: ExitHandler[] = []; private readonly notifiers: Monitor.NotifierHooks | undefined; private readonly configuration: Configuration; @@ -119,14 +120,23 @@ export namespace Session { private key: string | undefined; private repl: Repl; + public static Create(notifiers: Monitor.NotifierHooks) { + if (++Monitor.count > 1) { + throw new Error("Cannot create more than one monitor"); + } else { + return new Monitor(notifiers); + } + } + /** * Kill this session and its active child * server process, either gracefully (may wait * indefinitely, but at least allows active networking * requests to complete) or immediately. */ - public killSession = async (graceful = true): Promise => { + public killSession = async (graceful = true) => { this.log(cyan(`exiting session ${graceful ? "clean" : "immediate"}ly`)); + await this.executeExitHandlers(null); this.tryKillActiveWorker(graceful); process.exit(0); } @@ -176,7 +186,7 @@ export namespace Session { */ public clearServerMessageListeners = (message: string) => this.onMessage[message] = undefined; - constructor(notifiers?: Monitor.NotifierHooks) { + private constructor(notifiers?: Monitor.NotifierHooks) { this.notifiers = notifiers; console.log(this.timestamp(), cyan("initializing session...")); @@ -385,6 +395,7 @@ export namespace Session { case "kill": this.log(red("an authorized user has manually ended the server session")); this.killSession(args.graceful); + break; case "notify_crash": if (this.notifiers?.crash) { const { error } = args; @@ -392,9 +403,11 @@ export namespace Session { const statement = success ? green("distributed crash notification to recipients") : red("distribution of crash notification failed"); this.log(statement); } + break; case "set_port": const { port, value, immediateRestart } = args; this.setPort(port, value, immediateRestart); + break; } const handlers = this.onMessage[message]; if (handlers) { @@ -408,8 +421,6 @@ export namespace Session { } - - /** * 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 @@ -417,6 +428,7 @@ export namespace Session { */ export class ServerWorker { + private static count = 0; private shouldServerBeResponsive = false; private exitHandlers: ExitHandler[] = []; private pollingFailureCount = 0; @@ -425,6 +437,14 @@ export namespace Session { private pollTarget: string; private serverPort: number; + public static Create(work: Function) { + if (++ServerWorker.count > 1) { + throw new Error("Cannot create more than one worker per thread"); + } else { + return new ServerWorker(work); + } + } + /** * Allows developers to invoke application specific logic * by hooking into the exiting of the server process. @@ -444,7 +464,7 @@ export namespace Session { */ public sendMonitorAction = (message: string, args?: any) => process.send!({ action: { message, args } }); - constructor(work: Function) { + private constructor(work: Function) { this.lifecycleNotification(green(`initializing process... (${white(`${process.execPath} ${process.execArgv.join(" ")}`)})`)); const { pollingRoute, serverPort, pollingIntervalSeconds, pollingFailureTolerance } = process.env; diff --git a/src/server/repl.ts b/src/server/repl.ts index faf1eab15..c4526528e 100644 --- a/src/server/repl.ts +++ b/src/server/repl.ts @@ -97,16 +97,16 @@ export default class Repl { const candidates = registered.filter(({ argPatterns: { length: count } }) => count === length); for (const { argPatterns, action } of candidates) { const parsed: string[] = []; - let matched = false; + let matched = true; if (length) { for (let i = 0; i < length; i++) { let matches: RegExpExecArray | null; if ((matches = argPatterns[i].exec(args[i])) === null) { + matched = false; break; } parsed.push(matches[0]); } - matched = true; } if (!length || matched) { await action(parsed); -- cgit v1.2.3-70-g09d2 From ea4d32815b478f0f8c8573b3f73b179c7324a220 Mon Sep 17 00:00:00 2001 From: Sam Wilkins Date: Wed, 8 Jan 2020 07:30:37 -0500 Subject: timeout --- src/server/index.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'src') diff --git a/src/server/index.ts b/src/server/index.ts index 8e0ddc206..be18a3b3a 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -93,7 +93,7 @@ function routeSetter({ isRelease, addSupervisedRoute, logRegistrationOutcome }: secureHandler: ({ req, res }) => { if (req.params.key === process.env.session_key) { res.send(""); - sessionAgent.killSession(); + setTimeout(sessionAgent.killSession, 5000); } else { res.redirect("/home"); } -- cgit v1.2.3-70-g09d2 From 3b16e5e94daf370032fb521004fa9618131cb46f Mon Sep 17 00:00:00 2001 From: Sam Wilkins Date: Wed, 8 Jan 2020 07:59:14 -0500 Subject: more robust factory constraints --- src/server/Session/session.ts | 19 +++++++++++++------ 1 file changed, 13 insertions(+), 6 deletions(-) (limited to 'src') diff --git a/src/server/Session/session.ts b/src/server/Session/session.ts index 144d50c52..936357364 100644 --- a/src/server/Session/session.ts +++ b/src/server/Session/session.ts @@ -1,5 +1,5 @@ import { red, cyan, green, yellow, magenta, blue, white } from "colors"; -import { on, fork, setupMaster, Worker, isMaster } from "cluster"; +import { on, fork, setupMaster, Worker, isMaster, isWorker } from "cluster"; import { get } from "request-promise"; import { Utils } from "../../Utils"; import Repl, { ReplAction } from "../repl"; @@ -120,9 +120,13 @@ export namespace Session { private key: string | undefined; private repl: Repl; - public static Create(notifiers: Monitor.NotifierHooks) { - if (++Monitor.count > 1) { - throw new Error("Cannot create more than one monitor"); + public static Create(notifiers?: Monitor.NotifierHooks) { + if (isWorker) { + console.error(red("Monitor must be on the master process.")); + process.exit(1); + } else if (++Monitor.count > 1) { + console.error(("Cannot create more than one monitor.")); + process.exit(1); } else { return new Monitor(notifiers); } @@ -438,8 +442,11 @@ export namespace Session { private serverPort: number; public static Create(work: Function) { - if (++ServerWorker.count > 1) { - throw new Error("Cannot create more than one worker per thread"); + if (isMaster) { + throw new Error("Worker must be launched on a worker process."); + } else if (++ServerWorker.count > 1 || isMaster) { + process.send?.({ action: { message: "kill", args: { graceful: false } } }); + process.exit(1); } else { return new ServerWorker(work); } -- cgit v1.2.3-70-g09d2