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 simultaneous connections the server will handle before returning an error
+  * `options.resp` {Error} An error that will be passed to `res.send` when the limit is reached.
+  * `options.resp.statusCode` {Number} The status code to return when the limit is reached.
+  * `options.server.inflightReuqests` {Function} Should return the number of active connections to the server
+
 ## Accept Parser
 
 Parses out the `Accept` header, and ensures that the server can respond to what
@@ -439,6 +446,24 @@ 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
+const options = { limit: 600, server: server };
+options.resp = new require('restify-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

@@ -25,6 +25,7 @@ module.exports = {
     serveStatic: require('./static'),
     throttle: require('./throttle'),
     urlEncodedBodyParser: require('./formBodyParser'),
+    inflightRequestThrottle: require('./inflightRequestThrottle'),
 
 
     pre: {

lib/plugins/inflightRequestThrottle.js

@@ -0,0 +1,66 @@
+'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.resp the error object to pass to res.send when the
+ *    maximum concurrency is exceeded
+ * @param {Number} opts.resp.statusCode the status code to return when the
+ *    maximum concurrency is exceeded
+ * @param {Function} opts.server.inflightRequests A function that, when
+ *    invoked, returns the number of requests currently being handled by the
+ *    server.
+ * @returns {Function} middleware to be registered on server.pre
+ */
+function inflightRequestThrottle (opts) {
+    // Be nice to our users and let them drop the `new` keyword
+    if (!(this instanceof inflightRequestThrottle)) {
+        return new inflightRequestThrottle(opts); //jscs:ignore
+    }
+    var self = this;
+
+    // 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.resp !== undefined && opts.resp !== null) {
+        assert.ok(opts.resp instanceof Error, 'opts.resp must be an error');
+        assert.optionalNumber(opts.resp.statusCode, 'opts.resp.statusCode');
+    }
+
+    self._resp = opts.resp || defaultResponse;
+    self._limit = opts.limit;
+    self._server = opts.server;
+
+    // We need to bind in order to keep our `this` context when passed back
+    // out of the constructor.
+    return self.onRequest.bind(self);
+}
+
+inflightRequestThrottle.prototype.onRequest =
+function onRequest (req, res, next) {
+    var self = this;
+    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._resp);
+    }
+
+    return next();
+};
+
+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 resp = new Error('foo');
+        var p = P({ server: server, limit: 1, resp: resp });
+        function send (body) {
+            assert.equal(body, resp, '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 resp = new Error('foo');
+        resp.statusCode = 555;
+        server.pre(P({ server: server, limit: 1, resp: resp }));
+        var RESP;
+        server.get('/foo', function (req, res) {
+            if (RESP) {
+                res.send(999);
+            } else {
+                RESP = 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');
+                RESP.send(200);
+            });
+        });
+    });
+});