Move base64 handling into antietcd.js

README.md

@@ -27,6 +27,7 @@ Simplistic miniature etcd replacement based on [TinyRaft](https://git.yourcmc.ru
   - [/v3/lease/grant](#v3-lease-grant)
   - [/v3/lease/keepalive](#v3-lease-keepalive)
   - [/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)
 - [HTTP Error Codes](#http-error-codes)
 
@@ -453,6 +454,33 @@ 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
 
 Client-to-server message format:

anticluster.js

@@ -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';
     }

antietcd-app.js

@@ -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,9 @@ 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
+--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).
 
@@ -107,14 +110,6 @@ function parse()
             options[arg.substr(2)] = process.argv[++i];
         }
     }
-    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);

antietcd.js

@@ -17,19 +17,16 @@ const AntiPersistence = require('./antipersistence.js');
 const AntiCluster = require('./anticluster.js');
 const { runCallbacks, de64, b64, RequestError } = require('./common.js');
 
+const VERSION = '1.0.9';
+
 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;
-        }
+        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.client_id = 1;
         this.etctree = new EtcTree(true);
@@ -80,7 +77,7 @@ class AntiEtcd extends EventEmitter
         {
             this.server = http.createServer((req, res) => this._handleRequest(req, res));
         }
-        this.wss = new ws.WebSocketServer({ server: this.server });
+        this.wss = new ws.Server({ server: this.server });
         // eslint-disable-next-line no-unused-vars
         this.wss.on('connection', (conn, req) => this._startWebsocket(conn, null));
         this.server.listen(this.cfg.port || 2379, this.cfg.ip || undefined);
@@ -126,7 +123,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();
@@ -189,13 +186,13 @@ class AntiEtcd extends EventEmitter
                 if (e instanceof RequestError)
                 {
                     code = e.code;
-                    reply = e.message;
+                    reply = e.message+'\n';
                 }
                 else
                 {
                     console.error(e);
                     code = 500;
-                    reply = 'Internal error: '+e.message;
+                    reply = 'Internal error: '+e.message+'\n';
                 }
             }
             try
@@ -337,7 +334,7 @@ class AntiEtcd extends EventEmitter
             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, '+
-                    '/v3/lease/grant, /v3/lease/revoke, /v3/kv/lease/revoke, /v3/lease/keepalive');
+                    '/v3/lease/grant, /v3/lease/revoke, /v3/kv/lease/revoke, /v3/lease/keepalive, /v3/maintenance/status');
             }
             else
             {
@@ -353,7 +350,7 @@ class AntiEtcd extends EventEmitter
         {
             throw new RequestError(502, 'Server is stopping');
         }
-        if (path !== 'dump' && this.cluster)
+        if (this.cluster && path !== 'dump' && path != 'maintenance_status')
         {
             const res = await this.cluster.checkRaftState(
                 path,
@@ -417,9 +414,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(this._encodeWatch(params), (msg) => callback(this._encodeMsg(msg)));
+        const watch = this.etctree.api_create_watch(this._encodeWatch(params), (msg) => callback(this._encodeMsg(msg)), stream_id);
         if (!watch.created)
         {
             throw new RequestError(400, 'Requested watch revision is compacted', { compact_revision: watch.compact_revision });
@@ -436,7 +433,7 @@ class AntiEtcd extends EventEmitter
         {
             throw new RequestError(400, 'Watch not found');
         }
-        this.etctree.api_cancel_watch({ watch_id: mapped_id });
+        this.etctree.api_cancel_watch(mapped_id);
         delete this.api_watches[watch_id];
     }
 
@@ -524,6 +521,26 @@ class AntiEtcd extends EventEmitter
         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
     _handle_dump(data)
     {
@@ -542,7 +559,10 @@ 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(this._encodeWatch(create_request), (msg) => socket.send(JSON.stringify(this._encodeMsg(msg))));
+                client.send_cb = client.send_cb || (msg => socket.send(JSON.stringify(this._encodeMsg(msg))));
+                const watch = this.etctree.api_create_watch(
+                    this._encodeWatch(create_request), client.send_cb, (this.cfg.merge_watches ? 'C'+client_id : null)
+                );
                 if (!watch.created)
                 {
                     socket.send(JSON.stringify({ result: { header: { revision: this.etctree.mod_revision }, watch_id: create_request.watch_id, ...watch } }));
@@ -560,7 +580,7 @@ class AntiEtcd extends EventEmitter
             const mapped_id = client.watches[msg.cancel_request.watch_id];
             if (mapped_id)
             {
-                this.etctree.api_cancel_watch({ watch_id: mapped_id });
+                this.etctree.api_cancel_watch(mapped_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 } }));
             }
@@ -617,11 +637,18 @@ class AntiEtcd extends EventEmitter
         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 });
