SAN-3030 - Create rebuild worker

lib/models/mongo/context-version.js

@@ -1132,12 +1132,26 @@ ContextVersionSchema.methods.clearDockerHost = function (cb) {
  * @param {function} cb
  */
 ContextVersionSchema.methods.handleRecovery = function (cb) {
+ var logData = {
+ tx: true,
+ id: this._id
+ }
+ log.trace(logData, 'ContextVersionSchema.methods.handleRecovery')
  this.update({
  $set: {
  dockRemoved: false,
  dockRemovedNeedsUserConfirmation: this.dockRemoved
  }
- }, cb)
+ }, function (err) {
+ if (err) {
+ log.error(put({
+ err: err
+ }, logData), 'ContextVersionSchema.methods.handleRecovery - Error')
+ return cb(err)
+ }
+ log.trace(logData, 'ContextVersionSchema.methods.handleRecovery - Success')
+ cb(null)
+ })
 }
 
 ContextVersion = module.exports = mongoose.model('ContextVersions', ContextVersionSchema)

lib/models/mongo/instance.js

@@ -124,18 +124,72 @@ InstanceSchema.statics.findByBuild = function (build /* , args */) {
 }
 
 /**
- * finds all instances on the dock (based on dockerHost)
+ * finds all instances running or starting on the dock (based on dockerHost)
  * @param {String} dockerHost format: http://10.0.0.1:4242
  * @param {Function} cb (err, InstanceArray)
  */
