Compare commits

..

2 Commits

13 changed files with 422 additions and 379 deletions

View File

@ -27,7 +27,6 @@ Simplistic miniature etcd replacement based on [TinyRaft](https://git.yourcmc.ru
- [/v3/lease/grant](#v3-lease-grant) - [/v3/lease/grant](#v3-lease-grant)
- [/v3/lease/keepalive](#v3-lease-keepalive) - [/v3/lease/keepalive](#v3-lease-keepalive)
- [/v3/lease/revoke or /v3/kv/lease/revoke](#v3-lease-revoke-or-v3-kv-lease-revoke) - [/v3/lease/revoke or /v3/kv/lease/revoke](#v3-lease-revoke-or-v3-kv-lease-revoke)
- [/v3/maintenance/status](#v3-maintenance-status)
- [Websocket-based watch APIs](#websocket-based-watch-apis) - [Websocket-based watch APIs](#websocket-based-watch-apis)
- [HTTP Error Codes](#http-error-codes) - [HTTP Error Codes](#http-error-codes)
@ -49,9 +48,8 @@ Antietcd doesn't background itself, so use systemd or start-stop-daemon to run i
``` ```
node_modules/.bin/anticli [OPTIONS] put <key> [<value>] node_modules/.bin/anticli [OPTIONS] put <key> [<value>]
node_modules/.bin/anticli [OPTIONS] get <key> [-p|--prefix] [-v|--print-value-only] [-k|--keys-only] [--no-temp] node_modules/.bin/anticli [OPTIONS] get <key> [-p|--prefix] [-v|--print-value-only] [-k|--keys-only]
node_modules/.bin/anticli [OPTIONS] del <key> [-p|--prefix] node_modules/.bin/anticli [OPTIONS] del <key> [-p|--prefix]
node_modules/.bin/anticli [OPTIONS] load [--with-lease] < dump.json
``` ```
For `put`, if `<value>` is not specified, it will be read from STDIN. For `put`, if `<value>` is not specified, it will be read from STDIN.
@ -72,9 +70,6 @@ Options:
<dt>--timeout 1000</dt> <dt>--timeout 1000</dt>
<dd>Specify request timeout in milliseconds</dd> <dd>Specify request timeout in milliseconds</dd>
<dt>--json or --write-out=json</dt>
<dd>Print raw response in JSON</dd>
</dl> </dl>
## Options ## Options
@ -86,9 +81,6 @@ Options:
<dt>--port 2379</dt> <dt>--port 2379</dt>
<dd>Listen port</dd> <dd>Listen port</dd>
<dt>--ip 192.168.0.10</dt>
<dd>Listen address (by default listen on all interfaces)</dd>
<dt>--cert &lt;cert&gt;</dt> <dt>--cert &lt;cert&gt;</dt>
<dd>Use TLS with this certificate file (PEM format)</dd> <dd>Use TLS with this certificate file (PEM format)</dd>
@ -105,9 +97,6 @@ Specify &lt;ca&gt; = &lt;cert&gt; if your certificate is self-signed.</dd>
<dt>--ws_keepalive_interval 30000</dt> <dt>--ws_keepalive_interval 30000</dt>
<dd>Client websocket ping (keepalive) interval in milliseconds</dd> <dd>Client websocket ping (keepalive) interval in milliseconds</dd>
<dt>--use_base64 1</dt>
<dd>Use base64 encoding of keys and values, like in etcd (enabled by default).</dd>
</dl> </dl>
### Persistence ### Persistence
@ -454,33 +443,6 @@ type LeaseRevokeResponse = {
} }
``` ```
### /v3/maintenance/status
Request:
```{}```
Response:
```ts
type MaintenanceStatusResponse = {
header: {
member_id?: string,
revision: number,
compact_revision: number,
raft_term?: number,
},
version: string,
cluster?: { [string]: string },
leader?: string,
followers?: string[],
raftTerm?: string,
raftState?: 'leader'|'follower'|'candidate',
// dbSize actually reports process memory usage
dbSize: number,
}
```
### Websocket-based watch APIs ### Websocket-based watch APIs
Client-to-server message format: Client-to-server message format:

86
anticli.js Executable file → Normal file
View File

@ -4,7 +4,6 @@
// (c) Vitaliy Filippov, 2024 // (c) Vitaliy Filippov, 2024
// License: Mozilla Public License 2.0 or Vitastor Network Public License 1.1 // License: Mozilla Public License 2.0 or Vitastor Network Public License 1.1
const fs = require('fs');
const fsp = require('fs').promises; const fsp = require('fs').promises;
const http = require('http'); const http = require('http');
const https = require('https'); const https = require('https');
@ -16,14 +15,13 @@ License: Mozilla Public License 2.0 or Vitastor Network Public License 1.1
Usage: Usage:
anticli.js [OPTIONS] put <key> [<value>] anticli.js [OPTIONS] put <key> [<value>]
anticli.js [OPTIONS] get <key> [-p|--prefix] [-v|--print-value-only] [-k|--keys-only] [--no-temp] anticli.js [OPTIONS] get <key> [-p|--prefix] [-v|--print-value-only] [-k|--keys-only]
anticli.js [OPTIONS] del <key> [-p|--prefix] anticli.js [OPTIONS] del <key> [-p|--prefix]
anticli.js [OPTIONS] load [--with-lease] < dump.json
Options: Options:
[--endpoints|-e http://node1:2379,http://node2:2379,http://node3:2379] [--endpoints|-e http://node1:2379,http://node2:2379,http://node3:2379]
[--cert cert.pem] [--key key.pem] [--timeout 1000] [--json] [--cert cert.pem] [--key key.pem] [--timeout 1000]
`; `;
class AntiEtcdCli class AntiEtcdCli
@ -56,19 +54,6 @@ class AntiEtcdCli
{ {
options['keys_only'] = true; options['keys_only'] = true;
} }
else if (arg == '--with_lease')
{
options['with_lease'] = true;
}
else if (arg == '--write_out' && args[i+1] == 'json')
{
i++;
options['json'] = true;
}
else if (arg == '--json' || arg == '--write_out=json')
{
options['json'] = true;
}
else if (arg[0] == '-' && arg[1] !== '-') else if (arg[0] == '-' && arg[1] !== '-')
{ {
process.stderr.write('Unknown option '+arg); process.stderr.write('Unknown option '+arg);
@ -83,9 +68,9 @@ class AntiEtcdCli
cmd.push(arg); cmd.push(arg);
} }
} }
if (!cmd.length || cmd[0] != 'get' && cmd[0] != 'put' && cmd[0] != 'del' && cmd[0] != 'load') if (!cmd.length || cmd[0] != 'get' && cmd[0] != 'put' && cmd[0] != 'del')
{ {
process.stderr.write('Supported commands: get, put, del, load. Use --help to see details\n'); process.stderr.write('Supported commands: get, put, del. Use --help to see details\n');
process.exit(1); process.exit(1);
} }
return [ cmd, options ]; return [ cmd, options ];
@ -117,51 +102,12 @@ class AntiEtcdCli
{ {
await this.del(cmd.slice(1)); await this.del(cmd.slice(1));
} }
else if (cmd[0] == 'load')
{
await this.load();
}
// wait until output is fully flushed // wait until output is fully flushed
await new Promise(ok => process.stdout.write('', ok)); await new Promise(ok => process.stdout.write('', ok));
await new Promise(ok => process.stderr.write('', ok)); await new Promise(ok => process.stderr.write('', ok));
process.exit(0); process.exit(0);
} }
async load()
{
const dump = JSON.parse(await new Promise((ok, no) => fs.readFile(0, { encoding: 'utf-8' }, (err, res) => err ? no(err) : ok(res))));
if (!dump.responses && !dump.kvs)
{
console.error('dump should be /kv/txn or /kv/range response in json format');
process.exit(1);
}
const success = [];
for (const r of (dump.responses
? dump.responses.map(r => r.response_range).filter(r => r)
: [ dump ]))
{
for (const kv of r.kvs)
{
if (kv.value == null)
{
console.error('dump should contain values');
process.exit(1);
}
success.push({ request_put: { key: kv.key, value: kv.value, lease: this.options.with_lease ? kv.lease||undefined : undefined } });
}
}
const res = await this.request('/v3/kv/txn', { success });
if (this.options.json)
{
process.stdout.write(JSON.stringify(res));
return;
}
if (res.succeeded)
{
process.stdout.write('OK, loaded '+success.length+' values\n');
}
}
async get(keys) async get(keys)
{ {
if (this.options.prefix) if (this.options.prefix)
@ -170,22 +116,6 @@ class AntiEtcdCli
} }
const txn = { success: keys.map(key => ({ request_range: this.options.prefix ? { key: b64(key+'/'), range_end: b64(key+'0') } : { key: b64(key) } })) }; const txn = { success: keys.map(key => ({ request_range: this.options.prefix ? { key: b64(key+'/'), range_end: b64(key+'0') } : { key: b64(key) } })) };
const res = await this.request('/v3/kv/txn', txn); const res = await this.request('/v3/kv/txn', txn);
if (this.options.notemp)
{
// Skip temporary values (values with lease)
for (const r of res.responses||[])
{
if (r.response_range)
{
r.response_range.kvs = r.response_range.kvs.filter(kv => !kv.lease);
}
}
}
if (this.options.json)
{
process.stdout.write(JSON.stringify(keys.length == 1 ? res.responses[0].response_range : res));
return;
}
for (const r of res.responses||[]) for (const r of res.responses||[])
{ {
if (r.response_range) if (r.response_range)
@ -209,7 +139,7 @@ class AntiEtcdCli
{ {
if (value === undefined) if (value === undefined)
{ {
value = await new Promise((ok, no) => fs.readFile(0, { encoding: 'utf-8' }, (err, res) => err ? no(err) : ok(res))); value = await fsp.readFile(0, { encoding: 'utf-8' });
} }
const res = await this.request('/v3/kv/put', { key: b64(key), value: b64(value) }); const res = await this.request('/v3/kv/put', { key: b64(key), value: b64(value) });
if (res.header) if (res.header)
@ -245,18 +175,18 @@ class AntiEtcdCli
{ {
if (res.json.error) if (res.json.error)
{ {
process.stderr.write(cur_url+': '+res.json.error+'\n'); process.stderr.write(cur_url+': '+res.json.error);
process.exit(1); process.exit(1);
} }
return res.json; return res.json;
} }
if (res.body) if (res.body)
{ {
process.stderr.write(cur_url+': '+res.body+'\n'); process.stderr.write(cur_url+': '+res.body);
} }
if (res.error) if (res.error)
{ {
process.stderr.write(cur_url+': '+res.error+'\n'); process.stderr.write(cur_url+': '+res.error);
if (!res.response || !res.response.statusCode) if (!res.response || !res.response.statusCode)
{ {
// This URL is unavailable // This URL is unavailable

View File

@ -226,10 +226,7 @@ class AntiCluster
} }
else else
{ {
this._log( this._log('Got dump from '+client.raft_node_id+' with stored term '+res.term);
'Got dump from '+client.raft_node_id+' with stored term '+res.term+
', mod_revision '+res.mod_revision+', compact_revision '+res.compact_revision
);
} }
this.resync_state.dumps[client.raft_node_id] = res.error ? null : res; this.resync_state.dumps[client.raft_node_id] = res.error ? null : res;
this._continueResync(); this._continueResync();
@ -328,19 +325,16 @@ class AntiCluster
this.antietcd.stored_term = this.raft.term; this.antietcd.stored_term = this.raft.term;
this.synced = true; this.synced = true;
runCallbacks(this, 'wait_sync', []); runCallbacks(this, 'wait_sync', []);
this._log( this._log('Synchronized with followers, new term is '+this.raft.term);
'Synchronized with followers, new term is '+this.raft.term+
', mod_revision '+this.antietcd.etctree.mod_revision+', compact_revision '+this.antietcd.etctree.compact_revision
);
} }
_isWrite(path, data) _isWrite(path, data)
{ {
if (path == 'kv_txn') if (path == 'kv_txn')
{ {
return (data.compare && data.compare.length || return ((!data.compare || !data.compare.length) &&
data.success && data.success.filter(f => f.request_put || f.requestPut || f.request_delete_range || f.requestDeleteRange).length || (!data.success || !data.success.filter(f => f.request_put || f.requestPut || f.request_delete_range || f.requestDeleteRange).length) &&
data.failure && data.failure.filter(f => f.request_put || f.requestPut || f.request_delete_range || f.requestDeleteRange).length); (!data.failure || !data.failure.filter(f => f.request_put || f.requestPut || f.request_delete_range || f.requestDeleteRange).length));
} }
return path != 'kv_range'; return path != 'kv_range';
} }
@ -458,10 +452,7 @@ class AntiCluster
this.antietcd.stored_term = msg.term; this.antietcd.stored_term = msg.term;
this.synced = true; this.synced = true;
runCallbacks(this, 'wait_sync', []); runCallbacks(this, 'wait_sync', []);
this._log( this._log('Synchronized with leader, new term is '+msg.term);
'Synchronized with leader, new term is '+this.raft.term+
', mod_revision '+this.antietcd.etctree.mod_revision+', compact_revision '+this.antietcd.etctree.compact_revision
);
client.socket.send(JSON.stringify({ request_id: msg.request_id, reply: {} })); client.socket.send(JSON.stringify({ request_id: msg.request_id, reply: {} }));
} }
else else

16
antietcd-app.js Executable file → Normal file
View File

@ -12,10 +12,10 @@ License: Mozilla Public License 2.0 or Vitastor Network Public License 1.1
Usage: Usage:
${process.argv[0]} ${process.argv[1]} \n\ ${process.argv[0]} ${process.argv[1]} \
[--cert ssl.crt] [--key ssl.key] [--port 12379] \n\ [--cert ssl.crt] [--key ssl.key] [--port 12379] \
[--data data.gz] [--persist-filter ./filter.js] [--persist_interval 500] \n\ [--data data.gz] [--persist-filter ./filter.js] [--persist_interval 500] \
[--node_id node1 --cluster_key abcdef --cluster node1=http://localhost:12379,node2=http://localhost:12380,node3=http://localhost:12381] \n\ [--node_id node1 --cluster_key abcdef --cluster node1=http://localhost:12379,node2=http://localhost:12380,node3=http://localhost:12381] \
[other options] [other options]
Supported etcd REST APIs: Supported etcd REST APIs:
@ -30,8 +30,6 @@ HTTP:
--port 2379 --port 2379
Listen port Listen port
--ip 192.168.0.10
Listen address (by default listen on all interfaces)
--cert <cert> --cert <cert>
Use TLS with this certificate file (PEM format) Use TLS with this certificate file (PEM format)
--key <key> --key <key>
@ -43,11 +41,6 @@ HTTP:
Require TLS client certificates signed by <ca> or by default CA to connect. Require TLS client certificates signed by <ca> or by default CA to connect.
--ws_keepalive_interval 30000 --ws_keepalive_interval 30000
Client websocket ping (keepalive) interval in milliseconds Client websocket ping (keepalive) interval in milliseconds
--merge_watches 1
Antietcd merges all watcher events into a single websocket message to provide
more ordering/transaction guarantees. Set to 0 to disable this behaviour.
--use_base64 1
Use base64 encoding of keys and values, like in etcd (enabled by default).
Persistence: Persistence:
@ -110,6 +103,7 @@ function parse()
options[arg.substr(2)] = process.argv[++i]; options[arg.substr(2)] = process.argv[++i];
} }
} }
options['stale_read'] = options['stale_read'] === '1' || options['stale_read'] === 'yes' || options['stale_read'] === 'true';
if (options['persist_filter']) if (options['persist_filter'])
{ {
options['persist_filter'] = require(options['persist_filter'])(options); options['persist_filter'] = require(options['persist_filter'])(options);

View File

@ -15,18 +15,13 @@ const ws = require('ws');
const EtcTree = require('./etctree.js'); const EtcTree = require('./etctree.js');
const AntiPersistence = require('./antipersistence.js'); const AntiPersistence = require('./antipersistence.js');
const AntiCluster = require('./anticluster.js'); const AntiCluster = require('./anticluster.js');
const { runCallbacks, de64, b64, RequestError } = require('./common.js'); const { runCallbacks, RequestError } = require('./common.js');
const VERSION = '1.1.1';
class AntiEtcd extends EventEmitter class AntiEtcd extends EventEmitter
{ {
constructor(cfg) constructor(cfg)
{ {
super(); super();
cfg['merge_watches'] = !('merge_watches' in cfg) || is_true(cfg['merge_watches']);
cfg['stale_read'] = !('stale_read' in cfg) || is_true(cfg['stale_read']);
cfg['use_base64'] = !('use_base64' in cfg) || is_true(cfg['use_base64']);
this.clients = {}; this.clients = {};
this.client_id = 1; this.client_id = 1;
this.etctree = new EtcTree(true); this.etctree = new EtcTree(true);
@ -77,10 +72,10 @@ class AntiEtcd extends EventEmitter
{ {
this.server = http.createServer((req, res) => this._handleRequest(req, res)); this.server = http.createServer((req, res) => this._handleRequest(req, res));
} }
this.wss = new ws.Server({ server: this.server }); this.wss = new ws.WebSocketServer({ server: this.server });
// eslint-disable-next-line no-unused-vars // eslint-disable-next-line no-unused-vars
this.wss.on('connection', (conn, req) => this._startWebsocket(conn, null)); this.wss.on('connection', (conn, req) => this._startWebsocket(conn, null));
this.server.listen(this.cfg.port || 2379, this.cfg.ip || undefined); this.server.listen(this.cfg.port || 2379);
} }
async stop() async stop()
@ -123,7 +118,7 @@ class AntiEtcd extends EventEmitter
let done = 0; let done = 0;
await new Promise((allOk, allNo) => await new Promise((allOk, allNo) =>
{ {
res.map(promise => promise.then((/*r*/) => res.map(promise => promise.then(res =>
{ {
if ((++done) == res.length) if ((++done) == res.length)
allOk(); allOk();
@ -137,11 +132,11 @@ class AntiEtcd extends EventEmitter
if (!this.cluster) if (!this.cluster)
{ {
// Run deletion compaction without followers // Run deletion compaction without followers
const mod_revision = this.etctree.mod_revision; const mod_revision = this.antietcd.etctree.mod_revision;
if (mod_revision - this.etctree.compact_revision > (this.cfg.compact_revisions||1000)*2) if (mod_revision - this.antietcd.etctree.compact_revision > (this.cfg.compact_revisions||1000)*2)
{ {
const revision = mod_revision - (this.cfg.compact_revisions||1000); const revision = mod_revision - (this.cfg.compact_revisions||1000);
this.etctree.compact(revision); this.antietcd.etctree.compact(revision);
} }
} }
} }
@ -186,13 +181,13 @@ class AntiEtcd extends EventEmitter
if (e instanceof RequestError) if (e instanceof RequestError)
{ {
code = e.code; code = e.code;
reply = e.message+'\n'; reply = e.message;
} }
else else
{ {
console.error(e); console.error(e);
code = 500; code = 500;
reply = 'Internal error: '+e.message+'\n'; reply = 'Internal error: '+e.message;
} }
} }
try try
@ -334,7 +329,7 @@ class AntiEtcd extends EventEmitter
if ((e instanceof RequestError) && e.code == 404) if ((e instanceof RequestError) && e.code == 404)
{ {
throw new RequestError(404, 'Supported APIs: /v3/kv/txn, /v3/kv/range, /v3/kv/put, /v3/kv/deleterange, '+ 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, /v3/maintenance/status'); '/v3/lease/grant, /v3/lease/revoke, /v3/kv/lease/revoke, /v3/lease/keepalive');
} }
else else
{ {
@ -350,7 +345,7 @@ class AntiEtcd extends EventEmitter
{ {
throw new RequestError(502, 'Server is stopping'); throw new RequestError(502, 'Server is stopping');
} }
if (this.cluster && path !== 'dump' && path != 'maintenance_status') if (path !== 'dump' && this.cluster)
{ {
const res = await this.cluster.checkRaftState( const res = await this.cluster.checkRaftState(
path, path,
@ -414,9 +409,9 @@ class AntiEtcd extends EventEmitter
} }
// public watch API // public watch API
async create_watch(params, callback, stream_id) async create_watch(params, callback)
{ {
const watch = this.etctree.api_create_watch(this._encodeWatch(params), (msg) => callback(this._encodeMsg(msg)), stream_id); const watch = this.etctree.api_create_watch({ ...params, watch_id: null }, callback);
if (!watch.created) if (!watch.created)
{ {
throw new RequestError(400, 'Requested watch revision is compacted', { compact_revision: watch.compact_revision }); throw new RequestError(400, 'Requested watch revision is compacted', { compact_revision: watch.compact_revision });
@ -433,71 +428,31 @@ class AntiEtcd extends EventEmitter
{ {
throw new RequestError(400, 'Watch not found'); throw new RequestError(400, 'Watch not found');
} }
this.etctree.api_cancel_watch(mapped_id); this.etctree.api_cancel_watch({ watch_id: mapped_id });
delete this.api_watches[watch_id]; delete this.api_watches[watch_id];
} }
// internal handlers // internal handlers
async _handle_kv_txn(data) async _handle_kv_txn(data)
{ {
if (this.cfg.use_base64) return await this.etctree.api_txn(data);
{
for (const item of data.compare||[])
{
if (item.key != null)
item.key = de64(item.key);
}
for (const items of [ data.success, data.failure ])
{
for (const item of items||[])
{
const req = item.request_range || item.requestRange ||
item.request_put || item.requestPut ||
item.request_delete_range || item.requestDeleteRange;
if (req.key != null)
req.key = de64(req.key);
if (req.range_end != null)
req.range_end = de64(req.range_end);
if (req.value != null)
req.value = de64(req.value);
}
}
}
const result = await this.etctree.api_txn(data);
if (this.cfg.use_base64)
{
for (const item of result.responses||[])
{
if (item.response_range)
{
for (const kv of item.response_range.kvs)
{
if (kv.key != null)
kv.key = b64(kv.key);
if (kv.value != null)
kv.value = b64(kv.value);
}
}
}
}
return result;
} }
async _handle_kv_range(data) async _handle_kv_range(data)
{ {
const r = await this._handle_kv_txn({ success: [ { request_range: data } ] }); const r = await this.etctree.api_txn({ success: [ { request_range: data } ] });
return { header: r.header, ...r.responses[0].response_range }; return { header: r.header, ...r.responses[0].response_range };
} }
async _handle_kv_put(data) async _handle_kv_put(data)
{ {
const r = await this._handle_kv_txn({ success: [ { request_put: data } ] }); const r = await this.etctree.api_txn({ success: [ { request_put: data } ] });
return { header: r.header, ...r.responses[0].response_put }; return { header: r.header, ...r.responses[0].response_put };
} }
async _handle_kv_deleterange(data) async _handle_kv_deleterange(data)
{ {
const r = await this._handle_kv_txn({ success: [ { request_delete_range: data } ] }); const r = await this.etctree.api_txn({ success: [ { request_delete_range: data } ] });
return { header: r.header, ...r.responses[0].response_delete_range }; return { header: r.header, ...r.responses[0].response_delete_range };
} }
@ -521,26 +476,6 @@ class AntiEtcd extends EventEmitter
return this.etctree.api_keepalive_lease(data); return this.etctree.api_keepalive_lease(data);
} }
_handle_maintenance_status(/*data*/)
{
const raft = this.cluster && this.cluster.raft;
return {
header: {
member_id: this.cfg.node_id || undefined,
revision: this.etctree.mod_revision,
compact_revision: this.etctree.compact_revision || 0,
raft_term: raft && raft.term || undefined,
},
version: 'antietcd '+AntiEtcd.VERSION,
cluster: this.cfg.cluster || undefined,
leader: raft && raft.leader || undefined,
followers: raft && raft.followers || undefined,
raftTerm: raft && raft.term || undefined,
raftState: raft && raft.state || undefined,
dbSize: process.memoryUsage().heapUsed,
};
}
// eslint-disable-next-line no-unused-vars // eslint-disable-next-line no-unused-vars
_handle_dump(data) _handle_dump(data)
{ {
@ -559,9 +494,8 @@ class AntiEtcd extends EventEmitter
const create_request = msg.create_request; const create_request = msg.create_request;
if (!create_request.watch_id || !client.watches[create_request.watch_id]) if (!create_request.watch_id || !client.watches[create_request.watch_id])
{ {
client.send_cb = client.send_cb || (msg => socket.send(JSON.stringify(this._encodeMsg(msg))));
const watch = this.etctree.api_create_watch( const watch = this.etctree.api_create_watch(
this._encodeWatch(create_request), client.send_cb, (this.cfg.merge_watches ? 'C'+client_id : null) { ...create_request, watch_id: null }, (msg) => socket.send(JSON.stringify(msg))
); );
if (!watch.created) if (!watch.created)
{ {
@ -580,7 +514,7 @@ class AntiEtcd extends EventEmitter
const mapped_id = client.watches[msg.cancel_request.watch_id]; const mapped_id = client.watches[msg.cancel_request.watch_id];
if (mapped_id) if (mapped_id)
{ {
this.etctree.api_cancel_watch(mapped_id); this.etctree.api_cancel_watch({ watch_id: mapped_id });
delete client.watches[msg.cancel_request.watch_id]; delete client.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 } })); socket.send(JSON.stringify({ result: { header: { revision: this.etctree.mod_revision }, watch_id: msg.cancel_request.watch_id, canceled: true } }));
} }
@ -599,35 +533,6 @@ class AntiEtcd extends EventEmitter
} }
} }
_encodeWatch(create_request)
{
const req = { ...create_request, watch_id: null };
if (this.cfg.use_base64)
{
if (req.key != null)
req.key = de64(req.key);
if (req.range_end != null)
req.range_end = de64(req.range_end);
}
return req;
}
_encodeMsg(msg)
{
if (this.cfg.use_base64 && msg.result && msg.result.events)
{
return { ...msg, result: { ...msg.result, events: msg.result.events.map(ev => ({
...ev,
kv: !ev.kv ? ev.kv : {
...ev.kv,
key: b64(ev.kv.key),
value: b64(ev.kv.value),
},
})) } };
}
return msg;
}
_unsubscribeClient(client_id) _unsubscribeClient(client_id)
{ {
if (!this.clients[client_id]) if (!this.clients[client_id])
@ -637,18 +542,11 @@ class AntiEtcd extends EventEmitter
for (const watch_id in this.clients[client_id].watches) for (const watch_id in this.clients[client_id].watches)
{ {
const mapped_id = this.clients[client_id].watches[watch_id]; const mapped_id = this.clients[client_id].watches[watch_id];
this.etctree.api_cancel_watch(mapped_id); this.etctree.api_cancel_watch({ watch_id: mapped_id });
} }
} }
} }
function is_true(s)
{
return s === true || s === 1 || s === '1' || s === 'yes' || s === 'true' || s === 'on';
}
AntiEtcd.RequestError = RequestError; AntiEtcd.RequestError = RequestError;
AntiEtcd.VERSION = VERSION;
module.exports = AntiEtcd; module.exports = AntiEtcd;

View File

@ -8,7 +8,7 @@ const zlib = require('zlib');
const stableStringify = require('./stable-stringify.js'); const stableStringify = require('./stable-stringify.js');
const EtcTree = require('./etctree.js'); const EtcTree = require('./etctree.js');
const { runCallbacks } = require('./common.js'); const { de64, runCallbacks } = require('./common.js');
class AntiPersistence class AntiPersistence
{ {
@ -60,18 +60,20 @@ class AntiPersistence
if (ev.lease) if (ev.lease)
{ {
// Values with lease are never persisted // Values with lease are never persisted
if (this.prev_value[ev.key] !== undefined) const key = de64(ev.key);
if (this.prev_value[key] !== undefined)
{ {
delete this.prev_value[ev.key]; delete this.prev_value[key];
changed = true; changed = true;
} }
} }
else else
{ {
const filtered = this.cfg.persist_filter(ev.key, ev.value == null ? undefined : ev.value); const key = de64(ev.key);
if (!EtcTree.eq(filtered, this.prev_value[ev.key])) const filtered = this.cfg.persist_filter(key, ev.value == null ? undefined : de64(ev.value));
if (!EtcTree.eq(filtered, this.prev_value[key]))
{ {
this.prev_value[ev.key] = filtered; this.prev_value[key] = filtered;
changed = true; changed = true;
} }
} }
@ -114,7 +116,7 @@ class AntiPersistence
this.wait_persist = []; this.wait_persist = [];
try try
{ {
let dump = this.antietcd.etctree.dump(true, this.cfg.persist_filter); let dump = this.antietcd.etctree.dump(true);
dump['term'] = this.antietcd.stored_term; dump['term'] = this.antietcd.stored_term;
dump = stableStringify(dump); dump = stableStringify(dump);
dump = await new Promise((ok, no) => zlib.gzip(dump, (err, res) => err ? no(err) : ok(res))); dump = await new Promise((ok, no) => zlib.gzip(dump, (err, res) => err ? no(err) : ok(res)));

View File

@ -1,11 +1,10 @@
// (c) Vitaliy Filippov, 2024 // (c) Vitaliy Filippov, 2024
// License: Mozilla Public License 2.0 or Vitastor Network Public License 1.1 // License: Mozilla Public License 2.0 or Vitastor Network Public License 1.1
class RequestError extends Error class RequestError
{ {
constructor(code, text, details) constructor(code, text, details)
{ {
super();
this.code = code; this.code = code;
this.message = text; this.message = text;
this.details = details; this.details = details;
@ -19,13 +18,6 @@ function de64(k)
return Buffer.from(k, 'base64').toString(); return Buffer.from(k, 'base64').toString();
} }
function b64(k)
{
if (k == null) // null or undefined
return k;
return Buffer.from(k).toString('base64');
}
function runCallbacks(obj, key, new_value) function runCallbacks(obj, key, new_value)
{ {
const cbs = obj[key]; const cbs = obj[key];
@ -42,6 +34,5 @@ function runCallbacks(obj, key, new_value)
module.exports = { module.exports = {
RequestError, RequestError,
de64, de64,
b64,
runCallbacks, runCallbacks,
}; };

View File

@ -19,7 +19,7 @@ const { RequestError } = require('./common.js');
class EtcTree class EtcTree
{ {
constructor() constructor(use_base64)
{ {
this.state = {}; this.state = {};
this.leases = {}; this.leases = {};
@ -27,6 +27,7 @@ class EtcTree
this.watcher_id = 0; this.watcher_id = 0;
this.mod_revision = 0; this.mod_revision = 0;
this.compact_revision = 0; this.compact_revision = 0;
this.use_base64 = use_base64;
this.replicate = null; this.replicate = null;
this.paused = false; this.paused = false;
this.active_immediate = []; this.active_immediate = [];
@ -50,9 +51,23 @@ class EtcTree
this.replicate = replicate; this.replicate = replicate;
} }
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) _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); const { cur } = this._get_subtree(parts, false, false);
let check_value, ref_value; let check_value, ref_value;
if (chk.target === 'MOD') if (chk.target === 'MOD')
@ -103,9 +118,9 @@ class EtcTree
_get_range(req) _get_range(req)
{ {
const key = req.key; const key = this.de64(req.key);
const end = req.range_end; const end = this.de64(req.range_end);
if (end != null && (key !== '' && end !== '') && (key[key.length-1] != '/' || end[end.length-1] != '0' || if (end != null && (key[key.length-1] != '/' || end[end.length-1] != '0' ||
end.substr(0, end.length-1) !== key.substr(0, key.length-1))) end.substr(0, end.length-1) !== key.substr(0, key.length-1)))
{ {
throw new RequestError(501, 'Non-directory range queries are unsupported'); throw new RequestError(501, 'Non-directory range queries are unsupported');
@ -153,7 +168,7 @@ class EtcTree
dump(persistent_only, value_filter) dump(persistent_only, value_filter)
{ {
const snapshot = { const snapshot = {
state: this._copy_tree(this.state, null, persistent_only, value_filter) || {}, state: this._copy_tree(this.state, persistent_only, value_filter) || {},
mod_revision: this.mod_revision, mod_revision: this.mod_revision,
compact_revision: this.compact_revision, compact_revision: this.compact_revision,
}; };
@ -169,13 +184,13 @@ class EtcTree
return snapshot; return snapshot;
} }
_copy_tree(cur, key, no_lease, value_filter) _copy_tree(cur, no_lease, value_filter)
{ {
let nonempty = cur.value != null && (!no_lease || !cur.lease); let nonempty = cur.value != null && (!no_lease || !cur.lease);
let filtered; let filtered;
if (nonempty && value_filter) if (nonempty && value_filter)
{ {
filtered = value_filter(key === null ? '' : key, cur.value); filtered = value_filter(cur.value);
nonempty = nonempty && filtered != null; nonempty = nonempty && filtered != null;
} }
const copy = (nonempty ? { ...cur } : {}); const copy = (nonempty ? { ...cur } : {});
@ -189,7 +204,7 @@ class EtcTree
let has_children = false; let has_children = false;
for (const k in cur.children) for (const k in cur.children)
{ {
const child = this._copy_tree(cur.children[k], key === null ? k : key+'/'+k, no_lease, value_filter); const child = this._copy_tree(cur.children[k], no_lease, value_filter);
if (child) if (child)
{ {
copy.children[k] = child; copy.children[k] = child;
@ -431,11 +446,6 @@ class EtcTree
{ {
this._delete_range({ key }, next_revision, notifications); this._delete_range({ key }, next_revision, notifications);
} }
if (this.leases[id].timer_id)
{
clearTimeout(this.leases[id].timer_id);
this.leases[id].timer_id = null;
}
delete this.leases[id]; delete this.leases[id];
} }
@ -529,13 +539,13 @@ class EtcTree
} }
} }
api_create_watch(req, send, stream_id) api_create_watch(req, send)
{ {
const { parts, all } = this._get_range(req); const { parts, all } = this._get_range(req);
if (req.start_revision && this.compact_revision && this.compact_revision > req.start_revision) if (req.start_revision && this.compact_revision && this.compact_revision > req.start_revision)
{ {
// Deletions up to this.compact_revision are forgotten // Deletions up to this.compact_revision are forgotten
return { canceled: true, cancel_reason: 'Revisions up to '+this.compact_revision+' are compacted', compact_revision: this.compact_revision }; return { compact_revision: this.compact_revision };
} }
let watch_id = req.watch_id; let watch_id = req.watch_id;
if (watch_id instanceof Object) if (watch_id instanceof Object)
@ -551,7 +561,6 @@ class EtcTree
this.watchers[watch_id] = { this.watchers[watch_id] = {
paths: [], paths: [],
send, send,
stream_id,
}; };
} }
this.watchers[watch_id].paths.push(parts); this.watchers[watch_id].paths.push(parts);
@ -566,7 +575,7 @@ class EtcTree
cur.key_watchers = cur.key_watchers || []; cur.key_watchers = cur.key_watchers || [];
cur.key_watchers.push(watch_id); cur.key_watchers.push(watch_id);
} }
if (req.start_revision && req.start_revision <= this.mod_revision) if (req.start_revision && req.start_revision < this.mod_revision)
{ {
// Send initial changes // Send initial changes
const imm = setImmediate(() => const imm = setImmediate(() =>
@ -579,7 +588,7 @@ class EtcTree
}); });
this.active_immediate.push(imm); this.active_immediate.push(imm);
} }
return { header: { revision: this.mod_revision }, watch_id, created: true }; return { watch_id, created: true };
} }
_get_modified(events, cur, prefix, min_rev) _get_modified(events, cur, prefix, min_rev)
@ -588,9 +597,9 @@ class EtcTree
{ {
const ev = { const ev = {
type: cur.value == null ? 'DELETE' : 'PUT', type: cur.value == null ? 'DELETE' : 'PUT',
kv: cur.value == null ? { key: (prefix === null ? '' : prefix) } : { kv: cur.value == null ? { key: this.b64(prefix === null ? '' : prefix) } : {
key: prefix, key: this.b64(prefix),
value: cur.value, value: this.b64(cur.value),
mod_revision: cur.mod_revision, mod_revision: cur.mod_revision,
}, },
}; };
@ -653,15 +662,15 @@ class EtcTree
{ {
if (this.watchers[wid]) if (this.watchers[wid])
{ {
const stream_id = this.watchers[wid].stream_id || wid; by_watcher[wid] = by_watcher[wid] || { header: { revision: this.mod_revision }, events: {} };
by_watcher[stream_id] = by_watcher[stream_id] || { send: this.watchers[wid].send, events: {} }; by_watcher[wid].events[notif.key] = conv;
by_watcher[stream_id].events[notif.key] = conv;
} }
} }
} }
for (const stream_id in by_watcher) for (const wid in by_watcher)
{ {
by_watcher[stream_id].send({ result: { header: { revision: this.mod_revision }, events: Object.values(by_watcher[stream_id].events) } }); by_watcher[wid].events = Object.values(by_watcher[wid].events);
this.watchers[wid].send({ result: by_watcher[wid] });
} }
} }
@ -718,9 +727,9 @@ class EtcTree
_put(request_put, cur_revision, notifications) _put(request_put, cur_revision, notifications)
{ {
// FIXME: prev_kv, ignore_value(?), ignore_lease(?) // FIXME: prev_kv, ignore_value(?), ignore_lease(?)
const parts = this._key_parts(request_put.key); const parts = this._key_parts(this.de64(request_put.key));
const key = parts.join('/'); const key = parts.join('/');
const value = request_put.value; const value = this.de64(request_put.value);
const { cur, watchers } = this._get_subtree(parts, true, true); const { cur, watchers } = this._get_subtree(parts, true, true);
if (cur.key_watchers) if (cur.key_watchers)
{ {
@ -753,7 +762,7 @@ class EtcTree
cur.create_revision = cur_revision; cur.create_revision = cur_revision;
} }
cur.value = value; 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) if (cur.lease)
{ {
notify.lease = cur.lease; notify.lease = cur.lease;
@ -784,10 +793,10 @@ class EtcTree
} }
if (cur.value != null) if (cur.value != null)
{ {
const item = { key: (prefix === null ? '' : prefix) }; const item = { key: this.b64(prefix === null ? '' : prefix) };
if (!req.keys_only) if (!req.keys_only)
{ {
item.value = cur.value; item.value = this.b64(cur.value);
item.mod_revision = cur.mod_revision; item.mod_revision = cur.mod_revision;
//item.create_revision = cur.create_revision; //item.create_revision = cur.create_revision;
//item.version = cur.version; //item.version = cur.version;
@ -824,7 +833,7 @@ class EtcTree
this.mod_revision = cur_revision; this.mod_revision = cur_revision;
notifications.push({ notifications.push({
watchers: cur.key_watchers ? [ ...watchers, ...cur.key_watchers ] : watchers, watchers: cur.key_watchers ? [ ...watchers, ...cur.key_watchers ] : watchers,
key: (prefix === null ? '' : prefix), key: this.b64(prefix === null ? '' : prefix),
mod_revision: cur_revision, mod_revision: cur_revision,
}); });
} }

View File

@ -64,7 +64,7 @@ tests['read/write'] = async () =>
); );
expect( expect(
t.dump(false), t.dump(false),
{"state":{"children":{"":{"children":{"vitastor":{"children":{"config":{"children":{"global":{"version":2,"mod_revision":2,"create_revision":1,"value":{"hello":"world2"}}}}}}}}}},"mod_revision":2,"compact_revision":0,"leases":{}} {"state":{"children":{"":{"children":{"vitastor":{"children":{"config":{"children":{"global":{"version":2,"mod_revision":2,"create_revision":1,"value":{"hello":"world2"}}}}}}}}}},"mod_revision":2,"leases":{}}
); );
t.destroy(); t.destroy();
}; };
@ -80,7 +80,7 @@ tests['watch'] = async () =>
); );
expect( expect(
t.api_create_watch({ watch_id: 1, key: '/vitastor/', range_end: '/vitastor0' }, send), t.api_create_watch({ watch_id: 1, key: '/vitastor/', range_end: '/vitastor0' }, send),
{ header: { revision: 1 }, watch_id: 1, created: true } { watch_id: 1, created: true }
); );
expect(sent, []); expect(sent, []);
expect( expect(
@ -91,38 +91,6 @@ tests['watch'] = async () =>
t.destroy(); t.destroy();
}; };
tests['merge watch'] = async () =>
{
const t = new EtcTree();
const sent = [];
const send = (event) => sent.push(event);
expect(
await t.api_txn({ success: [ { request_put: { key: '/vitastor//config/pgs', value: { items: {} } } } ] }),
{ header: { revision: 1 }, succeeded: true, responses: [ { response_put: {} } ] }
);
expect(
t.api_create_watch({ watch_id: 1, key: '/vitastor/config/pgs' }, send, 'X1' /* stream_id */),
{ header: { revision: 1 }, watch_id: 1, created: true }
);
expect(
t.api_create_watch({ watch_id: 2, key: '/vitastor/pg/history/', range_end: '/vitastor/pg/history0' }, send, 'X1' /* stream_id */),
{ header: { revision: 1 }, watch_id: 2, created: true },
);
expect(sent, []);
expect(
await t.api_txn({ success: [
{ request_put: { key: '/vitastor/config/pgs', value: { items: { 1: { 1: { osd_set: [ 1, 2, 3 ] } } } } } },
{ request_put: { key: '/vitastor/pg/history/1/1', value: { all_peers: [ 1, 2, 3, 4, 5 ] } } },
] }),
{ header: { revision: 2 }, succeeded: true, responses: [ { response_put: {} }, { response_put: {} } ] }
);
expect(sent, [ { result: { header: { revision: 2 }, events: [
{ type: 'PUT', kv: { key: '/vitastor/config/pgs', value: { items: { 1: { 1: { osd_set: [ 1, 2, 3 ] } } } }, mod_revision: 2 } },
{ type: 'PUT', kv: { key: '/vitastor/pg/history/1/1', value: { all_peers: [ 1, 2, 3, 4, 5 ] }, mod_revision: 2 } },
] } } ]);
t.destroy();
};
tests['lease'] = async () => tests['lease'] = async () =>
{ {
const t = new EtcTree(); const t = new EtcTree();
@ -132,18 +100,18 @@ tests['lease'] = async () =>
expect(leaseID != null, true); expect(leaseID != null, true);
expect( expect(
await t.api_txn({ success: [ { request_put: { key: '/vitastor/osd/state/1', lease: leaseID, value: { ip: '1.2.3.4' } } } ] }), await t.api_txn({ success: [ { request_put: { key: '/vitastor/osd/state/1', lease: leaseID, value: { ip: '1.2.3.4' } } } ] }),
{ header: { revision: 2 }, succeeded: true, responses: [ { response_put: {} } ] } { header: { revision: 1 }, succeeded: true, responses: [ { response_put: {} } ] }
); );
expect( expect(
t.api_create_watch({ watch_id: 1, key: '/vitastor/', range_end: '/vitastor0' }, send), t.api_create_watch({ watch_id: 1, key: '/vitastor/', range_end: '/vitastor0' }, send),
{ header: { revision: 2 }, watch_id: 1, created: true } { watch_id: 1, created: true }
); );
expect(sent, []); expect(sent, []);
const dump = t.dump(false); const dump = t.dump(false);
const expires = dump.leases[leaseID].expires; const expires = dump.leases[leaseID].expires;
expect(dump, {"state":{"children":{"":{"children":{"vitastor":{"children":{"osd":{"children":{"state":{"children":{"1":{"lease":leaseID,"version":1,"mod_revision":2,"create_revision":2,"value":{"ip":"1.2.3.4"}}}}}}}}}}}},"mod_revision":2,"compact_revision":0,"leases":{[leaseID]:{"ttl":0.5,"expires":expires}}}); expect(dump, {"state":{"children":{"":{"children":{"vitastor":{"children":{"osd":{"children":{"state":{"children":{"1":{"lease":leaseID,"version":1,"mod_revision":1,"create_revision":1,"value":{"ip":"1.2.3.4"}}}}}}}}}}}},"mod_revision":1,"leases":{[leaseID]:{"ttl":0.5,"expires":expires}}});
await new Promise(ok => setTimeout(ok, 600)); await new Promise(ok => setTimeout(ok, 600));
expect(sent, [ { result: { header: { revision: 3 }, events: [ { type: 'DELETE', kv: { key: '/vitastor/osd/state/1', mod_revision: 3 } } ] } } ]); expect(sent, [ { result: { header: { revision: 2 }, events: [ { type: 'DELETE', kv: { key: '/vitastor/osd/state/1', mod_revision: 2 } } ] } } ]);
t.pause_leases(); t.pause_leases();
t.load(dump); t.load(dump);
expect(t.dump(false), dump); expect(t.dump(false), dump);
@ -163,7 +131,7 @@ tests['update'] = async () =>
expect(leaseID != null, true); expect(leaseID != null, true);
expect( expect(
await t1.api_txn({ success: [ { request_put: { key: '/vitastor/osd/state/1', lease: leaseID, value: { ip: '1.2.3.4' } } } ] }), await t1.api_txn({ success: [ { request_put: { key: '/vitastor/osd/state/1', lease: leaseID, value: { ip: '1.2.3.4' } } } ] }),
{ header: { revision: 2 }, succeeded: true, responses: [ { response_put: {} } ] } { header: { revision: 1 }, succeeded: true, responses: [ { response_put: {} } ] }
); );
expect( expect(
await t2.api_txn({ success: [ { request_put: { key: '/vitastor/osd/state/1', value: { ip: '1.2.3.6' } } } ] }), await t2.api_txn({ success: [ { request_put: { key: '/vitastor/osd/state/1', value: { ip: '1.2.3.6' } } } ] }),
@ -171,38 +139,19 @@ tests['update'] = async () =>
); );
expect( expect(
await t1.api_txn({ success: [ { request_put: { key: '/vitastor/osd/state/1', lease: leaseID, value: { ip: '1.2.3.5' } } } ] }), await t1.api_txn({ success: [ { request_put: { key: '/vitastor/osd/state/1', lease: leaseID, value: { ip: '1.2.3.5' } } } ] }),
{ header: { revision: 3 }, succeeded: true, responses: [ { response_put: {} } ] } { header: { revision: 2 }, succeeded: true, responses: [ { response_put: {} } ] }
); );
let dump2 = t2.dump(); let dump2 = t2.dump();
t2.load(t1.dump(), true); t2.load(t1.dump(), true);
t1.load(dump2, true); t1.load(dump2, true);
let dump = t2.dump(false); let dump = t2.dump(false);
let expires = dump.leases[leaseID].expires; let expires = dump.leases[leaseID].expires;
expect(dump, {"state":{"children":{"":{"children":{"vitastor":{"children":{"osd":{"children":{"state":{"children":{"1":{"lease":leaseID,"version":2,"mod_revision":3,"create_revision":2,"value":{"ip":"1.2.3.5"}}}}}}}}}}}},"mod_revision":3,"compact_revision":0,"leases":{[leaseID]:{"ttl":0.5,"expires":expires}}}); expect(dump, {"state":{"children":{"":{"children":{"vitastor":{"children":{"osd":{"children":{"state":{"children":{"1":{"lease":leaseID,"version":2,"mod_revision":2,"create_revision":1,"value":{"ip":"1.2.3.5"}}}}}}}}}}}},"mod_revision":2,"leases":{[leaseID]:{"ttl":0.5,"expires":expires}}});
expect(t1.dump(false), {"state":{"children":{"":{"children":{"vitastor":{"children":{"osd":{"children":{"state":{"children":{"1":{"lease":leaseID,"version":2,"mod_revision":3,"create_revision":2,"value":{"ip":"1.2.3.5"}}}}}}}}}}}},"mod_revision":3,"compact_revision":0,"leases":{[leaseID]:{"ttl":0.5,"expires":expires}}}); expect(t1.dump(false), {"state":{"children":{"":{"children":{"vitastor":{"children":{"osd":{"children":{"state":{"children":{"1":{"lease":leaseID,"version":2,"mod_revision":2,"create_revision":1,"value":{"ip":"1.2.3.5"}}}}}}}}}}}},"mod_revision":2,"leases":{[leaseID]:{"ttl":0.5,"expires":expires}}});
t1.destroy(); t1.destroy();
t2.destroy(); t2.destroy();
}; };
tests['dump filter'] = async () =>
{
const t1 = new EtcTree();
const leaseID = (await t1.api_grant_lease({ TTL: 0.5 })).ID;
expect(leaseID != null, true);
expect(
await t1.api_txn({ success: [
{ request_put: { key: '/vitastor/osd/state/1', lease: leaseID, value: { ip: '1.2.3.6' } } },
{ request_put: { key: '/vitastor/osd/stats/1', value: { time: 123, read: [ 10, 1000, 10459 ] } } },
] }),
{ header: { revision: 2 }, succeeded: true, responses: [ { response_put: {} }, { response_put: {} } ] }
);
expect(
t1.dump(true, (key, value) => (key.substr(0, '/vitastor/osd/stats'.length) === '/vitastor/osd/stats' ? { time: value.time } : value)),
{"state":{"children":{"":{"children":{"vitastor":{"children":{"osd":{"children":{"stats":{"children":{"1":{"version":1,"mod_revision":2,"create_revision":2,"value":{"time":123}}}}}}}}}}}},"mod_revision":2,"compact_revision":0}
);
t1.destroy();
};
tests['replicate watcher'] = async () => tests['replicate watcher'] = async () =>
{ {
const t = new EtcTree(); const t = new EtcTree();

140
model_simple.js Normal file
View File

@ -0,0 +1,140 @@
#!/usr/bin/nodejs
// "Stupid" gossip algorithm simulation tool
function test_simple(options)
{
options.total ||= 100;
options.gossip ||= 4;
options.msgcap ||= 5;
options.update ||= 0;
options.initial ||= 5;
let messages_sent = 0;
let tick = 1;
const known = {};
const lists = {};
const listsv2 = {};
for (let i = 1; i <= options.total; i++)
{
known[i] = {};
lists[i] = [];
for (let j = 1; j <= (options.update ? options.total : options.initial); j++)
{
known[i][j] = 1; // meta version 1
lists[i].push(j);
}
listsv2[i] = [];
}
let cmp_lists;
let cmp_n;
if (options.update)
{
// We want to update <options.update> nodes metadata to version 2
for (let i = 1; i <= options.update; i++)
{
known[i][i] = 2;
listsv2[i].push(i);
}
cmp_lists = listsv2;
cmp_n = options.update;
}
else
{
// We want <options.total-options.initial> to join <options.initial>
for (let i = 1; i <= options.initial; i++)
{
if (!known[i][i])
{
known[i][i] = 1;
lists[i].push(i);
}
for (let alive = options.initial+1; alive <= options.total; alive++)
{
if (!known[i][alive])
{
known[i][alive] = true;
lists[i].push(alive);
}
}
}
cmp_lists = lists;
cmp_n = options.total;
}
let in_sync = 0;
for (let i = 1; i <= options.total; i++)
{
if (cmp_lists[i].length == cmp_n)
{
in_sync++;
}
}
let avg_known = 0;
while (in_sync < options.total)
{
console.log('tick '+tick+': '+in_sync+' in sync, avg '+avg_known);
for (let i = 1; i <= options.total; i++)
{
const known_i = lists[i];
const send_to = [];
for (let j = 0; j < options.gossip; j++)
{
send_to.push(known_i[0|(Math.random()*known_i.length)]);
}
const send_what = [];
for (let j = 0; j < options.msgcap; j++)
{
// FIXME: Exclude duplicates, exclude <send_to>
send_what.push(known_i[0|(Math.random()*known_i.length)]);
}
for (const alive of send_what)
{
for (const to of send_to)
{
if (!known[to][alive] || known[i][alive] > known[to][alive])
{
known[to][alive] = known[i][alive];
cmp_lists[to].push(alive);
if (cmp_lists[to].length == cmp_n)
{
console.log('node '+to+': tick '+tick);
in_sync++;
}
}
}
}
messages_sent += send_what.length*send_to.length;
}
avg_known = 0;
for (let i = 1; i <= options.total; i++)
{
avg_known += cmp_lists[i].length;
}
avg_known /= options.total;
tick++;
}
console.log('tick '+tick+': '+in_sync+' in sync, avg '+avg_known);
console.log(messages_sent+' messages sent');
}
const options = {};
for (let i = 2; i < process.argv.length; i++)
{
if (process.argv[i] === '-h' || process.argv[i] === '--help')
{
console.error('USAGE: '+process.argv[0]+' '+process.argv[1]+` [OPTIONS]
--gossip 4 how many nodes to gossip with every tick
--msgcap 5 how many nodes to gossip about every tick
--total 1000 total nodes
--update 0 total nodes to update if testing update. if 0 then test joining
--initial 5 initial nodes in sync to test joining (when --update is 0)`);
process.exit();
}
else if (process.argv[i].substr(0, 2) == '--')
{
options[process.argv[i].substr(2)] = 0|process.argv[i+1];
i++;
}
}
test_simple(options);

177
model_update.js Normal file
View File

@ -0,0 +1,177 @@
#!/usr/bin/nodejs
// https://github.com/hashicorp/memberlist simulation tool
class LimQ
{
constructor(retransmit, maxlen)
{
this.buckets = [];
for (let i = 0; i < retransmit; i++)
{
this.buckets.push([]);
}
this.len = 0;
this.maxlen = maxlen;
}
push(item)
{
if (this.len >= this.maxlen)
return;
const b = this.buckets[this.buckets.length-1];
b.push(item);
}
shift(n)
{
let items = [];
let move = [];
for (let i = this.buckets.length-1; i >= 0 && items.length < n; i--)
{
const rm = this.buckets[i].splice(0, n-items.length);
items.push.apply(items, rm);
if (i > 0)
for (const e of rm)
move.push([ e, i-1 ]);
else
this.len -= rm.length;
}
for (const e of move)
{
this.buckets[e[1]].push(e[0]);
}
return items;
}
}
function test_memberlist(options)
{
options.gossip ||= 4;
options.msgcap ||= 5;
options.max_ticks ||= 100000;
options.total ||= 100;
options.retransmit ||= 12;
options.update ||= 0;
options.initial ||= 5;
let tick = 0;
let messages_sent = 0;
const queue = {};
const known = {}; // { node: { other_node: meta_version } }
const lists = {};
const listsv2 = {};
for (let i = 1; i <= options.total; i++)
{
known[i] = {};
lists[i] = [];
for (let j = 1; j <= (options.update ? options.total : options.initial); j++)
{
known[i][j] = 1; // meta version 1
lists[i].push(j);
}
listsv2[i] = [];
queue[i] = new LimQ(options.retransmit, options.max_queue);
}
let cmp_lists;
let cmp_n;
if (options.update)
{
// We want to update <options.update> nodes metadata to version 2
for (let i = 1; i <= options.update; i++)
{
known[i][i] = 2;
listsv2[i].push(i);
queue[i].push(i);
}
cmp_lists = listsv2;
cmp_n = options.update;
}
else
{
// We want <options.total-options.initial> to join <options.initial>
for (let i = 1; i <= options.initial; i++)
{
for (let alive = options.initial+1; alive <= options.total; alive++)
{
known[i][alive] = 1;
lists[i].push(alive);
queue[i].push(alive);
}
}
cmp_lists = lists;
cmp_n = options.total;
}
let in_sync = 0;
for (let i = 1; i <= options.total; i++)
{
if (cmp_lists[i].length == cmp_n)
{
in_sync++;
}
}
let avg_known = 0;
while (in_sync < options.total && tick < options.max_ticks)
{
console.log('tick '+tick+': '+in_sync+' in sync, avg '+avg_known);
for (let i = 1; i <= options.total; i++)
{
const known_i = lists[i];
for (let g = 0; g < options.gossip; g++)
{
const to = known_i[0|(Math.random()*known_i.length)];
let send_what = queue[i].shift(options.msgcap);
messages_sent += send_what.length;
for (const alive of send_what)
{
if (!known[to][alive] || known[i][alive] > known[to][alive])
{
known[to][alive] = known[i][alive];
cmp_lists[to].push(alive);
queue[to].push(alive);
const cur_updated = cmp_lists[to].length;
if (cur_updated == cmp_n)
{
console.log('node '+to+': synced at tick '+tick);
in_sync++;
}
}
}
}
}
avg_known = 0;
for (let i = 1; i <= options.total; i++)
{
avg_known += cmp_lists[i].length;
}
avg_known /= options.total;
tick++;
}
console.log('tick '+tick+': '+in_sync+' in sync, avg '+avg_known);
console.log(messages_sent+' messages sent');
}
const options = {};
for (let i = 2; i < process.argv.length; i++)
{
if (process.argv[i] === '-h' || process.argv[i] === '--help')
{
console.error('USAGE: '+process.argv[0]+' '+process.argv[1]+` [OPTIONS]
--gossip 4 how many nodes to gossip with every tick
--msgcap 5 how many "alive" messages fits in a single packet (meta size/UDP packet size in memberlist)
--max_ticks 100000 execution limit
--max_queue 1024 queue size limit
--total 100 total nodes
--retransmit 12 retransmission count. by default log(total)*4 in memberlist
--update 0 total nodes to update if testing update. if 0 then test joining
--initial 5 initial nodes in sync to test joining (when --update is 0)`);
process.exit();
}
else if (process.argv[i].substr(0, 2) == '--')
{
options[process.argv[i].substr(2)] = 0|process.argv[i+1];
i++;
}
}
test_memberlist(options);

View File

@ -1,6 +1,6 @@
{ {
"name": "antietcd", "name": "antietcd",
"version": "1.1.1", "version": "1.0.0",
"description": "Simplistic etcd replacement based on TinyRaft", "description": "Simplistic etcd replacement based on TinyRaft",
"main": "antietcd.js", "main": "antietcd.js",
"scripts": { "scripts": {

View File

@ -24,7 +24,7 @@ function vitastor_persist_filter(cfg)
catch (e) catch (e)
{ {
console.error('invalid JSON in '+key+' = '+value+': '+e); console.error('invalid JSON in '+key+' = '+value+': '+e);
value = '{}'; value = {};
} }
} }
else else