MQTT: Upgrade container to Mosquitto 2.0 and fix listener bug #1071 (#…

…1074)

Co-authored-by: Pierre-Gilles Leymarie <pierregilles.leymarie@gmail.com>

server/lib/system/index.js

@@ -15,6 +15,7 @@ const { pull } = require('./system.pull');
 const { exec } = require('./system.exec');
 const { createContainer } = require('./system.createContainer');
 const { restartContainer } = require('./system.restartContainer');
+const { removeContainer } = require('./system.removeContainer');
 const { getNetworkMode } = require('./system.getNetworkMode');
 
 const { shutdown } = require('./system.shutdown');
@@ -45,6 +46,7 @@ System.prototype.pull = pull;
 System.prototype.exec = exec;
 System.prototype.createContainer = createContainer;
 System.prototype.restartContainer = restartContainer;
+System.prototype.removeContainer = removeContainer;
 System.prototype.getNetworkMode = getNetworkMode;
 
 System.prototype.shutdown = shutdown;

server/lib/system/system.getNetworkMode.js

@@ -1,8 +1,6 @@
 const get = require('get-value');
 const { PlatformNotCompatible } = require('../../utils/coreErrors');
-const getConfig = require('../../utils/getConfig');
-
-const config = getConfig();
+const { exec } = require('../../utils/childProcess');
 
 /**
  * @description Get Gladys network into Docker environment.
@@ -16,9 +14,11 @@ async function getNetworkMode() {
   }
 
   if (!this.networkMode) {
-    const containers = await this.dockerode.listContainers();
-    const gladysContainer = containers.find((c) => c.Image.startsWith(config.dockerImage));
-    this.networkMode = get(gladysContainer, 'HostConfig.NetworkMode', { default: 'unknown' });
+    const cmdResult = await exec('head -1 /proc/self/cgroup | cut -d/ -f3');
+    const [containerId] = cmdResult.split('\n');
+    const gladysContainer = this.dockerode.getContainer(containerId);
+    const gladysContainerInspect = await gladysContainer.inspect();
+    this.networkMode = get(gladysContainerInspect, 'HostConfig.NetworkMode', { default: 'unknown' });
   }
 
   return this.networkMode;

server/lib/system/system.removeContainer.js

@@ -0,0 +1,22 @@
+const { PlatformNotCompatible } = require('../../utils/coreErrors');
+
+/**
+ * @description Remove an Docker container.
+ * @param {string} containerId - Container id.
+ * @param {Object} options - Options for removal (see https://docs.docker.com/engine/api/v1.37/#operation/ContainerDelete).
+ * @returns {Promise} The removed container.
+ * @example
+ * await removeContainer(options);
+ */
+async function removeContainer(containerId, options = {}) {
+  if (!this.dockerode) {
+    throw new PlatformNotCompatible('SYSTEM_NOT_RUNNING_DOCKER');
+  }
+
+  const container = await this.dockerode.getContainer(containerId);
+  return container.remove(options);
+}
+
+module.exports = {
+  removeContainer,
+};

server/services/mqtt/docker/eclipse-mosquitto-container.json