-InstanceSchema.statics.findInstancesByDockerHost = function (dockerHost, cb) {
- log.info({
+InstanceSchema.statics.findInstancesRunningOrStartingByDockerHost = function (dockerHost, cb) {
+ var query = {
+ 'container.dockerHost': dockerHost,
+ $or: [
+ { 'container.inspect.State.Starting': true },
+ { 'container.inspect.State.Running': true }
+ ]
+ }
+ var logData = {
  tx: true,
- dockerHost: dockerHost
- }, 'InstanceSchema.statics.findInstancesByDockerHost findOneByContainerId')
- Instance.find({
- 'container.dockerHost': dockerHost
- }, cb)
+ dockerHost: dockerHost,
+ query: query
+ }
+ log.info(logData, 'InstanceSchema.statics.findInstancesRunningOrStartingByDockerHost')
+ Instance.find(query, function (err, instances) {
+ if (err) {
+ log.error(put({ err: err }, logData), 'findInstancesRunningOrStartingByDockerHost error')
+ return cb(err)
+ }
+ log.trace(put({ count: instances.length }, logData), 'findInstancesRunningOrStartingByDockerHost success')
+ cb(null, instances)
+ })
+}
+
+/**
+ * Find instances that are currently building on the dock.
+ * Building on dock means:
+ * - instance has no container
+ * - contextVersion.dockerHost should match
+ * - contextVersion.build.started should be set
+ * - contextVersion.build.failed & contextVersion.build.completed should not exist
+ */
+InstanceSchema.statics.findInstancesBuildingOnDockerHost = function (dockerHost, cb) {
+ var query = {
+ 'container': {
+ $exists: false
+ },
+ 'contextVersion.dockerHost': dockerHost,
+ 'contextVersion.build.started': {
+ $exists: true
+ },
+ 'contextVersion.build.failed': {
+ $exists: false
+ },
+ 'contextVersion.build.completed': {
+ $exists: false
+ }
+ }
+ var logData = {
+ tx: true,
+ dockerHost: dockerHost,
+ query: query
+ }
+ log.info(logData, 'InstanceSchema.statics.findInstancesBuildingOnDockerHost')
+ Instance.find(query, function (err, instances) {
+ if (err) {
+ log.error(put({ err: err }, logData), 'findInstancesBuildingOnDockerHost error')
+ return cb(err)
+ }
+ log.trace(put({ count: instances.length }, logData), 'findInstancesBuildingOnDockerHost success')
+ cb(null, instances)
+ })
 }
 
 /**

lib/models/mongo/schemas/instance.js

@@ -249,6 +249,12 @@ InstanceSchema.index({
  'container.dockerHost': 1
 })
 
+InstanceSchema.index({
+ 'container.dockerHost': 1,
+ 'container.inspect.State.Starting': 1,
+ 'container.inspect.State.Running': 1
+})
+
 InstanceSchema.index({ masterPod: 1, 'owner.github': 1 })
 InstanceSchema.index({ masterPod: 1, 'owner.github': 1, 'contextVersion.context': 1 })
 

lib/models/rabbitmq/index.js

@@ -61,6 +61,7 @@ RabbitMQ.prototype.connect = function (cb) {
  'delete-instance-container',
  'deploy-instance',
  'instance.container.redeploy',
+ 'instance.rebuild',
  'metis-github-event',
  'on-image-builder-container-create',
  'on-image-builder-container-die',
@@ -117,6 +118,7 @@ RabbitMQ.prototype.loadWorkers = function () {
  'create-instance-container': require('workers/create-instance-container'),
  'pull-instance-image': require('workers/pull-instance-image'),
  'dock.removed': require('workers/dock.removed'),
+ 'instance.rebuild': require('workers/instance.rebuild'),
  'instance.container.redeploy': require('workers/instance.container.redeploy')
  }
  // ponos: subscribe to queues
@@ -347,6 +349,22 @@ RabbitMQ.prototype.createImageBuilderContainer = function (data) {
  this.hermesClient.publish('create-image-builder-container', data)
 }
 
+/**
+ * create an instance.rebuild job
+ * @param {Object} data
+ */
+RabbitMQ.prototype.publishInstanceRebuild = function (data) {
+ log.info({
+ tx: true,
+ data: data
+ }, 'RabbitMQ.prototype.publishInstanceRebuild')
+ var requiredKeys = [
+ 'instanceId'
+ ]
+ this._validate(data, requiredKeys, 'instance.rebuild')
+ this.hermesClient.publish('instance.rebuild', data)
+}
+
 /**
  * create an cluster-provision job and insert it into queue
  * @param {Object} data

lib/workers/dock.removed.js

@@ -8,7 +8,6 @@
 
 require('loadenv')()
 
-var keypather = require('keypather')()
 var rabbitMQ = require('models/rabbitmq')
 var Promise = require('bluebird')
 var ContextVersion = require('models/mongo/context-version')
@@ -23,9 +22,10 @@ module.exports = DockRemovedWorker
 
 /**
  * Main handler for docker unhealthy event
- * Should redeploy all containers on unhealthy dock
  * Should mark the dock removed on every context version that runs on that dock
- * Should mark stopping instances as stopped on that dock since the instances are technically stopped now.
+ * Should mark stopping instances as stopped on that dock since the instances are technically stopped now
+ * Should redeploy all running or starting containers on unhealthy dock
+ * Should rebuild all building containers
  * @param {Object} job - Job info
  * @returns {Promise}
  */
@@ -38,6 +38,7 @@ function DockRemovedWorker (job) {
  host: joi.string().required(),
  githubId: joi.number()
  })
+
  return joi.validateOrBoomAsync(job, schema)
  .catch(function (err) {
  throw new TaskFatalError(err)
@@ -49,51 +50,127 @@ function DockRemovedWorker (job) {
  })
  })
  .then(function () {
- return Promise.fromCallback(function (cb) {
- Instance.setStoppingAsStoppedByDockerHost(job.host, cb)
- })
+ var tasks = [
+ DockRemovedWorker._stopStoppingInstances(job),
+ DockRemovedWorker._redeployAndNotify(job),
+ DockRemovedWorker._rebuildAndNotify(job)
+ ]
+ // those 3 can go in parallel
+ return Promise.all(tasks)
  })
