Compare commits

...

4 Commits

5 changed files with 60 additions and 18 deletions

View File

@ -332,9 +332,9 @@ class AntiCluster
{
if (path == 'kv_txn')
{
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.failure || !data.failure.filter(f => f.request_put || f.requestPut || f.request_delete_range || f.requestDeleteRange).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.failure && data.failure.filter(f => f.request_put || f.requestPut || f.request_delete_range || f.requestDeleteRange).length);
}
return path != 'kv_range';
}

View File

@ -12,10 +12,10 @@ License: Mozilla Public License 2.0 or Vitastor Network Public License 1.1
Usage:
${process.argv[0]} ${process.argv[1]} \
[--cert ssl.crt] [--key ssl.key] [--port 12379] \
[--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] \
${process.argv[0]} ${process.argv[1]} \n\
[--cert ssl.crt] [--key ssl.key] [--port 12379] \n\
[--data data.gz] [--persist-filter ./filter.js] [--persist_interval 500] \n\
[--node_id node1 --cluster_key abcdef --cluster node1=http://localhost:12379,node2=http://localhost:12380,node3=http://localhost:12381] \n\
[other options]
Supported etcd REST APIs:
@ -43,6 +43,10 @@ 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
--no_merge_watch off
Antietcd merges all watcher events into a single websocket message to provide
more ordering guarantees. Specify 1/on/true/yes for this option to disable this
behaviour.
Persistence:

View File

@ -31,6 +31,10 @@ class AntiEtcd extends EventEmitter
this.cluster = null;
this.stored_term = 0;
this.cfg = cfg;
this.cfg.no_merge_watch = this.cfg.no_merge_watch == 1 ||
this.cfg.no_merge_watch == 'on' ||
this.cfg.no_merge_watch == 'yes' ||
this.cfg.no_merge_watch == 'true';
this.loading = false;
this.stopped = false;
this.inflight = 0;
@ -120,7 +124,7 @@ class AntiEtcd extends EventEmitter
let done = 0;
await new Promise((allOk, allNo) =>
{
res.map(promise => promise.then(res =>
res.map(promise => promise.then(r =>
{
if ((++done) == res.length)
allOk();
@ -411,9 +415,9 @@ class AntiEtcd extends EventEmitter
}
// public watch API
async create_watch(params, callback)
async create_watch(params, callback, stream_id)
{
const watch = this.etctree.api_create_watch({ ...params, watch_id: null }, callback);
const watch = this.etctree.api_create_watch({ ...params, watch_id: null }, callback, stream_id);
if (!watch.created)
{
throw new RequestError(400, 'Requested watch revision is compacted', { compact_revision: watch.compact_revision });
@ -478,7 +482,7 @@ class AntiEtcd extends EventEmitter
return this.etctree.api_keepalive_lease(data);
}
_handle_maintenance_status(data)
_handle_maintenance_status(/*data*/)
{
const raft = this.cluster && this.cluster.raft;
return {
@ -516,8 +520,9 @@ class AntiEtcd extends EventEmitter
const create_request = msg.create_request;
if (!create_request.watch_id || !client.watches[create_request.watch_id])
{
client.send_cb = client.send_cb || (msg => socket.send(JSON.stringify(msg)));
const watch = this.etctree.api_create_watch(
{ ...create_request, watch_id: null }, (msg) => socket.send(JSON.stringify(msg))
{ ...create_request, watch_id: null }, client.send_cb, (this.cfg.no_merge_watch ? null : 'C'+client_id)
);
if (!watch.created)
{

View File

@ -544,7 +544,7 @@ class EtcTree
}
}
api_create_watch(req, send)
api_create_watch(req, send, stream_id)
{
const { parts, all } = this._get_range(req);
if (req.start_revision && this.compact_revision && this.compact_revision > req.start_revision)
@ -566,6 +566,7 @@ class EtcTree
this.watchers[watch_id] = {
paths: [],
send,
stream_id,
};
}
this.watchers[watch_id].paths.push(parts);
@ -667,15 +668,15 @@ class EtcTree
{
if (this.watchers[wid])
{
by_watcher[wid] = by_watcher[wid] || { header: { revision: this.mod_revision }, events: {} };
by_watcher[wid].events[notif.key] = conv;
const stream_id = this.watchers[wid].stream_id || wid;
by_watcher[stream_id] = by_watcher[stream_id] || { send: this.watchers[wid].send, events: {} };
by_watcher[stream_id].events[notif.key] = conv;
}
}
}
for (const wid in by_watcher)
for (const stream_id in by_watcher)
{
by_watcher[wid].events = Object.values(by_watcher[wid].events);
this.watchers[wid].send({ result: by_watcher[wid] });
by_watcher[stream_id].send({ result: { header: { revision: this.mod_revision }, events: Object.values(by_watcher[stream_id].events) } });
}
}

View File

@ -91,6 +91,38 @@ tests['watch'] = async () =>
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 () =>
{
const t = new EtcTree();