-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Some refactor Packages updated and version-pinned More aggressive management of RabbitFallback
- Loading branch information
1 parent
7b5a82a
commit 52bb3fb
Showing
25 changed files
with
583 additions
and
512 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,36 @@ | ||
| { | ||
| "root": true, | ||
| "env": { | ||
| "es6": true, | ||
| "node": true | ||
| }, | ||
| "extends": [ | ||
| "eslint:recommended" | ||
| ], | ||
| "parserOptions": { | ||
| "ecmaVersion": 8 | ||
| }, | ||
| "rules": { | ||
| "quote-props": ["warn", "consistent-as-needed"], | ||
| "arrow-parens": 0, | ||
| "generator-star-spacing": 0, | ||
| "no-debugger": 0, | ||
| "array-element-newline": ["error", "consistent"], | ||
| "array-bracket-newline": ["error", "consistent"], | ||
| "array-bracket-spacing": ["error", "never"], | ||
| "no-console": "error", | ||
| "indent": ["error", 4, { "SwitchCase": 1 }], | ||
| "space-infix-ops": ["warn"], | ||
| "eol-last":["warn"], | ||
| "no-multi-spaces": ["warn", { "ignoreEOLComments": true }], | ||
| "no-multiple-empty-lines": ["error", { "max": 2, "maxEOF": 1 }], | ||
| "comma-dangle": ["warn", "never"], | ||
| "eqeqeq": ["warn"], | ||
| "semi": ["warn", "never"], | ||
| "keyword-spacing": ["warn"], | ||
| "space-before-blocks": ["warn"], | ||
| "comma-spacing": ["warn", { "before": false, "after": true }] | ||
| } | ||
| } | ||
|
|
||
|
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,133 +1,107 @@ | ||
| 'use strict'; | ||
|
|
||
| let config = require('./config'); | ||
| let express = require('express'); | ||
| let morgan = require('morgan'); | ||
| let bodyParser = require('body-parser'); | ||
| let Raven = require('raven'); | ||
| if (config.SENTRY_DSN) { | ||
| Raven.config(config.SENTRY_DSN) | ||
| .install(); | ||
| 'use strict' | ||
|
|
||
| // get configuration | ||
| const { | ||
| SENTRY_DSN, | ||
| MAX_POST_SIZE, | ||
| FALLBACK, | ||
| RABBIT | ||
| } = require('./config') | ||
|
|
||
| let serviceIsAvailable = false | ||
| const express = require('express') | ||
| const morgan = require('morgan') | ||
| const bodyParser = require('body-parser') | ||
| const Raven = require('raven') | ||
|
|
||
| //TODO Upgrade to sentry pls -- this is deprecated | ||
| if (SENTRY_DSN) { | ||
| Raven.config(SENTRY_DSN) | ||
| .install() | ||
| } | ||
|
|
||
| if (config.FALLBACK) { | ||
| // Use var intentionally so that rabbit doesn't need to be defined | ||
| var rabbit = require('./rabbit'); // eslint-disable-line no-var | ||
| } | ||
| const prom = require('./lib/prometheus') | ||
| const logger = require('./lib/logger') | ||
| const queue = require('./lib/queue') | ||
| const ServiceError = require('./lib/ServiceError') | ||
| const rabbitController = require('./lib/rabbit') | ||
|
|
||
| let prom = require('./prometheus'); | ||
|
|
||
| let logger = require('./logger'); | ||
| let pubsub = require('./pubsub'); | ||
| let queue = require('./queue'); | ||
|
|
||
| // bootstrap app | ||
| let app = express(); | ||
| app.use(morgan('dev')); | ||
| app.use(bodyParser.json({ limit: config.MAX_POST_SIZE })); | ||
| app.use(bodyParser.urlencoded({extended: false})); | ||
|
|
||
| // register routes | ||
| app.get('/', (req, res, next) => { | ||
| res.json({ping: 'pong'}); | ||
| }); | ||
|
|
||
| app.get('/healthz', (req, res, next) => { | ||
| res.json({ping: 'pong'}); | ||
| }); | ||
|
|
||
| app.post('/messages/:channel', (req, res, next) => { | ||
| let end = prom.requestSummary.startTimer(); | ||
| let channel = req.params.channel; | ||
| prom.receiveCount.inc({channel}); | ||
| let messages = req.body.messages || []; | ||
| queue.push(publishJob(channel, messages, end)); | ||
| res.json({success: true}); | ||
| }); | ||
|
|
||
|
|
||
| /** | ||
| * Return a job | ||
| * | ||
| * @param {String} channel | ||
| * @param {[Object]} messages | ||
| * @param {Function} end | ||
| * @param {Integer} retries | ||
| * | ||
| * @return {Function} | ||
| */ | ||
| function publishJob(channel, messages, end, retries=0) { | ||
| return function(cb) { | ||
| pubsub.publish(channel, messages).then((result)=>{ | ||
| let errors = result.errors; | ||
| if(errors.length > 0) { | ||
| prom.publishCount.inc({state: 'failed', channel}); | ||
| if(config.FALLBACK && retries >= 2) { | ||
| end(); | ||
| return rabbit.publish(channel, errors); | ||
| } | ||
| retries++; | ||
| queue.push(publishJob(channel, errors.map((error) => error.message), end, retries)); | ||
| } else { | ||
| prom.publishCount.inc({state: 'success', channel}); | ||
| end(); | ||
| } | ||
| cb(); | ||
| }).catch((error) => { | ||
| cb(error); | ||
| }); | ||
| }; | ||
| } | ||
| if (FALLBACK) rabbitController.start() | ||
|
|
||
| const app = express() | ||
|
|
||
| app.use(morgan('dev')) | ||
| app.use(bodyParser.json({ limit: MAX_POST_SIZE })) | ||
| app.use(bodyParser.urlencoded({ extended: false })) | ||
|
|
||
| // why is this here? :shrug: | ||
| app.get('/', async (req, res) => { | ||
| // logger.info('ping') | ||
| res.json({ ping: 'pong' }).end() | ||
| }) | ||
|
|
||
| // k8s health check endpoint | ||
| app.get('/healthz', (req, res) => { | ||
| if (!serviceIsAvailable) throw new ServiceError('This service is not currently accepting requests', 400) | ||
| res.json({ ping: 'pong' }) | ||
| } ) | ||
|
|
||
| // bind middleware | ||
| app.use((req, res, next) => { | ||
| let err = new Error('Not Found'); | ||
| err.status = 404; | ||
| next(err); | ||
| }); | ||
| // handle incoming message post requests | ||
| app.post('/messages/:channel', async (req, res) => { | ||
| const end = prom.requestSummary.startTimer() | ||
| const channel = req.params.channel | ||
| const messages = req.body.messages | ||
|
|
||
| // Early exit if the format is wrong | ||
| if (!Array.isArray(messages)) throw new ServiceError('`messages` property is expected to be an array', 400) | ||
|
|
||
| // Count this channel add | ||
| prom.receiveCount.inc({ channel }) | ||
| queue.addPublishJob(channel, messages, end) | ||
| res.json({ success: true }).end() | ||
| }) | ||
|
|
||
|
|
||
| // Anything requests that have not been dealt with by this point is 404 | ||
| app.use((req, res, next) => next(new ServiceError('Not Found', 404))) | ||
|
|
||
| // Express Error Handler | ||
| // eslint-disable-next-line no-unused-vars | ||
| app.use((err, req, res, next) => { | ||
| // set locals, only providing error in development | ||
| res.locals.message = err.message; | ||
| res.locals.error = req.app.get('env') === 'development' ? err : {}; | ||
|
|
||
| // return a json error response | ||
| let status = err.status || 500; | ||
| res.status(status); | ||
| res.json({ | ||
| status: status, | ||
| message: err.message, | ||
| }); | ||
| }); | ||
|
|
||
| let onExitHandler = () => { | ||
| logger.info('Preparing to shutdown application'); | ||
|
|
||
| logger.info('Stopping queue timer'); | ||
| clearInterval(queue.timer); | ||
|
|
||
| logger.info('Closing express server socket'); | ||
| // When the HTTP server closes we want to empty the job queue. | ||
| app.server.close(emptyQueue); | ||
| }; | ||
|
|
||
|
|
||
| /** | ||
| * Run the queue and exit if it is empty | ||
| */ | ||
| function emptyQueue() { | ||
| if(queue.length > 0) { | ||
| logger.info('Processing remaining jobs on queue'); | ||
| queue.start(emptyQueue); | ||
| } | ||
| // set locals, only providing error in development | ||
| res.locals.message = err.message | ||
| res.locals.error = req.app.get('env') === 'development' ? err : {} | ||
|
|
||
| // return a json error response | ||
| let status = err.status || 500 | ||
| res.status(status) | ||
| res.json({ | ||
| status, | ||
| message: err.message | ||
| }) | ||
| }) | ||
|
|
||
| // Graceful shutdown | ||
| let exitHandler = async () => { | ||
| serviceIsAvailable = false // Health endpoint will start reporting failure | ||
|
|
||
| logger.info('Shutdown Initiated.') | ||
|
|
||
| logger.info('Express server has shut down.') | ||
|
|
||
| Promise.all([ | ||
| queue.end(), //Empty queue | ||
| rabbitController.stop(RABBIT.SHUTDOWN_WAIT) //Wait for rabbit connection to close | ||
| ]).then(()=>{ | ||
| logger.info('PSRP Terminating.') | ||
|
|
||
| }).catch(()=>{}) | ||
| } | ||
|
|
||
| process.on('SIGTERM', () => { | ||
| onExitHandler(); | ||
| }); | ||
|
|
||
| process.on('SIGINT', () => { | ||
| onExitHandler(); | ||
| }); | ||
| // Make sure we exit cleanly | ||
| process.on('SIGTERM', exitHandler) // regular termination signal | ||
| process.on('SIGINT', exitHandler) // for ^C | ||
| // process.on('SIGUSR2', exitHandler) // for nodemon during dev | ||
|
|
||
| module.exports = app; | ||
| serviceIsAvailable = true // Health endpoint will start reporting success | ||
| module.exports = app |
Oops, something went wrong.