Create inflightRequestThrottle plugin

docs/api/plugins.md

@@ -153,6 +153,13 @@ event, e.g., `server.on('after', plugins.metrics());`:
   * `res` {Object} the response obj
   * `route` {Object} the route obj that serviced the request
 
+The module includes the following plugins to be used with restify's `pre` event:
+* `inflightRequestThrottle(options)` - limits the max number of inflight requests
+  * `options.limit` {Number} the maximum number of inflight requests the server will handle before returning an error
+  * `options.res` {Error} An error that will be passed to `res.send` when the limit is reached.
+  * `options.res.statusCode` {Number} The status code to return when the limit is reached.
+  * `options.server` {Object} The restify server that this module will throttle
+
 ## Accept Parser
 
 Parses out the `Accept` header, and ensures that the server can respond to what
@@ -439,6 +446,28 @@ uniform request distribution.  To enable this, you can pass in
 `options.tokensTable`, which is simply any Object that supports `put` and `get`
 with a `String` key, and an `Object` value.
 
+## Inflight Request Throttling
+
+```js
+var errors = require('restify-errors');
+var restify = require('restify');
+
+var server = restify.createServer();
+const options = { limit: 600, server: server };
+options.res = new errors.InternalServerError();
+server.pre(restify.plugins.inflightRequestThrottle(options));
+```
+
+The `inflightRequestThrottle` module allows you to specify an upper limit to
+the maximum number of inflight requests your server is able to handle. This
+is a simple heuristic for protecting against event loop contention between
+requests causing unacceptable latencies.
+
+The custom response is optional, and allows you to specify your own response
+and status code when rejecting incoming requests due to too many inflight
+requests. It defaults to `503 ServiceUnavailableError`.
+
+
 ## Conditional Request Handler
 
 ```js

lib/plugins/index.js

@@ -14,6 +14,7 @@ module.exports = {
     dateParser: require('./date'),
     fullResponse: require('./fullResponse'),
     gzipResponse: require('./gzip'),
+    inflightRequestThrottle: require('./inflightRequestThrottle'),
     jsonBodyParser: require('./jsonBodyParser'),
     jsonp: require('./jsonp'),
     multipartBodyParser: require('./multipartBodyParser'),

lib/plugins/inflightRequestThrottle.js

@@ -0,0 +1,61 @@
+'use strict';
+
+var assert = require('assert-plus');
+var ServiceUnavailableError = require('restify-errors').ServiceUnavailableError;
+var defaultResponse = new ServiceUnavailableError('resource exhausted');
+
+/**
+ * inflightRequestThrottle
+ *
+ * @param {Object} opts configure this plugin
+ * @param {Number} opts.limit maximum number of simultanenous
+ *    connections the server will handle before returning an error
+ * @param {Error} opts.res the error object to pass to res.send when the
+ *    maximum concurrency is exceeded
+ * @param {Number} opts.res.statusCode the status code to return when the
+ *    maximum concurrency is exceeded
+ * @param {Function} opts.server the instance of the restify server this module
+ *    will throttle.
+ * @returns {Function} middleware to be registered on server.pre
+ */
+function inflightRequestThrottle (opts) {
+
+    // Scrub input and populate our configuration
+    assert.object(opts, 'opts');
+    assert.number(opts.limit, 'opts.limit');
+    assert.object(opts.server, 'opts.server');
+    assert.func(opts.server.inflightRequests, 'opts.server.inflightRequests');
+
+    if (opts.res !== undefined && opts.res !== null) {
+        assert.ok(opts.res instanceof Error, 'opts.res must be an error');
+        assert.optionalNumber(opts.res.statusCode, 'opts.res.statusCode');
+    }
+
+    var self = {};
+    self._res = opts.res || defaultResponse;
+    self._limit = opts.limit;
+    self._server = opts.server;
+
+    function onRequest (req, res, next) {
+        var inflightRequests = self._server.inflightRequests();
+
+        if (inflightRequests > self._limit) {
+            req.log.trace({
+                plugin: 'inflightRequestThrottle',
+                inflightRequests: inflightRequests,
+                limit: self._limit
+            }, 'maximum inflight requests exceeded, rejecting request');
+            return res.send(self._res);
+        }
+
+        return next();
+    }
+
+    // We need to bind in order to keep our `this` context when passed back
+    // out of the constructor.
+    return onRequest;
+}
+
+inflightRequestThrottle.prototype.onRequest =
+
+module.exports = inflightRequestThrottle;

test/plugins/inflightRequestThrottle.test.js

@@ -0,0 +1,119 @@
+'use strict';
+
+var assert = require('chai').assert;
+var restify = require('../../lib/index.js');
+var restifyClients = require('restify-clients');
+var P = restify.plugins.inflightRequestThrottle;
+
+function fakeServer(count) {
+    return {
+        inflightRequests: function () {
+            return count;
+        }
+    };
+}
+
+describe('inlfightRequestThrottle', function () {
+
+    it('Unit: Should shed load', function (done) {
+        var logged = false;
+        var p = P({ server: fakeServer(10), limit: 1 });
+        function send (body) {
+            assert(logged, 'Should have emitted a log');
+            assert.equal(body.statusCode, 503, 'Defaults to 503 status');
+            assert(body instanceof Error, 'Defaults to error body');
+            done();
+        }
+        function next () {
+            assert(false, 'Should not call next');
+            done();
+        }
+        function trace () {
+            logged = true;
+        }
+        var log = { trace: trace };
+        var fakeReq = { log: log };
+        p(fakeReq, { send: send }, next);
+    });
+
+    it('Unit: Should support custom response', function (done) {
+        var server = fakeServer(10);
+        var res = new Error('foo');
+        var p = P({ server: server, limit: 1, res: res });
+        function send (body) {
+            assert.equal(body, res, 'Overrides body');
+            done();
+        }
+        function next () {
+            assert(false, 'Should not call next');
+            done();
+        }
+        var fakeReq = { log : { trace: function () {} } };
+        p(fakeReq, { send: send }, next);
+    });
+
+    it('Unit: Should let request through when not under load', function (done) {
+        var p = P({ server: fakeServer(1), limit: 2 });
+        function send () {
+            assert(false, 'Should not call send');
+            done();
+        }
+        function next () {
+            assert(true, 'Should call next');
+            done();
+        }
+        var fakeReq = { log : { trace: function () {} } };
+        p(fakeReq, { send: send }, next);
+    });
+
+    it('Integration: Should shed load', function (done) {
+        var server = restify.createServer();
+        var client = {
+            close: function () {}
+        };
+        var isDone = false;
+        var to;
+        function finish() {
+            if (isDone) {
+                return null;
+            }
+            clearTimeout(to);
+            isDone = true;
+            client.close();
+            server.close();
+            return done();
+        }
+        to = setTimeout(finish, 2000);
+        var err = new Error('foo');
+        err.statusCode = 555;
+        server.pre(P({ server: server, limit: 1, res: err }));
+        var RES;
+        server.get('/foo', function (req, res) {
+            if (RES) {
+                res.send(999);
+            } else {
+                RES = res;
+            }
+        });
+        server.listen(0, '127.0.0.1', function () {
+            client = restifyClients.createJsonClient({
+                url: 'http://127.0.0.1:' + server.address().port,
+                retry: false
+            });
+            client.get({ path: '/foo' }, function (e, _, res) {
+                assert(e === null || e === undefined,
+                    'First request isnt shed');
+                assert.equal(res.statusCode, 200, '200 returned on success');
+                finish();
+            });
+            client.get({ path: '/foo' }, function (e, _, res) {
+                assert(e, 'Second request is shed');
+                assert.equal(e.name,
+                    'InternalServerError', 'Default err returned');
+                assert.equal(res.statusCode, 555,
+                    'Default shed status code returned');
+                RES.send(200);
+            });
+        });
+    });
+});