Merge pull request #12 from uber/addhead

Add key used on lookup to head transmitted through proxy and emit head too

benchmarks/bench_ringpop_stat_cached_keys.js

@@ -17,7 +17,6 @@
 // LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 // THE SOFTWARE.
-var _ = require('underscore');
 var RingPop = require('../index.js');
 var Suite = require('benchmark').Suite;
 

benchmarks/bench_ringpop_stat_new_keys.js

@@ -17,7 +17,6 @@
 // LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 // THE SOFTWARE.
-var _ = require('underscore');
 var RingPop = require('../index.js');
 var Suite = require('benchmark').Suite;
 

index.js

@@ -19,6 +19,7 @@
 // THE SOFTWARE.
 'use strict';
 
+var _ = require('underscore');
 var EventEmitter = require('events').EventEmitter;
 var fs = require('fs');
 var metrics = require('metrics');
@@ -40,6 +41,7 @@ var Suspicion = require('./lib/swim.js').Suspicion;
 
 var HOST_PORT_PATTERN = /^(\d+.\d+.\d+.\d+):\d+$/;
 var MAX_JOIN_DURATION = 300000;
+var PROXY_REQ_PROPS = ['key', 'dest', 'req', 'res'];
 
 var AppRequiredError = TypedError({
     type: 'ringpop.options-app.required',
@@ -79,7 +81,14 @@ var InvalidLocalMemberError = TypedError({
 var OptionsRequiredError = TypedError({
     type: 'ringpop.options.required',
     message: 'Expected `options` argument to be passed.\n' +
-        'Must specify options for `RingPop({ ... })`.\n'
+        'Must specify options for `{method}`.\n',
+    method: null
+});
+
+var PropertyRequiredError = TypedError({
+    type: 'ringpop.options.property-required',
+    message: 'Expected `{property}` to be defined within options argument.',
+    property: null
 });
 
 var RedundantLeaveError = TypedError({
@@ -93,7 +102,7 @@ function RingPop(options) {
     }
 
     if (!options) {
-        throw OptionsRequiredError();
+        throw OptionsRequiredError({ method: 'RingPop' });
     }
 
     if (typeof options.app !== 'string' ||
@@ -795,10 +804,15 @@ RingPop.prototype.handleIncomingRequest =
         this.requestProxy.handleRequest(header, body, cb);
     };
 
-RingPop.prototype.proxyReq =
-    function proxyReq(destination, req, res, opts) {
-        this.requestProxy.proxyReq(destination, req, res, opts);
-    };
+RingPop.prototype.proxyReq = function proxyReq(opts) {
+    if (!opts) {
+        throw OptionsRequiredError({ method: 'proxyReq' });
+    }
+
+    this.validateProps(opts, PROXY_REQ_PROPS);
+
+    this.requestProxy.proxyReq(opts);
+};
 
 RingPop.prototype.handleOrProxy =
     function handleOrProxy(key, req, res, opts) {
@@ -807,8 +821,23 @@ RingPop.prototype.handleOrProxy =
         if (this.whoami() === dest) {
             return true;
         } else {
-            this.proxyReq(dest, req, res, opts);
+            this.proxyReq(_.extend(opts, {
+                key: key,
+                dest: dest,
+                req: req,
+                res: res,
+            }));
         }
     };
 
+RingPop.prototype.validateProps = function validateProps(opts, props) {
+    for (var i = 0; i < props.length; i++) {
+        var prop = props[i];
+
+        if (!opts[prop]) {
+            throw PropertyRequiredError({ property: prop });
+        }
+    }
+};
+
 module.exports = RingPop;

lib/members.js

@@ -152,23 +152,23 @@ Membership.evalOverride = function evalOverride(member, change) {
         // it affirms its "aliveness" and bumps its incarnation number.
         member.status = 'alive';
         member.incarnationNumber = +new Date();
-        return _.extend(member, { type: 'alive' });
+        return _.extend({ type: 'alive' }, member);
     } else if (Membership.isAliveOverride(member, change)) {
         member.status = 'alive';
         member.incarnationNumber = change.incarnationNumber || member.incarnationNumber;
-        return _.extend(member, { type: 'alive' });
+        return _.extend({ type: 'alive' }, member);
     } else if (Membership.isSuspectOverride(member, change)) {
         member.status = 'suspect';
         member.incarnationNumber = change.incarnationNumber || member.incarnationNumber;
-        return _.extend(member, { type: 'suspect' });
+        return _.extend({ type: 'suspect' }, member);
     } else if (Membership.isFaultyOverride(member, change)) {
         member.status = 'faulty';
         member.incarnationNumber = change.incarnationNumber || member.incarnationNumber;
-        return _.extend(member, { type: 'faulty' });
+        return _.extend({ type: 'faulty' }, member);
     } else if (Membership.isLeaveOverride(member, change)) {
         member.status = 'leave';
         member.incarnationNumber = change.incarnationNumber || member.incarnationNumber;
-        return _.extend(member, { type: 'leave' });
+        return _.extend({ type: 'leave' }, member);
     }
 };
 
@@ -229,7 +229,7 @@ Membership.prototype.addMember = function addMember(member, force, noEvent) {
     this.members.splice(this.getJoinPosition(), 0, newMember);
 
     if (!noEvent) {
-        this._emitUpdated(_.extend(newMember, { type: 'new' }));
+        this._emitUpdated(_.extend({ type: 'new' }, newMember));
     }
 };
 
@@ -258,9 +258,15 @@ Membership.prototype.computeChecksum = function computeChecksum() {
 };
 
 Membership.prototype.findMemberByAddress = function findMemberByAddress(address) {
-    return _.find(this.members, function(member) {
-        return member.address === address;
-    });
+    for (var i = 0; i < this.members.length; i++) {
+        var member = this.members[i];
+
+        if (member.address === address) {
+            return member;
+        }
+    }
+
+    return null;
 };
 
 Membership.prototype.generateChecksumString = function generateChecksumString() {

lib/request-proxy.js

@@ -44,14 +44,19 @@ function RequestProxy(ringpop) {
 
 var proto = RequestProxy.prototype;
 
-proto.proxyReq = function proxyReq(dest, req, res, opts) {
+proto.proxyReq = function proxyReq(opts) {
+    var key = opts.key;
+    var dest = opts.dest;
+    var req = opts.req;
+    var res = opts.res;
+
     var ringpop = this.ringpop;
     var url = req.url;
     var headers = req.headers;
     var method = req.method;
     var httpVersion = req.httpVersion;
 
-    var timeout = opts && opts.timeout ?
+    var timeout = opts.timeout ?
         opts.timeout : ringpop.proxyReqTimeout;
 
     body(req, onBody);
@@ -70,7 +75,8 @@ proto.proxyReq = function proxyReq(dest, req, res, opts) {
             headers: headers,
             method: method,
             httpVersion: httpVersion,
-            checksum: ringpop.membership.checksum
+            checksum: ringpop.membership.checksum,
+            ringpopKey: key
         });
 
         ringpop.channel.send(options, '/proxy/req',
@@ -123,7 +129,7 @@ proto.handleRequest = function handleRequest(head, body, cb) {
 
     var httpResponse = hammock.Response(onResponse);
 
-    ringpop.emit('request', httpRequest, httpResponse);
+    ringpop.emit('request', httpRequest, httpResponse, head);
 
     function onResponse(err, resp) {
         if (err) {

test/mock/index.js

@@ -21,5 +21,7 @@ module.exports = {
     channel: require('./channel.js'),
     logger: require('./logger.js'),
     membership: require('./membership.js'),
-    noop: require('./noop.js')
+    noop: require('./noop.js'),
+    requestProxy: require('./request-proxy.js'),
+    tchannel: require('./tchannel.js')
 };

test/mock/request-proxy.js

@@ -0,0 +1,26 @@
+// Copyright (c) 2015 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+'use strict';
+
+var noop = require('./noop.js');
+
+module.exports = {
+    proxyReq: noop
+};

test/mock/tchannel.js

@@ -0,0 +1,26 @@
+// Copyright (c) 2015 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+'use strict';
+
+var noop = require('./noop.js');
+
+module.exports = {
+    send: noop
+};

test/proxy_req_test.js

@@ -20,10 +20,13 @@
 
 'use strict';
 
+var _ = require('underscore');
 var test = require('tape');
 
 var allocRingpop = require('./lib/alloc-ringpop.js');
 var bootstrap = require('./lib/bootstrap.js');
+var mocks = require('./mock');
+var Ringpop = require('../index.js');
 
 test('proxyReq() proxies the request', function t(assert) {
     var left = allocRingpop('left');
@@ -43,3 +46,43 @@ test('proxyReq() proxies the request', function t(assert) {
         assert.end();
     });
 });
+
+test('proxyReq enforces required args', function t(assert) {
+    var ringpop = new Ringpop({
+        app: 'test',
+        hostPort: '127.0.0.1:3000'
+    });
+    ringpop.requestProxy = mocks.requestProxy;
+
+    function assertProxyReqThrows(opts) {
+        try {
+            ringpop.proxyReq(opts);
+            assert.fail('no exception thrown');
+        } catch (e) {
+            assert.pass('exception thrown');
+            return e;
+        }
+    }
+
+    function assertPropertyRequiredError(property, opts, newOpts) {
+        opts = _.extend(opts || {}, newOpts);
+
+        var exception = assertProxyReqThrows(opts);
+        assert.equals(exception.type, 'ringpop.options.property-required', 'err type is correct');
+        assert.equals(property, exception.property, 'err property is correct');
+
+        return opts
+    }
+
+    var exception = assertProxyReqThrows();
+    assert.equals(exception.type, 'ringpop.options.required', 'err type is correct');
+
+    var opts = assertPropertyRequiredError('key');
+    opts = assertPropertyRequiredError('dest', opts, { key: 'KEY0' });
+    opts = assertPropertyRequiredError('req', opts, { dest: '127.0.0.1:3000' });
+    opts = assertPropertyRequiredError('res', opts, { req: {} });
+
+    opts = _.extend(opts, { res: {} });
+    assert.doesNotThrow(function() { ringpop.proxyReq(opts); });
+    assert.end();
+});

test/request_proxy_test.js

@@ -0,0 +1,72 @@
+// Copyright (c) 2015 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+var allocRequest = require('./lib/alloc-request.js');
+var mocks = require('./mock');
+var RequestProxy = require('../lib/request-proxy.js');
+var Ringpop = require('../index.js');
+var test = require('tape');
+
+function createRingpop() {
+    return new Ringpop({
+        app: 'test',
+        hostPort: '127.0.0.1:3000',
+        channel: mocks.tchannel
+    });
+}
+
+function createRequestProxy() {
+    return new RequestProxy(createRingpop());
+}
+
+test('request proxy sends custom ringpop metadata in head', function t(assert) {
+    assert.plan(1);
+
+    var key = 'donaldduck';
+    var dest = 'disneyworld';
+
+    var proxy = createRequestProxy();
+    proxy.ringpop.channel.send = function(options, arg1, arg2, arg3, callback) {
+        var head = JSON.parse(arg2);
+        assert.equals(head.ringpopKey, key, 'sends key in head');
+        assert.end();
+    };
+    proxy.proxyReq({
+        key: key,
+        req: allocRequest({}),
+        dest: dest
+    });
+});
+
+test('request proxy emits head', function t(assert) {
+    assert.plan(3);
+
+    var proxy = createRequestProxy();
+    var headExpected = {
+        checksum: proxy.ringpop.membership.checksum,
+        ringpopKey: 'KEY0'
+    };
+    proxy.ringpop.on('request', function(req, res, head) {
+        assert.ok(req, 'req exists');
+        assert.ok(res, 'res exists');
+        assert.equals(head, headExpected, 'head is emitted');
+        assert.end();
+    });
+    proxy.handleRequest(headExpected, null, mocks.noop);
+});