Notify master about instance status update

Send updates to the master server when the running status of instances
change and add onInstanceStatusChanged hook to master plugins.

lib/link/messages.js

@@ -535,17 +535,28 @@ messages.debugWsMessage = new Event({
 
 messages.instanceInitialized = new Event({
 	type: "instance_initialized",
-	links: ["instance-slave"],
+	links: ["instance-slave", "slave-master"],
 	eventProperties: {
+		"instance_id": { type: "integer" },
 		"plugins": {
 			type: "object",
 			additionalProperties: { type: "string" },
 		}
 	},
 });
+messages.instanceStarted = new Event({
+	type: "instance_started",
+	links: ["instance-slave", "slave-master"],
+	eventProperties: {
+		"instance_id": { type: "integer" },
+	},
+});
 messages.instanceStopped = new Event({
 	type: 'instance_stopped',
-	links: ['instance-slave'],
+	links: ["instance-slave", "slave-master"],
+	eventProperties: {
+		"instance_id": { type: "integer" },
+	},
 });
 
 messages.updateInstances = new Event({
@@ -557,9 +568,10 @@ messages.updateInstances = new Event({
 			items: {
 				type: "object",
 				additionalProperties: false,
-				required: ["serialized_config"],
+				required: ["serialized_config", "status"],
 				properties: {
 					"serialized_config": { type: "object" },
+					"status": { enum: ["stopped", "initialized", "running"] },
 				},
 			},
 		},

lib/plugin.js

@@ -163,6 +163,21 @@ class BaseMasterPlugin {
 	 */
 	async init() { }
 
+	/**
+	 * Called when the status of an instance changes
+	 *
+	 * Invoked when the master server has received notice from a slave that
+	 * the running status of an instance has changed.  The possible statuses
+	 * are `stopped` meaning it is not loaded or running, `initialized`
+	 * meaning the instance has initialized and is starting up the Factorio
+	 * server, and `running` which means the Factorio server it manages is
+	 * up and running.
+	 *
+	 * @param {Object} instance - the instance that changed.
+	 * @param {string} prev - the previous status of the instance.
+	 */
+	async onInstanceStatusChanged(instance, prev) { }
+
 	/**
 	 * Called before collecting Prometheus metrics
 	 *

master.js

@@ -630,6 +630,36 @@ class SlaveConnection extends BaseConnection {
 		}
 	}
 
+	async instanceInitializedEventHandler(message, event) {
+		let instance = db.instances.get(message.data.instance_id);
+		let prev = instance.status;
+		instance.status = "initialized";
+		console.log(`Clusterio | Instance ${instance.config.get("instance.name")} Initialized`)
+		for (let masterPlugin of masterPlugins.values()) {
+			await masterPlugin.onInstanceStatusChanged(instance, prev);
+		}
+	}
+
+	async instanceStartedEventHandler(message, event) {
+		let instance = db.instances.get(message.data.instance_id);
+		let prev = instance.status;
+		instance.status = "running";
+		console.log(`Clusterio | Instance ${instance.config.get("instance.name")} Started`)
+		for (let masterPlugin of masterPlugins.values()) {
+			await masterPlugin.onInstanceStatusChanged(instance, prev);
+		}
+	}
+
+	async instanceStoppedEventHandler(message, event) {
+		let instance = db.instances.get(message.data.instance_id);
+		let prev = instance.status;
+		instance.status = "stopped";
+		console.log(`Clusterio | Instance ${instance.config.get("instance.name")} Stopped`)
+		for (let masterPlugin of masterPlugins.values()) {
+			await masterPlugin.onInstanceStatusChanged(instance, prev);
+		}
+	}
+
 	async updateInstancesEventHandler(message) {
 		// Push updated instance configs
 		for (let instance of db.instances.values()) {
@@ -645,7 +675,19 @@ class SlaveConnection extends BaseConnection {
 		for (let instance of message.data.instances) {
 			let instanceConfig = new config.InstanceConfig();
 			await instanceConfig.load(instance.serialized_config);
-			if (db.instances.has(instanceConfig.get("instance.id"))) {
+
+			let masterInstance = db.instances.get(instanceConfig.get("instance.id"))
+			if (masterInstance) {
+				// Already have this instance, update state instead
+				if (masterInstance.status !== instance.status) {
+					let prev = masterInstance.status;
+					masterInstance.status = instance.status;
+					if (prev !== undefined) {
+						for (let masterPlugin of masterPlugins.values()) {
+							await masterPlugin.onInstanceStatusChanged(instance, prev);
+						}
+					}
+				}
 				continue;
 			}
 

slave.js

@@ -112,7 +112,7 @@ class Instance extends link.Link{
 
 	notifyExit() {
 		this._running = false;
-		link.messages.instanceStopped.send(this);
+		link.messages.instanceStopped.send(this, { instance_id: this.config.get("instance.id") });
 
 		// Clear metrics this instance is exporting
 		for (let collector of prometheus.defaultRegistry.collectors) {
@@ -166,7 +166,7 @@ class Instance extends link.Link{
 		for (let [name, plugin] of this.plugins) {
 			plugins[name] = plugin.info.version;
 		}
-		link.messages.instanceInitialized.send(this, { plugins });
+		link.messages.instanceInitialized.send(this, { instance_id: this.config.get("instance.id"), plugins });
 	}
 
 	/**
@@ -340,6 +340,7 @@ class Instance extends link.Link{
 		}
 
 		this._running = true;
+		link.messages.instanceStarted.send(this, { instance_id: this.config.get("instance.id") });
 	}
 
 	/**
@@ -514,6 +515,7 @@ class InstanceConnection extends link.Link {
 		this.slave = slave;
 		this.instanceId = instanceId;
 		this.plugins = new Map();
+		this.status = "stopped";
 		link.attachAllMessages(this);
 
 		for (let pluginInfo of slave.pluginInfos) {
@@ -552,12 +554,21 @@ class InstanceConnection extends link.Link {
 		}
 	}
 
-	async instanceInitializedEventHandler(message) {
+	async instanceInitializedEventHandler(message, event) {
+		this.status = "initialized";
 		this.plugins = new Map(Object.entries(message.data.plugins));
+		this.forwardEventToMaster(message, event);
 	}
 
-	async instanceStoppedEventHandler(message) {
+	async instanceStartedEventHandler(message, event) {
+		this.status = "running";
+		this.forwardEventToMaster(message, event);
+	}
+
+	async instanceStoppedEventHandler(message, event) {
+		this.status = "stopped";
 		this.slave.instanceConnections.delete(this.instanceId);
+		this.forwardEventToMaster(message, event);
 	}
 }
 
@@ -616,6 +627,23 @@ class Slave extends link.Link {
 		this._disconnecting = false;
 		this._shuttingDown = false;
 
+		this.connector.on("connect", () => {
+			if (this._shuttingDown) {
+				return;
+			}
+
+			this.updateInstances().catch((err) => {
+				console.error("ERROR: Unexpected error updating instances");
+				console.error(err);
+				this.shutdown().catch((err) => {
+					setBlocking(true);
+					console.error("ERROR: Unexpected error during shutdown");
+					console.error(err);
+					process.exit(1)
+				});
+			});;
+		});
+
 		this.connector.on("close", () => {
 			if (this._shuttingDown) {
 				return;
@@ -835,18 +863,16 @@ class Slave extends link.Link {
 	async updateInstances() {
 		this.instanceInfos = await discoverInstances(this.config.get("slave.instances_directory"), console);
 		let list = [];
-		for (let instanceInfo of this.instanceInfos.values()) {
+		for (let [instanceId, instanceInfo] of this.instanceInfos) {
+			let instanceConnection = this.instanceConnections.get(instanceId);
 			list.push({
 				serialized_config: instanceInfo.config.serialize(),
+				status: instanceConnection ? instanceConnection.status : "stopped",
 			});
 		}
 		link.messages.updateInstances.send(this, { instances: list });
 	}
 
-	async start() {
-		await this.updateInstances();
-	}
-
 	async prepareDisconnectRequestHandler(message, request) {
 		this._disconnecting = true;
 		for (let instanceConnection of this.instanceConnections.values()) {
@@ -1089,7 +1115,6 @@ async function startSlave() {
 	});
 
 	await slaveConnector.connect();
-	await slave.start();
 
 	/*
 	} else if (command == "manage"){

test/lib/plugin.js

@@ -22,7 +22,7 @@ describe("lib/plugin", function() {
 			instancePlugin.onExit();
 			await instancePlugin.onOutput({});
 			instancePlugin.onMasterConnectionEvent("connect");
-			instancePlugin.onPrepareMasterDisconnect();
+			await instancePlugin.onPrepareMasterDisconnect();
 		})
 	});
 
@@ -33,10 +33,11 @@ describe("lib/plugin", function() {
 			await masterPlugin.init();
 		})
 		it("should define defaults for hooks", async function() {
+			await masterPlugin.onInstanceStatusChanged({}, "running", "initialized");
 			await masterPlugin.onMetrics();
 			await masterPlugin.onShutdown();
 			masterPlugin.onSlaveConnectionEvent({}, "connect");
-			masterPlugin.onPrepareSlaveDisconnect({});
+			await masterPlugin.onPrepareSlaveDisconnect({});
 		})
 	});