From 500a07025ec31c5a00ff419df2de93fbf1b24a3d Mon Sep 17 00:00:00 2001 From: Vitaliy Filippov Date: Tue, 30 Apr 2024 01:51:50 +0300 Subject: [PATCH] Add Anti-Etcd - etcd mock, already sufficient to run Vitastor tests --- antietcd.js | 290 ++++++++++++++++++++++++++++++++++++++++++++ etctree.js | 149 ++++++++++++++++------- etctree.spec.js | 148 ++++++++++++++-------- stable-stringify.js | 78 ++++++++++++ 4 files changed, 573 insertions(+), 92 deletions(-) create mode 100644 antietcd.js create mode 100644 stable-stringify.js diff --git a/antietcd.js b/antietcd.js new file mode 100644 index 0000000..2569fc6 --- /dev/null +++ b/antietcd.js @@ -0,0 +1,290 @@ +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); + } + } + // Only 5 APIs supported + 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(); diff --git a/etctree.js b/etctree.js index ea141a0..b0cd740 100644 --- a/etctree.js +++ b/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; diff --git a/etctree.spec.js b/etctree.spec.js index 659440d..4312434 100644 --- a/etctree.spec.js +++ b/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'); +} diff --git a/stable-stringify.js b/stable-stringify.js new file mode 100644 index 0000000..ea91fcd --- /dev/null +++ b/stable-stringify.js @@ -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;