Add simple persistence
parent
b559f9b555
commit
1f9cf6307a
123
antietcd.js
123
antietcd.js
|
|
@ -1,7 +1,10 @@
|
|||
const fs = require('fs');
|
||||
const fsp = require('fs').promises;
|
||||
const { URL } = require('url');
|
||||
const http = require('http');
|
||||
const https = require('https');
|
||||
const crypto = require('crypto');
|
||||
const zlib = require('zlib');
|
||||
|
||||
const ws = require('ws');
|
||||
|
||||
|
|
@ -19,27 +22,81 @@ class RequestError
|
|||
class AntiEtcd
|
||||
{
|
||||
constructor(cfg)
|
||||
{
|
||||
this.cfg = cfg;
|
||||
}
|
||||
|
||||
run()
|
||||
{
|
||||
this.clients = {};
|
||||
this.client_id = 1;
|
||||
this.etctree = new EtcTree(true);
|
||||
this.cfg = cfg;
|
||||
this.stopped = false;
|
||||
}
|
||||
|
||||
async run()
|
||||
{
|
||||
if (this.cfg.filename)
|
||||
{
|
||||
// Load data from disk
|
||||
const [ err, stat ] = await new Promise(ok => fs.stat(this.cfg.filename, (err, stat) => ok([ err, stat ])));
|
||||
if (!err)
|
||||
{
|
||||
const data = await fsp.readFile(this.cfg.filename);
|
||||
data = JSON.parse(zlib.Gunzip(data));
|
||||
this.etctree.load(data);
|
||||
}
|
||||
else if (err.code != ENOENT)
|
||||
{
|
||||
throw err;
|
||||
}
|
||||
// Set exit hook
|
||||
const on_stop_cb = () => this.on_stop();
|
||||
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));
|
||||
this.server.listen(this.cfg.port || 2379);
|
||||
}
|
||||
|
||||
async on_stop()
|
||||
{
|
||||
if (this.stopped)
|
||||
{
|
||||
return;
|
||||
}
|
||||
this.stopped = true;
|
||||
// Wait until all requests complete
|
||||
while (this.inflight > 0)
|
||||
{
|
||||
await new Promise(ok => this.waitInflight.push(ok));
|
||||
}
|
||||
await this.persist();
|
||||
}
|
||||
|
||||
async persist()
|
||||
{
|
||||
if (!this.cfg.filename)
|
||||
{
|
||||
return;
|
||||
}
|
||||
let dump = this.etctree.dump(true);
|
||||
dump = JSON.stringify(dump);
|
||||
dump = zlib.Gzip(dump);
|
||||
const fd = await fsp.open(this.cfg.filename+'.tmp', 'w');
|
||||
await fsp.write(fd, dump);
|
||||
await fsp.fsync(fd);
|
||||
await fsp.close(fd);
|
||||
await fsp.rename(this.cfg.filename+'.tmp', this.cfg.filename);
|
||||
}
|
||||
|
||||
fail(res, code, text)
|
||||
{
|
||||
res.writeHead(code);
|
||||
|
|
@ -51,7 +108,7 @@ class AntiEtcd
|
|||
{
|
||||
let data = [];
|
||||
req.on('data', (chunk) => data.push(chunk));
|
||||
req.on('end', () =>
|
||||
req.on('end', async () =>
|
||||
{
|
||||
data = Buffer.concat(data);
|
||||
let body = '';
|
||||
|
|
@ -76,7 +133,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,21 +150,28 @@ 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.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();
|
||||
}
|
||||
catch (e)
|
||||
{
|
||||
console.error(e);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
|
|
@ -163,7 +227,7 @@ class AntiEtcd
|
|||
});
|
||||
}
|
||||
|
||||
runHandler(req, data, res)
|
||||
async runHandler(req, data, res)
|
||||
{
|
||||
// v3/kv/txn
|
||||
// v3/kv/range
|
||||
|
|
@ -187,6 +251,10 @@ class AntiEtcd
|
|||
return cb.call(this, data);
|
||||
}
|
||||
}
|
||||
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');
|
||||
}
|
||||
|
|
@ -234,6 +302,11 @@ class AntiEtcd
|
|||
return this.etctree.api_keepalive_lease(data);
|
||||
}
|
||||
|
||||
handle_dump(data)
|
||||
{
|
||||
return this.etctree.dump();
|
||||
}
|
||||
|
||||
handleMessage(client_id, msg, socket)
|
||||
{
|
||||
if (msg.create_request)
|
||||
|
|
@ -285,4 +358,4 @@ class AntiEtcd
|
|||
}
|
||||
}
|
||||
|
||||
new AntiEtcd({ port: 12379 }).run();
|
||||
new AntiEtcd({ port: 12379 }).run().catch(console.error);
|
||||
|
|
|
|||
Loading…
Reference in New Issue