From 0df1b8e1fd79cd182e837656cfa1f9b16701cd62 Mon Sep 17 00:00:00 2001 From: Godmave Date: Mon, 5 Aug 2019 07:47:57 +0200 Subject: [PATCH] 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 --- client.js | 88 +++++++------ master.js | 193 ++++++++++++++-------------- package.json | 1 + sharedPlugins/clusterioMod/index.js | 2 +- sharedPlugins/globalChat/index.js | 2 +- 5 files changed, 148 insertions(+), 138 deletions(-) diff --git a/client.js b/client.js index cb1d9853..adcc8cab 100644 --- a/client.js +++ b/client.js @@ -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 diff --git a/master.js b/master.js index e08631dc..e8c47dff 100644 --- a/master.js +++ b/master.js @@ -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; })(); diff --git a/package.json b/package.json index 3c3ef4dd..2f817d0f 100644 --- a/package.json +++ b/package.json @@ -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" }, diff --git a/sharedPlugins/clusterioMod/index.js b/sharedPlugins/clusterioMod/index.js index 4b591025..9b2dc35d 100644 --- a/sharedPlugins/clusterioMod/index.js +++ b/sharedPlugins/clusterioMod/index.js @@ -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) { diff --git a/sharedPlugins/globalChat/index.js b/sharedPlugins/globalChat/index.js index 99622368..ad8a6c0e 100644 --- a/sharedPlugins/globalChat/index.js +++ b/sharedPlugins/globalChat/index.js @@ -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) {