diff --git a/packages/ctl/ctl.js b/packages/ctl/ctl.js index 9e9e27d5..15241642 100755 --- a/packages/ctl/ctl.js +++ b/packages/ctl/ctl.js @@ -203,13 +203,13 @@ instanceConfigCommands.add(new libCommand.Command({ instanceCommands.add(instanceConfigCommands); instanceCommands.add(new libCommand.Command({ - definition: ["assign ", "Assign instance to a slave", (yargs) => { + definition: ["assign [slave]", "Assign instance to a slave", (yargs) => { yargs.positional("instance", { describe: "Instance to assign", type: "string" }); - yargs.positional("slave", { describe: "Slave to assign to", type: "string" }); + yargs.positional("slave", { describe: "Slave to assign to or unassign if none", type: "string" }); }], handler: async function(args, control) { let instanceId = await libCommand.resolveInstance(control, args.instance); - let slaveId = await libCommand.resolveSlave(control, args.slave); + let slaveId = args.slave ? await libCommand.resolveSlave(control, args.slave) : null; await libLink.messages.assignInstanceCommand.send(control, { instance_id: instanceId, slave_id: slaveId, diff --git a/packages/lib/link/messages.js b/packages/lib/link/messages.js index 36454012..b9e011eb 100644 --- a/packages/lib/link/messages.js +++ b/packages/lib/link/messages.js @@ -411,19 +411,10 @@ messages.assignInstanceCommand = new Request({ permission: "core.instance.assign", requestProperties: { "instance_id": { type: "integer" }, - "slave_id": { type: "integer" }, + "slave_id": { type: ["integer", "null"] }, }, }); -messages.assignInstance = new Request({ - type: "assign_instance", - links: ["master-slave"], - requestProperties: { - "serialized_config": { type: "object" }, - }, - forwardTo: "instance", -}); - messages.startInstance = new Request({ type: "start_instance", links: ["control-master", "master-slave", "slave-instance"], @@ -655,6 +646,44 @@ messages.deleteUser = new Request({ }, }); + +// Internal requests +messages.updateInstances = new Request({ + type: "update_instances", + links: ["slave-master"], + requestProperties: { + "instances": { + type: "array", + items: { + type: "object", + additionalProperties: false, + required: ["serialized_config", "status"], + properties: { + "serialized_config": { type: "object" }, + "status": { enum: [ + "stopped", "starting", "running", "creating_save", "exporting_data", + ]}, + }, + }, + }, + }, +}); + +messages.assignInstance = new Request({ + type: "assign_instance", + links: ["master-slave"], + requestProperties: { + "serialized_config": { type: "object" }, + }, + forwardTo: "instance", +}); + +messages.unassignInstance = new Request({ + type: "unassigne_instance", + links: ["master-slave"], + forwardTo: "instance", +}); + messages.getMetrics = new Request({ type: "get_metrics", links: ["master-slave", "slave-instance"], @@ -846,27 +875,6 @@ messages.instanceStatusChanged = new Event({ }, }); -messages.updateInstances = new Event({ - type: "update_instances", - links: ["slave-master"], - eventProperties: { - "instances": { - type: "array", - items: { - type: "object", - additionalProperties: false, - required: ["serialized_config", "status"], - properties: { - "serialized_config": { type: "object" }, - "status": { enum: [ - "stopped", "starting", "running", "creating_save", "exporting_data", - ]}, - }, - }, - }, - }, -}); - messages.instanceOutput = new Event({ type: "instance_output", links: ["instance-slave", "slave-master", "master-control"], diff --git a/packages/master/master.js b/packages/master/master.js index 367117e8..9aa24e2c 100755 --- a/packages/master/master.js +++ b/packages/master/master.js @@ -717,29 +717,43 @@ class ControlConnection extends BaseConnection { } async assignInstanceCommandRequestHandler(message, request) { - let instance = db.instances.get(message.data.instance_id); + let { slave_id, instance_id } = message.data; + let instance = db.instances.get(instance_id); if (!instance) { - throw new libErrors.RequestError(`Instance with ID ${message.data.instance_id} does not exist`); + throw new libErrors.RequestError(`Instance with ID ${instance_id} does not exist`); } - // XXX: Should the instance be stopped if it's running on another slave? - - let connection = slaveConnections.get(message.data.slave_id); - if (!connection) { - // The case of the slave not getting the assign instance message - // still have to be handled, so it's not a requirement that the - // target slave be connected to the master while doing the - // assignment, but it is IMHO a better user experience if this - // is the case. - throw new libErrors.RequestError("Target slave is not connected to the master server"); + // Check if target slave is connected + let newSlaveConnection; + if (slave_id !== null) { + newSlaveConnection = slaveConnections.get(slave_id); + if (!newSlaveConnection) { + // The case of the slave not getting the assign instance message + // still have to be handled, so it's not a requirement that the + // target slave be connected to the master while doing the + // assignment, but it is IMHO a better user experience if this + // is the case. + throw new libErrors.RequestError("Target slave is not connected to the master server"); + } } - instance.config.set("instance.assigned_slave", message.data.slave_id); + // Unassign from currently assigned slave if it is connected. + let currentAssignedSlave = instance.config.get("instance.assigned_slave"); + if (currentAssignedSlave !== null && slave_id !== currentAssignedSlave) { + let oldSlaveConnection = slaveConnections.get(currentAssignedSlave); + if (oldSlaveConnection && !oldSlaveConnection.connector.closing) { + await libLink.messages.unassignInstance.send(oldSlaveConnection, { instance_id }); + } + } - return await libLink.messages.assignInstance.send(connection, { - instance_id: instance.config.get("instance.id"), - serialized_config: instance.config.serialize(), - }); + // Assign to target + instance.config.set("instance.assigned_slave", slave_id); + if (slave_id !== null) { + await libLink.messages.assignInstance.send(newSlaveConnection, { + instance_id, + serialized_config: instance.config.serialize(), + }); + } } async setInstanceOutputSubscriptionsRequestHandler(message) { @@ -972,13 +986,23 @@ class SlaveConnection extends BaseConnection { async instanceStatusChangedEventHandler(message, event) { let instance = db.instances.get(message.data.instance_id); + + // It's possible to get updates from an instance that does not exist + // or is not assigned to the slave it originated from if it was + // reassigned or deleted while the connection to the slave it was + // originally on was down at the time. + if (!instance || instance.config.get("instance.assigned_slave") !== this._id) { + console.log(`Clusterio | Got bogus update for instance id ${message.data.instance_id}`); + return; + } + let prev = instance.status; instance.status = message.data.status; console.log(`Clusterio | Instance ${instance.config.get("instance.name")} State: ${instance.status}`); await libPlugin.invokeHook(masterPlugins, "onInstanceStatusChanged", instance, prev); } - async updateInstancesEventHandler(message) { + async updateInstancesRequestHandler(message) { // Push updated instance configs for (let instance of db.instances.values()) { if (instance.config.get("instance.assigned_slave") === this._id) { @@ -996,10 +1020,21 @@ class SlaveConnection extends BaseConnection { let masterInstance = db.instances.get(instanceConfig.get("instance.id")); if (masterInstance) { + // Check if this instance is assigned somewhere else. + if (masterInstance.config.get("instance.assigned_slave") !== this._id) { + await libLink.messages.unassignInstance.send(this, { + instance_id: masterInstance.config.get("instance.id"), + }); + continue; + } + // Already have this instance, update state instead if (masterInstance.status !== instance.status) { let prev = masterInstance.status; masterInstance.status = instance.status; + console.log( + `Clusterio | Instance ${instanceConfig.get("instance.name")} State: ${instance.status}` + ); await libPlugin.invokeHook(masterPlugins, "onInstanceStatusChanged", instance, prev); } continue; diff --git a/packages/slave/slave.js b/packages/slave/slave.js index 4e57fb73..ad3283bb 100755 --- a/packages/slave/slave.js +++ b/packages/slave/slave.js @@ -755,15 +755,16 @@ class Instance extends libLink.Link{ * Searches for instances in the provided directory * * Looks through all sub-dirs of the provided directory for valid - * instance definitions and updates the mapping of instance id to + * instance definitions and return a mapping of instance id to * instanceInfo objects. * - * @param {Map} instanceInfos - - * mapping between instance id and information about this instance. * @param {string} instancesDir - Directory containing instances * @param {Object} logger - console like logging interface. + * @returns {Map} + * mapping between instance id and information about this instance. */ -async function discoverInstances(instanceInfos, instancesDir, logger) { +async function discoverInstances(instancesDir, logger) { + let instanceInfos = new Map(); for (let entry of await fs.readdir(instancesDir, { withFileTypes: true })) { if (entry.isDirectory()) { let instanceConfig = new libConfig.InstanceConfig(); @@ -781,11 +782,6 @@ async function discoverInstances(instanceInfos, instancesDir, logger) { continue; } - // Ignore instances we have already discovered before. - if (instanceInfos.has(instanceConfig.get("instance.id"))) { - continue; - } - let instancePath = path.join(instancesDir, entry.name); logger.log(`found instance ${instanceConfig.get("instance.name")} in ${instancePath}`); instanceInfos.set(instanceConfig.get("instance.id"), { @@ -794,6 +790,7 @@ async function discoverInstances(instanceInfos, instancesDir, logger) { }); } } + return instanceInfos; } class InstanceConnection extends libLink.Link { @@ -915,6 +912,7 @@ class Slave extends libLink.Link { this.config = slaveConfig; this.instanceConnections = new Map(); + this.discoveredInstanceInfos = new Map(); this.instanceInfos = new Map(); this.adminlist = new Set(); @@ -1124,20 +1122,28 @@ class Slave extends libLink.Link { console.log(`Clusterio | Updated config for ${instanceInfo.path}`); } else { - let instanceConfig = new libConfig.InstanceConfig(); - await instanceConfig.load(serialized_config); + instanceInfo = this.discoveredInstanceInfos.get(instance_id); + if (instanceInfo) { + instanceInfo.config.update(serialized_config, true); - // XXX: race condition on multiple simultanious calls - let instanceDir = await this._findNewInstanceDir(instanceConfig.get("instance.name")); + } else { + let instanceConfig = new libConfig.InstanceConfig(); + await instanceConfig.load(serialized_config); + + // XXX: race condition on multiple simultanious calls + let instanceDir = await this._findNewInstanceDir(instanceConfig.get("instance.name")); + + await Instance.create(instanceDir, this.config.get("slave.factorio_directory")); + instanceInfo = { + path: instanceDir, + config: instanceConfig, + }; + + this.discoveredInstanceInfos.set(instance_id, instanceInfo); + } - await Instance.create(instanceDir, this.config.get("slave.factorio_directory")); - instanceInfo = { - path: instanceDir, - config: instanceConfig, - }; this.instanceInfos.set(instance_id, instanceInfo); - console.log(`Clusterio | assigned instance ${instanceConfig.get("instance.name")}`); - + console.log(`Clusterio | assigned instance ${instanceInfo.config.get("instance.name")}`); } // Somewhat hacky, but in the event of a lost session the status is @@ -1160,6 +1166,20 @@ class Slave extends libLink.Link { ); } + async unassignInstanceRequestHandler(message) { + let instanceId = message.data.instance_id; + let instanceInfo = this.instanceInfos.get(instanceId); + if (instanceInfo) { + let instanceConnection = this.instanceConnections.get(instanceId); + if (instanceConnection && ["starting", "running"].includes(instanceConnection.status)) { + await libLink.messages.stopInstance.send(instanceConnection, { instance_id: instanceId }); + } + + this.instanceInfos.delete(instanceId); + console.log(`Clusterio | unassigned instance ${instanceInfo.config.get("instance.name")}`); + } + } + /** * Initialize and connect an unloaded instance * @@ -1249,13 +1269,14 @@ class Slave extends libLink.Link { throw new libErrors.RequestError(`Instance with ID ${instanceId} is running`); } - let instanceInfo = this.instanceInfos.get(instanceId); + let instanceInfo = this.discoveredInstanceInfos.get(instanceId); if (!instanceInfo) { throw new libErrors.RequestError(`Instance with ID ${instanceId} does not exist`); } - await fs.remove(instanceInfo.path); + this.discoveredInstanceInfos.delete(instanceId); this.instanceInfos.delete(instanceId); + await fs.remove(instanceInfo.path); } /** @@ -1265,16 +1286,16 @@ class Slave extends libLink.Link { * the slave and master server with the new list of instances. */ async updateInstances() { - await discoverInstances(this.instanceInfos, this.config.get("slave.instances_directory"), console); + this.discoveredInstanceInfos = await discoverInstances(this.config.get("slave.instances_directory"), console); let list = []; - for (let [instanceId, instanceInfo] of this.instanceInfos) { + for (let [instanceId, instanceInfo] of this.discoveredInstanceInfos) { let instanceConnection = this.instanceConnections.get(instanceId); list.push({ serialized_config: instanceInfo.config.serialize(), status: instanceConnection ? instanceConnection.status : "stopped", }); } - libLink.messages.updateInstances.send(this, { instances: list }); + await libLink.messages.updateInstances.send(this, { instances: list }); } async prepareDisconnectRequestHandler(message, request) { diff --git a/test/slave.js b/test/slave.js index a14c40f1..47ff8f53 100644 --- a/test/slave.js +++ b/test/slave.js @@ -152,8 +152,7 @@ describe("Slave testing", function() { it("should discover test instance", async function() { let logger = { log: () => {}, error: () => {} }; let instancePath = path.join("test", "file", "instances"); - let instanceInfos = new Map(); - await slave._discoverInstances(instanceInfos, instancePath, logger); + let instanceInfos = await slave._discoverInstances(instancePath, logger); let referenceConfig = new libConfig.InstanceConfig(); await referenceConfig.init();