291 lines
9.0 KiB
JavaScript
291 lines
9.0 KiB
JavaScript
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();
|