Compare commits
2 Commits
0ad0df3926
...
695cc21cf7
Author | SHA1 | Date |
---|---|---|
Vitaliy Filippov | 695cc21cf7 | |
Vitaliy Filippov | c1278cf749 |
|
@ -0,0 +1,289 @@
|
|||
const fs = require('fs');
|
||||
const { URL } = require('url');
|
||||
const http = require('http');
|
||||
const https = require('https');
|
||||
|
||||
const ws = require('ws');
|
||||
|
||||
const EtcTree = require('./etctree.js');
|
||||
const TinyRaft = require('./tinyraft.js');
|
||||
|
||||
class RequestError
|
||||
{
|
||||
constructor(code, text)
|
||||
{
|
||||
this.code = code;
|
||||
this.message = text;
|
||||
}
|
||||
}
|
||||
|
||||
class AntiEtcd
|
||||
{
|
||||
constructor(cfg)
|
||||
{
|
||||
this.cfg = cfg;
|
||||
}
|
||||
|
||||
run()
|
||||
{
|
||||
this.clients = {};
|
||||
this.client_id = 1;
|
||||
this.etctree = new EtcTree(true);
|
||||
if (this.cfg.cert)
|
||||
{
|
||||
this.tls = { key: fs.readFileSync(this.cfg.key), cert: fs.readFileSync(this.cfg.cert) };
|
||||
this.server = https.createServer(this.tls, (req, res) => this.handleRequest(req, res));
|
||||
}
|
||||
else
|
||||
this.server = http.createServer((req, res) => this.handleRequest(req, res));
|
||||
this.wss = new ws.WebSocketServer({ server: this.server });
|
||||
this.wss.on('connection', (conn, req) => this.startWebsocket(conn, req));
|
||||
this.server.listen(this.cfg.port || 2379);
|
||||
}
|
||||
|
||||
fail(res, code, text)
|
||||
{
|
||||
res.writeHead(code);
|
||||
res.write(text);
|
||||
res.end();
|
||||
}
|
||||
|
||||
handleRequest(req, res)
|
||||
{
|
||||
let data = [];
|
||||
req.on('data', (chunk) => data.push(chunk));
|
||||
req.on('end', () =>
|
||||
{
|
||||
data = Buffer.concat(data);
|
||||
let body = '';
|
||||
let code = 200;
|
||||
let reply;
|
||||
try
|
||||
{
|
||||
if (req.headers['content-type'] != 'application/json')
|
||||
{
|
||||
throw new RequestError(400, 'content-type should be application/json');
|
||||
}
|
||||
body = data.toString();
|
||||
try
|
||||
{
|
||||
data = data.length ? JSON.parse(data) : {};
|
||||
}
|
||||
catch (e)
|
||||
{
|
||||
throw new RequestError(400, 'body should be valid JSON');
|
||||
}
|
||||
if (!(data instanceof Object) || data instanceof Array)
|
||||
{
|
||||
throw new RequestError(400, 'body should be JSON object');
|
||||
}
|
||||
reply = this.runHandler(req, data, res);
|
||||
reply = JSON.stringify(reply);
|
||||
}
|
||||
catch (e)
|
||||
{
|
||||
if (e instanceof RequestError)
|
||||
{
|
||||
code = e.code;
|
||||
reply = e.message;
|
||||
}
|
||||
else
|
||||
{
|
||||
console.error(e);
|
||||
code = 500;
|
||||
reply = 'Internal error: '+e.message;
|
||||
}
|
||||
}
|
||||
// Access log
|
||||
console.log(
|
||||
new Date().toISOString()+
|
||||
' '+(req.headers['x-forwarded-for'] || req.socket.remoteAddress)+
|
||||
' '+req.method+' '+req.url+' '+code+'\n '+body.replace(/\n/g, '\\n')+
|
||||
'\n '+reply.replace(/\n/g, '\\n')
|
||||
);
|
||||
// FIXME: Access log :req[X-Forwarded-For] [:date[clf]] pid=:pid ":method :url HTTP/:http-version" :status :res[content-length] ":referrer" ":user-agent" - :response-time ms
|
||||
reply = Buffer.from(reply);
|
||||
res.writeHead(200, {
|
||||
'Content-Type': 'application/json',
|
||||
'Content-Length': reply.length,
|
||||
});
|
||||
res.write(reply);
|
||||
res.end();
|
||||
});
|
||||
}
|
||||
|
||||
startWebsocket(socket, req)
|
||||
{
|
||||
const client_id = this.client_id++;
|
||||
this.clients[client_id] = {
|
||||
socket,
|
||||
alive: true,
|
||||
watches: {},
|
||||
};
|
||||
socket.on('pong', () => this.clients[client_id].alive = true);
|
||||
socket.on('error', console.error);
|
||||
const pinger = setInterval(() =>
|
||||
{
|
||||
if (!this.clients[client_id])
|
||||
{
|
||||
return;
|
||||
}
|
||||
if (!this.clients[client_id].alive)
|
||||
{
|
||||
return socket.terminate();
|
||||
}
|
||||
this.clients[client_id].alive = false;
|
||||
socket.ping(() => {});
|
||||
}, 30000);
|
||||
socket.on('message', (msg) =>
|
||||
{
|
||||
try
|
||||
{
|
||||
msg = JSON.parse(msg);
|
||||
}
|
||||
catch (e)
|
||||
{
|
||||
socket.send(JSON.stringify({ error: 'bad-json' }));
|
||||
return;
|
||||
}
|
||||
if (!msg)
|
||||
{
|
||||
socket.send(JSON.stringify({ error: 'empty-message' }));
|
||||
}
|
||||
else
|
||||
{
|
||||
this.handleMessage(client_id, msg, socket);
|
||||
}
|
||||
});
|
||||
socket.on('close', () =>
|
||||
{
|
||||
this.unsubscribeClient(client_id);
|
||||
clearInterval(pinger);
|
||||
delete this.clients[client_id];
|
||||
socket.terminate();
|
||||
});
|
||||
}
|
||||
|
||||
runHandler(req, data, res)
|
||||
{
|
||||
// v3/kv/txn
|
||||
// v3/kv/range
|
||||
// v3/kv/put
|
||||
// v3/kv/deleterange
|
||||
// v3/lease/grant
|
||||
// v3/lease/keepalive
|
||||
// v3/lease/revoke O_o
|
||||
// v3/kv/lease/revoke O_o
|
||||
const requestUrl = new URL(req.url, 'http://'+(req.headers.host || 'localhost'));
|
||||
if (requestUrl.pathname.substr(0, 4) == '/v3/')
|
||||
{
|
||||
const path = requestUrl.pathname.substr(4).replace(/\/+$/, '').replace(/\/+/g, '_');
|
||||
const cb = this['handle_'+path];
|
||||
if (cb)
|
||||
{
|
||||
if (req.method != 'POST')
|
||||
{
|
||||
throw new RequestError(405, 'Please use POST method');
|
||||
}
|
||||
return cb.call(this, data);
|
||||
}
|
||||
}
|
||||
throw new RequestError(404, 'Supported APIs: /v3/kv/txn, /v3/kv/range, /v3/kv/put, /v3/kv/deleterange, '+
|
||||
'/v3/lease/grant, /v3/lease/revoke, /v3/kv/lease/revoke, /v3/lease/keepalive');
|
||||
}
|
||||
|
||||
handle_kv_txn(data)
|
||||
{
|
||||
return this.etctree.api_txn(data);
|
||||
}
|
||||
|
||||
handle_kv_range(data)
|
||||
{
|
||||
const r = this.etctree.api_txn({ success: [ { request_range: data } ] });
|
||||
return { header: r.header, ...r.responses[0].response_range };
|
||||
}
|
||||
|
||||
handle_kv_put(data)
|
||||
{
|
||||
const r = this.etctree.api_txn({ success: [ { request_put: data } ] });
|
||||
return { header: r.header, ...r.responses[0].response_put };
|
||||
}
|
||||
|
||||
handle_kv_deleterange(data)
|
||||
{
|
||||
const r = this.etctree.api_txn({ success: [ { request_delete_range: data } ] });
|
||||
return { header: r.header, ...r.responses[0].response_delete_range };
|
||||
}
|
||||
|
||||
handle_lease_grant(data)
|
||||
{
|
||||
return this.etctree.api_grant_lease(data);
|
||||
}
|
||||
|
||||
handle_lease_revoke(data)
|
||||
{
|
||||
return this.etctree.api_revoke_lease(data);
|
||||
}
|
||||
|
||||
handle_kv_lease_revoke(data)
|
||||
{
|
||||
return this.etctree.api_revoke_lease(data);
|
||||
}
|
||||
|
||||
handle_lease_keepalive(data)
|
||||
{
|
||||
return this.etctree.api_keepalive_lease(data);
|
||||
}
|
||||
|
||||
handleMessage(client_id, msg, socket)
|
||||
{
|
||||
if (msg.create_request)
|
||||
{
|
||||
// FIXME progress_notify, filters, prev_kv
|
||||
const create_request = msg.create_request;
|
||||
if (!create_request.watch_id || !this.clients[client_id].watches[create_request.watch_id])
|
||||
{
|
||||
const watch = this.etctree.api_create_watch(
|
||||
{ ...create_request, watch_id: null }, (msg) => socket.send(JSON.stringify(msg))
|
||||
);
|
||||
if (!watch.created)
|
||||
{
|
||||
socket.send(JSON.stringify({ result: { header: { revision: this.etctree.mod_revision }, watch_id: create_request.watch_id, ...watch } }));
|
||||
}
|
||||
else
|
||||
{
|
||||
create_request.watch_id ||= watch.watch_id;
|
||||
this.clients[client_id].watches[create_request.watch_id] = watch.watch_id;
|
||||
socket.send(JSON.stringify({ result: { header: { revision: this.etctree.mod_revision }, watch_id: create_request.watch_id, created: true } }));
|
||||
}
|
||||
}
|
||||
}
|
||||
else if (msg.cancel_request)
|
||||
{
|
||||
const mapped_id = this.clients[client_id].watches[msg.cancel_request.watch_id];
|
||||
if (mapped_id)
|
||||
{
|
||||
this.etctree.api_cancel_watch({ watch_id: mapped_id });
|
||||
delete this.clients[client_id].watches[msg.cancel_request.watch_id];
|
||||
socket.send(JSON.stringify({ result: { header: { revision: this.etctree.mod_revision }, watch_id: msg.cancel_request.watch_id, canceled: true } }));
|
||||
}
|
||||
}
|
||||
else if (msg.progress_request)
|
||||
{
|
||||
socket.send(JSON.stringify({ result: { header: { revision: this.etctree.mod_revision } } }));
|
||||
}
|
||||
}
|
||||
|
||||
unsubscribeClient(client_id)
|
||||
{
|
||||
if (!this.clients[client_id])
|
||||
return;
|
||||
for (const watch_id in this.clients[client_id].watches)
|
||||
{
|
||||
const mapped_id = this.clients[client_id].watches[watch_id];
|
||||
this.etctree.api_cancel_watch({ watch_id: mapped_id });
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
new AntiEtcd({ port: 12379 }).run();
|
149
etctree.js
149
etctree.js
|
@ -1,4 +1,5 @@
|
|||
const crypto = require('crypto');
|
||||
const stableStringify = require('./stable-stringify.js');
|
||||
|
||||
/*type TreeNode = {
|
||||
value?: any,
|
||||
|
@ -13,38 +14,53 @@ const crypto = require('crypto');
|
|||
|
||||
class EtcTree
|
||||
{
|
||||
constructor()
|
||||
constructor(use_base64)
|
||||
{
|
||||
this.state = {};
|
||||
this.leases = {};
|
||||
this.watchers = {};
|
||||
this.watcher_id = 0;
|
||||
this.mod_revision = 0;
|
||||
this.use_base64 = use_base64;
|
||||
}
|
||||
|
||||
de64(k)
|
||||
{
|
||||
if (k == null) // null or undefined
|
||||
return k;
|
||||
return this.use_base64 ? Buffer.from(k, 'base64').toString() : k;
|
||||
}
|
||||
|
||||
b64(k)
|
||||
{
|
||||
if (k == null) // null or undefined
|
||||
return k;
|
||||
return this.use_base64 ? Buffer.from(k).toString('base64') : k;
|
||||
}
|
||||
|
||||
check(chk)
|
||||
{
|
||||
const parts = this.key_parts(chk.key);
|
||||
const parts = this.key_parts(this.de64(chk.key));
|
||||
const { cur } = this.get_subtree(parts, false, false);
|
||||
let check_value, ref_value;
|
||||
if (chk.target === 'MOD')
|
||||
{
|
||||
check_value = cur.mod_revision || 0;
|
||||
check_value = cur && cur.mod_revision || 0;
|
||||
ref_value = chk.mod_revision || 0;
|
||||
}
|
||||
else if (chk.target === 'CREATE')
|
||||
{
|
||||
check_value = cur.create_revision || 0;
|
||||
check_value = cur && cur.create_revision || 0;
|
||||
ref_value = chk.create_revision || 0;
|
||||
}
|
||||
else if (chk.target === 'VERSION')
|
||||
{
|
||||
check_value = cur.version || 0;
|
||||
check_value = cur && cur.version || 0;
|
||||
ref_value = chk.version || 0;
|
||||
}
|
||||
else if (chk.target === 'LEASE')
|
||||
{
|
||||
check_value = cur.lease;
|
||||
check_value = cur && cur.lease;
|
||||
ref_value = chk.lease;
|
||||
}
|
||||
else
|
||||
|
@ -64,14 +80,14 @@ class EtcTree
|
|||
|
||||
key_parts(key)
|
||||
{
|
||||
const parts = key.replace(/\/\/+/g, '/').replace(/^\/|\/$/g, '');
|
||||
const parts = key.replace(/\/\/+/g, '/').replace(/\/$/g, ''); // trim beginning?
|
||||
return parts === '' ? [] : parts.split('/');
|
||||
}
|
||||
|
||||
get_range(req)
|
||||
{
|
||||
const key = req.key;
|
||||
const end = req.range_end;
|
||||
const key = this.de64(req.key);
|
||||
const end = this.de64(req.range_end);
|
||||
if (end != null && (key[key.length-1] != '/' || end[end.length-1] != '0' ||
|
||||
end.substr(0, end.length-1) !== key.substr(0, key.length-1)))
|
||||
{
|
||||
|
@ -125,11 +141,11 @@ class EtcTree
|
|||
let id;
|
||||
while (!id || this.leases[id])
|
||||
{
|
||||
id = crypto.randomBytes(8).toString('base64');
|
||||
id = crypto.randomBytes(8).toString('hex');
|
||||
}
|
||||
const timer_id = setTimeout(() => this.api_revoke_lease({ ID: id }), req.TTL*1000);
|
||||
this.leases[id] = { ttl: req.TTL, timer_id, keys: {} };
|
||||
return { ID: id };
|
||||
return { header: { revision: this.mod_revision }, ID: id, TTL: req.TTL };
|
||||
}
|
||||
|
||||
api_keepalive_lease(req)
|
||||
|
@ -140,9 +156,10 @@ class EtcTree
|
|||
throw new Error('unknown lease');
|
||||
}
|
||||
clearTimeout(this.leases[id].timer_id);
|
||||
const ttl = this.leases[id].TTL;
|
||||
const ttl = this.leases[id].ttl;
|
||||
this.leases[id].timer_id = setTimeout(() => this.api_revoke_lease({ ID: id }), ttl*1000);
|
||||
return { TTL: ttl };
|
||||
// extra wrapping in { result: ... }
|
||||
return { result: { header: { revision: this.mod_revision }, ID: id, TTL: ''+ttl } };
|
||||
}
|
||||
|
||||
api_revoke_lease(req)
|
||||
|
@ -158,14 +175,16 @@ class EtcTree
|
|||
this.txn_action({ request_delete_range: { key } }, next_revision, notifications);
|
||||
}
|
||||
this.notify(notifications);
|
||||
return { header: { revision: this.mod_revision } };
|
||||
}
|
||||
|
||||
api_create_watch(req, send)
|
||||
{
|
||||
const { parts, all } = this.get_range(req);
|
||||
if (req.start_revision && req.start_revision < this.mod_revision)
|
||||
if (req.start_revision && this.compact_revision && this.compact_revision > req.start_revision)
|
||||
{
|
||||
throw new Error('history storage is not implemented');
|
||||
// Deletions up to this.compact_revision are forgotten
|
||||
return { compact_revision: this.compact_revision };
|
||||
}
|
||||
let watch_id = req.watch_id;
|
||||
if (watch_id instanceof Object)
|
||||
|
@ -195,9 +214,41 @@ class EtcTree
|
|||
cur.key_watchers = cur.key_watchers || [];
|
||||
cur.key_watchers.push(watch_id);
|
||||
}
|
||||
if (req.start_revision && req.start_revision < this.mod_revision)
|
||||
{
|
||||
// Send initial changes
|
||||
setImmediate(() =>
|
||||
{
|
||||
const events = [];
|
||||
const { cur } = this.get_subtree([], false, false);
|
||||
this.get_modified(events, cur, null, req.start_revision);
|
||||
send({ result: { header: { revision: this.mod_revision }, events } });
|
||||
});
|
||||
}
|
||||
return { watch_id, created: true };
|
||||
}
|
||||
|
||||
get_modified(events, cur, prefix, min_rev)
|
||||
{
|
||||
if (cur.mod_revision >= min_rev)
|
||||
{
|
||||
events.push({
|
||||
type: cur.value == null ? 'DELETE' : 'PUT',
|
||||
kv: cur.value == null ? { key: this.b64(prefix === null ? '' : prefix) } : {
|
||||
key: this.b64(prefix),
|
||||
value: this.b64(cur.value),
|
||||
},
|
||||
});
|
||||
}
|
||||
if (cur.children)
|
||||
{
|
||||
for (const k in cur.children)
|
||||
{
|
||||
this.get_modified(events, cur.children[k], prefix === null ? k : prefix+'/'+k, min_rev);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
api_cancel_watch(watch_id)
|
||||
{
|
||||
if (this.watchers[watch_id])
|
||||
|
@ -224,6 +275,7 @@ class EtcTree
|
|||
}
|
||||
delete this.watchers[watch_id];
|
||||
}
|
||||
return { canceled: true };
|
||||
}
|
||||
|
||||
api_txn({ compare, success, failure })
|
||||
|
@ -237,7 +289,7 @@ class EtcTree
|
|||
responses.push(this.txn_action(req, next_revision, notifications));
|
||||
}
|
||||
this.notify(notifications);
|
||||
return { revision: this.mod_revision, succeeded: !failed, responses };
|
||||
return { header: { revision: this.mod_revision }, succeeded: !failed, responses };
|
||||
}
|
||||
|
||||
notify(notifications)
|
||||
|
@ -251,58 +303,61 @@ class EtcTree
|
|||
{
|
||||
const watchers = notif.watchers;
|
||||
delete notif.watchers;
|
||||
const conv = { type: ('value' in notif) ? 'PUT' : 'DELETE', kv: notif };
|
||||
for (const wid of watchers)
|
||||
{
|
||||
if (this.watchers[wid])
|
||||
{
|
||||
by_watcher[wid] = by_watcher[wid] || { header: { revision: this.mod_revision }, events: {} };
|
||||
by_watcher[wid].events[notif.key] = notif;
|
||||
by_watcher[wid].events[notif.key] = conv;
|
||||
}
|
||||
}
|
||||
}
|
||||
for (const wid in by_watcher)
|
||||
{
|
||||
by_watcher[wid].events = Object.values(by_watcher[wid].events);
|
||||
this.watchers[wid].send(by_watcher[wid]);
|
||||
this.watchers[wid].send({ result: by_watcher[wid] });
|
||||
}
|
||||
}
|
||||
|
||||
txn_action(req, cur_revision, notifications)
|
||||
{
|
||||
if (req.request_range)
|
||||
if (req.request_range || req.requestRange)
|
||||
{
|
||||
// FIXME: limit, revision(-), sort_order, sort_target, serializable(-), keys_only,
|
||||
const request_range = req.request_range || req.requestRange;
|
||||
// FIXME: limit, revision(-), sort_order, sort_target, serializable(-),
|
||||
// count_only, min_mod_revision, max_mod_revision, min_create_revision, max_create_revision
|
||||
const { parts, all } = this.get_range(req.request_range);
|
||||
const { parts, all } = this.get_range(request_range);
|
||||
const { cur } = this.get_subtree(parts, false, false);
|
||||
const kvs = [];
|
||||
if (cur)
|
||||
{
|
||||
this.get_all(kvs, cur, all, parts.join('/'), req.request_range);
|
||||
this.get_all(kvs, cur, all, parts.join('/') || null, request_range);
|
||||
}
|
||||
return { kvs };
|
||||
return { response_range: { kvs } };
|
||||
}
|
||||
else if (req.request_put)
|
||||
else if (req.request_put || req.requestPut)
|
||||
{
|
||||
const request_put = req.request_put || req.requestPut;
|
||||
// FIXME: prev_kv, ignore_value(?), ignore_lease(?)
|
||||
const parts = this.key_parts(req.request_put.key);
|
||||
const parts = this.key_parts(this.de64(request_put.key));
|
||||
const key = parts.join('/');
|
||||
const value = req.request_put.value;
|
||||
const value = this.de64(request_put.value);
|
||||
const { cur, watchers } = this.get_subtree(parts, true, true);
|
||||
if (!eq(cur.value, value) || cur.lease != req.request_put.lease)
|
||||
if (!eq(cur.value, value) || cur.lease != request_put.lease)
|
||||
{
|
||||
if (cur.lease && this.leases[cur.lease])
|
||||
{
|
||||
delete this.leases[cur.lease].keys[key];
|
||||
}
|
||||
if (req.request_put.lease)
|
||||
if (request_put.lease)
|
||||
{
|
||||
if (!this.leases[req.request_put.lease])
|
||||
if (!this.leases[request_put.lease])
|
||||
{
|
||||
throw new Error('unknown lease: '+req.request_put.lease);
|
||||
throw new Error('unknown lease: '+request_put.lease);
|
||||
}
|
||||
cur.lease = req.request_put.lease;
|
||||
this.leases[req.request_put.lease].keys[key] = true;
|
||||
cur.lease = request_put.lease;
|
||||
this.leases[request_put.lease].keys[key] = true;
|
||||
}
|
||||
this.mod_revision = cur_revision;
|
||||
cur.version = (cur.version||0) + 1;
|
||||
|
@ -312,25 +367,29 @@ class EtcTree
|
|||
cur.create_revision = cur_revision;
|
||||
}
|
||||
cur.value = value;
|
||||
const notify = { watchers, key, value, mod_revision: cur.mod_revision };
|
||||
const notify = { watchers, key: this.b64(key), value: this.b64(value), mod_revision: cur.mod_revision };
|
||||
if (cur.lease)
|
||||
{
|
||||
notify.lease = cur.lease;
|
||||
}
|
||||
notifications.push(notify);
|
||||
}
|
||||
return {};
|
||||
return { response_put: {} };
|
||||
}
|
||||
else if (req.request_delete_range)
|
||||
else if (req.request_delete_range || req.requestDeleteRange)
|
||||
{
|
||||
const request_delete_range = req.request_delete_range || req.requestDeleteRange;
|
||||
// FIXME: prev_kv
|
||||
const { parts, all } = this.get_range(req.request_delete_range);
|
||||
const { parts, all } = this.get_range(request_delete_range);
|
||||
const { cur, watchers } = this.get_subtree(parts, false, true);
|
||||
const prevcount = notifications.length;
|
||||
if (cur)
|
||||
{
|
||||
this.delete_all(notifications, watchers, cur, all, parts.join('/'), cur_revision);
|
||||
this.delete_all(notifications, watchers, cur, all, parts.join('/') || null, cur_revision);
|
||||
}
|
||||
return { response_delete_range: { deleted: notifications.length-prevcount } };
|
||||
}
|
||||
return {};
|
||||
}
|
||||
|
||||
get_all(kvs, cur, all, prefix, req)
|
||||
|
@ -343,18 +402,18 @@ class EtcTree
|
|||
{
|
||||
if (req.keys_only)
|
||||
{
|
||||
kvs.push({ key: prefix, mod_revision: cur.mod_revision });
|
||||
kvs.push({ key: this.b64(prefix === null ? '' : prefix), mod_revision: cur.mod_revision });
|
||||
}
|
||||
else
|
||||
{
|
||||
kvs.push({ key: prefix, value: cur.value, mod_revision: cur.mod_revision });
|
||||
kvs.push({ key: this.b64(prefix === null ? '' : prefix), value: this.b64(cur.value), mod_revision: cur.mod_revision });
|
||||
}
|
||||
}
|
||||
if (all && cur.children)
|
||||
{
|
||||
for (let k in cur.children)
|
||||
{
|
||||
this.get_all(kvs, cur.children[k], true, prefix === '' ? k : prefix+'/'+k, req);
|
||||
this.get_all(kvs, cur.children[k], true, prefix === null ? k : prefix+'/'+k, req);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -364,18 +423,20 @@ class EtcTree
|
|||
if (cur.value != null)
|
||||
{
|
||||
// Do not actually forget the key until the deletion is confirmed by all replicas
|
||||
// ...and until it's not required by watchers
|
||||
cur.value = null;
|
||||
cur.version = 0;
|
||||
cur.create_revision = null;
|
||||
cur.mod_revision = cur_revision;
|
||||
this.mod_revision = cur_revision;
|
||||
notifications.push({ watchers, key: prefix, mod_revision: cur_revision });
|
||||
notifications.push({ watchers, key: this.b64(prefix === null ? '' : prefix), mod_revision: cur_revision });
|
||||
}
|
||||
if (all && cur.children)
|
||||
{
|
||||
for (let k in cur.children)
|
||||
{
|
||||
const subw = cur.children[k].watchers ? [ ...watchers, ...cur.children[k].watchers ] : watchers;
|
||||
this.delete_all(notifications, subw, cur.children[k], true, prefix === '' ? k : prefix+'/'+k, cur_revision);
|
||||
this.delete_all(notifications, subw, cur.children[k], true, prefix === null ? k : prefix+'/'+k, cur_revision);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -385,9 +446,11 @@ function eq(a, b)
|
|||
{
|
||||
if (a instanceof Object || b instanceof Object)
|
||||
{
|
||||
return JSON.stringify(a) === JSON.stringify(b);
|
||||
return stableStringify(a) === stableStringify(b);
|
||||
}
|
||||
return a == b;
|
||||
}
|
||||
|
||||
EtcTree.eq = eq;
|
||||
|
||||
module.exports = EtcTree;
|
||||
|
|
148
etctree.spec.js
148
etctree.spec.js
|
@ -1,59 +1,109 @@
|
|||
const EtcTree = require('./etctree.js');
|
||||
|
||||
describe('EtcTree', () =>
|
||||
const tests = {};
|
||||
|
||||
let cur_test = '';
|
||||
|
||||
const expect = (a, b) =>
|
||||
{
|
||||
it('should return the inserted value', () =>
|
||||
if (!EtcTree.eq(a, b))
|
||||
{
|
||||
const t = new EtcTree();
|
||||
expect(t.api_txn({ success: [ { request_put: { key: '/vitastor//config/global', value: { hello: 'world' } } } ] }))
|
||||
.toEqual({ succeeded: true, revision: 1, responses: [ {} ] });
|
||||
expect(t.api_txn({ success: [ { request_range: { key: '/vitastor/config/global' } } ] }))
|
||||
.toEqual({ succeeded: true, revision: 1, responses: [ { kvs: [ { key: 'vitastor/config/global', mod_revision: 1, value: { hello: 'world' } } ] } ] });
|
||||
expect(t.api_txn({ success: [ { request_range: { key: '/vitastor/config/', range_end: '/vitastor/config0' } } ] }))
|
||||
.toEqual({ succeeded: true, revision: 1, responses: [ { kvs: [ { key: 'vitastor/config/global', mod_revision: 1, value: { hello: 'world' } } ] } ] });
|
||||
expect(t.api_txn({ success: [ { request_range: { key: '/vitasto/', range_end: '/vitasto0' } } ] }))
|
||||
.toEqual({ succeeded: true, revision: 1, responses: [ { kvs: [] } ] });
|
||||
expect(t.api_txn({
|
||||
process.stderr.write(cur_test+' test:\nexpected: '+JSON.stringify(b)+'\nreal: '+JSON.stringify(a)+'\n'+new Error().stack.replace(/^.*\n.*\n/, '')+'\n');
|
||||
process.exit(1);
|
||||
}
|
||||
};
|
||||
|
||||
tests['read/write'] = () =>
|
||||
{
|
||||
const t = new EtcTree();
|
||||
expect(
|
||||
t.api_txn({ success: [ { request_put: { key: '/vitastor//config/global', value: { hello: 'world' } } } ] }),
|
||||
{ header: { revision: 1 }, succeeded: true, responses: [ { response_put: {} } ] }
|
||||
);
|
||||
expect(
|
||||
t.api_txn({ success: [ { request_range: { key: '/vitastor/config/global' } } ] }),
|
||||
{ header: { revision: 1 }, succeeded: true, responses: [ { response_range: {
|
||||
kvs: [ { key: '/vitastor/config/global', mod_revision: 1, value: { hello: 'world' } } ],
|
||||
} } ] }
|
||||
);
|
||||
expect(
|
||||
t.api_txn({ success: [ { request_range: { key: '/vitastor/config/', range_end: '/vitastor/config0' } } ] }),
|
||||
{ header: { revision: 1 }, succeeded: true, responses: [ { response_range: {
|
||||
kvs: [ { key: '/vitastor/config/global', mod_revision: 1, value: { hello: 'world' } } ],
|
||||
} } ] }
|
||||
);
|
||||
expect(
|
||||
t.api_txn({ success: [ { request_range: { key: '/vitasto/', range_end: '/vitasto0' } } ] }),
|
||||
{ header: { revision: 1 }, succeeded: true, responses: [ { response_range: { kvs: [] } } ] }
|
||||
);
|
||||
expect(
|
||||
t.api_txn({
|
||||
compare: [ { key: '/vitastor/config/global', target: 'MOD', mod_revision: 1, result: 'LESS' } ],
|
||||
success: [ { request_put: { key: '/vitastor//config/global', value: { hello: 'world' } } } ],
|
||||
failure: [ { request_range: { key: 'vitastor/config/global' } } ],
|
||||
})).toEqual({ succeeded: false, revision: 1, responses: [ { kvs: [ { key: 'vitastor/config/global', mod_revision: 1, value: { hello: 'world' } } ] } ] });
|
||||
expect(t.api_txn({
|
||||
failure: [ { request_range: { key: '/vitastor/config/global' } } ],
|
||||
}),
|
||||
{ header: { revision: 1 }, succeeded: false, responses: [ { response_range: {
|
||||
kvs: [ { key: '/vitastor/config/global', mod_revision: 1, value: { hello: 'world' } } ],
|
||||
} } ] }
|
||||
);
|
||||
expect(
|
||||
t.api_txn({
|
||||
compare: [ { key: '/vitastor/config/global', target: 'MOD', mod_revision: 2, result: 'LESS' } ],
|
||||
success: [ { request_put: { key: '/vitastor//config/global', value: { hello: 'world2' } } } ]
|
||||
})).toEqual({ succeeded: true, revision: 2, responses: [ {} ] });
|
||||
expect(t.api_txn({ success: [ { request_range: { key: '/vitastor/config/', range_end: '/vitastor/config0' } } ] }))
|
||||
.toEqual({ succeeded: true, revision: 2, responses: [ { kvs: [ { key: 'vitastor/config/global', mod_revision: 2, value: { hello: 'world2' } } ] } ] });
|
||||
});
|
||||
}),
|
||||
{ header: { revision: 2 }, succeeded: true, responses: [ { response_put: {} } ] }
|
||||
);
|
||||
expect(
|
||||
t.api_txn({ success: [ { request_range: { key: '/vitastor/config/', range_end: '/vitastor/config0' } } ] }),
|
||||
{ header: { revision: 2 }, succeeded: true, responses: [ { response_range: {
|
||||
kvs: [ { key: '/vitastor/config/global', mod_revision: 2, value: { hello: 'world2' } } ],
|
||||
} } ] }
|
||||
);
|
||||
};
|
||||
|
||||
it('should watch', () =>
|
||||
{
|
||||
const t = new EtcTree();
|
||||
const sent = [];
|
||||
const send = (event) => sent.push(event);
|
||||
expect(t.api_txn({ success: [ { request_put: { key: '/vitastor//config/global', value: { hello: 'world' } } } ] }))
|
||||
.toEqual({ succeeded: true, revision: 1, responses: [ {} ] });
|
||||
expect(t.api_create_watch({ watch_id: 1, key: '/vitastor/', range_end: '/vitastor0' }, send))
|
||||
.toEqual({ watch_id: 1, created: true });
|
||||
expect(sent).toEqual([]);
|
||||
expect(t.api_txn({ success: [ { request_put: { key: '/vitastor/osd/state/1', value: { ip: '1.2.3.4' } } } ] }))
|
||||
.toEqual({ succeeded: true, revision: 2, responses: [ {} ] });
|
||||
expect(sent).toEqual([ { header: { revision: 2 }, events: [ { key: 'vitastor/osd/state/1', mod_revision: 2, value: { ip: '1.2.3.4' } } ] } ]);
|
||||
});
|
||||
tests['watch'] = () =>
|
||||
{
|
||||
const t = new EtcTree();
|
||||
const sent = [];
|
||||
const send = (event) => sent.push(event);
|
||||
expect(
|
||||
t.api_txn({ success: [ { request_put: { key: '/vitastor//config/global', value: { hello: 'world' } } } ] }),
|
||||
{ header: { revision: 1 }, succeeded: true, responses: [ { response_put: {} } ] }
|
||||
);
|
||||
expect(
|
||||
t.api_create_watch({ watch_id: 1, key: '/vitastor/', range_end: '/vitastor0' }, send),
|
||||
{ watch_id: 1, created: true }
|
||||
);
|
||||
expect(sent, []);
|
||||
expect(
|
||||
t.api_txn({ success: [ { request_put: { key: '/vitastor/osd/state/1', value: { ip: '1.2.3.4' } } } ] }),
|
||||
{ header: { revision: 2 }, succeeded: true, responses: [ { response_put: {} } ] }
|
||||
);
|
||||
expect(sent, [ { result: { header: { revision: 2 }, events: [ { type: 'PUT', kv: { key: '/vitastor/osd/state/1', value: { ip: '1.2.3.4' }, mod_revision: 2 } } ] } } ]);
|
||||
};
|
||||
|
||||
it('should lease', async () =>
|
||||
{
|
||||
const t = new EtcTree();
|
||||
const sent = [];
|
||||
const send = (event) => sent.push(event);
|
||||
const leaseID = t.api_grant_lease({ TTL: 0.5 }).ID;
|
||||
expect(leaseID).not.toBeNull();
|
||||
expect(t.api_txn({ success: [ { request_put: { key: '/vitastor/osd/state/1', lease: leaseID, value: { ip: '1.2.3.4' } } } ] }))
|
||||
.toEqual({ succeeded: true, revision: 1, responses: [ {} ] });
|
||||
expect(t.api_create_watch({ watch_id: 1, key: '/vitastor/', range_end: '/vitastor0' }, send))
|
||||
.toEqual({ watch_id: 1, created: true });
|
||||
expect(sent).toEqual([]);
|
||||
await new Promise(ok => setTimeout(ok, 600));
|
||||
expect(sent).toEqual([ { header: { revision: 2 }, events: [ { key: 'vitastor/osd/state/1', mod_revision: 2 } ] } ]);
|
||||
});
|
||||
});
|
||||
tests['lease'] = async () =>
|
||||
{
|
||||
const t = new EtcTree();
|
||||
const sent = [];
|
||||
const send = (event) => sent.push(event);
|
||||
const leaseID = t.api_grant_lease({ TTL: 0.5 }).ID;
|
||||
expect(leaseID != null, true);
|
||||
expect(
|
||||
t.api_txn({ success: [ { request_put: { key: '/vitastor/osd/state/1', lease: leaseID, value: { ip: '1.2.3.4' } } } ] }),
|
||||
{ header: { revision: 1 }, succeeded: true, responses: [ { response_put: {} } ] }
|
||||
);
|
||||
expect(
|
||||
t.api_create_watch({ watch_id: 1, key: '/vitastor/', range_end: '/vitastor0' }, send),
|
||||
{ watch_id: 1, created: true }
|
||||
);
|
||||
expect(sent, []);
|
||||
await new Promise(ok => setTimeout(ok, 600));
|
||||
expect(sent, [ { result: { header: { revision: 2 }, events: [ { type: 'DELETE', kv: { key: '/vitastor/osd/state/1', mod_revision: 2 } } ] } } ]);
|
||||
};
|
||||
|
||||
for (cur_test in tests)
|
||||
{
|
||||
tests[cur_test]();
|
||||
console.log(cur_test+' test: OK');
|
||||
}
|
||||
|
|
|
@ -0,0 +1,78 @@
|
|||
// Copyright (c) Vitaliy Filippov, 2019+
|
||||
// License: MIT
|
||||
|
||||
function stableStringify(obj, opts)
|
||||
{
|
||||
if (!opts)
|
||||
opts = {};
|
||||
if (typeof opts === 'function')
|
||||
opts = { cmp: opts };
|
||||
let space = opts.space || '';
|
||||
if (typeof space === 'number')
|
||||
space = Array(space+1).join(' ');
|
||||
const cycles = (typeof opts.cycles === 'boolean') ? opts.cycles : false;
|
||||
const cmp = opts.cmp && (function (f)
|
||||
{
|
||||
return function (node)
|
||||
{
|
||||
return function (a, b)
|
||||
{
|
||||
let aobj = { key: a, value: node[a] };
|
||||
let bobj = { key: b, value: node[b] };
|
||||
return f(aobj, bobj);
|
||||
};
|
||||
};
|
||||
})(opts.cmp);
|
||||
const seen = new Map();
|
||||
return (function stringify (parent, key, node, level)
|
||||
{
|
||||
const indent = space ? ('\n' + new Array(level + 1).join(space)) : '';
|
||||
const colonSeparator = space ? ': ' : ':';
|
||||
if (node === undefined)
|
||||
{
|
||||
return;
|
||||
}
|
||||
if (typeof node !== 'object' || node === null)
|
||||
{
|
||||
return JSON.stringify(node);
|
||||
}
|
||||
if (node instanceof Array)
|
||||
{
|
||||
const out = [];
|
||||
for (let i = 0; i < node.length; i++)
|
||||
{
|
||||
const item = stringify(node, i, node[i], level+1) || JSON.stringify(null);
|
||||
out.push(indent + space + item);
|
||||
}
|
||||
return '[' + out.join(',') + indent + ']';
|
||||
}
|
||||
else
|
||||
{
|
||||
if (seen.has(node))
|
||||
{
|
||||
if (cycles)
|
||||
return JSON.stringify('__cycle__');
|
||||
throw new TypeError('Converting circular structure to JSON');
|
||||
}
|
||||
else
|
||||
seen.set(node, true);
|
||||
const keys = Object.keys(node).sort(cmp && cmp(node));
|
||||
const out = [];
|
||||
for (let i = 0; i < keys.length; i++)
|
||||
{
|
||||
const key = keys[i];
|
||||
const value = stringify(node, key, node[key], level+1);
|
||||
if (!value)
|
||||
continue;
|
||||
const keyValue = JSON.stringify(key)
|
||||
+ colonSeparator
|
||||
+ value;
|
||||
out.push(indent + space + keyValue);
|
||||
}
|
||||
seen.delete(node);
|
||||
return '{' + out.join(',') + indent + '}';
|
||||
}
|
||||
})({ '': obj }, '', obj, 0);
|
||||
}
|
||||
|
||||
module.exports = stableStringify;
|
75
tinyraft.js
75
tinyraft.js
|
@ -11,6 +11,14 @@
|
|||
//
|
||||
// Supports leader expiration like in NuRaft:
|
||||
// https://github.com/eBay/NuRaft/blob/master/docs/leadership_expiration.md
|
||||
//
|
||||
// Also supports leader priorities, similar to NuRaft but even simpler:
|
||||
// If a node receives a VoteRequest message with larger term but with smaller
|
||||
// priority than its own, it immediately starts a new voting round.
|
||||
// It guarantees that a node with non-maximum priority can't become leader
|
||||
// without being re-elected.
|
||||
// If all priorities are equal (or just zero), the election algorithm
|
||||
// becomes identical to the basic algorithm without priorities.
|
||||
|
||||
const EventEmitter = require('events');
|
||||
|
||||
|
@ -32,6 +40,7 @@ class TinyRaft extends EventEmitter
|
|||
// heartbeatTimeout?: number,
|
||||
// leadershipTimeout?: number,
|
||||
// initialTerm?: number,
|
||||
// leaderPriority?: number,
|
||||
// }
|
||||
constructor(config)
|
||||
{
|
||||
|
@ -43,6 +52,7 @@ class TinyRaft extends EventEmitter
|
|||
this.randomTimeout = config.randomTimeout > 0 ? Number(config.randomTimeout) : this.electionTimeout;
|
||||
this.heartbeatTimeout = Number(config.heartbeatTimeout) || 1000;
|
||||
this.leadershipTimeout = Number(config.leadershipTimeout) || 0;
|
||||
this.leaderPriority = Number(config.leaderPriority) || undefined;
|
||||
if (!this.nodeId || this.nodeId instanceof Object ||
|
||||
!(this.nodes instanceof Array) || this.nodes.filter(n => !n || n instanceof Object).length > 0 ||
|
||||
!(this.send instanceof Function))
|
||||
|
@ -85,7 +95,7 @@ class TinyRaft extends EventEmitter
|
|||
{
|
||||
if (node != this.nodeId)
|
||||
{
|
||||
this.send(node, { type: VOTE_REQUEST, term: this.term, leader: this.leader });
|
||||
this.send(node, { type: VOTE_REQUEST, term: this.term, leader: this.leader, priority: this.leaderPriority });
|
||||
}
|
||||
}
|
||||
// Next term will start right after this one times out
|
||||
|
@ -111,11 +121,11 @@ class TinyRaft extends EventEmitter
|
|||
{
|
||||
if (this.state == LEADER)
|
||||
{
|
||||
for (const node of this.votes[this.nodeId])
|
||||
for (const node of this.followers)
|
||||
{
|
||||
if (node != this.nodeId)
|
||||
{
|
||||
this.send(node, { type: PING, term: this.term });
|
||||
this.send(node, { type: PING, term: this.term, priority: this.leaderPriority });
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -125,36 +135,45 @@ class TinyRaft extends EventEmitter
|
|||
{
|
||||
if (msg.type == VOTE_REQUEST)
|
||||
{
|
||||
this.onReceiveVoteRequest(from, msg);
|
||||
this._onReceiveVoteRequest(from, msg);
|
||||
}
|
||||
else if (msg.type == VOTE)
|
||||
{
|
||||
this.onReceiveVote(from, msg);
|
||||
this._onReceiveVote(from, msg);
|
||||
}
|
||||
else if (msg.type == PING)
|
||||
{
|
||||
this.onReceivePing(from, msg);
|
||||
this._onReceivePing(from, msg);
|
||||
}
|
||||
else if (msg.type == PONG)
|
||||
{
|
||||
this.onReceivePong(from, msg);
|
||||
this._onReceivePong(from, msg);
|
||||
}
|
||||
}
|
||||
|
||||
onReceiveVoteRequest(from, msg)
|
||||
_onReceiveVoteRequest(from, msg)
|
||||
{
|
||||
if (msg.term > this.term && msg.leader)
|
||||
{
|
||||
this.leader = msg.leader;
|
||||
this.term = msg.term;
|
||||
this.state = CANDIDATE;
|
||||
this._nextTerm(this.heartbeatTimeout*2 + this.electionTimeout);
|
||||
this.emit('change', { state: this.state, term: this.term, leader: this.leader });
|
||||
if (this.leaderPriority && (msg.priority||0) < this.leaderPriority)
|
||||
{
|
||||
this.term = msg.term;
|
||||
this.start();
|
||||
}
|
||||
else
|
||||
{
|
||||
this.leader = msg.leader;
|
||||
this.term = msg.term;
|
||||
this.state = CANDIDATE;
|
||||
this._nextTerm(this.heartbeatTimeout*2 + this.electionTimeout);
|
||||
this.emit('change', { state: this.state, term: this.term, leader: this.leader });
|
||||
}
|
||||
}
|
||||
this.send(from, { type: VOTE, term: this.term, leader: this.leader });
|
||||
const prio = (this.leader == this.nodeId ? this.leaderPriority : undefined);
|
||||
this.send(from, { type: VOTE, term: this.term, leader: this.leader, leaderPriority: prio });
|
||||
}
|
||||
|
||||
onReceiveVote(from, msg)
|
||||
_onReceiveVote(from, msg)
|
||||
{
|
||||
if (!msg.leader || msg.term < this.term)
|
||||
{
|
||||
|
@ -165,7 +184,7 @@ class TinyRaft extends EventEmitter
|
|||
this.term = msg.term;
|
||||
this.leader = msg.leader;
|
||||
// Repeat VOTE to the leader to join it
|
||||
this.send(this.leader, { type: VOTE, term: this.term, leader: this.leader });
|
||||
this.send(this.leader, { type: VOTE, term: this.term, leader: this.leader, priority: msg.priority });
|
||||
}
|
||||
// add <from> as voter for <msg.leader>
|
||||
this.votes[msg.leader] = this.votes[msg.leader] || [];
|
||||
|
@ -192,14 +211,8 @@ class TinyRaft extends EventEmitter
|
|||
this.state = LEADER;
|
||||
this._nextTerm(this.leadershipTimeout > 0 ? this.leadershipTimeout : -1);
|
||||
this.followers = this.votes[this.nodeId];
|
||||
for (const follower of this.followers)
|
||||
{
|
||||
if (follower != this.nodeId)
|
||||
{
|
||||
// Send a heartbeat to confirm leadership
|
||||
this.send(follower, { type: PING, term: this.term });
|
||||
}
|
||||
}
|
||||
// Send a heartbeat to confirm leadership
|
||||
this._heartbeat();
|
||||
this.emit('change', { state: this.state, term: this.term, leader: this.nodeId, followers: this.votes[this.nodeId] });
|
||||
}
|
||||
else
|
||||
|
@ -211,7 +224,7 @@ class TinyRaft extends EventEmitter
|
|||
{
|
||||
this.followers = this.votes[this.nodeId];
|
||||
// Send a heartbeat to confirm leadership
|
||||
this.send(from, { type: PING, term: this.term });
|
||||
this.send(from, { type: PING, term: this.term, priority: this.leaderPriority });
|
||||
this.emit('change', { state: this.state, term: this.term, leader: this.nodeId, followers: this.votes[this.nodeId] });
|
||||
}
|
||||
else if (this._isVotingFailed())
|
||||
|
@ -220,7 +233,7 @@ class TinyRaft extends EventEmitter
|
|||
}
|
||||
}
|
||||
|
||||
onReceivePing(from, msg)
|
||||
_onReceivePing(from, msg)
|
||||
{
|
||||
if (this.state == CANDIDATE && this.term == msg.term && from == this.leader)
|
||||
{
|
||||
|
@ -237,7 +250,7 @@ class TinyRaft extends EventEmitter
|
|||
}
|
||||
}
|
||||
|
||||
onReceivePong(from, msg)
|
||||
_onReceivePong(from, msg)
|
||||
{
|
||||
if (this.state != LEADER)
|
||||
{
|
||||
|
@ -281,12 +294,14 @@ class TinyRaft extends EventEmitter
|
|||
if (this.state == LEADER)
|
||||
{
|
||||
this.votes[this.nodeId] = this.votes[this.nodeId].filter(n => nodes.indexOf(n) >= 0);
|
||||
this.emit('change', { state: this.state, term: this.term, leader: this.nodeId, followers: this.votes[this.nodeId] });
|
||||
if (this.votes[this.nodeId].length < (1 + (0 | this.nodes.length/2)))
|
||||
this.start();
|
||||
else
|
||||
this.emit('change', { state: this.state, term: this.term, leader: this.nodeId, followers: this.votes[this.nodeId] });
|
||||
}
|
||||
else if (this.state == FOLLOWER && nodes.indexOf(this.leader) < 0 || this.state == CANDIDATE)
|
||||
{
|
||||
r.nextTerm(-1);
|
||||
r.start();
|
||||
this.start();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
@ -153,6 +153,7 @@ async function testAddNode()
|
|||
|
||||
async function testLeadershipExpiration()
|
||||
{
|
||||
console.log('testLeadershipExpiration');
|
||||
const partitions = {};
|
||||
const nodes = newNodes(5, partitions, cfg => cfg.leadershipTimeout = 1500);
|
||||
// Check that 5 nodes are in quorum after 2000ms
|
||||
|
@ -182,6 +183,7 @@ async function testLeadershipExpiration()
|
|||
|
||||
async function testRestart()
|
||||
{
|
||||
console.log('testRestart');
|
||||
const nodes = newNodes(5, {}, cfg => cfg.initialTerm = 1000);
|
||||
let leaderChanges = 0, prevLeader = null;
|
||||
nodes[2].on('change', (st) =>
|
||||
|
@ -231,12 +233,92 @@ async function testRestart()
|
|||
console.log('testRestart: OK');
|
||||
}
|
||||
|
||||
async function testChangeNodes()
|
||||
{
|
||||
console.log('testChangeNodes');
|
||||
console.log('starting nodes 1-5');
|
||||
const nodes = newNodes(5, {}, cfg => cfg.initialTerm = 1000);
|
||||
// Check that 5 nodes are in quorum after 2000ms
|
||||
await new Promise(ok => setTimeout(ok, 2000));
|
||||
checkQuorum(nodes, 5);
|
||||
// Stop node 4
|
||||
console.log('stopping node 4');
|
||||
nodes[4].stop();
|
||||
delete nodes[4];
|
||||
// Wait 1000ms
|
||||
await new Promise(ok => setTimeout(ok, 1000));
|
||||
// Change nodes from 1 2 3 4 5 to 1 2 3 5 6
|
||||
console.log('starting node 6');
|
||||
newNode(6, nodes);
|
||||
nodes[6].start();
|
||||
nodes[1].setNodes([ 1, 2, 3, 5, 6 ]);
|
||||
nodes[2].setNodes([ 1, 2, 3, 5, 6 ]);
|
||||
nodes[3].setNodes([ 1, 2, 3, 5, 6 ]);
|
||||
nodes[5].setNodes([ 1, 2, 3, 5, 6 ]);
|
||||
nodes[6].setNodes([ 1, 2, 3, 5, 6 ]);
|
||||
// Check that 5 nodes are in quorum after 2000ms
|
||||
await new Promise(ok => setTimeout(ok, 2000));
|
||||
checkQuorum(nodes, 5);
|
||||
// Clean up
|
||||
for (const id in nodes)
|
||||
{
|
||||
if (nodes[id])
|
||||
nodes[id].stop();
|
||||
}
|
||||
console.log('testChangeNodes: OK');
|
||||
}
|
||||
|
||||
async function testLeaderPriority()
|
||||
{
|
||||
console.log('testLeaderPriority');
|
||||
console.log('starting nodes 1-5');
|
||||
const nodes = newNodes(5, {}, cfg => cfg.leaderPriority = cfg.nodeId+1);
|
||||
// Check that 5 nodes are in quorum after 2000ms
|
||||
await new Promise(ok => setTimeout(ok, 2000));
|
||||
checkQuorum(nodes, 5);
|
||||
if (nodes[1].leader != 5)
|
||||
{
|
||||
throw new Error('leader is not 5');
|
||||
}
|
||||
// Stop node 5
|
||||
console.log('stopping node 5');
|
||||
nodes[5].stop();
|
||||
delete nodes[5];
|
||||
// Wait 2000ms and check that the leader is now 4
|
||||
await new Promise(ok => setTimeout(ok, 2000));
|
||||
checkQuorum(nodes, 4);
|
||||
if (nodes[1].leader != 4)
|
||||
{
|
||||
throw new Error('leader is not 4');
|
||||
}
|
||||
// Stop node 4
|
||||
console.log('stopping node 4');
|
||||
nodes[4].stop();
|
||||
delete nodes[4];
|
||||
// Wait 2000ms and check that the leader is now 3
|
||||
await new Promise(ok => setTimeout(ok, 2000));
|
||||
checkQuorum(nodes, 3);
|
||||
if (nodes[1].leader != 3)
|
||||
{
|
||||
throw new Error('leader is not 3');
|
||||
}
|
||||
// Clean up
|
||||
for (const id in nodes)
|
||||
{
|
||||
if (nodes[id])
|
||||
nodes[id].stop();
|
||||
}
|
||||
console.log('testLeaderPriority: OK');
|
||||
}
|
||||
|
||||
async function run()
|
||||
{
|
||||
await testStartThenRemoveNode();
|
||||
await testAddNode();
|
||||
await testLeadershipExpiration();
|
||||
await testRestart();
|
||||
await testChangeNodes();
|
||||
await testLeaderPriority();
|
||||
process.exit(0);
|
||||
}
|
||||
|
||||
|
|
Loading…
Reference in New Issue