From 3211fbe9859fe540895648281e0ce22c044bb40d Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Mon, 11 May 2020 18:10:11 +0200 Subject: [PATCH] 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 | 18 +++++++++++++++--- lib/plugin.js | 15 +++++++++++++++ master.js | 44 ++++++++++++++++++++++++++++++++++++++++++- slave.js | 45 ++++++++++++++++++++++++++++++++++---------- test/lib/plugin.js | 5 +++-- 5 files changed, 111 insertions(+), 16 deletions(-) diff --git a/lib/link/messages.js b/lib/link/messages.js index 547d7470..f1fba2b5 100644 --- a/lib/link/messages.js +++ b/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"] }, }, }, }, diff --git a/lib/plugin.js b/lib/plugin.js index ed907ce5..81a66330 100644 --- a/lib/plugin.js +++ b/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 * diff --git a/master.js b/master.js index d6b558c7..f624d279 100644 --- a/master.js +++ b/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; } diff --git a/slave.js b/slave.js index f4ee2692..eabefd02 100644 --- a/slave.js +++ b/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"){ diff --git a/test/lib/plugin.js b/test/lib/plugin.js index d7e34a90..5e915e8a 100644 --- a/test/lib/plugin.js +++ b/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({}); }) });