Forward Factorio logs to clusterctl

Add instance output event that is forwarded by the master server to
control connections that are subscribing to them.  Add code subscribing
to the relevant instance output when dealing with instances in
clusterctl.

client.js

@@ -109,11 +109,6 @@ class Instance {
 	async createSave() {
 		console.log("Creating save .....");
 
-		// XXX Move to constructor and send to master
-		this.server.on('output', function(output) {
-			console.log("Fact: " + output.message);
-		});
-
 		await this.server.create("world");
 		console.log("Clusterio | Successfully created save");
 	}
@@ -188,10 +183,6 @@ class Instance {
 		}
 		await factorio.patch(this.path("saves", latestSave), modules);
 
-		this.server.on('output', (output) => {
-			console.log("Fact: " + output.message);
-		});
-
 		this.server.on('rcon-ready', () => {
 			console.log("Clusterio | RCON connection established");
 			// Temporary measure for backwards compatibility
@@ -317,6 +308,7 @@ class Slave extends link.Client {
 		// XXX: race condition on multiple simultanious calls
 		let instance = await Instance.create(id, instanceDir, this.config.factorioDir, options);
 		this.instances.set(id, instance);
+		this.hookInstance(instance);
 		this.updateInstances();
 	}
 
@@ -376,6 +368,7 @@ class Slave extends link.Client {
 				console.log(`found instance ${instanceConfig.name} in ${instancePath}`);
 				let instance = new Instance(instancePath, this.config.factorioDir, instanceConfig);
 				await instance.init();
+				this.hookInstance(instance); // XXX this is the wrong place for this
 				instances.set(instanceConfig.id, instance);
 			}
 		}
@@ -383,6 +376,13 @@ class Slave extends link.Client {
 		return instances;
 	}
 
+	hookInstance(instance) {
+		instance.server.on('output', (output) => {
+			link.events.instanceOutput.send(this, { instance_id: instance.config.id, output })
+		});
+	}
+
+
 	updateInstances() {
 		let list = [];
 		for (let instance of this.instances.values()) {

clusterctl.js

@@ -4,11 +4,71 @@ const fs = require("fs-extra");
 const yargs = require("yargs");
 const version = require("./package").version;
 const asTable = require("as-table").configure({delimiter: ' | '});
+const chalk = require("chalk");
 
 const link = require("lib/link");
 const errors = require("lib/errors");
 
 
+/**
+ * Format a parsed Factorio output message with colors
+ *
+ * Formats a parsed Factorio output from lib/factorio into a readable
+ * colorized output using terminal escape codes that can be printed.
+ */
+function formatOutputColored(output) {
+	let time = "";
+	if (output.format === 'seconds') {
+		time = chalk.dim(output.time.padStart(8)) + ' ';
+	} else if (output.format === 'date') {
+		time = chalk.dim(output.time) + ' ';
+	}
+
+	let info = "";
+	if (output.type === 'log') {
+		let level = output.level;
+		if (level === "Info") {
+			level = chalk.bold.blueBright(level);
+		} else if (output.level === "Warning") {
+			level = chalk.bold.yellowBright(level);
+		} else if (output.level === "Error") {
+			level = chalk.bold.redBright(level);
+		}
+
+		info = level + ' ' + chalk.dim(output.file) + ': ';
+
+	} else if (output.type === 'action') {
+		info = '[' + chalk.yellow(output.action) + '] ';
+	}
+
+	return time + info + output.message;
+}
+
+/**
+ * Format a parsed Factorio output message
+ *
+ * Formats a parsed Factorio output from lib/factorio into a readable
+ * output string that can be printed
+ */
+function formatOutput(output) {
+	let time = "";
+	if (output.format === 'seconds') {
+		time = output.time.padStart(8) + ' ';
+	} else if (output.format === 'date') {
+		time = output.time + ' ';
+	}
+
+	let info = "";
+	if (output.type === 'log') {
+		info = output.level + ' ' + output.file + ': ';
+
+	} else if (output.type === 'action') {
+		info = '[' + output.action + '] ';
+	}
+
+	return time + info + output.message;
+}
+
 /**
  * Represens a command that can be runned by clusterctl
  */
@@ -115,8 +175,10 @@ commands.push(new Command({
 		});
 	}],
 	handler: async function(args, control) {
+		let instanceId = await resolveInstance(control, args.instance);
+		await link.requests.setInstanceOutputSubscriptions.send(control, { instance_ids: [instanceId] });
 		let response = await link.requests.createSave.send(control, {
-			instance_id: await resolveInstance(control, args.instance),
+			instance_id: instanceId,
 		});
 		console.log(response);
 	},
@@ -129,8 +191,10 @@ commands.push(new Command({
 		});
 	}],
 	handler: async function(args, control) {
+		let instanceId = await resolveInstance(control, args.instance);
+		await link.requests.setInstanceOutputSubscriptions.send(control, { instance_ids: [instanceId] });
 		let response = await link.requests.startInstance.send(control, {
-			instance_id: await resolveInstance(control, args.instance),
+			instance_id: instanceId,
 		});
 		console.log(response);
 	},
@@ -143,8 +207,10 @@ commands.push(new Command({
 		});
 	}],
 	handler: async function(args, control) {
+		let instanceId = await resolveInstance(control, args.instance);
+		await link.requests.setInstanceOutputSubscriptions.send(control, { instance_ids: [instanceId] });
 		let response = await link.requests.stopInstance.send(control, {
-			instance_id: await resolveInstance(control, args.instance),
+			instance_id: instanceId,
 		});
 		console.log(response);
 	},
@@ -208,6 +274,15 @@ class Control extends link.Client {
 			version: version,
 		});
 	}
