Add tailing of files and move socketio connect after plugin load (#191)

#add tailing of files to prevent lost data between factorio and clusterio
#move socketio connect to the bottom after the plugins are loaded 
  to prevent plugins not getting the connection event
#add some more transfer compression in different places
This commit is contained in:
Godmave authored and GitHub committed 2019-08-05 07:47:57 +02:00
1 parent f76c84c3f1
commit 0df1b8e1fd
5 files changed
+148 -138

No files matched your search

+50 -38
View File
@@ -1,4 +1,5 @@
const fs = require('fs-extra');
const Tail = require('tail').Tail;
const https = require('follow-redirects').https;
const needle = require("needle");
const child_process = require('child_process');
@@ -55,6 +56,7 @@ if(!config.masterAuthToken || typeof config.masterAuthToken !== "string"){
You can retrieve your auth token from the master in secret-api-token.txt after running it once.");
}
const needleOptionsWithTokenAuthHeader = {
compressed:true,
headers: {
'x-access-token': config.masterAuthToken
},
@@ -582,9 +584,6 @@ async function instanceManagement(instanceconfig) {
for(let i = 0; i < pluginsToLoad.length; i++){
let pluginLoadStarted = Date.now();
let log = function(message) {
console.log("Clusterio | "+ pluginsToLoad[i].name + " | " + message);
}
let combinedConfig = deepmerge(instanceconfig,config,{clone:true});
let pluginConfig = pluginsToLoad[i];
@@ -596,7 +595,7 @@ async function instanceManagement(instanceconfig) {
let pluginClass = require(path.resolve(pluginConfig.pluginPath, "index.js"));
plugins[i] = new pluginClass(combinedConfig, async function(data, callback){
if(data && data.toString('utf8')[0] != "/") {
log(data.toString('utf8'));
console.log("Clusterio | "+ pluginsToLoad[i].name + " | " + data.toString('utf8'));
return true;
} else if (data && data.toString('utf8')[0] == "/"){
return messageInterface(data.toString('utf8'), callback);
@@ -622,37 +621,50 @@ async function instanceManagement(instanceconfig) {
global.subscribedFiles[pluginConfig.scriptOutputFileSubscription] = true;
console.log("Clusterio | Registered file subscription on "+instancedirectory + "/script-output/" + pluginConfig.scriptOutputFileSubscription);
fs.watch(instancedirectory + "/script-output/" + pluginConfig.scriptOutputFileSubscription, fileChangeHandler);
// run once in case a plugin wrote out information before the plugin loaded fully
// delay, so the socket got enough time to connect
setTimeout(()=> {
fileChangeHandler(false, pluginConfig.scriptOutputFileSubscription);
}, 500);
// send file contents to plugin for processing
function fileChangeHandler(eventType, filename) {
if(filename != null){
setTimeout(
()=>{
// get array of lines in file
let stuff = fs.readFileSync(instancedirectory + "/script-output/" + filename, "utf8").split("\n");
// if you found anything, reset the file
if (stuff[0]) {
fs.writeFileSync(instancedirectory + "/script-output/" + filename, "");
}
for(let o = 0; o < stuff.length; o++) {
if(stuff[o] && !stuff[o].includes('\u0000\u0000')) {
try{
plugins[i].scriptOutput(stuff[o]);
}catch(e){console.error(e)}
}
}
},
pluginConfig.fileReadDelay || 0
);
}
}
}
if(!pluginConfig.fileReadDelay || pluginConfig.fileReadDelay == 0) {
// only wipe the file on restart for now, should most likely be rotated during runtime too
fs.writeFileSync(instancedirectory + "/script-output/" + pluginConfig.scriptOutputFileSubscription, "");
let tail = new Tail(instancedirectory + "/script-output/" + pluginConfig.scriptOutputFileSubscription);
tail.on("line", function (data) {
plugins[i].scriptOutput(data);
});
} else {
fs.watch(instancedirectory + "/script-output/" + pluginConfig.scriptOutputFileSubscription, fileChangeHandler);
// run once in case a plugin wrote out information before the plugin loaded fully
// delay, so the socket got enough time to connect
setTimeout(() => {
fileChangeHandler(false, pluginConfig.scriptOutputFileSubscription);
}, 500);
// send file contents to plugin for processing
function fileChangeHandler(eventType, filename) {
if (filename != null) {
setTimeout(
() => {
// get array of lines in file
let stuff = fs.readFileSync(instancedirectory + "/script-output/" + filename, "utf8").split("\n");
// if you found anything, reset the file
if (stuff[0]) {
fs.writeFileSync(instancedirectory + "/script-output/" + filename, "");
}
for (let o = 0; o < stuff.length; o++) {
if (stuff[o] && !stuff[o].includes('\u0000\u0000')) {
try {
plugins[i].scriptOutput(stuff[o]);
} catch (e) {
console.error(e)
}
}
}
},
pluginConfig.fileReadDelay || 0
);
}
}
}
}
console.log(`Clusterio | Loaded plugin ${pluginsToLoad[i].name} in ${Date.now() - pluginLoadStarted}ms`);
} else {
// this plugin doesn't have a client portion. Maybe it runs on the master only?
@@ -675,7 +687,7 @@ async function instanceManagement(instanceconfig) {
mods:modHashes,
instanceName: instance,
info: instanceconfig.info,
}
};
if(playerCount){
payload.playerCount = playerCount.replace(/(\r\n\t|\n|\r\t)/gm, "");
} else {
@@ -722,7 +734,7 @@ async function instanceManagement(instanceconfig) {
let payload = {
modName: modHashes[i].modName,
hash: modHashes[i].hash,
}
};
needle.post(config.masterIP + ":" + config.masterPort + '/api/checkMod', payload, needleOptionsWithTokenAuthHeader, function (err, response, body) {
if(err) console.error("Unable to contact master server /api/checkMod! Please check your config.json.");
if(response && body && body == "found") {
@@ -765,7 +777,7 @@ function hashMods(instanceName, callback) {
hashedMods[hashedMods.length] = {
modName: modName,
hash: hash,
}
};
// check if this callback has ran once for each mod
if(hashedMods.length == instanceMods.length) {
callback(hashedMods);
@@ -788,7 +800,7 @@ function hashMods(instanceName, callback) {
let name = instanceMods[i];
let options = {
files:path,
}
};
// options {files:[array of paths]}
hashFiles(options, function(error, hash) {
// hash will be a string if no error occurred
+95 -98
View File
@@ -40,14 +40,13 @@ const deepmerge = require("deepmerge");
const path = require("path");
const fs = require("fs-extra");
const jwt = require('jsonwebtoken');
const bcrypt = require('bcryptjs');
const crypto = require('crypto');
const base64url = require('base64url');
const moment = require("moment");
const request = require("request")
const request = require("request");
// constants
console.log(`Requiring config from ${args.config || './config'}`)
console.log(`Requiring config from ${args.config || './config'}`);
const config = require(args.config || './config');
config.databaseDirectory = args.databaseDirectory || config.databaseDirectory || "./database";
const masterModFolder = path.join(config.databaseDirectory, "/masterMods/");
@@ -56,8 +55,6 @@ mkdirp.sync(masterModFolder);
// homebrew modules
const getFactorioLocale = require("lib/getFactorioLocale");
const stringUtils = require("lib/stringUtils");
const fileOps = require("lib/fileOps");
// homemade express middleware for token auth
const authenticate = require("lib/authenticate")(config);
@@ -68,7 +65,7 @@ function randomStringAsBase64Url(size) {
}
if (!fs.existsSync("secret-api-token.txt")) {
config.masterAuthSecret = randomStringAsBase64Url(256);
fs.writeFileSync("config.json",JSON.stringify(config, null, 4));
fs.writeFileSync("config.json",JSON.stringify(config, null, 4));
fs.writeFileSync("secret-api-token.txt", jwt.sign({ id: "api" }, config.masterAuthSecret, {
expiresIn: 86400*365 // expires in 1 year
}));
@@ -176,18 +173,13 @@ const prometheusUPSGauge = new Prometheus.Gauge({
help: 'UPS of the current server',
labelNames: ["instanceID", "instanceName"],
});
const prometheusTickGauge = new Prometheus.Gauge({
name: prometheusPrefix+'tick_gauge',
help: 'Servers current gametick',
labelNames: ["instanceID", "instanceName"],
});
const prometheusMasterInventoryGauge = new Prometheus.Gauge({
name: prometheusPrefix+'master_inventory_gauge',
help: 'Amount of items stored on master',
labelNames: ["itemName"],
});
setInterval(()=>{
fs.writeFileSync("database/items.json", JSON.stringify(db.items));
fs.writeFileSync("database/items.json", JSON.stringify(db.items));
},config.autosaveInterval || 60000);
/**
GET Prometheus metrics endpoint. Returns performance and usage metrics in a prometheus readable format.
@@ -198,7 +190,7 @@ GET Prometheus metrics endpoint. Returns performance and usage metrics in a prom
app.get('/metrics', (req, res) => {
endpointHitCounter.labels(req.route.path).inc();
res.set('Content-Type', Prometheus.register.contentType);
/// gather some static metrics
registerMoreMetrics();
res.end(Prometheus.register.metrics());
@@ -248,7 +240,7 @@ const db = {};
} catch (e){
db.items = {};
}
})()
})();
db.items.addItem = function(object) {
if(object.name == "addItem" || object.name == "removeItem") {
@@ -262,7 +254,7 @@ db.items.addItem = function(object) {
}
return true;
}
}
};
db.items.removeItem = function(object) {
if(object.name == "addItem" || object.name == "removeItem") {
console.error("Fuck you, that would screw everything up if you named your item that.");
@@ -275,7 +267,7 @@ db.items.removeItem = function(object) {
}
return true;
}
}
};
// store slaves and inventory in a .json full of JSON data
process.on('SIGINT', shutdown); // ctrl + c
@@ -308,8 +300,8 @@ async function shutdown() {
console.log("Clusterio cleanly exited in "+(Date.now()-exitStartTime)+"ms");
process.exit(0);
} catch(e) {
console.log(e)
console.log("Clusterio failed to exit cleanly. Time elapsed: "+(Date.now()-exitStartTime)+"ms")
console.log(e);
console.log("Clusterio failed to exit cleanly. Time elapsed: "+(Date.now()-exitStartTime)+"ms");
process.exit(1)
}
}
@@ -351,7 +343,7 @@ app.post("/api/editSlaveMeta", authenticate.middleware, function(req,res) {
endpointHitCounter.labels(req.route.path).inc();
// request.body should be an object
// {instanceID, pass, meta:{x,y,z}}
if(req.body && req.body.instanceID && req.body.meta){
if(slaves[req.body.instanceID]){
if(!slaves[req.body.instanceID].meta) {
@@ -360,8 +352,8 @@ app.post("/api/editSlaveMeta", authenticate.middleware, function(req,res) {
slaves[req.body.instanceID].meta = deepmerge(slaves[req.body.instanceID].meta, req.body.meta, {clone:true});
let metaPortion = JSON.stringify(req.body.meta);
if(metaPortion.length > 50) {
metaPortion = metaPortion.substring(0,20) + "...";
}
metaPortion = metaPortion.substring(0,20) + "...";
}
// console.log("Updating slave "+slaves[req.body.instanceID].instanceName+": " + slaves[req.body.instanceID].mac + " : " + slaves[req.body.instanceID].serverPort+" at " + slaves[req.body.instanceID].publicIP, metaPortion);
res.sendStatus(200);
} else {
@@ -387,7 +379,7 @@ app.post("/api/getSlaveMeta", function (req, res) {
}
res.send(JSON.stringify(slaves[req.body.instanceID].meta))
} else {
res.status(404)
res.status(404);
res.send('{"status": 404, "info": "Slave not registered"}')
}
} else {
@@ -440,8 +432,8 @@ app.post("/api/uploadMod", authenticate.middleware, function(req,res) {
});
} else {
res.send('No files were uploaded.');
return;
}
}
});
/**
Prepare the slaveCache for both /api/slaves calls
@@ -480,15 +472,15 @@ POST endpoint for getting the rconPasswords of all our slaves
@alias /api/rconPasswords
*/
app.post("/api/rconPasswords", authenticate.middleware, function(req, res) {
endpointHitCounter.labels(req.route.path).inc();
res.header("Access-Control-Allow-Origin", "*");
res.header("Access-Control-Allow-Headers", "Origin, X-Requested-With, Content-Type, Accept");
let slaveCache = getSlaves();
let validKeys = ['rconPort', 'rconPassword', 'publicIP'];
for(key in slaveCache.cache) {
slaveCache.cache[key] = {'publicIP': slaveCache.cache[key].publicIP, 'rconPort': slaveCache.cache[key].rconPort, 'rconPassword': slaveCache.cache[key].rconPassword};
}
res.send(slaveCache.cache);
endpointHitCounter.labels(req.route.path).inc();
res.header("Access-Control-Allow-Origin", "*");
res.header("Access-Control-Allow-Headers", "Origin, X-Requested-With, Content-Type, Accept");
let slaveCache = getSlaves();
let validKeys = ['rconPort', 'rconPassword', 'publicIP'];
for(key in slaveCache.cache) {
slaveCache.cache[key] = {'publicIP': slaveCache.cache[key].publicIP, 'rconPort': slaveCache.cache[key].rconPort, 'rconPassword': slaveCache.cache[key].rconPassword};
}
res.send(slaveCache.cache);
});
/*
var recievedItemStatistics = new averagedTimeSeries({
@@ -559,20 +551,20 @@ app.post("/api/place", authenticate.middleware, function(req, res) {
name:req.body.name,
count:req.body.count
});
// Attempt confirming
res.send("success");
} else {
res.send("failure");
}
});
const routes_api_remove = require("./routes/api/remove.js")
var neuralDole = null
const routes_api_remove = require("./routes/api/remove.js");
var neuralDole = null;
if(config.useNeuralNetDoleDivider)//Only initialize neural network when it's enabled, otherwise it might override gauge
neuralDole=new routes_api_remove.neuralDole({
items: db.items, gaugePrefix: prometheusPrefix
})
});
/**
POST endpoint to remove items from DB when client orders items.
@@ -597,7 +589,7 @@ app.post("/api/remove", authenticate.middleware, function(req, res) {
object.instanceName = "unknown";
}
if(slaves[object.instanceID]) object.instanceName = slaves[object.instanceID].instanceName;
let item = db.items[object.name]
let item = db.items[object.name];
// console.dir(doc);
if((item === undefined)
||((
@@ -613,7 +605,7 @@ app.post("/api/remove", authenticate.middleware, function(req, res) {
let numberToRemove = Math.min(Math.abs(Number(object.count)),Number(item));
db.items.removeItem({count: numberToRemove, name: object.name});
res.send({count: numberToRemove, name: object.name});
// track statistics and do graphing things
prometheusImportGauge.labels(object.instanceID, object.name).inc(Number(numberToRemove) || 0);
let sentItemStatistics = sentItemStatisticsBySlaveID[object.instanceID];
@@ -702,19 +694,19 @@ app.get("/api/inventoryAsObject", function(req, res) {
*/
// wrap a request in an promise
async function downloadPage(url) {
return new Promise((resolve, reject) => {
request(url, (error, response, body) => {
resolve(body);
});
});
}
async function downloadPage(url) {
return new Promise((resolve, reject) => {
request(url, (error, response, body) => {
resolve(body);
});
});
}
app.get("/api/modmeta", async function(req, res) {
endpointHitCounter.labels(req.route.path).inc();
res.header("Access-Control-Allow-Origin", "*");
res.setHeader('Content-Type', 'application/json');
let modData = await downloadPage("https://mods.factorio.com/api/mods/" + req.query.modname);
res.header("Access-Control-Allow-Origin", "*");
res.setHeader('Content-Type', 'application/json');
let modData = await downloadPage("https://mods.factorio.com/api/mods/" + req.query.modname);
res.send(modData);
});
@@ -731,36 +723,36 @@ Requires x-access-token header to be set. Find you api token in secret-api-token
*/
app.post("/api/runCommand", authenticate.middleware, function(req,res) {
endpointHitCounter.labels(req.route.path).inc();
// validate request.body
let body = req.body;
// validate request.body
let body = req.body;
if (
body.broadcast
&& body.command
&& typeof body.command == "string"
&& body.command[0] == "/")
{
let instanceResponses = [];
for (let instanceID in wsSlaves) {
// skip loop if the property is from prototype
if (!wsSlaves.hasOwnProperty(instanceID)) continue;
wsSlaves[instanceID].runCommand(body.command);
}
res.status(200).send({auth: true, message: "success", response: "Cluster wide messaging initiated"});
} else if (
body.instanceID
&& wsSlaves[body.instanceID]
&& body.command
&& typeof body.command == "string"
&& body.command[0] == "/")
{
// execute command
wsSlaves[body.instanceID].runCommand(body.command, data => {
res.status(200).send({auth: true, message: "success", data});
});
} else {
res.status(400).send({auth: true, message: "Error: invalid request.body"});
}
if (
body.broadcast
&& body.command
&& typeof body.command == "string"
&& body.command[0] == "/")
{
let instanceResponses = [];
for (let instanceID in wsSlaves) {
// skip loop if the property is from prototype
if (!wsSlaves.hasOwnProperty(instanceID)) continue;
wsSlaves[instanceID].runCommand(body.command);
}
res.status(200).send({auth: true, message: "success", response: "Cluster wide messaging initiated"});
} else if (
body.instanceID
&& wsSlaves[body.instanceID]
&& body.command
&& typeof body.command == "string"
&& body.command[0] == "/")
{
// execute command
wsSlaves[body.instanceID].runCommand(body.command, data => {
res.status(200).send({auth: true, message: "success", data});
});
} else {
res.status(400).send({auth: true, message: "Error: invalid request.body"});
}
});
/**
GET endpoint. Returns factorio's base locale as a JSON object.
@@ -797,16 +789,17 @@ if(args.masterPort || config.masterPort){
}
/* Websockets for remoteMap */
var io = require("socket.io")(server);
var io = require("socket.io")();
const ioMetrics = require("socket.io-prometheus");
ioMetrics(io);
var slaveMappers = {};
class slaveMapper {
constructor(instanceID, socket) {
this.instanceID = instanceID;
this.socket = socket;
this.lastBeat = Date.now();
this.socket.on("heartbeat", () => {
prometheusWsUsageCounter.labels('heartbeat', this.instanceID).inc();
// we aren't ready to die yet apparently
@@ -821,7 +814,7 @@ class slaveMapper {
prometheusWsUsageCounter.labels('sendEntity', this.instanceID).inc();
Object.keys(mapRequesters).forEach(requesterName => {
let requester = mapRequesters[requesterName];
if(requester.instanceID == this.instanceID){
// this mapRequester is listening to this slaveMapper, so we send it updates
requester.socket.emit("displayEntity", entity);
@@ -830,6 +823,8 @@ class slaveMapper {
});
}
}
var mapRequesters = {};
class mapRequester {
constructor(requesterID, socket, instanceID){
@@ -837,7 +832,7 @@ class mapRequester {
this.socket = socket;
this.instanceID = instanceID;
this.lastBeat = Date.now();
this.socket.on("heartbeat", () => {
prometheusWsUsageCounter.labels('heartbeat', "other").inc();
// we aren't ready to die yet apparently
@@ -869,6 +864,8 @@ class mapRequester {
});
}
}
/* Websockets for send and recieve combinators */
var wsSlaves = {};
class wsSlave {
@@ -876,7 +873,7 @@ class wsSlave {
this.instanceID = instanceID;
this.socket = socket;
this.lastBeat = Date.now();
this.socket.on("heartbeat", () => {
prometheusWsUsageCounter.labels('heartbeat', this.instanceID).inc();
this.lastBeat = Date.now();
@@ -898,7 +895,7 @@ class wsSlave {
delete this.commandsWaitingForReturn[resp.commandID];
}
});
this.socket.on("gameChat", data => {
prometheusWsUsageCounter.labels('gameChat', this.instanceID).inc();
if(typeof data === "object"){
@@ -909,7 +906,7 @@ class wsSlave {
});
}
});
this.socket.on("alert", alert => {
prometheusWsUsageCounter.labels('alert', this.instanceID).inc();
if(typeof alert === "object"){
@@ -927,6 +924,7 @@ class wsSlave {
if(commandID) this.commandsWaitingForReturn[commandID] = {callback, timestamp: Date.now()};
}
}
io.on('connection', function (socket) {
// cleanup dead sockets from disconnected people
let terminatedConnections = 0;
@@ -941,10 +939,10 @@ io.on('connection', function (socket) {
});
});
if(terminatedConnections > 0) console.log("SOCKET | There are currently "+currentConnections+" websocket connections, deleting "+terminatedConnections+" on timeout");
// tell our friend that we are listening
setTimeout(()=>socket.emit('hello', { hello: 'world' }), 5000);
/* initial processing for remoteMap */
socket.on('registerSlaveMapper', function (data) {
prometheusWsUsageCounter.labels('registerSlaveMapper', "other").inc();
@@ -959,7 +957,7 @@ io.on('connection', function (socket) {
socket.emit("mapRequesterReady", true);
console.log("remoteMap | SOCKET registered map requester for "+data.instanceID);
});
/* Websockets for send and recieve combinators */
socket.on("registerSlave", function(data) {
prometheusWsUsageCounter.labels('registerSlave', "other").inc();
@@ -981,16 +979,8 @@ io.on('connection', function (socket) {
});
// handle plugins on the master
pluginManagement();
masterPlugins = [];
async function pluginManagement(){
let startPluginLoad = Date.now();
masterPlugins = await getPlugins();
console.log("All plugins loaded in "+(Date.now() - startPluginLoad)+"ms");
return;
}
async function getPlugins(){
let startPluginLoad = Date.now();
const pluginManager = require("lib/manager/pluginManager.js")(config);
let plugins = [];
let pluginsToLoad = await pluginManager.getPlugins();
@@ -1012,7 +1002,7 @@ async function getPlugins(){
}),
pluginConfig,
});
console.log("Loaded plugin "+pluginConfig.name+" in "+(Date.now() - pluginStartedLoading)+"ms");
}
}
@@ -1021,8 +1011,15 @@ async function getPlugins(){
let plugin = plugins[i];
if(plugin.main.onLoadFinish && typeof plugin.main.onLoadFinish == "function") await plugin.main.onLoadFinish({plugins});
}
console.log("All plugins loaded in "+(Date.now() - startPluginLoad)+"ms");
return plugins;
}
var masterPlugins = await getPlugins();
io.attach(server);
module.exports = app;
})();
+1
View File
@@ -70,6 +70,7 @@
"socket.io-client": "^2.0.4",
"socket.io-prometheus": "^0.2.1",
"sync-request": "^4.0.1",
"tail": "^2.0.2",
"typescript-collections": "^1.3.2",
"underscore": "^1.8.3"
},
+1 -1
View File
@@ -307,7 +307,7 @@ module.exports = class remoteCommands {
return new Promise((resolve, reject) => {
let instance = this.instances[instanceID];
if(!instance){
needle.get(this.config.masterIP+":"+this.config.masterPort+ '/api/slaves', (err, response) => {
needle.get(this.config.masterIP+":"+this.config.masterPort+ '/api/slaves', { compressed: true }, (err, response) => {
if(err || response.statusCode != 200) {
console.log("Unable to get JSON master/api/slaves, master might be unaccessible");
} else if (response && response.body) {
+1 -1
View File
@@ -55,7 +55,7 @@ module.exports = class remoteCommands {
return new Promise((resolve, reject) => {
let instance = this.instances[instanceID];
if(!instance){
needle.get(this.config.masterIP+":"+this.config.masterPort+ '/api/slaves', (err, response) => {
needle.get(this.config.masterIP+":"+this.config.masterPort+ '/api/slaves', { compressed: true }, (err, response) => {
if(err || response.statusCode != 200) {
console.log("Unable to get JSON master/api/slaves, master might be unaccessible");
} else if (response && response.body) {