ETCD3

docker-compose.yml

@@ -1,6 +1,5 @@
 etcd:
-  image: quay.io/coreos/etcd
+  image: quay.io/coreos/etcd:v3.2.13
   ports:
     - "2379:2379"
-    - "4001:4001"
   command: etcd --listen-client-urls 'http://0.0.0.0:2379' --advertise-client-urls 'http://0.0.0.0:2379'

package.json

@@ -7,11 +7,14 @@
     "clean": "rm -rf .nyc_output coverage lib",
     "prepublish": "npm run build",
     "build": "babel src --out-dir lib -s --copy-files",
-    "test": "npm run lint && npm run nyc",
+    "build:watch": "npm run build -- --watch",
+    "test": "npm run nyc",
+    "test:dev": "npm run lint && npm run ava",
     "lint": "eslint src/ *.js *.json --ext .js,.json",
     "lint:fix": "npm run lint -- --fix",
     "ava": "NODE_ENV=test ava",
-    "nyc": "NODE_ENV=test nyc ava",
+    "ava:watch": "npm run ava -- --watch",
+    "nyc": "NODE_ENV=test nyc npm run test:dev",
     "coveralls": "nyc report --reporter=text-lcov | coveralls"
   },
   "repository": {
@@ -29,13 +32,16 @@
   "bin": {
     "squirrel-sync": "lib/bin/sync.js",
     "squirrel-watch": "lib/bin/watch.js",
-    "squirrel-dump": "lib/bin/dump.js"
+    "squirrel-dump": "lib/bin/dump.js",
+    "squirrel-restore": "lib/bin/restore.js"
   },
   "homepage": "https://github.com/coorpacademy/squirrel#readme",
   "dependencies": {
     "debug": "^3.1.0",
+    "etcd3": "^0.2.7",
     "lodash": "^4.17.2",
     "minimist": "^1.2.0",
+    "mkdirp": "^0.5.1",
     "node-etcd": "^5.0.1",
     "rxjs": "^5.0.0-beta.7"
   },
@@ -74,7 +80,7 @@
     ],
     "exclude": [
       "**/test/**/*",
-      "src/bin"
+      "**/bin/*.js"
     ],
     "cache": true,
     "sourceMap": false,

src/api.js

@@ -1,17 +1,12 @@
-import {join, relative} from 'path';
-import {find, getOr, keys, startsWith, curry} from 'lodash/fp';
+import {getOr, keys, pipe} from 'lodash/fp';
 import createDebug from 'debug';
-import {stringify, parseValue} from './parse';
-import {set$, del$} from './util/etcd';
 
 const debug = createDebug('squirrel');
 
