feat: simple leader election using zookeeper

Author: hyperlink

.editorconfig

@@ -0,0 +1,12 @@
+root = true
+
+[*]
+indent_style = space
+indent_size = 2
+end_of_line = lf
+charset = utf-8
+trim_trailing_whitespace = true
+insert_final_newline = true
+
+[*.md]
+trim_trailing_whitespace = false

.travis.yml

@@ -0,0 +1,24 @@
+language: node_js
+cache:
+  directories:
+    - node_modules
+notifications:
+  email: false
+node_js:
+  - '8'
+  - '7'
+  - '6'
+sudo: required
+services:
+  - docker
+before_install:
+  - docker --version
+  - sudo apt-get update
+  - sudo apt-get -y -o Dpkg::Options::="--force-confnew" install docker-engine
+before_script:
+  - npm prune
+after_success:
+  - npm run semantic-release
+branches:
+  except:
+    - /^v\d+\.\d+\.\d+$/

docker-compose.yml

@@ -0,0 +1,6 @@
+version: '2'
+services:
+  zookeeper:
+    image: jplock/zookeeper
+    ports:
+      - "2181:2181"

index.js

@@ -0,0 +1,117 @@
+'use strict'
+
+const debug = require('debug')('elector')
+const zookeeper = require('node-zookeeper-client')
+const async = require('async')
+const _ = require('lodash')
+const EventEmitter = require('events')
+
+class Elector extends EventEmitter {
+  constructor (zkOptions, electionPath = '/election') {
+    super()
+    this.id = null
+    this.electionPath = electionPath
+    this.client = zookeeper.createClient(zkOptions.host)
+    this.client.once('connected', () => this._onConnect())
+  }
+
+  connect () {
+    this.client.connect()
+  }
+
+  _onConnect () {
+    debug('connected!')
+    async.waterfall(
+      [
+        callback => {
+          debug('creating election path "%s"', this.electionPath)
+          this.client.mkdirp(this.electionPath, callback)
+        },
+
+        (path, callback) => {
+          debug('created path: "%s"', path)
+          debug('creating EPHEMERAL_SEQUENTIAL node')
+          this.client.create(
+            `${this.electionPath}/p_`,
+            zookeeper.CreateMode.EPHEMERAL_SEQUENTIAL,
+            callback
+          )
+        },
+
+        (path, callback) => {
+          debug('newly created znode is "%s"', path)
+          this.id = _.last(path.split('/'))
+          debug('my candidateId is', this.id)
+          this.emit('candidateId', this.id)
+          this._listCandidates(callback)
+        }
+      ],
+      (error, candidates) => {
+        if (error) {
+          return console.error(error)
+        }
+        debug('received candidates', candidates)
+        this._pickLeader(candidates)
+      }
+    )
+  }
+
+  _pickLeader (candidates) {
+    const previousLeadershipState = this.isLeader
+    this.isLeader = _.first(candidates.sort()) === this.id
+
+    if (previousLeadershipState !== this.isLeader) {
+      if (this.isLeader) {
+        debug('%s I am the leader!', this.id)
+        this.emit('leader')
+      } else {
+        this.emit('follower')
+        debug('%s I am a follower', this.id)
+      }
+    } else {
+      debug('%s state is the same', this.id)
+    }
+  }
+
+  _onCandidateChange (event) {
+    if (this.disconnecting) {
+      return
+    }
+
+    this._listCandidates((error, candidates) => {
+      if (error) {
+        debug(error)
+        this.emit('error', error)
+        return
+      }
+      debug('new candidates', candidates)
+      this._pickLeader(candidates)
+    })
+  }
+
+  _listCandidates (callback) {
+    this.client.getChildren(
+      this.electionPath,
+      event => this._onCandidateChange(event),
+      (error, candidates, stats) => {
+        if (error) {
+          return callback(error)
+        }
+        callback(null, candidates)
+      }
+    )
+  }
+
+  disconnect () {
+    debug('disconnecting')
+    this.disconnecting = true
+    this.client.remove(`${this.electionPath}/${this.id}`, error => {
+      this.client.close()
+      if (error) {
+        return debug(error)
+      }
+    })
+  }
+}
+
+module.exports = Elector

package.json

@@ -0,0 +1,51 @@
+{
+  "name": "elector",
+  "version": "0.0.0-development",
+  "description": "simple zookeeper based leader election",
+  "main": "index.js",
+  "scripts": {
+    "test": "standard && docker-compose up -d && mocha",
+    "format": "prettier-standard '*.js'",
+    "semantic-release": "semantic-release pre && npm publish && semantic-release post"
+  },
+  "repository": {
+    "type": "git",
+    "url": "https://github.com/hyperlink/elector.git"
+  },
+  "keywords": [
+    "zookeeper",
+    "leadership",
+    "election"
+  ],
+  "engines": {
+    "node": ">6.0.0"
+  },
+  "author": "Xiaoxin Lu <[email protected]>",
+  "license": "MIT",
+  "bugs": {
+    "url": "https://github.com/hyperlink/elector/issues"
+  },
+  "homepage": "https://github.com/hyperlink/elector#readme",
+  "dependencies": {
+    "async": "^2.4.1",
+    "debug": "^2.6.8",
+    "lodash": "^4.17.4",
+    "node-zookeeper-client": "^0.2.2"
+  },
+  "devDependencies": {
+    "husky": "^0.13.4",
+    "lint-staged": "^3.6.1",
+    "mocha": "^3.4.2",
+    "prettier-standard": "^5.1.0",
+    "semantic-release": "^6.3.6",
+    "standard": "^10.0.2"
+  },
+  "lint-staged": {
+    "linters": {
+      "src/**/*.js": [
+        "prettier-standard",
+        "git add"
+      ]
+    }
+  }
+}

test/index.js

@@ -0,0 +1,57 @@
+/* eslint-env mocha */
+
+'use strict'
+
+const Elector = require('../')
+const async = require('async')
+const _ = require('lodash')
+const assert = require('assert')
+
+describe('Elector', function () {
+  let electors = []
+
+  afterEach(function () {
+    for (let elector of electors) {
+      elector.disconnect()
+    }
+  })
+
+  it('should elect one node out of the five as leader', function (done) {
+    electors = _.times(5, createElector)
+
+    assert(electors.length, 5)
+
+    const leaders = []
+    const followers = []
+
+    async.each(
+      electors,
+      function (elector, callback) {
+        elector.connect()
+        function cb (error) {
+          if (error) {
+            return callback(error)
+          }
+          if (this.isLeader) {
+            leaders.push(this.id)
+          } else {
+            followers.push(this.id)
+          }
+          callback(null)
+        }
+        elector.on('leader', cb)
+        elector.on('follower', cb)
+        elector.on('error', callback)
+      },
+      function (error) {
+        assert.equal(leaders.length, 1)
+        assert.equal(followers.length, 4)
+        done(error)
+      }
+    )
+  })
+})
+
+function createElector () {
+  return new Elector({ host: 'localhost:2181' })
+}