Compare commits
7 Commits
1734b422cb
...
84533dc9ef
Author | SHA1 | Date |
---|---|---|
Vitaliy Filippov | 84533dc9ef | |
Vitaliy Filippov | 5f645c5e44 | |
Vitaliy Filippov | 0c75cd1d63 | |
Vitaliy Filippov | 2e89aa8b17 | |
Vitaliy Filippov | b4e0ebd600 | |
Vitaliy Filippov | 2b6cc135d3 | |
Vitaliy Filippov | 5aa81a5c45 |
42
anticli.js
42
anticli.js
|
@ -83,6 +83,7 @@ class AntiEtcdCli
|
|||
{
|
||||
await this.del(cmd.slice(1));
|
||||
}
|
||||
process.exit(0);
|
||||
}
|
||||
|
||||
async get(keys)
|
||||
|
@ -93,7 +94,7 @@ 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 res = await this.request('/v3/kv/txn', txn);
|
||||
for (const r of (res||{}).responses||[])
|
||||
for (const r of res.responses||[])
|
||||
{
|
||||
if (r.response_range)
|
||||
{
|
||||
|
@ -119,7 +120,7 @@ class AntiEtcdCli
|
|||
value = await fsp.readFile(0, { encoding: 'utf-8' });
|
||||
}
|
||||
const res = await this.request('/v3/kv/put', { key: b64(key), value: b64(value) });
|
||||
if (res && res.header)
|
||||
if (res.header)
|
||||
{
|
||||
console.log('OK');
|
||||
}
|
||||
|
@ -133,7 +134,7 @@ class AntiEtcdCli
|
|||
}
|
||||
const txn = { success: keys.map(key => ({ request_delete_range: this.options.prefix ? { key: b64(key+'/'), range_end: b64(key+'0') } : { key: b64(key) } })) };
|
||||
const res = await this.request('/v3/kv/txn', txn);
|
||||
for (const r of (res||{}).responses||[])
|
||||
for (const r of res.responses||[])
|
||||
{
|
||||
if (r.response_delete_range)
|
||||
{
|
||||
|
@ -147,16 +148,33 @@ class AntiEtcdCli
|
|||
for (const url of this.options.endpoints)
|
||||
{
|
||||
const cur_url = url.replace(/\/+$/, '')+path;
|
||||
try
|
||||
const res = await POST(cur_url, body, this.options.timeout||1000);
|
||||
let ok = false;
|
||||
if (res.json)
|
||||
{
|
||||
return (await POST(cur_url, body, this.options.timeout||1000)).json;
|
||||
}
|
||||
catch (e)
|
||||
if (res.json.error)
|
||||
{
|
||||
console.error(cur_url+': '+e.message);
|
||||
console.error(cur_url+': '+res.json.error);
|
||||
process.exit(1);
|
||||
}
|
||||
return res.json;
|
||||
}
|
||||
if (res.body)
|
||||
{
|
||||
console.error(cur_url+': '+res.body);
|
||||
}
|
||||
if (res.error)
|
||||
{
|
||||
console.error(cur_url+': '+res.error);
|
||||
if (!res.response || !res.response.statusCode)
|
||||
{
|
||||
// This URL is unavailable
|
||||
continue;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
break;
|
||||
}
|
||||
process.exit(1);
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -188,9 +206,9 @@ function POST(url, body, timeout)
|
|||
res.on('data', chunk => { res_body += chunk; });
|
||||
res.on('end', () =>
|
||||
{
|
||||
if (res.statusCode != 200)
|
||||
if (res.statusCode != 200 || !/application\/json/i.exec(res.headers['content-type']))
|
||||
{
|
||||
ok({ error: res_body, code: res.statusCode });
|
||||
ok({ response: res, body: res_body, code: res.statusCode });
|
||||
return;
|
||||
}
|
||||
try
|
||||
|
@ -200,7 +218,7 @@ function POST(url, body, timeout)
|
|||
}
|
||||
catch (e)
|
||||
{
|
||||
ok({ error: e, response: res, body: res_body });
|
||||
ok({ response: res, error: e, body: res_body });
|
||||
}
|
||||
});
|
||||
});
|
||||
|
|
|
@ -59,6 +59,7 @@ class AntiCluster
|
|||
this.antietcd.clients[client_id].raft_node_id = node_id;
|
||||
this.antietcd.clients[client_id].addr = socket._socket.remoteAddress+':'+socket._socket.remotePort;
|
||||
socket.send(JSON.stringify({ identify: { key: this.cfg.cluster_key, node_id: this.cfg.node_id } }));
|
||||
this.raft.start();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
|
32
antietcd.js
32
antietcd.js
|
@ -86,13 +86,29 @@ class AntiEtcd
|
|||
|
||||
async persistAndReplicate(msg)
|
||||
{
|
||||
if (this.persistence)
|
||||
{
|
||||
await this.persistence.persistChange(msg);
|
||||
}
|
||||
let res = [];
|
||||
if (this.cluster)
|
||||
{
|
||||
await this.cluster.replicateChange(msg);
|
||||
// We have to guarantee that replication is processed sequentially
|
||||
// So we have to send messages without first awaiting for anything!
|
||||
res.push(this.cluster.replicateChange(msg));
|
||||
}
|
||||
if (this.persistence)
|
||||
{
|
||||
res.push(this.persistence.persistChange(msg));
|
||||
}
|
||||
if (res.length)
|
||||
{
|
||||
res = await Promise.allSettled(res);
|
||||
const errors = res.filter(r => r.status == 'rejected');
|
||||
if (errors.length)
|
||||
{
|
||||
for (const e of errors)
|
||||
{
|
||||
console.error(e.reason);
|
||||
}
|
||||
throw errors[0].reason;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -106,6 +122,7 @@ class AntiEtcd
|
|||
data = Buffer.concat(data);
|
||||
let body = '';
|
||||
let code = 200;
|
||||
let ctype = 'text/plain; charset=utf-8';
|
||||
let reply;
|
||||
try
|
||||
{
|
||||
|
@ -128,6 +145,7 @@ class AntiEtcd
|
|||
}
|
||||
reply = await this.runHandler(req, data, res);
|
||||
reply = JSON.stringify(reply);
|
||||
ctype = 'application/json';
|
||||
}
|
||||
catch (e)
|
||||
{
|
||||
|
@ -153,8 +171,8 @@ class AntiEtcd
|
|||
'\n '+reply.replace(/\n/g, '\\n')
|
||||
);
|
||||
reply = Buffer.from(reply);
|
||||
res.writeHead(200, {
|
||||
'Content-Type': 'application/json',
|
||||
res.writeHead(code, {
|
||||
'Content-Type': ctype,
|
||||
'Content-Length': reply.length,
|
||||
});
|
||||
res.write(reply);
|
||||
|
|
|
@ -2,6 +2,7 @@ const fs = require('fs');
|
|||
const fsp = require('fs').promises;
|
||||
const zlib = require('zlib');
|
||||
|
||||
const stableStringify = require('./stable-stringify.js');
|
||||
const EtcTree = require('./etctree.js');
|
||||
const { de64, runCallbacks } = require('./common.js');
|
||||
|
||||
|
@ -52,10 +53,10 @@ class AntiPersistence
|
|||
let changed = false;
|
||||
for (const ev of msg.events)
|
||||
{
|
||||
if (ev.kv.lease)
|
||||
if (ev.lease)
|
||||
{
|
||||
// Values with lease are never persisted
|
||||
const key = de64(ev.kv.key);
|
||||
const key = de64(ev.key);
|
||||
if (this.prev_value[key] !== undefined)
|
||||
{
|
||||
delete this.prev_value[key];
|
||||
|
@ -64,15 +65,14 @@ class AntiPersistence
|
|||
}
|
||||
else
|
||||
{
|
||||
const key = de64(ev.kv.key);
|
||||
const filtered = this.cfg.persist_filter(key, ev.type === 'DELETE' ? undefined : de64(ev.kv.value));
|
||||
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]))
|
||||
{
|
||||
this.prev_value[key] = filtered;
|
||||
changed = true;
|
||||
}
|
||||
}
|
||||
changed = true;
|
||||
}
|
||||
if (!changed)
|
||||
{
|
||||
|
@ -114,7 +114,7 @@ class AntiPersistence
|
|||
{
|
||||
let dump = this.antietcd.etctree.dump(true);
|
||||
dump['term'] = this.antietcd.stored_term;
|
||||
dump = JSON.stringify(dump);
|
||||
dump = stableStringify(dump);
|
||||
dump = await new Promise((ok, no) => zlib.gzip(dump, (err, res) => err ? no(err) : ok(res)));
|
||||
const fh = await fsp.open(this.cfg.data+'.tmp', 'w');
|
||||
await fh.writeFile(dump);
|
||||
|
@ -124,7 +124,7 @@ class AntiPersistence
|
|||
}
|
||||
catch (e)
|
||||
{
|
||||
console.error(e);
|
||||
console.error('Error persisting data to disk: '+e);
|
||||
process.exit(1);
|
||||
}
|
||||
runCallbacks(this, 'wait_persist', null);
|
||||
|
|
21
etctree.js
21
etctree.js
|
@ -232,7 +232,7 @@ class EtcTree
|
|||
if (!snapshot.leases[id])
|
||||
{
|
||||
// Revoke without replicating and notifying
|
||||
this._sync_revoke_lease(id, notifications);
|
||||
this._sync_revoke_lease(id, notifications, this.mod_revision);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -269,10 +269,10 @@ 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, value: cur_old.value, mod_revision: cur_old.mod_revision };
|
||||
if (cur_old.lease)
|
||||
const notify = { watchers: key_watchers, key, value: cur_new.value, mod_revision: cur_new.mod_revision };
|
||||
if (cur_new.lease)
|
||||
{
|
||||
notify.lease = cur_old.lease;
|
||||
notify.lease = cur_new.lease;
|
||||
}
|
||||
notifications.push(notify);
|
||||
}
|
||||
|
@ -366,6 +366,7 @@ class EtcTree
|
|||
}
|
||||
const expires = Date.now() + req.TTL*1000;
|
||||
this.leases[id] = { ttl: req.TTL, expires, timer_id: null, keys: {} };
|
||||
this.mod_revision++;
|
||||
this._set_expire(id);
|
||||
if (this.replicate)
|
||||
{
|
||||
|
@ -389,6 +390,7 @@ class EtcTree
|
|||
}
|
||||
const ttl = this.leases[id].ttl;
|
||||
lease.expires = Date.now() + ttl*1000;
|
||||
this.mod_revision++;
|
||||
this._set_expire(id);
|
||||
if (this.replicate)
|
||||
{
|
||||
|
@ -424,13 +426,12 @@ class EtcTree
|
|||
return true;
|
||||
}
|
||||
|
||||
_sync_revoke_lease(id, notifications)
|
||||
_sync_revoke_lease(id, notifications, next_revision)
|
||||
{
|
||||
if (!this.leases[id])
|
||||
{
|
||||
throw new Error('unknown lease');
|
||||
}
|
||||
const next_revision = this.mod_revision + 1;
|
||||
for (const key in this.leases[id].keys)
|
||||
{
|
||||
this._delete_range({ key }, next_revision, notifications);
|
||||
|
@ -447,7 +448,8 @@ class EtcTree
|
|||
return null;
|
||||
throw new Error('unknown lease');
|
||||
}
|
||||
this._sync_revoke_lease(req.ID, notifications);
|
||||
this.mod_revision++;
|
||||
this._sync_revoke_lease(req.ID, notifications, this.mod_revision);
|
||||
if (this.replicate)
|
||||
{
|
||||
await this.notify_replicator(notifications, [ { id: req.ID } ]);
|
||||
|
@ -470,6 +472,7 @@ class EtcTree
|
|||
|
||||
async apply_replication(msg)
|
||||
{
|
||||
this.mod_revision = msg.header.revision;
|
||||
const notifications = [];
|
||||
if ((msg.leases||[]).length)
|
||||
{
|
||||
|
@ -481,7 +484,7 @@ class EtcTree
|
|||
}
|
||||
else
|
||||
{
|
||||
this._sync_revoke_lease(lease.id, notifications);
|
||||
this._sync_revoke_lease(lease.id, notifications, this.mod_revision);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -495,7 +498,7 @@ class EtcTree
|
|||
}
|
||||
else
|
||||
{
|
||||
this._put({ key: ev.key, value: ev.value }, ev.mod_revision, notifications);
|
||||
this._put({ key: ev.key, value: ev.value, lease: ev.lease }, ev.mod_revision, notifications);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
Loading…
Reference in New Issue