+
+	async instanceOutputEventHandler(message) {
+		let { instance_id, output } = message.data;
+		if (process.platform === "win32") {
+			console.log(formatOutput(output));
+		} else {
+			console.log(formatOutputColored(output));
+		}
+	}
 }
 
 async function findCredentials() {

lib/link/messages.js

@@ -260,6 +260,18 @@ requests.listInstances = new Request({
 	},
 });
 
+requests.setInstanceOutputSubscriptions = new Request({
+	type: 'set_instance_output_subscriptions',
+	sources: ['control'],
+	targets: ['master'],
+	requestProperties: {
+		"instance_ids": {
+			type: "array",
+			items: { type: "integer" },
+		},
+	},
+});
+
 requests.createInstanceCommand = new Request({
 	type: 'create_instance',
 	sources: ['control'],
@@ -397,6 +409,31 @@ events.updateInstances = new Event({
 	},
 });
 
+events.instanceOutput = new Event({
+	type: 'instance_output',
+	sources: ['slave', 'master'],
+	targets: ['master', 'control'],
+	eventProperties: {
+		"instance_id": { type: "integer" },
+		"output": {
+			type: "object",
+			additionalProperties: false,
+			required: ["source", "received", "format", "type", "message"],
+			properties: {
+				"source": { type: "string" },
+				"received": { type: "number" },
+				"format": { type: "string" },
+				"time": { type: "string" },
+				"level": { type: "string" },
+				"file": { type: "string" },
+				"type": { type: "string" },
+				"action": { type: "string" },
+				"message": { type: "string" },
+			},
+		},
+	},
+});
+
 
 /**
  * Attaches all requests and events to the given link

master.js

@@ -283,6 +283,8 @@ class ControlConnection extends BaseConnection {
 				controlConnections.splice(index, 1);
 			}
 		});
+
+		this.instanceOutputSubscriptions = new Set();
 	}
 
 	async listSlavesRequestHandler(message) {
@@ -334,6 +336,10 @@ class ControlConnection extends BaseConnection {
 			}
 		});
 	}
+
+	async setInstanceOutputSubscriptionsRequestHandler(message) {
+		this.instanceOutputSubscriptions = new Set(message.data.instance_ids);
+	}
 }
 
 var slaveConnections = new Map();
@@ -403,6 +409,15 @@ class SlaveConnection extends BaseConnection {
 			});
 		}
 	}
+
+	async instanceOutputEventHandler(message) {
+		let { instance_id, output } = message.data;
+		for (let controlConnection of controlConnections) {
+			if (controlConnection.instanceOutputSubscriptions.has(instance_id)) {
+				link.events.instanceOutput.send(controlConnection, message.data);
+			}
+		}
+	}
 }
 
 // Authentication middleware for socket.io connections