-const createAPI = (store, client, options = {cwd: '/'}) => {
+const createAPI = (store, client) => {
   const getAll = index => {
     debug(`getAll => ${index}`);
-    return store('indexes').then(indexes => {
-      return keys(indexes && indexes[index]);
-    });
+    return store('indexes').then(pipe(getOr({}, index), keys));
   };
 
   const getByRaw = (index, key) => {
@@ -24,36 +19,19 @@ const createAPI = (store, client, options = {cwd: '/'}) => {
     return getByRaw(index, key).then(getOr(null, ['value']));
   };
 
-  const _get = curry((_path, node) => {
-    if (!node) return null;
-    if (relative(node.key, _path) === '') return node;
-
-    return _get(
-      _path,
-      find(function(child) {
-        return startsWith(child.key, _path);
-      }, node.nodes)
-    );
-  });
-
-  const get = path => {
-    debug(`get => ${path}`);
-    return store('node').then(_get(path));
+  const get = key => {
+    debug(`get => ${key}`);
+    return store('records').then(getOr(null, key));
   };
 
-  const set = (_path, value) => {
-    const fullPath = join(options.cwd, _path);
-    debug(`set => ${fullPath}`);
-    return set$(client, fullPath, stringify(value))
-      .pluck('node', 'value')
-      .map(parseValue)
-      .toPromise();
+  const set = (key, value) => {
+    debug(`set => ${key}`);
+    return client.put(key).value(JSON.stringify(value));
   };
 
-  const del = _path => {
-    const fullPath = join(options.cwd, _path);
-    debug(`del => ${fullPath}`);
-    return del$(client, fullPath).toPromise();
+  const del = key => {
+    debug(`del => ${key}`);
+    return client.delete().key(key);
   };
 
   return {

src/bin/dump.js

@@ -2,20 +2,17 @@
 
 import {resolve} from 'path';
 import minimist from 'minimist';
-import createFetch$ from '../fetch';
-import createSave from '../save';
-import makeEtcdClient from './etcd';
+import createEtcd from './helper/etcd';
+import dump from './helper/dump';
 
 const argz = minimist(process.argv.slice(2));
 
-const pathETCD = resolve('/', argz._[0]);
-const pathFS = resolve(process.cwd(), argz._[1]);
+const outDir = resolve(process.cwd(), argz._[0]);
+const namespace = argz._[1] || '';
 
-const client = makeEtcdClient(argz);
+const client = createEtcd(argz);
+const namespacedClient = client.namespace(namespace);
 
-const fetch$ = createFetch$(client, pathETCD).pluck('node');
-
-createSave(pathFS)(fetch$)
-  .toPromise()
-  .then(() => process.stdout.write('The end.\n'))
-  .catch(err => process.stderr.write(`${err.stack}\n`));
+dump(namespacedClient, outDir)
+  .then(console.log) // eslint-disable-line no-console
+  .catch(console.error); // eslint-disable-line no-console

src/bin/helper/dump.js

@@ -0,0 +1,28 @@
+import {promisify} from 'util';
+import {escape as qsEscape} from 'querystring';
+import {join} from 'path';
+import {writeFile} from 'fs';
+import {Observable} from 'rxjs';
+import mkdirp from 'mkdirp';
+import {parseRangeResponse} from '../../etcd/command';
+
+const mkdirpP = promisify(mkdirp);
+const writeFileP = promisify(writeFile);
+
+const dump = async (client, outDir) => {
+  await mkdirpP(outDir);
+
+  return Observable.fromPromise(client.getAll().exec())
+    .map(parseRangeResponse)
+    .concatMap(Observable.from)
+    .mergeMap(record =>
+      Observable.fromPromise(
+        writeFileP(join(outDir, qsEscape(record.key)), JSON.stringify(record.value, null, 2), {
+          encoding: 'utf8'
+        })
+      )
+    )
+    .toPromise();
+};
+
+export default dump;

src/bin/helper/etcd.js

@@ -0,0 +1,38 @@
+import {readFileSync} from 'fs';
+import {resolve} from 'path';
+import {Etcd3} from 'etcd3';
+
+const createEtcd = (
+  {
+    hosts = 'http://127.0.0.1:2379',
+    username,
+    password,
+    rootCertificate,
+
+    namespace = ''
+  } = {}
+) => {
+  const auth =
+    !!username || !!password
+      ? {
+          username,
+          password
+        }
+      : null;
+
+  const credentials = rootCertificate
+    ? {
+        rootCertificate: readFileSync(resolve(process.cwd(), rootCertificate))
+      }
+    : null;
+
+  const client = new Etcd3({
+    hosts: hosts.split(','),
+    auth,
+    credentials
+  });
+
+  return client;
+};
+
+export default createEtcd;

src/bin/helper/restore.js

@@ -0,0 +1,32 @@
+import {promisify} from 'util';
+import {unescape as qsUnescape} from 'querystring';
+import {join} from 'path';
+import {readFile, readdir} from 'fs';
+import mkdirp from 'mkdirp';
+import {Observable} from 'rxjs';
+
+const mkdirpP = promisify(mkdirp);
+const readFileP = promisify(readFile);
+const readdirP = promisify(readdir);
+
+const restore = async (client, inDir) => {
+  await mkdirpP(inDir);
+
+  return Observable.fromPromise(readdirP(inDir))
+    .concatMap(Observable.from)
+    .mergeMap(file =>
+      Observable.fromPromise(
+        readFileP(join(inDir, file), {
+          encoding: 'utf8'
+        })
+      )
+        .map(JSON.parse)
+        .map(content => [qsUnescape(file), content])
+    )
+    .mergeMap(([key, value]) =>
+      Observable.fromPromise(client.put(key).value(JSON.stringify(value, null, 4)))
+    )
+    .toPromise();
+};
+
+export default restore;

src/bin/helper/test/dump.js

@@ -0,0 +1,40 @@
+import {tmpdir} from 'os';
+import {join} from 'path';
+import {promisify} from 'util';
+import {readFile, readdir} from 'fs';
+import test from 'ava';
+import {Etcd3} from 'etcd3';
+import dump from '../dump';
+
+const readFileP = promisify(readFile);
+const readdirP = promisify(readdir);
+
+test('should create dump dir', async t => {
+  const tmpPath = join(tmpdir(), `squirrel-test-dump-${Date.now()}`);
+
+  const client = new Etcd3();
+  client.mock({
+    exec: (...argz) =>
+      Promise.resolve({
+        kvs: [
+          {
+            key: Buffer.from('foo'),
+            value: Buffer.from('{"foo": "foo"}'),
+            mod_revision: '1'
+          },
+          {
+            key: Buffer.from('bar'),
+            value: Buffer.from('{"bar": "bar"}'),
+            mod_revision: '1'
+          }
+        ]
+      })
+  });
+
+  await dump(client, tmpPath);
+
+  t.deepEqual(await readdirP(tmpPath), ['bar', 'foo']);
+
+  t.deepEqual(await readFileP(join(tmpPath, 'foo'), {encoding: 'utf8'}), '{\n  "foo": "foo"\n}');
+  t.deepEqual(await readFileP(join(tmpPath, 'bar'), {encoding: 'utf8'}), '{\n  "bar": "bar"\n}');
+});

src/bin/helper/test/etcd.js

@@ -0,0 +1,35 @@
+import {readFileSync} from 'fs';
+import {join} from 'path';
+import test from 'ava';
+import createEtcd from '../etcd';
+
+test('should create Etcd3 client', t => {
+  const client = createEtcd();
+
+  t.truthy(client.pool.options.hosts);
+  t.falsy(client.pool.options.auth);
+  t.falsy(client.pool.options.credentials);
+});
+
+test('should create Etcd3 client with options', t => {
+  const hosts = 'https://localhost:1,https://localhost:2';
+  const username = 'root';
+  const password = 'admin';
+  const rootCertificate = join(__dirname, 'fixtures/etcd.ca');
+
+  const client = createEtcd({
+    hosts,
+    username,
+    password,
+    rootCertificate
+  });
+
+  t.deepEqual(client.pool.options.hosts, hosts.split(','));
+  t.deepEqual(client.pool.options.auth, {
+    username,
+    password
+  });
+  t.deepEqual(client.pool.options.credentials, {
+    rootCertificate: readFileSync(rootCertificate)
+  });
+});

src/bin/helper/test/fixtures/dump/bar

@@ -0,0 +1,3 @@
+{
+  "bar": "bar"
+}

src/bin/helper/test/fixtures/dump/foo

@@ -0,0 +1,3 @@
+{
+  "foo": "foo"
+}

src/bin/helper/test/fixtures/etcd.ca

@@ -0,0 +1,3 @@
+-----BEGIN CERTIFICATE-----
+foo
+-----END CERTIFICATE-----

src/bin/helper/test/restore.js

@@ -0,0 +1,34 @@
+import {join} from 'path';
+import test from 'ava';
+import {Etcd3} from 'etcd3';
+import restore from '../restore';
+
+test('should restore dump dir', async t => {
+  t.plan(6);
+
+  const tmpPath = join(__dirname, 'fixtures/dump');
+
+  const puts = {
+    bar: {
+      key: Buffer.from('bar'),
+      value: Buffer.from('{\n    "bar": "bar"\n}')
+    },
+    foo: {
+      key: Buffer.from('foo'),
+      value: Buffer.from('{\n    "foo": "foo"\n}')
+    }
+  };
+
+  const client = new Etcd3();
+  client.mock({
+    exec: (service, method, params) => {
+      t.is(service, 'KV');
+      t.is(method, 'put');
+      const expected = puts[params.key.toString()];
+      t.deepEqual(params, expected);
+      return Promise.resolve();
+    }
+  });
+
+  await restore(client, tmpPath);
+});