diff options
| -rw-r--r-- | session.config.json | 3 | ||||
| -rw-r--r-- | src/server/ActionUtilities.ts | 23 | ||||
| -rw-r--r-- | src/server/Session/session.ts | 105 | ||||
| -rw-r--r-- | src/server/Session/session_config_schema.ts | 4 | ||||
| -rw-r--r-- | src/server/index.ts | 32 |
5 files changed, 110 insertions, 57 deletions
diff --git a/session.config.json b/session.config.json index 211422868..57ca9e3cc 100644 --- a/session.config.json +++ b/session.config.json @@ -5,5 +5,6 @@ "socket": 4321 }, "pollingRoute": "/serverHeartbeat", - "pollingIntervalSeconds": 15 + "pollingIntervalSeconds": 15, + "pollingFailureTolerance": 0 }
\ No newline at end of file diff --git a/src/server/ActionUtilities.ts b/src/server/ActionUtilities.ts index 30aed32e6..a93566fb1 100644 --- a/src/server/ActionUtilities.ts +++ b/src/server/ActionUtilities.ts @@ -118,25 +118,34 @@ export namespace Email { } }); + export interface DispatchFailure { + recipient: string; + error: Error; + } + export async function dispatchAll(recipients: string[], subject: string, content: string) { - const failures: string[] = []; + const failures: DispatchFailure[] = []; await Promise.all(recipients.map(async (recipient: string) => { - if (!await Email.dispatch(recipient, subject, content)) { - failures.push(recipient); + let error: Error | null; + if ((error = await Email.dispatch(recipient, subject, content)) !== null) { + failures.push({ + recipient, + error + }); } })); - return failures; + return failures.length ? failures : undefined; } - export async function dispatch(recipient: string, subject: string, content: string): Promise<boolean> { + export async function dispatch(recipient: string, subject: string, content: string): Promise<Error | null> { const mailOptions = { to: recipient, from: 'brownptcdash@gmail.com', subject, text: `Hello ${recipient.split("@")[0]},\n\n${content}` } as MailOptions; - return new Promise<boolean>(resolve => { - smtpTransport.sendMail(mailOptions, (dispatchError: Error | null) => resolve(dispatchError === null)); + return new Promise<Error | null>(resolve => { + smtpTransport.sendMail(mailOptions, resolve); }); } diff --git a/src/server/Session/session.ts b/src/server/Session/session.ts index cf2231b1f..06f22d855 100644 --- a/src/server/Session/session.ts +++ b/src/server/Session/session.ts @@ -1,6 +1,5 @@ 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"; @@ -8,8 +7,6 @@ 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 @@ -30,6 +27,7 @@ export namespace Session { ports: { [description: string]: number }; pollingRoute: string; pollingIntervalSeconds: number; + pollingFailureTolerance: number; [key: string]: any; } @@ -39,7 +37,8 @@ export namespace Session { workerIdentifier: magenta("__server__:"), ports: { server: 3000 }, pollingRoute: "/", - pollingIntervalSeconds: 30 + pollingIntervalSeconds: 30, + pollingFailureTolerance: 1 }; interface MasterExtensions { @@ -48,8 +47,8 @@ export namespace Session { } export interface NotifierHooks { - key?: (key: string) => boolean | Promise<boolean>; - crash?: (error: Error) => boolean | Promise<boolean>; + key?: (key: string, masterLog: (...optionalParams: any[]) => void) => boolean | Promise<boolean>; + crash?: (error: Error, masterLog: (...optionalParams: any[]) => void) => boolean | Promise<boolean>; } export interface SessionAction { @@ -57,14 +56,20 @@ 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<void>; - export type ActionHandler = (action: SessionAction) => void | Promise<void>; + export type ActionHandler = (action: SessionAction, hooks: SessionHooks) => void | Promise<void>; export interface EmailTemplate { subject: string; body: string; } - function loadAndValidateConfiguration(): any { + function loadAndValidateConfiguration(): Configuration { try { const configuration: Configuration = JSON.parse(readFileSync('./session.config.json', 'utf8')); const options = { @@ -121,7 +126,7 @@ export namespace Session { */ export async function initializeMonitorThread(notifiers?: NotifierHooks): Promise<MasterExtensions> { let activeWorker: Worker; - const childMessageHandlers: { [message: string]: (action: SessionAction, args: any) => void } = {}; + const childMessageHandlers: { [message: string]: ActionHandler } = {}; // read in configuration .json file only once, in the master thread // pass down any variables the pertinent to the child processes as environment variables @@ -131,7 +136,8 @@ export namespace Session { ports, pollingRoute, showServerOutput, - pollingIntervalSeconds + pollingIntervalSeconds, + pollingFailureTolerance } = loadAndValidateConfiguration(); const masterLog = (...optionalParams: any[]) => console.log(timestamp(), masterIdentifier, ...optionalParams); @@ -141,7 +147,7 @@ export namespace Session { let key: string | undefined; if (notifiers && notifiers.key) { key = Utils.GenerateGuid(); - const success = await notifiers.key(key); + const success = await notifiers.key(key, masterLog); const statement = success ? green("distributed session key to recipients") : red("distribution of session key failed"); masterLog(statement); } @@ -161,7 +167,7 @@ export namespace Session { // 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 + // attempts to kills the active worker ungracefully, unless otherwise specified const tryKillActiveWorker = (graceful = false): boolean => { if (activeWorker && !activeWorker.isDead()) { if (graceful) { @@ -174,17 +180,22 @@ export namespace Session { return false; }; - const restart = () => { + const restartServer = () => { // indicate to the worker that we are 'expecting' this restart activeWorker.send({ setResponsiveness: false }); - tryKillActiveWorker(); + tryKillActiveWorker(true); + }; + + const killSession = (graceful = true) => { + tryKillActiveWorker(graceful); + process.exit(0); }; const setPort = (port: string, value: number, immediateRestart: boolean) => { if (value > 1023 && value < 65536) { ports[port] = value; if (immediateRestart) { - restart(); + restartServer(); } } else { masterLog(red(`${port} is an invalid port number`)); @@ -197,12 +208,13 @@ export namespace Session { tryKillActiveWorker(); activeWorker = fork({ pollingRoute, + pollingFailureTolerance, serverPort: ports.server, socketPort: ports.socket, pollingIntervalSeconds, session_key: key }); - masterLog(`spawned new server worker with process id ${activeWorker.process.pid}`); + masterLog(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) { @@ -211,12 +223,11 @@ export namespace Session { switch (message) { case "kill": masterLog(red("an authorized user has manually ended the server session")); - tryKillActiveWorker(true); - process.exit(0); + killSession(); case "notify_crash": if (notifiers && notifiers.crash) { const { error } = args; - const success = await notifiers.crash(error); + const success = await notifiers.crash(error, masterLog); const statement = success ? green("distributed crash notification to recipients") : red("distribution of crash notification failed"); masterLog(statement); } @@ -226,7 +237,7 @@ export namespace Session { default: const handler = childMessageHandlers[message]; if (handler) { - handler(action, args); + handler({ message, args }, { restartServer, killSession, masterLog }); } } } else if (lifecycle) { @@ -245,9 +256,19 @@ export namespace Session { // 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("exit", [/clean|force/], args => killSession(args[0] === "clean")); + repl.registerCommand("restart", [], restartServer); repl.registerCommand("set", [/[a-zA-Z]+/, "port", /\d+/, /true|false/], args => setPort(args[0], Number(args[2]), args[3] === "true")); + repl.registerCommand("set", [/polling/, /interval/, /\d+/], args => { + const newPollingIntervalSeconds = Math.floor(Number(args[2])); + if (newPollingIntervalSeconds < 0) { + masterLog(red("the polling interval must be a non-negative integer")); + } else { + if (newPollingIntervalSeconds !== pollingIntervalSeconds) { + activeWorker.send({ newPollingIntervalSeconds }); + } + } + }); // finally, set things in motion by spawning off the first child (server) process spawn(); @@ -267,19 +288,26 @@ export namespace Session { export async function initializeWorkerThread(work: Function): Promise<(handler: ExitHandler) => void> { 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 - process.send?.({ lifecycle: green("compiling and initializing...") }); + lifecycleNotification(green("compiling and initializing...")); // updates the local value of listening to the value sent from master - process.on("message", ({ setResponsiveness }) => shouldServerBeResponsive = setResponsiveness); + process.on("message", ({ setResponsiveness, newPollingIntervalSeconds }) => { + if (setResponsiveness) { + shouldServerBeResponsive = setResponsiveness; + } + if (newPollingIntervalSeconds) { + pollingIntervalSeconds = newPollingIntervalSeconds; + } + }); // 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?.({ @@ -290,19 +318,18 @@ export namespace Session { }); 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) }); + lifecycleNotification(red(`crash event detected @ ${new Date().toUTCString()}`)); + 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 { - pollingIntervalSeconds, - pollingRoute, - serverPort - } = process.env; + 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}`; @@ -321,9 +348,15 @@ export namespace Session { // 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); + if (shouldServerBeResponsive) { + if (++pollingFailureCount > pollingFailureTolerance) { + activeExit(error); + } else { + lifecycleNotification(yellow(`the server has encountered ${pollingFailureCount} of ${pollingFailureTolerance} tolerable failures`)); + } + } } - }, 1000 * Number(pollingIntervalSeconds)); + }, 1000 * pollingIntervalSeconds); }); // controlled, asynchronous infinite recursion achieves a persistent poll that does not submit a new request until the previous has completed pollServer(); diff --git a/src/server/Session/session_config_schema.ts b/src/server/Session/session_config_schema.ts index 76af04b9f..5a85a45e3 100644 --- a/src/server/Session/session_config_schema.ts +++ b/src/server/Session/session_config_schema.ts @@ -30,6 +30,10 @@ export const configurationSchema: Schema = { type: "number", minimum: 1, maximum: 86400 + }, + pollingFailureTolerance: { + type: "number", + minimum: 0, } } };
\ No newline at end of file diff --git a/src/server/index.ts b/src/server/index.ts index 5e411aa3a..28b2885a1 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -22,7 +22,7 @@ import { log_execution, Email } from "./ActionUtilities"; import GeneralGoogleManager from "./ApiManagers/GeneralGoogleManager"; import GooglePhotosManager from "./ApiManagers/GooglePhotosManager"; import { Logger } from "./ProcessFactory"; -import { yellow } from "colors"; +import { yellow, red } from "colors"; import { Session } from "./Session/session"; import { isMaster } from "cluster"; import { execSync } from "child_process"; @@ -152,17 +152,19 @@ async function launchServer() { */ async function launchMonitoredSession() { if (isMaster) { - const recipients = ["samuel_wilkins@brown.edu"]; + const notificationRecipients = ["samuel_wilkins@brown.edu"]; const signature = "-Dash Server Session Manager"; - const customizer = await Session.initializeMonitorThread({ - key: async (key: string) => { + 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(recipients, "Server Termination Key", content); - return failures.length === 0; + 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 (error: Error) => { - const subject = "Dash Web Server Crash"; - const { name, message, stack } = error; + 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}`, @@ -171,12 +173,16 @@ async function launchMonitoredSession() { "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(recipients, subject, content); - return failures.length === 0; + 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; } }); - customizer.addReplCommand("pull", [], () => execSync("git pull", { stdio: ["ignore", "inherit", "inherit"] })); - customizer.addReplCommand("solr", [/start|stop/g], args => SolrManager.SetRunning(args[0] === "start")); + 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")); |