Compare commits

...

2 Commits

6 changed files with 127 additions and 48 deletions

View File

@ -100,6 +100,9 @@ Specify &lt;ca&gt; = &lt;cert&gt; if your certificate is self-signed.</dd>
<dt>--ws_keepalive_interval 30000</dt>
<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>
### Persistence

View File

@ -43,6 +43,8 @@ HTTP:
Require TLS client certificates signed by <ca> or by default CA to connect.
--ws_keepalive_interval 30000
Client websocket ping (keepalive) interval in milliseconds
--use_base64 1
Use base64 encoding of keys and values, like in etcd (enabled by default).
Persistence:
@ -105,7 +107,14 @@ function parse()
options[arg.substr(2)] = process.argv[++i];
}
}
options['stale_read'] = options['stale_read'] === '1' || options['stale_read'] === 'yes' || options['stale_read'] === 'true';
if ('stale_read' in options)
{
options['stale_read'] = options['stale_read'] === '1' || options['stale_read'] === 'yes' || options['stale_read'] === 'true';
}
if ('use_base64' in options)
{
options['use_base64'] = options['use_base64'] === '1' || options['use_base64'] === 'yes' || options['use_base64'] === 'true';
}
if (options['persist_filter'])
{
options['persist_filter'] = require(options['persist_filter'])(options);

View File

@ -15,13 +15,21 @@ const ws = require('ws');
const EtcTree = require('./etctree.js');
const AntiPersistence = require('./antipersistence.js');
const AntiCluster = require('./anticluster.js');
const { runCallbacks, RequestError } = require('./common.js');
const { runCallbacks, de64, b64, RequestError } = require('./common.js');
class AntiEtcd extends EventEmitter
{
constructor(cfg)
{
super();
if (!('use_base64' in cfg))
{
cfg.use_base64 = 1;
}
if (!('stale_read' in cfg))
{
cfg.stale_read = 1;
}
this.clients = {};
this.client_id = 1;
this.etctree = new EtcTree(true);
@ -411,7 +419,7 @@ class AntiEtcd extends EventEmitter
// public watch API
async create_watch(params, callback)
{
const watch = this.etctree.api_create_watch({ ...params, watch_id: null }, callback);
const watch = this.etctree.api_create_watch(this._encodeWatch(params), (msg) => callback(this._encodeMsg(msg)));
if (!watch.created)
{
throw new RequestError(400, 'Requested watch revision is compacted', { compact_revision: watch.compact_revision });
@ -435,24 +443,64 @@ class AntiEtcd extends EventEmitter
// internal handlers
async _handle_kv_txn(data)
{
return await this.etctree.api_txn(data);
if (this.cfg.use_base64)
{
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)
{
const r = await this.etctree.api_txn({ success: [ { request_range: data } ] });
const r = await this._handle_kv_txn({ success: [ { request_range: data } ] });
return { header: r.header, ...r.responses[0].response_range };
}
async _handle_kv_put(data)
{
const r = await this.etctree.api_txn({ success: [ { request_put: data } ] });
const r = await this._handle_kv_txn({ success: [ { request_put: data } ] });
return { header: r.header, ...r.responses[0].response_put };
}
async _handle_kv_deleterange(data)
{
const r = await this.etctree.api_txn({ success: [ { request_delete_range: data } ] });
const r = await this._handle_kv_txn({ success: [ { request_delete_range: data } ] });
return { header: r.header, ...r.responses[0].response_delete_range };
}
@ -494,9 +542,7 @@ class AntiEtcd extends EventEmitter
const create_request = msg.create_request;
if (!create_request.watch_id || !client.watches[create_request.watch_id])
{
const watch = this.etctree.api_create_watch(
{ ...create_request, watch_id: null }, (msg) => socket.send(JSON.stringify(msg))
);
const watch = this.etctree.api_create_watch(this._encodeWatch(create_request), (msg) => socket.send(JSON.stringify(this._encodeMsg(msg))));
if (!watch.created)
{
socket.send(JSON.stringify({ result: { header: { revision: this.etctree.mod_revision }, watch_id: create_request.watch_id, ...watch } }));
@ -533,6 +579,35 @@ 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)
{
if (!this.clients[client_id])

View File

@ -8,7 +8,7 @@ const zlib = require('zlib');
const stableStringify = require('./stable-stringify.js');
const EtcTree = require('./etctree.js');
const { de64, runCallbacks } = require('./common.js');
const { runCallbacks } = require('./common.js');
class AntiPersistence
{
@ -60,20 +60,18 @@ class AntiPersistence
if (ev.lease)
{
// Values with lease are never persisted
const key = de64(ev.key);
if (this.prev_value[key] !== undefined)
if (this.prev_value[ev.key] !== undefined)
{
delete this.prev_value[key];
delete this.prev_value[ev.key];
changed = true;
}
}
else
{
const key = de64(ev.key);
const filtered = this.cfg.persist_filter(key, ev.value == null ? undefined : de64(ev.value));
if (!EtcTree.eq(filtered, this.prev_value[key]))
const filtered = this.cfg.persist_filter(ev.key, ev.value == null ? undefined : ev.value);
if (!EtcTree.eq(filtered, this.prev_value[ev.key]))
{
this.prev_value[key] = filtered;
this.prev_value[ev.key] = filtered;
changed = true;
}
}

View File

@ -1,10 +1,11 @@
// (c) Vitaliy Filippov, 2024
// License: Mozilla Public License 2.0 or Vitastor Network Public License 1.1
class RequestError
class RequestError extends Error
{
constructor(code, text, details)
{
super();
this.code = code;
this.message = text;
this.details = details;
@ -18,6 +19,13 @@ function de64(k)
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)
{
const cbs = obj[key];
@ -34,5 +42,6 @@ function runCallbacks(obj, key, new_value)
module.exports = {
RequestError,
de64,
b64,
runCallbacks,
};

View File

@ -19,7 +19,7 @@ const { RequestError } = require('./common.js');
class EtcTree
{
constructor(use_base64)
constructor()
{
this.state = {};
this.leases = {};
@ -27,7 +27,6 @@ class EtcTree
this.watcher_id = 0;
this.mod_revision = 0;
this.compact_revision = 0;
this.use_base64 = use_base64;
this.replicate = null;
this.paused = false;
this.active_immediate = [];
@ -51,23 +50,9 @@ class EtcTree
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)
{
const parts = this._key_parts(this.de64(chk.key));
const parts = this._key_parts(chk.key);
const { cur } = this._get_subtree(parts, false, false);
let check_value, ref_value;
if (chk.target === 'MOD')
@ -118,8 +103,8 @@ class EtcTree
_get_range(req)
{
const key = this.de64(req.key);
const end = this.de64(req.range_end);
const key = req.key;
const end = 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)))
{
@ -444,7 +429,7 @@ class EtcTree
}
for (const key in this.leases[id].keys)
{
this._delete_range({ key: this.use_base64 ? this.b64(key) : key }, next_revision, notifications);
this._delete_range({ key }, next_revision, notifications);
}
delete this.leases[id];
}
@ -597,9 +582,9 @@ class EtcTree
{
const ev = {
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),
kv: cur.value == null ? { key: (prefix === null ? '' : prefix) } : {
key: prefix,
value: cur.value,
mod_revision: cur.mod_revision,
},
};
@ -727,9 +712,9 @@ class EtcTree
_put(request_put, cur_revision, notifications)
{
// FIXME: prev_kv, ignore_value(?), ignore_lease(?)
const parts = this._key_parts(this.de64(request_put.key));
const parts = this._key_parts(request_put.key);
const key = parts.join('/');
const value = this.de64(request_put.value);
const value = request_put.value;
const { cur, watchers } = this._get_subtree(parts, true, true);
if (cur.key_watchers)
{
@ -762,7 +747,7 @@ class EtcTree
cur.create_revision = cur_revision;
}
cur.value = value;
const notify = { watchers, key: this.b64(key), value: this.b64(value), mod_revision: cur.mod_revision };
const notify = { watchers, key, value, mod_revision: cur.mod_revision };
if (cur.lease)
{
notify.lease = cur.lease;
@ -793,10 +778,10 @@ class EtcTree
}
if (cur.value != null)
{
const item = { key: this.b64(prefix === null ? '' : prefix) };
const item = { key: (prefix === null ? '' : prefix) };
if (!req.keys_only)
{
item.value = this.b64(cur.value);
item.value = cur.value;
item.mod_revision = cur.mod_revision;
//item.create_revision = cur.create_revision;
//item.version = cur.version;
@ -833,7 +818,7 @@ class EtcTree
this.mod_revision = cur_revision;
notifications.push({
watchers: cur.key_watchers ? [ ...watchers, ...cur.key_watchers ] : watchers,
key: this.b64(prefix === null ? '' : prefix),
key: (prefix === null ? '' : prefix),
mod_revision: cur_revision,
});
}