Merge pull request #22 from uber/feature/discovery

Feature/discovery

Author: thanodnl

README.md

@@ -28,6 +28,38 @@ Command-line tools for Ringpop
 
     -h, --help     output usage information
     -V, --version  output the version number
+
+  Discovery:
+
+    Most of the commands can discover the ring via
+    a discoverUri like this: 'ringpop://127.0.0.1:3000'.
+    If no protocol is specified 'ringpop://' will be
+    used.
+
+    Supported protocols are:
+
+     - ringpop://
+       Discover the ring by connecting to a host of
+       the ring.
+
+       Example: ringpop://127.0.0.1:3000
+
+     - file://
+       Discover the ring by reading a json file
+       containing an array of host:port combinations
+
+       Example: file:///absolute/path/to/file
+       Example: file://./relative/path
+       File content: ["127.0.0.1:3000"]
+
+     - hyperbahn://
+       Discover the ring by querying hyperbahn for
+       the members of a service. When no hyperbahn
+       ip and port are given 127.0.0.1:21300 will be
+       used.
+
+       Example: hyperbahn:///ringpop
+       Example: hyperbahn://hyperbahn-ip:port/ringpop
 ```
 
 ## Tests

checksums.js

@@ -29,19 +29,19 @@ function main() {
     program
         .description('Prints membership checksums')
         .option('--tchannel-v1')
-        .usage('[options] <hostport or bootstrapfile>');
+        .usage('[options] <discoveryUri>');
     program.parse(process.argv);
 
-    var address = program.args[0];
+    var discoveryUri = program.args[0];
 
-    if (!address) {
-        console.error('Error: hostport or bootstrapfile is required');
+    if (!discoveryUri) {
+        console.error('Error: discoveryUri is required');
         process.exit(1);
     }
 
     var clusterManager = new ClusterManager({
         useTChannelV1: program.useTChannelV1,
-        coordAddr: address
+        discoveryUri: discoveryUri
     });
     clusterManager.fetchStats(function onStats(err) {
         if (err) {

commands.js

@@ -19,22 +19,22 @@
 // THE SOFTWARE.
 'use strict';
 
-function ReuseCommand(tchannelV1, coordinator, member, limit) {
+function ReuseCommand(tchannelV1, discoveryUri, member, limit) {
     this.useTChannelV1 = tchannelV1;
-    this.coordinator = coordinator;
+    this.discoveryUri = discoveryUri;
     this.member = member;
     this.limit = limit;
 }
 
-function StatusCommand(tchannelV1, coordinator, quiet) {
+function StatusCommand(tchannelV1, discoveryUri, quiet) {
     this.useTChannelV1 = tchannelV1;
-    this.coordinator = coordinator;
+    this.discoveryUri = discoveryUri;
     this.quiet = quiet;
 }
 
-function PartitionCommand(tchannelV1, coordinatorOrFile, quiet) {
+function PartitionCommand(tchannelV1, discoveryUri, quiet) {
     this.useTChannelV1 = tchannelV1;
-    this.coordinatorOrFile = coordinatorOrFile;
+    this.discoveryUri = discoveryUri;
     this.quiet = quiet;
 }
 

count.js

@@ -31,13 +31,13 @@ function main() {
         .option('-m --members', 'Count of members')
         .option('-p --partitions', 'Count of partitions')
         .option('--tchannel-v1')
-        .usage('[options] <hostport or bootstrapfile>');
+        .usage('[options] <discoveryUri>');
     program.parse(process.argv);
 
-    var coord = program.args[0];
+    var discoveryUri = program.args[0];
 
-    if (!coord) {
-        console.error('Error: hostport or path to bootstrap file is required');
+    if (!discoveryUri) {
+        console.error('Error: discoveryUri is required');
         process.exit(1);
     }
 
@@ -48,7 +48,7 @@ function main() {
 
     var clusterManager = new ClusterManager({
         useTChannelV1: program.tchannelV1,
-        coordAddr: coord
+        discoveryUri: discoveryUri
     });
     clusterManager.fetchStats(function onStats(err) {
         if (err) {

dump.js

@@ -30,13 +30,13 @@ function main() {
         .description('Dumps membership information to file')
         .option('-f --file <file>', 'File to dump to')
         .option('--tchannel-v1')
-        .usage('[options] <hostport or bootstrapfile>');
+        .usage('[options] <discoveryUri>');
     program.parse(process.argv);
 
-    var coord = program.args[0];
+    var discoveryUri = program.args[0];
 
-    if (!coord) {
-        console.error('Error: hostport or path to bootstrap file is required');
+    if (!discoveryUri) {
+        console.error('Error: discovery discoveryUri is required');
         process.exit(1);
     }
 
@@ -47,7 +47,7 @@ function main() {
 
     var clusterManager = new ClusterManager({
         useTChannelV1: program.tchannelV1,
-        coordAddr: coord
+        discoveryUri: discoveryUri
     });
     clusterManager.fetchStats(onStats);
 

lib/cluster.js

@@ -29,13 +29,15 @@ var Stats = require('./stats.js');
 var format = require('util').format;
 var startsWith = require('./util.js').startsWith;
 
+var discover = require('./discover').discover;
+
 function Cluster(opts) {
     opts = opts || {
         dumpTo: 'ringpop-admin-stats.dump'
     };
 
     this.useTChannelV1 = opts.useTChannelV1;
-    this.coordAddr = opts.coordAddr;
+    this.discoveryUri = opts.discoveryUri;
     this.adminClient = new AdminClient({
         useTChannelV1: this.useTChannelV1
     });
@@ -84,29 +86,8 @@ Cluster.prototype.getPartitionCount = function getPartitionCount() {
 };
 
 Cluster.prototype.getSeedList = function getSeedList(callback) {
-    var self = this;
-    // figure out if the coordinator is a file
-    fs.exists(self.coordAddr, function (exists) {
-        if (exists) {
-            // treat the coordinator as a bootstrap file
-            fs.readFile(self.coordAddr, function (err, body) {
-                if (err) {
-                    return callback(err);
-                }
-
-                try {
-                    body = JSON.parse(body);
-                } catch (e) {
-                    return callback(e);
-                }
-
-                // seed the list with the hosts in the bootstrap file
-                return callback(null, body);
-            });
-        } else {
-            return callback(null, [self.coordAddr]);
-        }
-    });
+    // use the discover library to fetch the seedlist
+    discover(this.discoveryUri, callback);
 };
 
 Cluster.prototype.fetchStats = function fetchStats(callback) {
@@ -128,7 +109,8 @@ Cluster.prototype.fetchStats = function fetchStats(callback) {
 
     this.getSeedList(function (err, seeds) {
         if (err) {
-            return callback(err);
+            callback(err);
+            return;
         }
         queueMembers(seeds);
     });
@@ -179,7 +161,7 @@ Cluster.prototype.fetchStats = function fetchStats(callback) {
 
     function onComplete(err, allStats) {
         if (allStats.length === 0) {
-            var addr = self.coordAddr;
+            var addr = self.discoveryUri;
             var msg = format('Failed to connect to ringpop listening on %s.', addr);
 
             // Check if user tries to connect to localhost or 127.0.0.1.
@@ -221,13 +203,13 @@ Cluster.prototype.fetchStats = function fetchStats(callback) {
 
 Cluster.prototype.lookup = function lookup(key, callback) {
     var self = this;
-    this.fetchStats(function onStats(err) {
+
+    this.getSeedList(function (err, seeds) {
         if (err) {
             callback(err);
             return;
         }
-
-        self.adminClient.lookup(self.coordAddr, key, callback);
+        self.adminClient.lookup(seeds[0], key, callback);
     });
 };
 

lib/discover/file.js

@@ -0,0 +1,23 @@
+var fs = require('fs');
+
+module.exports.discover = function fileDiscover(urlObj, uri, callback) {
+    var filename = urlObj.host + urlObj.pathname;
+    fs.readFile(filename, function (err, data) {
+        if (err) {
+            callback(err);
+            return;
+        }
+
+        // assume the file is a json list of hosts
+        try {
+            data = JSON.parse(data);
+        } catch (e) {
+            callback(e);
+            return;
+        }
+
+        // TODO add type tests for the data read from the file
+        callback(null, data);
+        return;
+    });
+};

lib/discover/hyperbahn.js

@@ -0,0 +1,78 @@
+var TChannelAsThrift = require('tchannel/as/thrift');
+var TChannel = require('tchannel');
+var path = require('path');
+
+module.exports.discover = function hyperbahnDiscover(urlObj, uri, callback) {
+    var host;
+    if (urlObj.host) {
+        host = urlObj.host;
+    } else {
+        host = '127.0.0.1:21300';
+    }
+
+    var client = TChannel();
+    var hChannel = client.makeSubChannel({
+        serviceName: 'hyperbahn',
+        peers: [host]
+    });
+    var tchannelAsThrift = TChannelAsThrift({
+        channel: hChannel,
+        entryPoint: path.join(__dirname, 'hyperbahn.thrift')
+    });
+
+    tchannelAsThrift.waitForIdentified({
+        host: host
+    }, function onIdentified(err) {
+        if (err) {
+            callback(err);
+            return;
+        }
+
+        tchannelAsThrift.request({
+            host: host,
+            timeout: 5000,
+
+            serviceName: 'hyperbahn',
+            headers: {
+                cn: 'hyperbahn'
+            },
+            hasNoParent: true
+        }).send('Hyperbahn::discover', {}, {
+            query: {
+                serviceName: urlObj.pathname.substr(1) // remove first slash
+            }
+        }, function (err, resp) {
+            if (err) {
+                callback(err);
+                return;
+            }
+
+            // parse the hyperbahn response
+            var seeds;
+            try {
+                seeds = resp.body.peers.map(function(peer) {
+                    switch (peer.ip.type) {
+                        case 'ipv4':
+                            return [intToIP(peer.ip.ipv4), peer.port].join(':');
+                        default:
+                            throw new Error('Hyperbahn returned peers with an invalid IP type: \'' + peer.ip.type + '\'');
+                    }
+                });
+            } catch (e) {
+                callback(e);
+                return;
+            }
+
+            callback(null, seeds);
+        });
+    });
+};
+
+function intToIP(int) {
+    var part1 = int & 255;
+    var part2 = ((int >> 8) & 255);
+    var part3 = ((int >> 16) & 255);
+    var part4 = ((int >> 24) & 255);
+
+    return part4 + "." + part3 + "." + part2 + "." + part1;
+}

lib/discover/hyperbahn.thrift

@@ -0,0 +1,35 @@
+exception NoPeersAvailable {
+    1: required string message
+    2: required string serviceName
+}
+
+exception InvalidServiceName {
+    1: required string message
+    2: required string serviceName
+}
+
+struct DiscoveryQuery {
+    1: required string serviceName
+}
+
+union IpAddress {
+  1: i32 ipv4
+}
+
+struct ServicePeer {
+  1: required IpAddress ip
+  2: required i32 port
+}
+
+struct DiscoveryResult {
+  1: required list<ServicePeer> peers
+}
+
+service Hyperbahn {
+    DiscoveryResult discover(
+        1: required DiscoveryQuery query
+    ) throws (
+        1: NoPeersAvailable noPeersAvailable
+        2: InvalidServiceName invalidServiceName
+    )
+}

lib/discover/index.js

@@ -0,0 +1,20 @@
+var url = require('url');
+
+module.exports.protocols = {
+    "hyperbahn:": require('./hyperbahn').discover,
+    "file:": require('./file').discover,
+    "ringpop:": require('./ringpop').discover,
+    // by default fallback on ringpop discovery
+    default: require('./ringpop').discover
+};
+
+// find a list of initial nodes to connect to based on the discover string
+module.exports.discover = function discover(uri, callback) {
+    var urlObj = url.parse(uri);
+    var handler = module.exports.protocols[urlObj.protocol];
+    if (typeof handler === 'undefined') {
+        handler = module.exports.protocols.default;
+    }
+
+    handler(urlObj, uri, callback);
+};