@@ -1,6 +1,6 @@
 {
   "name": "eclipse-mosquitto",
-  "Image": "eclipse-mosquitto:latest",
+  "Image": "eclipse-mosquitto:2",
   "ExposedPorts": { "1883/tcp": {} },
   "HostConfig": {
     "Binds": ["/var/lib/gladysassistant/mosquitto:/mosquitto/config"],

server/services/mqtt/docker/eclipse-mosquitto-env.sh

@@ -28,5 +28,10 @@ else
   echo "eclipse-mosquitto configuration file already exists."
 fi
 
+# Check for breaking change
+if ! grep -q "listener 1883" "$mosquitto_config_file"; then
+  echo "listener 1883" >> $mosquitto_config_file
+fi
+
 # Create passwd file if not exists
 touch ${mosquitto_passwd_file}

server/services/mqtt/lib/constants.js

@@ -3,6 +3,7 @@ const CONFIGURATION = {
   MQTT_USERNAME_KEY: 'MQTT_USERNAME',
   MQTT_PASSWORD_KEY: 'MQTT_PASSWORD',
   MQTT_EMBEDDED_BROKER_KEY: 'MQTT_EMBEDDED_BROKER',
+  MQTT_MOSQUITTO_VERSION: 'MQTT_MOSQUITTO',
 };
 
 const DEFAULT = {
@@ -15,6 +16,7 @@ const DEFAULT = {
     IN_PROGRESS: 'IN_PROGRESS',
     ERROR: 'ERROR',
   },
+  MOSQUITTO_VERSION: '2',
 };
 
 module.exports = {

server/services/mqtt/lib/getConfiguration.js

@@ -17,6 +17,7 @@ async function getConfiguration() {
 
   let useEmbeddedBroker = false;
   let networkModeValid = false;
+  let mosquittoVersion = null;
 
   // Look for broker docker image
   if (dockerBased) {
@@ -36,6 +37,8 @@ async function getConfiguration() {
       },
     });
     brokerContainerAvailable = dockerImages.length > 0;
+
+    mosquittoVersion = await this.gladys.variable.getValue(CONFIGURATION.MQTT_MOSQUITTO_VERSION, this.serviceId);
   }
 
   return {
@@ -46,6 +49,7 @@ async function getConfiguration() {
     dockerBased,
     brokerContainerAvailable,
     networkModeValid,
+    mosquittoVersion,
   };
 }
 

server/services/mqtt/lib/index.js

@@ -10,6 +10,7 @@ const { status } = require('./status');
 const { getConfiguration } = require('./getConfiguration');
 const { saveConfiguration } = require('./saveConfiguration');
 const { installContainer } = require('./installContainer');
+const { updateContainer } = require('./updateContainer');
 const { checkDockerNetwork } = require('./checkDockerNetwork');
 const { setValue } = require('./setValue');
 
@@ -44,6 +45,7 @@ MqttHandler.prototype.status = status;
 MqttHandler.prototype.getConfiguration = getConfiguration;
 MqttHandler.prototype.saveConfiguration = saveConfiguration;
 MqttHandler.prototype.installContainer = installContainer;
+MqttHandler.prototype.updateContainer = updateContainer;
 MqttHandler.prototype.checkDockerNetwork = checkDockerNetwork;
 MqttHandler.prototype.setValue = setValue;
 

server/services/mqtt/lib/init.js

@@ -11,6 +11,10 @@ async function init() {
   });
 
   const configuration = await this.getConfiguration();
+
+  // Check for container configuration
+  await this.updateContainer(configuration);
+
   await this.connect(configuration);
 }
 

server/services/mqtt/lib/installContainer.js