- .then(function () {
- return Promise.fromCallback(function (cb) {
- Instance.findInstancesByDockerHost(job.host, cb)
- })
+}
+
+/**
+ * Mark all stopping instances as stopped
+ * @param {Object} job job data
+ * @returns {Promise}
+ */
+DockRemovedWorker._stopStoppingInstances = function (job) {
+ return Instance.setStoppingAsStoppedByDockerHostAsync(job.host)
+}
+
+/**
+ * Redeploy instances that should be redeployed and send notification for each of them
+ * @param {Object} job job data
+ * @returns {Promise}
+ */
+DockRemovedWorker._redeployAndNotify = function (job) {
+ return DockRemovedWorker._redeploy(job)
+ .then(function (instances) {
+ return DockRemovedWorker._updateFrontendInstances(instances)
+ })
+}
+
+/**
+ * Rebuild instances that should be rebuild and send notification for each of them
+ * @param {Object} job job data
+ * @returns {Promise}
+ */
+DockRemovedWorker._rebuildAndNotify = function (job) {
+ return DockRemovedWorker._rebuild(job)
+ .then(function (instances) {
+ return DockRemovedWorker._updateFrontendInstances(instances)
  })
+}
+
+/**
+ * Find all instances that should be redeployed (running or starting) and create new job for each of them
+ * @param {Object} job job data
+ * @returns {Promise}
+ */
+DockRemovedWorker._redeploy = function (job) {
+ var logData = {
+ tx: true,
+ job: job
+ }
+ log.info(logData, 'DockRemovedWorker._redeploy')
+ return Instance.findInstancesRunningOrStartingByDockerHostAsync(job.host)
  .then(function (instances) {
- log.trace(logData, 'DockRemovedWorker - Instances found ' + instances.length)
- if (instances.length > 0) {
- DockRemovedWorker._redeployContainers(instances)
- return DockRemovedWorker._updateFrontendInstances(instances)
- }
+ log.trace(logData, '_redeploy found instances to redeploy')
+ DockRemovedWorker._redeployContainers(instances)
+ return instances
  })
 }
 
 /**
- * should find running instances and submit job to redeploy them
- * @param {Array} instances array of instances on the dock
+ * should redeploy all instances
+ * @param {Array} instances instances array to redeploy
+ * @returns {Promise}
  */
 DockRemovedWorker._redeployContainers = function (instances) {
- instances = instances || []
- var instancesToRedeploy = instances.filter(function (instance) {
- var container = instance.container
- if (!container) {
- return false
- }
- return keypather.get(container, 'inspect.State.Running') === true
- })
- // we creating job for all instances right now
- // job will check internally if it's valid
- // if job failed validation it will quit
- // we can change that later on and fetch more data here and then create
- // jobs only for valid data
- instancesToRedeploy.forEach(function (instance) {
+ instances.forEach(function (instance) {
  rabbitMQ.redeployInstanceContainer({
  instanceId: instance._id,
  sessionUserGithubId: process.env.HELLO_RUNNABLE_GITHUB_ID
  })
  })
 }
 
+/**
+ * Find all instances that should be rebuild and create new job for each of them
+ * @param {Object} job job data
+ * @returns {Promise}
+ */
+DockRemovedWorker._rebuild = function (job) {
+ var logData = {
+ tx: true,
+ job: job
+ }
+ log.info(logData, 'DockRemovedWorker._rebuild')
+ return Instance.findInstancesBuildingOnDockerHostAsync(job.host)
+ .then(function (instances) {
+ log.trace(logData, '_rebuild found instances to rebuild')
+ DockRemovedWorker._rebuildInstances(instances)
+ return instances
+ })
+}
+
+/**
+ * should rebuild all instances with not completed builds (and not failed)
+ * @param {Array} instances array of instances to be rebuild
+ * @returns {Promise}
+ * @private
+ */
+DockRemovedWorker._rebuildInstances = function (instances) {
+ instances.forEach(function (instance) {
+ var payload = {
+ instanceId: instance._id
+ }
+ rabbitMQ.publishInstanceRebuild(payload)
+ })
+}
+
+/**
+ * send events to update frontend instances
+ * @param {Array} instances array of instances that were updated
+ * @returns {Promise}
+ * @private
+ */
 DockRemovedWorker._updateFrontendInstances = function (instances) {
+ log.info({
+ tx: true,
+ count: instances.length
+ }, 'DockRemovedWorker._updateFrontendInstances')
  return Promise.all(
  instances
  .map(function (instance) {

lib/workers/instance.container.redeploy.js

@@ -42,7 +42,6 @@ function InstanceContainerRedeployWorker (job) {
  tx: true,
  data: job
  }
-
  var schema = joi.object({
  instanceId: joi.string().required(),
  sessionUserGithubId: joi.number().required(),