+            this.etctree.api_cancel_watch(mapped_id);
         }
     }
 }
 
+function is_true(s)
+{
+    return s === true || s === 1 || s === '1' || s === 'yes' || s === 'true' || s === 'on';
+}
+
 AntiEtcd.RequestError = RequestError;
 
+AntiEtcd.VERSION = VERSION;
+
 module.exports = AntiEtcd;

etctree.js

@@ -264,7 +264,7 @@ class EtcTree
             {
                 cur_old.value = cur_new.value;
                 const key_watchers = (cur_old.key_watchers ? [ ...watchers, ...(cur_old.key_watchers||[]) ] : watchers);
-                const notify = { watchers: key_watchers, key: this.b64(key), value: this.b64(cur_new.value), mod_revision: cur_new.mod_revision };
+                const notify = { watchers: key_watchers, key, value: cur_new.value, mod_revision: cur_new.mod_revision };
                 if (cur_new.lease)
                 {
                     notify.lease = cur_new.lease;
@@ -431,6 +431,11 @@ class EtcTree
         {
             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];
     }
 
@@ -524,7 +529,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)
@@ -546,6 +551,7 @@ class EtcTree
             this.watchers[watch_id] = {
                 paths: [],
                 send,
+                stream_id,
             };
         }
         this.watchers[watch_id].paths.push(parts);
@@ -647,15 +653,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) } });
         }
     }
 

etctree.spec.js

@@ -64,7 +64,7 @@ tests['read/write'] = async () =>
     );
     expect(
         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,"leases":{}}
+        {"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":{}}
     );
     t.destroy();
 };
@@ -80,7 +80,7 @@ tests['watch'] = async () =>
     );
     expect(
         t.api_create_watch({ watch_id: 1, key: '/vitastor/', range_end: '/vitastor0' }, send),
-        { watch_id: 1, created: true }
+        { header: { revision: 1 }, watch_id: 1, created: true }
     );
     expect(sent, []);
     expect(
@@ -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();
@@ -100,18 +132,18 @@ tests['lease'] = async () =>
     expect(leaseID != null, true);
     expect(
         await 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: {} } ] }
+        { header: { revision: 2 }, succeeded: true, responses: [ { response_put: {} } ] }
     );
     expect(
         t.api_create_watch({ watch_id: 1, key: '/vitastor/', range_end: '/vitastor0' }, send),
-        { watch_id: 1, created: true }
+        { header: { revision: 2 }, watch_id: 1, created: true }
     );
     expect(sent, []);
     const dump = t.dump(false);
     const expires = dump.leases[leaseID].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}}});
+    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}}});
     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 } } ] } } ]);
+    expect(sent, [ { result: { header: { revision: 3 }, events: [ { type: 'DELETE', kv: { key: '/vitastor/osd/state/1', mod_revision: 3 } } ] } } ]);
     t.pause_leases();
     t.load(dump);
     expect(t.dump(false), dump);
@@ -131,7 +163,7 @@ tests['update'] = async () =>
     expect(leaseID != null, true);
     expect(
         await t1.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: {} } ] }
+        { header: { revision: 2 }, succeeded: true, responses: [ { response_put: {} } ] }
     );
     expect(
         await t2.api_txn({ success: [ { request_put: { key: '/vitastor/osd/state/1', value: { ip: '1.2.3.6' } } } ] }),
@@ -139,15 +171,15 @@ tests['update'] = async () =>
     );
     expect(
         await t1.api_txn({ success: [ { request_put: { key: '/vitastor/osd/state/1', lease: leaseID, value: { ip: '1.2.3.5' } } } ] }),
-        { header: { revision: 2 }, succeeded: true, responses: [ { response_put: {} } ] }
+        { header: { revision: 3 }, succeeded: true, responses: [ { response_put: {} } ] }
     );
     let dump2 = t2.dump();
     t2.load(t1.dump(), true);
     t1.load(dump2, true);
     let dump = t2.dump(false);
     let expires = dump.leases[leaseID].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":2,"create_revision":1,"value":{"ip":"1.2.3.5"}}}}}}}}}}}},"mod_revision":2,"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":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":3,"create_revision":2,"value":{"ip":"1.2.3.5"}}}}}}}}}}}},"mod_revision":3,"compact_revision":0,"leases":{[leaseID]:{"ttl":0.5,"expires":expires}}});
     t1.destroy();
     t2.destroy();
 };

package.json

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