Implement simple persistence
parent
e8b600f536
commit
3596ecd92c
261
antietcd.js
261
antietcd.js
|
|
@ -1,4 +1,6 @@
|
|||
const fs = require('fs');
|
||||
#!/usr/bin/node
|
||||
|
||||
const fsp = require('fs').promises;
|
||||
const { URL } = require('url');
|
||||
const http = require('http');
|
||||
const https = require('https');
|
||||
|
|
@ -6,53 +8,83 @@ const https = require('https');
|
|||
const ws = require('ws');
|
||||
|
||||
const EtcTree = require('./etctree.js');
|
||||
|
||||
class RequestError
|
||||
{
|
||||
constructor(code, text)
|
||||
{
|
||||
this.code = code;
|
||||
this.message = text;
|
||||
}
|
||||
}
|
||||
const AntiPersistence = require('./antipersistence.js');
|
||||
const { runCallbacks, RequestError } = require('./common.js');
|
||||
|
||||
class AntiEtcd
|
||||
{
|
||||
constructor(cfg)
|
||||
{
|
||||
this.cfg = cfg;
|
||||
}
|
||||
|
||||
run()
|
||||
{
|
||||
this.clients = {};
|
||||
this.client_id = 1;
|
||||
this.etctree = new EtcTree(true);
|
||||
this.persistence = null;
|
||||
this.stored_term = 0;
|
||||
this.cfg = cfg;
|
||||
this.loading = false;
|
||||
this.stopped = false;
|
||||
this.inflight = 0;
|
||||
this.wait_inflight = [];
|
||||
}
|
||||
|
||||
async run()
|
||||
{
|
||||
if (this.cfg.data)
|
||||
{
|
||||
this.etctree.set_replicate_watcher(msg => this.persistence.persistChange(msg));
|
||||
}
|
||||
if (this.cfg.data)
|
||||
{
|
||||
this.persistence = new AntiPersistence(this);
|
||||
// Load data from disk
|
||||
await this.persistence.load();
|
||||
// Set exit hook
|
||||
const on_stop_cb = () => this.onStop();
|
||||
process.on('SIGINT', on_stop_cb);
|
||||
process.on('SIGTERM', on_stop_cb);
|
||||
process.on('SIGQUIT', on_stop_cb);
|
||||
}
|
||||
if (this.cfg.cert)
|
||||
{
|
||||
this.tls = { key: fs.readFileSync(this.cfg.key), cert: fs.readFileSync(this.cfg.cert) };
|
||||
this.tls = { key: await fsp.readFile(this.cfg.key), cert: await fsp.readFile(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));
|
||||
// eslint-disable-next-line no-unused-vars
|
||||
this.wss.on('connection', (conn, req) => this.startWebsocket(conn, null));
|
||||
this.server.listen(this.cfg.port || 2379);
|
||||
}
|
||||
|
||||
fail(res, code, text)
|
||||
async onStop()
|
||||
{
|
||||
res.writeHead(code);
|
||||
res.write(text);
|
||||
res.end();
|
||||
if (this.stopped)
|
||||
{
|
||||
return;
|
||||
}
|
||||
this.stopped = true;
|
||||
// Wait until all requests complete
|
||||
while (this.inflight > 0)
|
||||
{
|
||||
await new Promise(ok => this.wait_inflight.push(ok));
|
||||
}
|
||||
if (this.persistence)
|
||||
{
|
||||
await this.persistence.persist();
|
||||
}
|
||||
process.exit(0);
|
||||
}
|
||||
|
||||
handleRequest(req, res)
|
||||
{
|
||||
let data = [];
|
||||
req.on('data', (chunk) => data.push(chunk));
|
||||
req.on('end', () =>
|
||||
req.on('end', async () =>
|
||||
{
|
||||
this.inflight++;
|
||||
data = Buffer.concat(data);
|
||||
let body = '';
|
||||
let code = 200;
|
||||
|
|
@ -76,7 +108,7 @@ class AntiEtcd
|
|||
{
|
||||
throw new RequestError(400, 'body should be JSON object');
|
||||
}
|
||||
reply = this.runHandler(req, data, res);
|
||||
reply = await this.runHandler(req, data, res);
|
||||
reply = JSON.stringify(reply);
|
||||
}
|
||||
catch (e)
|
||||
|
|
@ -93,34 +125,47 @@ class AntiEtcd
|
|||
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();
|
||||
try
|
||||
{
|
||||
// Access log
|
||||
console.log(
|
||||
new Date().toISOString()+
|
||||
' '+(req.headers['x-forwarded-for'] || (req.socket.remoteAddress + ':' + req.socket.remotePort))+
|
||||
' '+req.method+' '+req.url+' '+code+'\n '+body.replace(/\n/g, '\\n')+
|
||||
'\n '+reply.replace(/\n/g, '\\n')
|
||||
);
|
||||
reply = Buffer.from(reply);
|
||||
res.writeHead(200, {
|
||||
'Content-Type': 'application/json',
|
||||
'Content-Length': reply.length,
|
||||
});
|
||||
res.write(reply);
|
||||
res.end();
|
||||
}
|
||||
catch (e)
|
||||
{
|
||||
console.error(e);
|
||||
}
|
||||
this.inflight--;
|
||||
if (!this.inflight)
|
||||
{
|
||||
runCallbacks(this, 'wait_inflight', []);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
startWebsocket(socket, req)
|
||||
startWebsocket(socket, reconnect)
|
||||
{
|
||||
const client_id = this.client_id++;
|
||||
this.clients[client_id] = {
|
||||
id: client_id,
|
||||
addr: socket._socket ? socket._socket.remoteAddress+':'+socket._socket.remotePort : '',
|
||||
socket,
|
||||
alive: true,
|
||||
watches: {},
|
||||
};
|
||||
socket.on('pong', () => this.clients[client_id].alive = true);
|
||||
socket.on('error', console.error);
|
||||
socket.on('error', e => console.error(e.syscall === 'connect' ? e.message : e));
|
||||
const pinger = setInterval(() =>
|
||||
{
|
||||
if (!this.clients[client_id])
|
||||
|
|
@ -160,10 +205,16 @@ class AntiEtcd
|
|||
clearInterval(pinger);
|
||||
delete this.clients[client_id];
|
||||
socket.terminate();
|
||||
if (reconnect)
|
||||
{
|
||||
reconnect();
|
||||
}
|
||||
});
|
||||
return client_id;
|
||||
}
|
||||
|
||||
runHandler(req, data, res)
|
||||
// eslint-disable-next-line no-unused-vars
|
||||
async runHandler(req, data, res)
|
||||
{
|
||||
// v3/kv/txn
|
||||
// v3/kv/range
|
||||
|
|
@ -173,6 +224,10 @@ class AntiEtcd
|
|||
// v3/lease/keepalive
|
||||
// v3/lease/revoke O_o
|
||||
// v3/kv/lease/revoke O_o
|
||||
if (this.stopped)
|
||||
{
|
||||
throw new RequestError(502, 'Server is stopping');
|
||||
}
|
||||
const requestUrl = new URL(req.url, 'http://'+(req.headers.host || 'localhost'));
|
||||
if (requestUrl.pathname.substr(0, 4) == '/v3/')
|
||||
{
|
||||
|
|
@ -184,33 +239,42 @@ class AntiEtcd
|
|||
{
|
||||
throw new RequestError(405, 'Please use POST method');
|
||||
}
|
||||
return cb.call(this, data);
|
||||
const res = cb.call(this, data);
|
||||
if (res instanceof Promise)
|
||||
{
|
||||
return await res;
|
||||
}
|
||||
return res;
|
||||
}
|
||||
}
|
||||
else if (requestUrl.pathname == '/dump')
|
||||
{
|
||||
return this.handle_dump(data);
|
||||
}
|
||||
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)
|
||||
async handle_kv_txn(data)
|
||||
{
|
||||
return this.etctree.api_txn(data);
|
||||
return await this.etctree.api_txn(data);
|
||||
}
|
||||
|
||||
handle_kv_range(data)
|
||||
async handle_kv_range(data)
|
||||
{
|
||||
const r = this.etctree.api_txn({ success: [ { request_range: data } ] });
|
||||
const r = await this.etctree.api_txn({ success: [ { request_range: data } ] });
|
||||
return { header: r.header, ...r.responses[0].response_range };
|
||||
}
|
||||
|
||||
handle_kv_put(data)
|
||||
async handle_kv_put(data)
|
||||
{
|
||||
const r = this.etctree.api_txn({ success: [ { request_put: data } ] });
|
||||
const r = await this.etctree.api_txn({ success: [ { request_put: data } ] });
|
||||
return { header: r.header, ...r.responses[0].response_put };
|
||||
}
|
||||
|
||||
handle_kv_deleterange(data)
|
||||
async handle_kv_deleterange(data)
|
||||
{
|
||||
const r = this.etctree.api_txn({ success: [ { request_delete_range: data } ] });
|
||||
const r = await this.etctree.api_txn({ success: [ { request_delete_range: data } ] });
|
||||
return { header: r.header, ...r.responses[0].response_delete_range };
|
||||
}
|
||||
|
||||
|
|
@ -234,13 +298,20 @@ class AntiEtcd
|
|||
return this.etctree.api_keepalive_lease(data);
|
||||
}
|
||||
|
||||
// eslint-disable-next-line no-unused-vars
|
||||
handle_dump(data)
|
||||
{
|
||||
return { ...this.etctree.dump(), term: this.stored_term };
|
||||
}
|
||||
|
||||
handleMessage(client_id, msg, socket)
|
||||
{
|
||||
const client = this.clients[client_id];
|
||||
console.log(new Date().toISOString()+' '+client.addr+' '+(client.raft_node_id || '-')+' -> '+JSON.stringify(msg));
|
||||
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])
|
||||
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))
|
||||
|
|
@ -251,19 +322,19 @@ class AntiEtcd
|
|||
}
|
||||
else
|
||||
{
|
||||
create_request.watch_id ||= watch.watch_id;
|
||||
this.clients[client_id].watches[create_request.watch_id] = watch.watch_id;
|
||||
create_request.watch_id = create_request.watch_id || watch.watch_id;
|
||||
client.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];
|
||||
const mapped_id = client.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];
|
||||
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 } }));
|
||||
}
|
||||
}
|
||||
|
|
@ -287,4 +358,80 @@ class AntiEtcd
|
|||
}
|
||||
}
|
||||
|
||||
new AntiEtcd({ port: 12379 }).run();
|
||||
function vitastor_persist_filter(prefix)
|
||||
{
|
||||
return (key, value) =>
|
||||
{
|
||||
if (key.substr(0, prefix.length+'/osd/stats/'.length) == prefix+'/osd/stats/')
|
||||
{
|
||||
if (value)
|
||||
{
|
||||
try
|
||||
{
|
||||
value = JSON.parse(value);
|
||||
value = JSON.stringify({
|
||||
bitmap_granularity: value.bitmap_granularity || undefined,
|
||||
data_block_size: value.data_block_size || undefined,
|
||||
host: value.host || undefined,
|
||||
immediate_commit: value.immediate_commit || undefined,
|
||||
});
|
||||
}
|
||||
catch (e)
|
||||
{
|
||||
console.error('invalid JSON in '+key+' = '+value+': '+e);
|
||||
value = {};
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
value = undefined;
|
||||
}
|
||||
return value;
|
||||
}
|
||||
else if (key.substr(0, prefix.length+'/osd/'.length) == prefix+'/osd/' ||
|
||||
key.substr(0, prefix.length+'/inode/stats/'.length) == prefix+'/inode/stats/' ||
|
||||
key.substr(0, prefix.length+'/pg/stats/'.length) == prefix+'/pg/stats/' ||
|
||||
key.substr(0, prefix.length+'/pool/stats/'.length) == prefix+'/pool/stats/' ||
|
||||
key == prefix+'/stats')
|
||||
{
|
||||
return undefined;
|
||||
}
|
||||
return value;
|
||||
};
|
||||
}
|
||||
|
||||
function parse()
|
||||
{
|
||||
const options = {
|
||||
persist_filter: vitastor_persist_filter('/vitastor'),
|
||||
};
|
||||
for (let i = 2; i < process.argv.length; i++)
|
||||
{
|
||||
const arg = process.argv[i].toLowerCase().replace(/^--(.+)$/, (m, m1) => '--'+m1.replace(/-/g, '_'));
|
||||
if (arg === '-h' || arg === '--help')
|
||||
{
|
||||
console.error(
|
||||
'USAGE:\n '+process.argv[0]+' '+process.argv[1]+' [OPTIONS]\n'+
|
||||
'OPTIONS:\n'+
|
||||
' [--cert ssl.crt] [--key ssl.key] [--port 12379]\n'+
|
||||
' [--data data.gz] [--vitastor-persist-filter /vitastor] [--no-persist-filter] [--persist_interval 500]\n'
|
||||
);
|
||||
process.exit();
|
||||
}
|
||||
else if (arg == '--no_persist_filter')
|
||||
{
|
||||
options['persist_filter'] = null;
|
||||
}
|
||||
else if (arg == '--vitastor_persist_filter')
|
||||
{
|
||||
options['persist_filter'] = vitastor_persist_filter(process.argv[++i]||'');
|
||||
}
|
||||
else if (arg.substr(0, 2) == '--' && arg != '--persist_filter')
|
||||
{
|
||||
options[arg.substr(2)] = process.argv[++i];
|
||||
}
|
||||
}
|
||||
return options;
|
||||
}
|
||||
|
||||
new AntiEtcd(parse()).run().catch(console.error);
|
||||
|
|
|
|||
|
|
@ -0,0 +1,134 @@
|
|||
const fs = require('fs');
|
||||
const fsp = require('fs').promises;
|
||||
const zlib = require('zlib');
|
||||
|
||||
const EtcTree = require('./etctree.js');
|
||||
const { de64, runCallbacks } = require('./common.js');
|
||||
|
||||
class AntiPersistence
|
||||
{
|
||||
constructor(antietcd)
|
||||
{
|
||||
this.cfg = antietcd.cfg;
|
||||
this.antietcd = antietcd;
|
||||
this.prev_value = {};
|
||||
this.persist_timer = null;
|
||||
this.wait_persist = null;
|
||||
}
|
||||
|
||||
async load()
|
||||
{
|
||||
// eslint-disable-next-line no-unused-vars
|
||||
const [ err, stat ] = await new Promise(ok => fs.stat(this.cfg.data, (err, stat) => ok([ err, stat ])));
|
||||
if (!err)
|
||||
{
|
||||
let data = await fsp.readFile(this.cfg.data);
|
||||
data = await new Promise((ok, no) => zlib.gunzip(data, (err, res) => err ? no(err) : ok(res)));
|
||||
data = JSON.parse(data);
|
||||
this.loading = true;
|
||||
this.antietcd.etctree.load(data);
|
||||
this.loading = false;
|
||||
this.antietcd.stored_term = data['term'] || 0;
|
||||
}
|
||||
else if (err.code != 'ENOENT')
|
||||
{
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
async persistChange(msg)
|
||||
{
|
||||
if (this.loading)
|
||||
{
|
||||
return;
|
||||
}
|
||||
if (!msg.events || !msg.events.length)
|
||||
{
|
||||
// lease-only changes don't need to be persisted
|
||||
return;
|
||||
}
|
||||
if (this.cfg.persist_filter)
|
||||
{
|
||||
let changed = false;
|
||||
for (const ev of msg.events)
|
||||
{
|
||||
if (ev.kv.lease)
|
||||
{
|
||||
// Values with lease are never persisted
|
||||
const key = de64(ev.kv.key);
|
||||
if (this.prev_value[key] !== undefined)
|
||||
{
|
||||
delete this.prev_value[key];
|
||||
changed = true;
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
const key = de64(ev.kv.key);
|
||||
const filtered = this.cfg.persist_filter(key, ev.type === 'DELETE' ? undefined : de64(ev.kv.value));
|
||||
if (!EtcTree.eq(filtered, this.prev_value[key]))
|
||||
{
|
||||
this.prev_value[key] = filtered;
|
||||
changed = true;
|
||||
}
|
||||
}
|
||||
changed = true;
|
||||
}
|
||||
if (!changed)
|
||||
{
|
||||
return;
|
||||
}
|
||||
}
|
||||
await this.schedulePersist();
|
||||
}
|
||||
|
||||
async schedulePersist()
|
||||
{
|
||||
if (!this.cfg.persist_interval)
|
||||
{
|
||||
await this.persist();
|
||||
return;
|
||||
}
|
||||
if (!this.persist_timer)
|
||||
{
|
||||
this.persist_timer = setTimeout(() =>
|
||||
{
|
||||
this.persist_timer = null;
|
||||
this.persist().catch(console.error);
|
||||
}, this.cfg.persist_interval);
|
||||
}
|
||||
}
|
||||
|
||||
async persist()
|
||||
{
|
||||
if (!this.cfg.data)
|
||||
{
|
||||
return;
|
||||
}
|
||||
while (this.wait_persist)
|
||||
{
|
||||
await new Promise(ok => this.wait_persist.push(ok));
|
||||
}
|
||||
this.wait_persist = [];
|
||||
try
|
||||
{
|
||||
let dump = this.antietcd.etctree.dump(true);
|
||||
dump['term'] = this.antietcd.stored_term;
|
||||
dump = JSON.stringify(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);
|
||||
await fh.sync();
|
||||
await fh.close();
|
||||
await fsp.rename(this.cfg.data+'.tmp', this.cfg.data);
|
||||
}
|
||||
catch (e)
|
||||
{
|
||||
console.error(e);
|
||||
process.exit(1);
|
||||
}
|
||||
runCallbacks(this, 'wait_persist', null);
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = AntiPersistence;
|
||||
|
|
@ -0,0 +1,34 @@
|
|||
class RequestError
|
||||
{
|
||||
constructor(code, text)
|
||||
{
|
||||
this.code = code;
|
||||
this.message = text;
|
||||
}
|
||||
}
|
||||
|
||||
function de64(k)
|
||||
{
|
||||
if (k == null) // null or undefined
|
||||
return k;
|
||||
return Buffer.from(k, 'base64').toString();
|
||||
}
|
||||
|
||||
function runCallbacks(obj, key, new_value)
|
||||
{
|
||||
const cbs = obj[key];
|
||||
obj[key] = new_value;
|
||||
if (cbs)
|
||||
{
|
||||
for (const cb of cbs)
|
||||
{
|
||||
cb();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
RequestError,
|
||||
de64,
|
||||
runCallbacks,
|
||||
};
|
||||
Loading…
Reference in New Issue