@@ -9,13 +9,15 @@ const containerDescriptor = require('../docker/eclipse-mosquitto-container.json'
 
 /**
  * @description Get MQTT configuration.
+ * @param {boolean} saveConfiguration - Save new configuration.
  * @returns {Promise} Current MQTT network configuration.
  * @example
  * installContainer();
  */
-async function installContainer() {
+async function installContainer(saveConfiguration = true) {
   logger.info('MQTT broker is being installed as Docker container...');
 
+  let container;
   try {
     logger.info(`Check Gladys network...`);
     const networkModeValid = await this.checkDockerNetwork();
@@ -32,7 +34,7 @@ async function installContainer() {
     logger.trace(brokerEnv);
 
     logger.info(`Creating container...`);
-    const container = await this.gladys.system.createContainer(containerDescriptor);
+    container = await this.gladys.system.createContainer(containerDescriptor);
     logger.trace(container);
 
     logger.info('MQTT broker successfully installed as Docker container');
@@ -54,12 +56,16 @@ async function installContainer() {
     throw e;
   }
 
-  await this.saveConfiguration({
-    mqttUrl: 'mqtt://localhost',
-    mqttUsername: 'gladys',
-    mqttPassword: generate(20, { number: true, lowercase: true, uppercase: true }),
-    useEmbeddedBroker: true,
-  });
+  if (saveConfiguration) {
+    await this.saveConfiguration({
+      mqttUrl: 'mqtt://localhost',
+      mqttUsername: 'gladys',
+      mqttPassword: generate(20, { number: true, lowercase: true, uppercase: true }),
+      useEmbeddedBroker: true,
+    });
+  }
+
+  return container;
 }
 
 module.exports = {

server/services/mqtt/lib/saveConfiguration.js

@@ -1,5 +1,5 @@
 const { promisify } = require('util');
-const { CONFIGURATION } = require('./constants');
+const { CONFIGURATION, DEFAULT } = require('./constants');
 const { NotFoundError } = require('../../../utils/coreErrors');
 const containerParams = require('../docker/eclipse-mosquitto-container.json');
 
@@ -65,6 +65,13 @@ async function saveConfiguration({ mqttUrl, mqttUsername, mqttPassword, useEmbed
     await this.gladys.system.restartContainer(container.id);
     // wait 5 seconds for the container to restart
     await sleep(5 * 1000);
+
+    await updateOrDestroyVariable(
+      variable,
+      CONFIGURATION.MQTT_MOSQUITTO_VERSION,
+      DEFAULT.MOSQUITTO_VERSION,
+      this.serviceId,
+    );
   }
 
   return this.connect({ mqttUrl, mqttUsername, mqttPassword, useEmbeddedBroker });

server/services/mqtt/lib/updateContainer.js

@@ -0,0 +1,47 @@
+const logger = require('../../../utils/logger');
+const { CONFIGURATION, DEFAULT } = require('./constants');
+const containerParams = require('../docker/eclipse-mosquitto-container.json');
+
+/**
+ * @description Updates MQTT container configuration according to required changes.
+ * @param {Object} configuration - MQTT service configuration.
+ * @returns {Promise} Current MQTT network configuration.
+ * @example
+ * updateContainer({ mqttUrl, mqttPort });
+ */
+async function updateContainer(configuration) {
+  logger.info('MQTT: checking for required changes...');
+
+  // Check for port listener option
+  const { brokerContainerAvailable, mosquittoVersion } = configuration;
+  if (brokerContainerAvailable && !mosquittoVersion) {
+    logger.info('MQTT: update to mosquitto v2 required...');
+    const dockerContainers = await this.gladys.system.getContainers({
+      all: true,
+      filters: { name: [containerParams.name] },
+    });
+
+    // Remove non versionned container
+    if (dockerContainers.length !== 0) {
+      const [container] = dockerContainers;
+      await this.gladys.system.removeContainer(container.id, { force: true });
+    }
+
+    // Reinstall container with explicit version
+    const newContainer = await this.installContainer(false);
+    await this.gladys.system.restartContainer(newContainer.id);
+
+    await this.gladys.variable.setValue(
+      CONFIGURATION.MQTT_MOSQUITTO_VERSION,
+      DEFAULT.MOSQUITTO_VERSION,
+      this.serviceId,
+    );
+    logger.info('MQTT: update to mosquitto v2 done');
+  }
+
+  return configuration;
+}
+
+module.exports = {
+  updateContainer,
+};

server/test/lib/system/DockerodeMock.test.js

@@ -18,7 +18,9 @@ Docker.prototype.listImages = fake.resolves(images);
 Docker.prototype.createContainer = fake.resolves({ id: containers[0].Id });
 
 Docker.prototype.getContainer = fake.returns({
+  inspect: fake.resolves({ HostConfig: { NetworkMode: 'host' } }),
   restart: fake.resolves(true),
+  remove: fake.resolves(true),
   exec: ({ Cmd }) => {
     const mockedStream = new stream.Readable();
     return fake.resolves({

server/test/lib/system/system.getNetworkMode.test.js

@@ -8,8 +8,13 @@ const proxyquire = require('proxyquire').noCallThru();
 const { PlatformNotCompatible } = require('../../../utils/coreErrors');
 const DockerodeMock = require('./DockerodeMock.test');
 
+const getNetworkMode = proxyquire('../../../lib/system/system.getNetworkMode', {
+  '../../utils/childProcess': { exec: () => 'containerId' },
+});
+
 const System = proxyquire('../../../lib/system', {
   dockerode: DockerodeMock,
+  './system.getNetworkMode': getNetworkMode,
 });
 
 const sequelize = {
@@ -51,7 +56,8 @@ describe('system.getNetworkMode', () => {
   });
 
   it('should check network', async () => {
-    await system.getNetworkMode();
-    assert.calledOnce(system.dockerode.listContainers);
+    const network = await system.getNetworkMode();
+    expect(network).to.eq('host');
+    assert.calledOnce(system.dockerode.getContainer);
   });
 });

server/test/lib/system/system.removeContainer.test.js

@@ -0,0 +1,68 @@
+const { expect } = require('chai');
+const sinon = require('sinon');
+
+const { fake, assert } = sinon;
+
+const proxyquire = require('proxyquire').noCallThru();
+
+const { PlatformNotCompatible } = require('../../../utils/coreErrors');
+const DockerodeMock = require('./DockerodeMock.test');
+
+const System = proxyquire('../../../lib/system', {
+  dockerode: DockerodeMock,
+});
+
+const sequelize = {
+  close: fake.resolves(null),
+};
+
+const event = {
+  on: fake.resolves(null),
+  emit: fake.resolves(null),
+};
+
+const config = {
+  tempFolder: '/tmp/gladys',
+};
+
+describe('system.removeContainer', () => {
+  let system;
+
+  beforeEach(async () => {
+    system = new System(sequelize, event, config);
+    await system.init();
+    // Reset all fakes invoked within init call
+    sinon.reset();
+  });
+
+  afterEach(() => {
+    sinon.reset();
+  });
+
+  it('should failed as not on docker env', async () => {
+    system.dockerode = undefined;
+
+    try {
+      await system.removeContainer('my-container');
+      assert.fail('should have fail');
+    } catch (e) {
+      expect(e).be.instanceOf(PlatformNotCompatible);
+
+      assert.notCalled(sequelize.close);
+      assert.notCalled(event.on);
+      assert.notCalled(event.emit);
+    }
+  });
+
+  it('should removeContainer command with success', async () => {
+    const result = await system.removeContainer('my-container');
+
+    expect(result).to.be.eq(true);
+
+    assert.notCalled(sequelize.close);
+    assert.notCalled(event.on);
+    assert.notCalled(event.emit);
+
+    assert.calledOnce(system.dockerode.getContainer);
+  });
+});