From 326a10dc5d7f7ad67a781864870d25b32f468dcd Mon Sep 17 00:00:00 2001 From: Wyatt Johnson Date: Thu, 20 Dec 2018 17:01:10 +0000 Subject: [PATCH] [next] Concurrency (#2136) * feat: added concurrency options * fix: Default CONCURRENCY to 2 for development --- .vscode/launch.json | 46 ++++---- package-lock.json | 18 ++- package.json | 6 +- src/core/server/config.ts | 8 ++ src/core/server/index.ts | 117 ++++++++++++++------ src/core/server/logger.ts | 4 + src/core/server/queue/Task.ts | 10 +- src/core/server/queue/tasks/mailer/index.ts | 7 ++ src/index.ts | 67 ++++++++++- 9 files changed, 217 insertions(+), 66 deletions(-) diff --git a/.vscode/launch.json b/.vscode/launch.json index 9c4f6d92d..b47e212d0 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -1,23 +1,27 @@ { - // Use IntelliSense to learn about possible attributes. - // Hover to view descriptions of existing attributes. - // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387 - "version": "0.2.0", - "configurations": [ - { - "type": "node", - "request": "launch", - "name": "Launch Program", - "cwd": "${workspaceFolder}", - "program": "${workspaceFolder}/node_modules/.bin/ts-node", - "args": [ - "--project", - "${workspaceFolder}/src/tsconfig.json", - "-r", - "tsconfig-paths/register", - "${workspaceFolder}/src/index.ts" - ], - "outputCapture": "std" - } - ] + // Use IntelliSense to learn about possible attributes. + // Hover to view descriptions of existing attributes. + // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387 + "version": "0.2.0", + "configurations": [ + { + "type": "node", + "request": "launch", + "name": "Launch Program", + "cwd": "${workspaceFolder}", + "args": [ + "-r", + "ts-node/register", + "-r", + "tsconfig-paths/register", + "${workspaceFolder}/src/index.ts" + ], + "env": { + "TS_NODE_PROJECT": "${workspaceFolder}/src/tsconfig.json", + // Set to 1 here beacuse clustering gets all weird with logging + output. + "CONCURRENCY": "1" + }, + "outputCapture": "std" + } + ] } diff --git a/package-lock.json b/package-lock.json index cabb710ed..4ab3c7e8b 100644 --- a/package-lock.json +++ b/package-lock.json @@ -2650,6 +2650,12 @@ "integrity": "sha512-78AdXtlhpCHT0K3EytMpn4JNxaf5tbqbLcbIRoQIHzpTIyjpxLQKRoxU55ujBXAtg3Nl2h/XWvfDa9dsMOd0pQ==", "dev": true }, + "@types/throng": { + "version": "4.0.2", + "resolved": "https://registry.npmjs.org/@types/throng/-/throng-4.0.2.tgz", + "integrity": "sha512-7tgh3R6vwtjj01URmhWXFSkWnm4wDJjsqLm8WPwIWadYjfsKAFi0HqabMQCU2JJ4TbeSGkb51qv27bBPN5Bubw==", + "dev": true + }, "@types/tlds": { "version": "1.199.0", "resolved": "https://registry.npmjs.org/@types/tlds/-/tlds-1.199.0.tgz", @@ -7140,7 +7146,7 @@ }, "colors": { "version": "1.1.2", - "resolved": "http://registry.npmjs.org/colors/-/colors-1.1.2.tgz", + "resolved": "https://registry.npmjs.org/colors/-/colors-1.1.2.tgz", "integrity": "sha1-FopHAXVran9RoSzgyXv6KMCE7WM=", "dev": true }, @@ -7948,7 +7954,7 @@ }, "css-color-names": { "version": "0.0.4", - "resolved": "http://registry.npmjs.org/css-color-names/-/css-color-names-0.0.4.tgz", + "resolved": "https://registry.npmjs.org/css-color-names/-/css-color-names-0.0.4.tgz", "integrity": "sha1-gIrcLnnPhHOAabZGyyDsJ762KeA=", "dev": true }, @@ -26228,6 +26234,14 @@ "integrity": "sha1-iQN8vJLFarGJJua6TLsgDhVnKmo=", "dev": true }, + "throng": { + "version": "4.0.0", + "resolved": "https://registry.npmjs.org/throng/-/throng-4.0.0.tgz", + "integrity": "sha1-mDxroZk7WOroWZmKpof/6I34TBc=", + "requires": { + "lodash.defaults": "^4.0.1" + } + }, "through": { "version": "2.3.8", "resolved": "https://registry.npmjs.org/through/-/through-2.3.8.tgz", diff --git a/package.json b/package.json index 7f7487861..22be86f6b 100644 --- a/package.json +++ b/package.json @@ -31,7 +31,7 @@ "generate:schema": "node ./scripts/generateSchemaTypes.js", "docz": "docz", "start": "node dist/index.js", - "start:development": "ts-node --project ./src/tsconfig.json -r tsconfig-paths/register ./src/index.ts", + "start:development": "CONCURRENCY=2 TS_NODE_PROJECT=./src/tsconfig.json node -r ts-node/register -r tsconfig-paths/register ./src/index.ts", "start:webpackDevServer": "ts-node ./scripts/start.ts", "lint": "npm-run-all --parallel lint:* tscheck:*", "lint:server": "tslint --project ./src/tsconfig.json", @@ -58,8 +58,8 @@ "content-security-policy-builder": "^2.0.0", "convict": "^4.3.1", "cors": "^2.8.4", - "dompurify": "^1.0.8", "dataloader": "^1.4.0", + "dompurify": "^1.0.8", "dotenv": "^6.0.0", "dotenv-expand": "^4.2.0", "express": "^4.16.3", @@ -105,6 +105,7 @@ "react-relay-network-modern": "^2.4.0", "striptags": "^3.1.1", "subscriptions-transport-ws": "^0.9.12", + "throng": "^4.0.0", "tlds": "^1.203.1", "uuid": "^3.3.2" }, @@ -172,6 +173,7 @@ "@types/relay-runtime": "^1.3.6", "@types/sane": "^2.0.0", "@types/sinon": "^5.0.1", + "@types/throng": "^4.0.2", "@types/tlds": "^1.199.0", "@types/uuid": "^3.4.3", "@types/vinyl": "^2.0.2", diff --git a/src/core/server/config.ts b/src/core/server/config.ts index b37a2a5d6..6ea751f30 100644 --- a/src/core/server/config.ts +++ b/src/core/server/config.ts @@ -1,5 +1,6 @@ import convict from "convict"; import Joi from "joi"; +import os from "os"; // Add custom format for the mongo uri scheme. convict.addFormat({ @@ -55,6 +56,13 @@ const config = convict({ env: "ENABLE_GRAPHIQL", arg: "enableGraphiQL", }, + concurrency: { + doc: "The number of worker nodes to spawn to handle traffic", + format: Number, + default: os.cpus().length, + env: "CONCURRENCY", + arg: "concurrency", + }, port: { doc: "The port to bind.", format: "port", diff --git a/src/core/server/index.ts b/src/core/server/index.ts index 0a016f09f..22cca5f7f 100644 --- a/src/core/server/index.ts +++ b/src/core/server/index.ts @@ -1,18 +1,22 @@ import express, { Express } from "express"; import http from "http"; +import { Db } from "mongodb"; +import { + attachSubscriptionHandlers, + createApp, + listenAndServe, +} from "talk-server/app"; +import config, { Config } from "talk-server/config"; import getManagementSchema from "talk-server/graph/management/schema"; import { Schemas } from "talk-server/graph/schemas"; import getTenantSchema from "talk-server/graph/tenant/schema"; -import { createQueue } from "talk-server/queue"; -import TenantCache from "talk-server/services/tenant/cache"; - +import logger from "talk-server/logger"; +import { createQueue, TaskQueue } from "talk-server/queue"; import { createJWTSigningConfig } from "talk-server/services/jwt"; -import { attachSubscriptionHandlers, createApp, listenAndServe } from "./app"; -import config, { Config } from "./config"; -import logger from "./logger"; -import { createMongoDB } from "./services/mongodb"; -import { createRedisClient } from "./services/redis"; +import { createMongoDB } from "talk-server/services/mongodb"; +import { AugmentedRedis, createRedisClient } from "talk-server/services/redis"; +import TenantCache from "talk-server/services/tenant/cache"; export interface ServerOptions { config?: Config; @@ -36,6 +40,24 @@ class Server { // the requested port. public httpServer: http.Server; + // queue stores a reference to the queues that can process operations. + private queue: TaskQueue; + + // redis stores the redis connection used by the application. + private redis: AugmentedRedis; + + // mongo stores the mongo connection used by the application. + private mongo: Db; + + // tenantCache stores the tenant cache used by the application. + private tenantCache: TenantCache; + + // connected when true, indicates that `connect()` was already called. + private connected: boolean = false; + + // processing when true, indicates that `process()` was already called. + private processing: boolean = false; + constructor(options: ServerOptions) { this.parentApp = express(); @@ -52,6 +74,51 @@ class Server { }; } + public async connect() { + // Guard against double connecting. + if (this.connected) { + throw new Error("server has already connected"); + } + this.connected = true; + + // Setup MongoDB. + this.mongo = await createMongoDB(config); + + // Setup Redis. + this.redis = createRedisClient(config); + + // Create the TenantCache. + this.tenantCache = new TenantCache( + this.mongo, + await createRedisClient(this.config), + config + ); + + // Prime the tenant cache so it'll be ready to serve now. + await this.tenantCache.primeAll(); + + // Create the Job Queue. + this.queue = createQueue({ + config: this.config, + mongo: this.mongo, + tenantCache: this.tenantCache, + }); + } + + /** + * process will start the job processors. + */ + public async process() { + // Guard against double connecting. + if (this.processing) { + throw new Error("server has already processing"); + } + this.processing = true; + + this.queue.mailer.process(); + this.queue.scraper.process(); + } + /** * start orchestrates the application by starting it and returning a promise * when the server has started. @@ -59,43 +126,29 @@ class Server { * @param parent the optional express application to bind the server to. */ public async start(parent?: Express) { + // Guard against not being connected. + if (!this.connected) { + throw new Error("server has not connected yet"); + } + const port = this.config.get("port"); // Ensure we have an app to bind to. parent = parent ? parent : this.parentApp; - // Setup MongoDB. - const mongo = await createMongoDB(config); - - // Setup Redis. - const redis = createRedisClient(config); - // Create the signing config. const signingConfig = createJWTSigningConfig(this.config); - // Create the TenantCache. - const tenantCache = new TenantCache( - mongo, - await createRedisClient(config), - config - ); - - // Prime the tenant cache so it'll be ready to serve now. - await tenantCache.primeAll(); - - // Create the Job Queue. - const queue = createQueue({ config, mongo, tenantCache }); - // Create the Talk App, branching off from the parent app. const app: Express = await createApp({ parent, - queue, - mongo, - redis, + mongo: this.mongo, + redis: this.redis, + signingConfig, + tenantCache: this.tenantCache, + queue: this.queue, config: this.config, schemas: this.schemas, - signingConfig, - tenantCache, }); // Start the application and store the resulting http.Server. diff --git a/src/core/server/logger.ts b/src/core/server/logger.ts index e5ec52f22..5716f4dc2 100644 --- a/src/core/server/logger.ts +++ b/src/core/server/logger.ts @@ -1,5 +1,6 @@ import bunyan, { LogLevelString, stdSerializers as serializers } from "bunyan"; import PrettyStream from "bunyan-prettystream"; +import cluster from "cluster"; import config from "talk-server/config"; @@ -19,6 +20,9 @@ function getStreams() { const logger = bunyan.createLogger({ name: "talk", + // Attach the cluster node information to the log entries. + clusterNode: cluster.worker ? `worker.${cluster.worker.id}` : "master", + // Include file references in log entries. src: true, serializers, diff --git a/src/core/server/queue/Task.ts b/src/core/server/queue/Task.ts index 2834456c1..48c8ad55a 100644 --- a/src/core/server/queue/Task.ts +++ b/src/core/server/queue/Task.ts @@ -18,9 +18,6 @@ export default class Task { this.queue = new Queue(options.jobName, options.queue); this.options = options; this.log = logger.child({ jobName: options.jobName }); - - // Sets up and attaches the job processor to the queue. - this.setupAndAttachProcessor(); } /** @@ -39,7 +36,12 @@ export default class Task { this.log.trace({ jobID: job.id }, "added job to queue"); return job; } - private setupAndAttachProcessor() { + + /** + * process will connect the queue to the processor so that it may process the + * job requests. + */ + public process() { this.queue.process(async (job: Job) => { const log = this.log.child({ jobID: job.id }); diff --git a/src/core/server/queue/tasks/mailer/index.ts b/src/core/server/queue/tasks/mailer/index.ts index df8184ba8..9fd105d28 100644 --- a/src/core/server/queue/tasks/mailer/index.ts +++ b/src/core/server/queue/tasks/mailer/index.ts @@ -193,6 +193,13 @@ export class Mailer { }, }); } + + /** + * process maps the interface to the task process function. + */ + public process() { + return this.task.process(); + } } export const createMailerTask = ( diff --git a/src/index.ts b/src/index.ts index 985b3476e..12b0de76e 100644 --- a/src/index.ts +++ b/src/index.ts @@ -5,21 +5,78 @@ import dotenv from "dotenv"; dotenv.config(); import express from "express"; - -import logger from "talk-server/logger"; +import throng from "throng"; import createTalk from "./core"; +import Server from "./core/server"; +import logger from "./core/server/logger"; // Create the app that will serve as the mounting point for the Talk Server. const app = express(); -async function bootstrap() { +// worker will start the worker process. +async function worker(server: Server) { try { - // Create the server instance. - const server = await createTalk(); + logger.debug("started server worker"); + + // Connect the server to databases. + await server.connect(); // Start the server. await server.start(app); + } catch (err) { + logger.error({ err }, "can not start server in worker mode"); + } +} + +// master will start the master process. +async function master(server: Server) { + try { + // Connect the server to databases. + await server.connect(); + + // Process jobs. + await server.process(); + } catch (err) { + logger.error({ err }, "can not start server in master mode"); + } +} + +// bootstrap will create a new Talk server, and start it up. +async function bootstrap() { + try { + logger.debug("starting bootstrap"); + + // Create the server instance. + const server = await createTalk(); + + // Determine the number of workers. + const workerCount = server.config.get("concurrency"); + + if (workerCount === 1) { + logger.debug( + { workerCount }, + "not utilizing cluster as concurrency level is 1" + ); + + // Connect the server to databases. + await server.connect(); + + // Process jobs. + await server.process(); + + // Start the server. + await server.start(app); + } else { + logger.debug({ workerCount }, "spawning workers to handle traffic"); + + // Launch the server start within throng. + throng({ + workers: workerCount, + start: () => worker(server), + master: () => master(server), + }); + } } catch (err) { logger.error({ err }, "can not bootstrap server"); }