mirror of
https://github.com/balena-os/balena-supervisor.git
synced 2025-04-24 21:09:47 +00:00
Proxyvisor: implement the Proxyvisor for the multicontainer supervisor
This will be quickly replaced by a newer version with a different API, but for now we needed to maintain backwards compatibility (see #508). This proxyvisor handles dependent apps and devices with a multicontainer parent app. It also switches to the new update mechanism by inferring and applying updates step by step. Signed-off-by: Pablo Carranza Velez <pablo@resin.io>
This commit is contained in:
Pablo Carranza Velez
parent
195697a7e1
commit
bc191ee86c
@ -1,28 +1,17 @@
|
||||
Promise = require 'bluebird'
|
||||
dockerUtils = require './docker-utils'
|
||||
{ docker } = dockerUtils
|
||||
_ = require 'lodash'
|
||||
express = require 'express'
|
||||
fs = Promise.promisifyAll require 'fs'
|
||||
{ resinApi, request } = require './request'
|
||||
knex = require './db'
|
||||
_ = require 'lodash'
|
||||
deviceRegister = require 'resin-register-device'
|
||||
randomHexString = require './lib/random-hex-string'
|
||||
utils = require './utils'
|
||||
device = require './device'
|
||||
{ request } = require './lib/request'
|
||||
constants = require './lib/constants'
|
||||
path = require 'path'
|
||||
mkdirp = Promise.promisify(require('mkdirp'))
|
||||
bodyParser = require 'body-parser'
|
||||
appConfig = require './config'
|
||||
PUBNUB = require 'pubnub'
|
||||
execAsync = Promise.promisify(require('child_process').exec)
|
||||
url = require 'url'
|
||||
|
||||
pubnub = PUBNUB.init(appConfig.pubnub)
|
||||
|
||||
isDefined = _.negate(_.isUndefined)
|
||||
|
||||
exports.router = router = express.Router()
|
||||
router.use(bodyParser())
|
||||
|
||||
parseDeviceFields = (device) ->
|
||||
device.id = parseInt(device.deviceId)
|
||||
device.appId = parseInt(device.appId)
|
||||
@ -32,71 +21,110 @@ parseDeviceFields = (device) ->
|
||||
device.targetEnvironment = JSON.parse(device.targetEnvironment ? '{}')
|
||||
return _.omit(device, 'markedForDeletion', 'logs_channel')
|
||||
|
||||
# TODO move to lib/validation
|
||||
validStringOrUndefined = (s) ->
|
||||
_.isUndefined(s) or !_.isEmpty(s)
|
||||
validObjectOrUndefined = (o) ->
|
||||
_.isUndefined(o) or _.isObject(o)
|
||||
|
||||
router.get '/v1/devices', (req, res) ->
|
||||
knex('dependentDevice').select()
|
||||
tarDirectory = (appId) ->
|
||||
return "/data/dependent-assets/#{appId}"
|
||||
|
||||
tarFilename = (appId, commit) ->
|
||||
return "#{appId}-#{commit}.tar"
|
||||
|
||||
tarPath = (appId, commit) ->
|
||||
return "#{tarDirectory(appId)}/#{tarFilename(appId, commit)}"
|
||||
|
||||
getTarArchive = (source, destination) ->
|
||||
fs.lstatAsync(destination)
|
||||
.catch ->
|
||||
mkdirp(path.dirname(destination))
|
||||
.then ->
|
||||
execAsync("tar -cvf '#{destination}' *", cwd: source)
|
||||
|
||||
cleanupTars = (appId, commit) ->
|
||||
if commit?
|
||||
fileToKeep = tarFilename(appId, commit)
|
||||
else
|
||||
fileToKeep = null
|
||||
dir = tarDirectory(appId)
|
||||
fs.readdirAsync(dir)
|
||||
.catchReturn([])
|
||||
.then (files) ->
|
||||
if fileToKeep?
|
||||
files = _.filter files, (file) ->
|
||||
return file isnt fileToKeep
|
||||
Promise.map files, (file) ->
|
||||
if !fileToKeep? or (file isnt fileToKeep)
|
||||
fs.unlinkAsync(path.join(dir, file))
|
||||
|
||||
formatTargetAsState = (device) ->
|
||||
return {
|
||||
appId: parseInt(device.appId)
|
||||
commit: device.targetCommit
|
||||
environment: device.targetEnvironment
|
||||
config: device.targetConfig
|
||||
}
|
||||
|
||||
formatCurrentAsState = (device) ->
|
||||
return {
|
||||
appId: parseInt(device.appId)
|
||||
commit: device.commit
|
||||
environment: device.environment
|
||||
config: device.config
|
||||
}
|
||||
|
||||
class ProxyvisorRouter
|
||||
constructor: (@proxyvisor) ->
|
||||
{ @config, @logger, @db, @docker } = @proxyvisor
|
||||
@router = express.Router()
|
||||
@router.use(bodyParser.urlencoded(extended: true))
|
||||
@router.use(bodyParser.json())
|
||||
@router.get '/v1/devices', (req, res) =>
|
||||
@db.models('dependentDevice').select()
|
||||
.map(parseDeviceFields)
|
||||
.then (devices) ->
|
||||
res.json(devices)
|
||||
.catch (err) ->
|
||||
res.status(503).send(err?.message or err or 'Unknown error')
|
||||
|
||||
router.post '/v1/devices', (req, res) ->
|
||||
@router.post '/v1/devices', (req, res) =>
|
||||
{ appId, device_type } = req.body
|
||||
|
||||
if !appId? or _.isNaN(parseInt(appId)) or parseInt(appId) <= 0
|
||||
res.status(400).send('appId must be a positive integer')
|
||||
return
|
||||
device_type = 'generic-amd64' if !device_type?
|
||||
|
||||
Promise.join(
|
||||
utils.getConfig('apiKey')
|
||||
utils.getConfig('userId')
|
||||
device.getID()
|
||||
randomHexString.generate()
|
||||
(apiKey, userId, deviceId, logsChannel) ->
|
||||
uuid = deviceRegister.generateUniqueKey()
|
||||
d =
|
||||
user: userId
|
||||
application: req.body.appId
|
||||
uuid: uuid
|
||||
device_type: device_type
|
||||
device: deviceId
|
||||
registered_at: Math.floor(Date.now() / 1000)
|
||||
logs_channel: logsChannel
|
||||
status: 'Provisioned'
|
||||
resinApi.post
|
||||
resource: 'device'
|
||||
body: d
|
||||
customOptions:
|
||||
apikey: apiKey
|
||||
.timeout(appConfig.apiTimeout)
|
||||
.then (dev) ->
|
||||
@proxyvisor.apiBinder.provisionDependentDevice(d)
|
||||
.then (dev) =>
|
||||
# If the response has id: null then something was wrong in the request
|
||||
# but we don't know precisely what.
|
||||
if !dev.id?
|
||||
res.status(400).send('Provisioning failed, invalid appId or credentials')
|
||||
return
|
||||
deviceForDB = {
|
||||
uuid: uuid
|
||||
appId: appId
|
||||
device_type: d.device_type
|
||||
uuid: dev.uuid
|
||||
appId
|
||||
device_type: dev.device_type
|
||||
deviceId: dev.id
|
||||
name: dev.name
|
||||
status: d.status
|
||||
logs_channel: d.logs_channel
|
||||
status: dev.status
|
||||
logs_channel: dev.logs_channel
|
||||
}
|
||||
knex('dependentDevice').insert(deviceForDB)
|
||||
@db.models('dependentDevice').insert(deviceForDB)
|
||||
.then ->
|
||||
res.status(201).send(dev)
|
||||
)
|
||||
.catch (err) ->
|
||||
console.error("Error on #{req.method} #{url.parse(req.url).pathname}", err, err.stack)
|
||||
res.status(503).send(err?.message or err or 'Unknown error')
|
||||
|
||||
router.get '/v1/devices/:uuid', (req, res) ->
|
||||
@router.get '/v1/devices/:uuid', (req, res) =>
|
||||
uuid = req.params.uuid
|
||||
knex('dependentDevice').select().where({ uuid })
|
||||
@db.models('dependentDevice').select().where({ uuid })
|
||||
.then ([ device ]) ->
|
||||
return res.status(404).send('Device not found') if !device?
|
||||
return res.status(410).send('Device deleted') if device.markedForDeletion
|
||||
@ -105,7 +133,7 @@ router.get '/v1/devices/:uuid', (req, res) ->
|
||||
console.error("Error on #{req.method} #{url.parse(req.url).pathname}", err, err.stack)
|
||||
res.status(503).send(err?.message or err or 'Unknown error')
|
||||
|
||||
router.post '/v1/devices/:uuid/logs', (req, res) ->
|
||||
@router.post '/v1/devices/:uuid/logs', (req, res) =>
|
||||
uuid = req.params.uuid
|
||||
m = {
|
||||
message: req.body.message
|
||||
@ -113,86 +141,74 @@ router.post '/v1/devices/:uuid/logs', (req, res) ->
|
||||
}
|
||||
m.isSystem = req.body.isSystem if req.body.isSystem?
|
||||
|
||||
knex('dependentDevice').select().where({ uuid })
|
||||
.then ([ device ]) ->
|
||||
@db.models('dependentDevice').select().where({ uuid })
|
||||
.then ([ device ]) =>
|
||||
return res.status(404).send('Device not found') if !device?
|
||||
return res.status(410).send('Device deleted') if device.markedForDeletion
|
||||
pubnub.publish({ channel: "device-#{device.logs_channel}-logs", message: m })
|
||||
@logger.log(m, { channel: "device-#{device.logs_channel}-logs" })
|
||||
res.status(202).send('OK')
|
||||
.catch (err) ->
|
||||
console.error("Error on #{req.method} #{url.parse(req.url).pathname}", err, err.stack)
|
||||
res.status(503).send(err?.message or err or 'Unknown error')
|
||||
|
||||
validStringOrUndefined = (s) ->
|
||||
_.isUndefined(s) or !_.isEmpty(s)
|
||||
validObjectOrUndefined = (o) ->
|
||||
_.isUndefined(o) or _.isObject(o)
|
||||
|
||||
router.put '/v1/devices/:uuid', (req, res) ->
|
||||
@router.put '/v1/devices/:uuid', (req, res) =>
|
||||
uuid = req.params.uuid
|
||||
{ status, is_online, commit, environment, config } = req.body
|
||||
{ status, is_online, commit, releaseId, environment, config } = req.body
|
||||
validateDeviceFields = ->
|
||||
if isDefined(is_online) and !_.isBoolean(is_online)
|
||||
res.status(400).send('is_online must be a boolean')
|
||||
return
|
||||
return 'is_online must be a boolean'
|
||||
if !validStringOrUndefined(status)
|
||||
res.status(400).send('status must be a non-empty string')
|
||||
return
|
||||
return 'status must be a non-empty string'
|
||||
if !validStringOrUndefined(commit)
|
||||
res.status(400).send('commit must be a non-empty string')
|
||||
return
|
||||
return 'commit must be a non-empty string'
|
||||
if !validStringOrUndefined(releaseId)
|
||||
return 'commit must be a non-empty string'
|
||||
if !validObjectOrUndefined(environment)
|
||||
res.status(400).send('environment must be an object')
|
||||
return
|
||||
return 'environment must be an object'
|
||||
if !validObjectOrUndefined(config)
|
||||
res.status(400).send('config must be an object')
|
||||
return 'config must be an object'
|
||||
return null
|
||||
requestError = validateDeviceFields()
|
||||
if requestError?
|
||||
res.status(400).send(requestError)
|
||||
return
|
||||
|
||||
environment = JSON.stringify(environment) if isDefined(environment)
|
||||
config = JSON.stringify(config) if isDefined(config)
|
||||
|
||||
fieldsToUpdateOnDB = _.pickBy({ status, is_online, commit, config, environment }, isDefined)
|
||||
fieldsToUpdateOnAPI = _.pick(fieldsToUpdateOnDB, 'status', 'is_online', 'commit')
|
||||
fieldsToUpdateOnDB = _.pickBy({ status, is_online, commit, releaseId, config, environment }, isDefined)
|
||||
fieldsToUpdateOnAPI = _.pick(fieldsToUpdateOnDB, 'status', 'is_online', 'commit', 'releaseId')
|
||||
|
||||
if _.isEmpty(fieldsToUpdateOnDB)
|
||||
res.status(400).send('At least one device attribute must be updated')
|
||||
return
|
||||
|
||||
Promise.join(
|
||||
utils.getConfig('apiKey')
|
||||
knex('dependentDevice').select().where({ uuid })
|
||||
(apiKey, [ device ]) ->
|
||||
throw new Error('apikey not found') if !apiKey?
|
||||
@db.models('dependentDevice').select().where({ uuid })
|
||||
.then ([ device ]) =>
|
||||
return res.status(404).send('Device not found') if !device?
|
||||
return res.status(410).send('Device deleted') if device.markedForDeletion
|
||||
throw new Error('Device is invalid') if !device.deviceId?
|
||||
Promise.try ->
|
||||
Promise.try =>
|
||||
if !_.isEmpty(fieldsToUpdateOnAPI)
|
||||
resinApi.patch
|
||||
resource: 'device'
|
||||
id: device.deviceId
|
||||
body: fieldsToUpdateOnAPI
|
||||
customOptions:
|
||||
apikey: apiKey
|
||||
.timeout(appConfig.apiTimeout)
|
||||
.then ->
|
||||
knex('dependentDevice').update(fieldsToUpdateOnDB).where({ uuid })
|
||||
.then ->
|
||||
@proxyvisor.apiBinder.patchDevice(device.deviceId, fieldsToUpdateOnAPI)
|
||||
.then =>
|
||||
@db.models('dependentDevice').update(fieldsToUpdateOnDB).where({ uuid })
|
||||
.then =>
|
||||
@db.models('dependentDevice').select().where({ uuid })
|
||||
.then ([ device ]) ->
|
||||
res.json(parseDeviceFields(device))
|
||||
)
|
||||
.catch (err) ->
|
||||
console.error("Error on #{req.method} #{url.parse(req.url).pathname}", err, err.stack)
|
||||
res.status(503).send(err?.message or err or 'Unknown error')
|
||||
|
||||
tarPath = ({ appId, commit }) ->
|
||||
return '/tmp/' + appId + '-' + commit + '.tar'
|
||||
|
||||
router.get '/v1/dependent-apps/:appId/assets/:commit', (req, res) ->
|
||||
knex('dependentApp').select().where(_.pick(req.params, 'appId', 'commit'))
|
||||
.then ([ app ]) ->
|
||||
@router.get '/v1/dependent-apps/:appId/assets/:commit', (req, res) =>
|
||||
@db.models('dependentApp').select().where(_.pick(req.params, 'appId', 'commit'))
|
||||
.then ([ app ]) =>
|
||||
return res.status(404).send('Not found') if !app
|
||||
dest = tarPath(app)
|
||||
dest = tarPath(app.appId, app.commit)
|
||||
fs.lstatAsync(dest)
|
||||
.catch ->
|
||||
Promise.using docker.imageRootDirMounted(app.imageId), (rootDir) ->
|
||||
.catch =>
|
||||
Promise.using @docker.imageRootDirMounted(app.image), (rootDir) ->
|
||||
getTarArchive(rootDir + '/assets', dest)
|
||||
.then ->
|
||||
res.sendFile(dest)
|
||||
@ -200,8 +216,8 @@ router.get '/v1/dependent-apps/:appId/assets/:commit', (req, res) ->
|
||||
console.error("Error on #{req.method} #{url.parse(req.url).pathname}", err, err.stack)
|
||||
res.status(503).send(err?.message or err or 'Unknown error')
|
||||
|
||||
router.get '/v1/dependent-apps', (req, res) ->
|
||||
knex('dependentApp').select()
|
||||
@router.get '/v1/dependent-apps', (req, res) =>
|
||||
@db.models('dependentApp').select()
|
||||
.map (app) ->
|
||||
return {
|
||||
id: parseInt(app.appId)
|
||||
@ -215,81 +231,41 @@ router.get '/v1/dependent-apps', (req, res) ->
|
||||
console.error("Error on #{req.method} #{url.parse(req.url).pathname}", err, err.stack)
|
||||
res.status(503).send(err?.message or err or 'Unknown error')
|
||||
|
||||
getTarArchive = (path, destination) ->
|
||||
fs.lstatAsync(path)
|
||||
.then ->
|
||||
execAsync("tar -cvf '#{destination}' *", cwd: path)
|
||||
module.exports = class Proxyvisor
|
||||
constructor: ({ @config, @logger, @db, @docker, @images, @applications }) ->
|
||||
@acknowledgedState = {}
|
||||
@lastRequestForDevice = {}
|
||||
@_router = new ProxyvisorRouter(this)
|
||||
@router = @_router.router
|
||||
@validActions = [ 'updateDependentTargets', 'sendDependentHooks', 'removeDependentApp' ]
|
||||
|
||||
# TODO: deduplicate code from compareForUpdate in application.coffee
|
||||
exports.fetchAndSetTargetsForDependentApps = (state, fetchFn, apiKey) ->
|
||||
knex('dependentApp').select()
|
||||
.then (localDependentApps) ->
|
||||
# Compare to see which to fetch, and which to delete
|
||||
remoteApps = _.mapValues state.apps, (app, appId) ->
|
||||
conf = app.config ? {}
|
||||
return {
|
||||
appId: appId
|
||||
parentAppId: app.parentApp
|
||||
imageId: app.image
|
||||
commit: app.commit
|
||||
config: JSON.stringify(conf)
|
||||
name: app.name
|
||||
}
|
||||
localApps = _.keyBy(localDependentApps, 'appId')
|
||||
bindToAPI: (apiBinder) =>
|
||||
@apiBinder = apiBinder
|
||||
|
||||
toBeDownloaded = _.filter remoteApps, (app, appId) ->
|
||||
return app.commit? and app.imageId? and !_.some(localApps, imageId: app.imageId)
|
||||
toBeRemoved = _.filter localApps, (app, appId) ->
|
||||
return app.commit? and !_.some(remoteApps, imageId: app.imageId)
|
||||
toBeDeletedFromDB = _(localApps).reject((app, appId) -> remoteApps[appId]?).map('appId').value()
|
||||
Promise.map toBeDownloaded, (app) ->
|
||||
deltaSource = null
|
||||
if localApps[app.appId]?
|
||||
deltaSource = localApps[app.appId].imageId
|
||||
else if !_.isEmpty(localDependentApps)
|
||||
deltaSource = localDependentApps[0].imageId
|
||||
fetchFn(app, { deltaSource, setDeviceUpdateState: false })
|
||||
.then ->
|
||||
Promise.map toBeRemoved, (app) ->
|
||||
fs.unlinkAsync(tarPath(app))
|
||||
.then ->
|
||||
docker.getImage(app.imageId).remove()
|
||||
.catch (err) ->
|
||||
console.error('Could not remove image/artifacts for dependent app', err, err.stack)
|
||||
.then ->
|
||||
Promise.props(
|
||||
_.mapValues remoteApps, (app, appId) ->
|
||||
knex('dependentApp').update(app).where({ appId })
|
||||
.then (n) ->
|
||||
knex('dependentApp').insert(app) if n == 0
|
||||
)
|
||||
.then ->
|
||||
knex('dependentDevice').del().whereIn('appId', toBeDeletedFromDB)
|
||||
.then ->
|
||||
knex('dependentApp').del().whereIn('appId', toBeDeletedFromDB)
|
||||
.then ->
|
||||
knex('dependentDevice').update({ markedForDeletion: true }).whereNotIn('uuid', _.keys(state.devices))
|
||||
.then ->
|
||||
Promise.all _.map state.devices, (device, uuid) ->
|
||||
executeStepAction: (step) =>
|
||||
Promise.try =>
|
||||
actions = {
|
||||
updateDependentTargets: =>
|
||||
@config.getMany([ 'currentApiKey', 'apiTimeout' ])
|
||||
.then ({ currentApiKey, apiTimeout }) =>
|
||||
# - take each of the step.devices and update dependentDevice with it (targetCommit, targetEnvironment, targetConfig)
|
||||
# - if update returns 0, then use APIBinder to fetch the device, then store it to the db
|
||||
# - set markedForDeletion: true for devices that are not in the step.devices list
|
||||
# - update dependentApp with step.app
|
||||
Promise.map step.devices, (device) =>
|
||||
uuid = device.uuid
|
||||
# Only consider one app per dependent device for now
|
||||
appId = _(device.apps).keys().head()
|
||||
targetCommit = state.apps[appId].commit
|
||||
targetEnvironment = JSON.stringify(device.apps[appId].environment ? {})
|
||||
targetConfig = JSON.stringify(device.apps[appId].config ? {})
|
||||
knex('dependentDevice').update({ targetEnvironment, targetConfig, targetCommit, name: device.name }).where({ uuid })
|
||||
.then (n) ->
|
||||
targetCommit = device.apps[appId].commit
|
||||
targetEnvironment = JSON.stringify(device.apps[appId].environment)
|
||||
targetConfig = JSON.stringify(device.apps[appId].config)
|
||||
@db.models('dependentDevice').update({ appId, targetEnvironment, targetConfig, targetCommit, name: device.name }).where({ uuid })
|
||||
.then (n) =>
|
||||
return if n != 0
|
||||
# If the device is not in the DB it means it was provisioned externally
|
||||
# so we need to fetch it.
|
||||
resinApi.get
|
||||
resource: 'device'
|
||||
options:
|
||||
filter:
|
||||
uuid: uuid
|
||||
customOptions:
|
||||
apikey: apiKey
|
||||
.timeout(appConfig.apiTimeout)
|
||||
.then ([ dev ]) ->
|
||||
@apiBinder.fetchDevice(uuid, currentApiKey, apiTimeout)
|
||||
.then (dev) =>
|
||||
deviceForDB = {
|
||||
uuid: uuid
|
||||
appId: appId
|
||||
@ -303,74 +279,354 @@ exports.fetchAndSetTargetsForDependentApps = (state, fetchFn, apiKey) ->
|
||||
targetConfig
|
||||
targetEnvironment
|
||||
}
|
||||
knex('dependentDevice').insert(deviceForDB)
|
||||
.catch (err) ->
|
||||
console.error('Error fetching dependent apps', err, err.stack)
|
||||
@db.models('dependentDevice').insert(deviceForDB)
|
||||
.then =>
|
||||
@db.models('dependentDevice').where({ appId: step.appId }).whereNotIn('uuid', _.map(step.devices, 'uuid')).update({ markedForDeletion: true })
|
||||
.then =>
|
||||
@normaliseDependentAppForDB(step.app)
|
||||
.then (appForDB) =>
|
||||
@db.upsertModel('dependentApp', appForDB, { appId: step.appId })
|
||||
.then ->
|
||||
cleanupTars(step.appId, step.app.commit)
|
||||
|
||||
getHookEndpoint = (appId) ->
|
||||
knex('dependentApp').select('parentAppId').where({ appId })
|
||||
.then ([ { parentAppId } ]) ->
|
||||
utils.getKnexApp(parentAppId)
|
||||
.then (parentApp) ->
|
||||
conf = JSON.parse(parentApp.config)
|
||||
dockerUtils.getImageEnv(parentApp.imageId)
|
||||
.then (imageEnv) ->
|
||||
return imageEnv.RESIN_DEPENDENT_DEVICES_HOOK_ADDRESS ?
|
||||
conf.RESIN_DEPENDENT_DEVICES_HOOK_ADDRESS ?
|
||||
"#{appConfig.proxyvisorHookReceiver}/v1/devices/"
|
||||
sendDependentHooks: =>
|
||||
Promise.join(
|
||||
@config.get('apiTimeout')
|
||||
@getHookEndpoint(step.appId)
|
||||
(apiTimeout, endpoint) =>
|
||||
Promise.mapSeries step.devices, (device) =>
|
||||
Promise.try =>
|
||||
if @lastRequestForDevice[device.uuid]?
|
||||
diff = Date.now() - @lastRequestForDevice[device.uuid]
|
||||
if diff < 30000
|
||||
Promise.delay(30001 - diff)
|
||||
.then =>
|
||||
@lastRequestForDevice[device.uuid] = Date.now()
|
||||
if device.markedForDeletion
|
||||
@sendDeleteHook(device, apiTimeout, endpoint)
|
||||
else
|
||||
@sendUpdate(device, apiTimeout, endpoint)
|
||||
)
|
||||
|
||||
formatTargetAsState = (device) ->
|
||||
removeDependentApp: =>
|
||||
# find step.app and delete it from the DB
|
||||
# find devices with step.appId and delete them from the DB
|
||||
@db.transaction (trx) ->
|
||||
trx('dependentApp').where({ appId: step.appId }).del()
|
||||
.then ->
|
||||
trx('dependentDevice').where({ appId: step.appId }).del()
|
||||
.then ->
|
||||
cleanupTars(step.appId)
|
||||
|
||||
}
|
||||
throw new Error("Invalid proxyvisor action #{step.action}") if !actions[step.action]?
|
||||
actions[step.action]()
|
||||
|
||||
getCurrentStates: =>
|
||||
Promise.join(
|
||||
@db.models('dependentApp').select().map(@normaliseDependentAppFromDB)
|
||||
@db.models('dependentDevice').select()
|
||||
(apps, devicesFromDB) ->
|
||||
devices = _.map devicesFromDB, (device) ->
|
||||
dev = {
|
||||
uuid: device.uuid
|
||||
name: device.name
|
||||
lock_expiry_date: device.lock_expiry_date
|
||||
markedForDeletion: device.markedForDeletion
|
||||
apps: {}
|
||||
}
|
||||
dev.apps[device.appId] = {
|
||||
commit: device.commit
|
||||
config: JSON.parse(device.config)
|
||||
environment: JSON.parse(device.environment)
|
||||
targetCommit: device.targetCommit
|
||||
targetEnvironment: JSON.parse(device.targetEnvironment)
|
||||
targetConfig: JSON.parse(device.targetConfig)
|
||||
}
|
||||
return dev
|
||||
return { apps, devices }
|
||||
)
|
||||
|
||||
normaliseDependentAppForDB: (app) =>
|
||||
if app.image?
|
||||
image = @images.normalise(app.image)
|
||||
else
|
||||
image = null
|
||||
dbApp = {
|
||||
appId: app.appId
|
||||
name: app.name
|
||||
commit: app.commit
|
||||
releaseId: app.releaseId
|
||||
parentApp: app.parentApp
|
||||
image: image
|
||||
config: JSON.stringify(app.config ? {})
|
||||
environment: JSON.stringify(app.environment ? {})
|
||||
}
|
||||
return Promise.props(dbApp)
|
||||
|
||||
normaliseDependentDeviceTargetForDB: (device, appCommit) ->
|
||||
Promise.try ->
|
||||
apps = _.clone(device.apps ? {})
|
||||
_.forEach apps, (app) ->
|
||||
app.commit ?= appCommit
|
||||
app.config ?= {}
|
||||
app.environment ?= {}
|
||||
apps = JSON.stringify(apps)
|
||||
outDevice = {
|
||||
uuid: device.uuid
|
||||
name: device.name
|
||||
apps
|
||||
}
|
||||
return outDevice
|
||||
|
||||
setTargetInTransaction: (dependent, trx) =>
|
||||
Promise.try =>
|
||||
if dependent?.apps?
|
||||
appsArray = _.map dependent.apps, (app, appId) ->
|
||||
appClone = _.clone(app)
|
||||
appClone.appId = appId
|
||||
return appClone
|
||||
Promise.map(appsArray, @normaliseDependentAppForDB)
|
||||
.then (appsForDB) =>
|
||||
Promise.map appsForDB, (app) =>
|
||||
@db.upsertModel('dependentAppTarget', app, { appId: app.appId }, trx)
|
||||
.then ->
|
||||
trx('dependentAppTarget').whereNotIn('appId', _.map(appsForDB, 'appId')).del()
|
||||
.then =>
|
||||
if dependent?.devices?
|
||||
devicesArray = _.map dependent.devices, (dev, uuid) ->
|
||||
devClone = _.clone(dev)
|
||||
devClone.uuid = uuid
|
||||
return devClone
|
||||
Promise.map devicesArray, (device) =>
|
||||
appId = _.keys(device.apps)[0]
|
||||
@normaliseDependentDeviceTargetForDB(device, dependent.apps[appId]?.commit)
|
||||
.then (devicesForDB) =>
|
||||
Promise.map devicesForDB, (device) =>
|
||||
@db.upsertModel('dependentDeviceTarget', device, { uuid: device.uuid }, trx)
|
||||
.then ->
|
||||
trx('dependentDeviceTarget').whereNotIn('uuid', _.map(devicesForDB, 'uuid')).del()
|
||||
|
||||
normaliseDependentAppFromDB: (app) ->
|
||||
Promise.try ->
|
||||
outApp = {
|
||||
appId: app.appId
|
||||
name: app.name
|
||||
commit: app.commit
|
||||
releaseId: app.releaseId
|
||||
image: app.image
|
||||
config: JSON.parse(app.config)
|
||||
environment: JSON.parse(app.environment)
|
||||
parentApp: app.parentApp
|
||||
}
|
||||
return outApp
|
||||
|
||||
normaliseDependentDeviceTargetFromDB: (device) ->
|
||||
Promise.try ->
|
||||
outDevice = {
|
||||
uuid: device.uuid
|
||||
name: device.name
|
||||
apps: JSON.parse(device.apps)
|
||||
}
|
||||
return outDevice
|
||||
|
||||
normaliseDependentDeviceFromDB: (device) ->
|
||||
Promise.try ->
|
||||
outDevice = _.clone(device)
|
||||
_.forEach [ 'environment', 'config', 'targetEnvironment', 'targetConfig' ], (prop) ->
|
||||
outDevice[prop] = JSON.parse(device[prop])
|
||||
return outDevice
|
||||
|
||||
getTarget: =>
|
||||
Promise.props({
|
||||
apps: @db.models('dependentAppTarget').select().map(@normaliseDependentAppFromDB)
|
||||
devices: @db.models('dependentDeviceTarget').select().map(@normaliseDependentDeviceTargetFromDB)
|
||||
})
|
||||
|
||||
imagesInUse: (current, target) ->
|
||||
images = []
|
||||
if current.dependent?.apps?
|
||||
_.forEach current.dependent.apps, (app) ->
|
||||
images.push app.image
|
||||
if target.dependent?.apps?
|
||||
_.forEach target.dependent.apps, (app) ->
|
||||
images.push app.image
|
||||
return images
|
||||
|
||||
_imageAvailable: (image, available) ->
|
||||
_.some(available, (availableImage) -> availableImage.name == image)
|
||||
|
||||
_getHookStep: (currentDevices, appId) =>
|
||||
hookStep = {
|
||||
action: 'sendDependentHooks'
|
||||
devices: []
|
||||
appId
|
||||
}
|
||||
_.forEach currentDevices, (device) =>
|
||||
if device.markedForDeletion
|
||||
hookStep.devices.push({
|
||||
uuid: device.uuid
|
||||
markedForDeletion: true
|
||||
})
|
||||
else
|
||||
targetState = {
|
||||
appId
|
||||
commit: device.apps[appId].targetCommit
|
||||
config: device.apps[appId].targetConfig
|
||||
environment: device.apps[appId].targetEnvironment
|
||||
}
|
||||
currentState = {
|
||||
appId
|
||||
commit: device.apps[appId].commit
|
||||
config: device.apps[appId].config
|
||||
environment: device.apps[appId].environment
|
||||
}
|
||||
if device.apps[appId].targetCommit? and !_.isEqual(targetState, currentState) and !_.isEqual(targetState, @acknowledgedState[device.uuid])
|
||||
hookStep.devices.push({
|
||||
uuid: device.uuid
|
||||
target: targetState
|
||||
})
|
||||
return hookStep
|
||||
|
||||
_compareDevices: (currentDevices, targetDevices, appId) ->
|
||||
currentDeviceTargets = _.map currentDevices, (dev) ->
|
||||
return null if dev.markedForDeletion
|
||||
devTarget = _.clone(dev)
|
||||
delete devTarget.markedForDeletion
|
||||
devTarget.apps = {}
|
||||
devTarget.apps[appId] = {
|
||||
commit: dev.apps[appId].targetCommit
|
||||
environment: dev.apps[appId].targetEnvironment
|
||||
config: dev.apps[appId].targetConfig
|
||||
}
|
||||
return devTarget
|
||||
currentDeviceTargets = _.filter(currentDeviceTargets, (dev) -> !_.isNull(dev))
|
||||
return !_.isEmpty(_.xorWith(currentDeviceTargets, targetDevices, _.isEqual))
|
||||
|
||||
imageForDependentApp: (app) ->
|
||||
return {
|
||||
commit: device.targetCommit
|
||||
environment: device.targetEnvironment
|
||||
config: device.targetConfig
|
||||
name: app.image
|
||||
imageId: app.imageId
|
||||
appId: app.appId
|
||||
dependent: true
|
||||
}
|
||||
|
||||
do ->
|
||||
acknowledgedState = {}
|
||||
sendUpdate = (device, endpoint) ->
|
||||
stateToSend = {
|
||||
appId: parseInt(device.appId)
|
||||
commit: device.targetCommit
|
||||
environment: JSON.parse(device.targetEnvironment)
|
||||
config: JSON.parse(device.targetConfig)
|
||||
}
|
||||
getRequiredSteps: (availableImages, current, target, stepsInProgress) =>
|
||||
steps = []
|
||||
Promise.try =>
|
||||
targetApps = _.keyBy(target.dependent?.apps ? [], 'appId')
|
||||
targetAppIds = _.keys(targetApps)
|
||||
currentApps = _.keyBy(current.dependent?.apps ? [], 'appId')
|
||||
currentAppIds = _.keys(currentApps)
|
||||
allAppIds = _.union(targetAppIds, currentAppIds)
|
||||
|
||||
toBeDownloaded = _.filter targetAppIds, (appId) =>
|
||||
return targetApps[appId].commit? and targetApps[appId].image? and !@_imageAvailable(targetApps[appId].image, availableImages)
|
||||
|
||||
appIdsToCheck = _.filter allAppIds, (appId) ->
|
||||
# - if a step is in progress for this appId, ignore
|
||||
!_.some(steps.concat(stepsInProgress), (step) -> step.appId == appId)
|
||||
_.forEach appIdsToCheck, (appId) =>
|
||||
# - if there's current but not target, push a removeDependentApp step
|
||||
if !targetApps[appId]?
|
||||
steps.push({
|
||||
action: 'removeDependentApp'
|
||||
appId
|
||||
})
|
||||
return
|
||||
|
||||
# - if toBeDownloaded includes this app, push a fetch step
|
||||
if _.includes(toBeDownloaded, appId)
|
||||
steps.push({
|
||||
action: 'fetch'
|
||||
appId
|
||||
image: @imageForDependentApp(targetApps[appId])
|
||||
})
|
||||
return
|
||||
|
||||
devicesForApp = (devices) ->
|
||||
_.filter devices, (d) ->
|
||||
_.includes(_.keys(d.apps), appId)
|
||||
|
||||
currentDevices = devicesForApp(current.dependent.devices)
|
||||
targetDevices = devicesForApp(target.dependent.devices)
|
||||
|
||||
devicesDiffer = @_compareDevices(currentDevices, targetDevices, appId)
|
||||
# - if current doesn't match target, or the devices differ, push an updateDependentTargets step
|
||||
if !_.isEqual(currentApps[appId], targetApps[appId]) or devicesDiffer
|
||||
steps.push({
|
||||
action: 'updateDependentTargets'
|
||||
devices: targetDevices
|
||||
app: targetApps[appId]
|
||||
appId
|
||||
})
|
||||
return
|
||||
|
||||
# if we got to this point, the current app is up to date and devices have the
|
||||
# correct targetCommit, targetEnvironment and targetConfig.
|
||||
hookStep = @_getHookStep(currentDevices, appId)
|
||||
if !_.isEmpty(hookStep.devices)
|
||||
steps.push(hookStep)
|
||||
.then ->
|
||||
return steps
|
||||
|
||||
getHookEndpoint: (appId) =>
|
||||
@db.models('dependentApp').select('parentApp').where({ appId })
|
||||
.then ([ { parentApp } ]) =>
|
||||
@applications.getTargetApp(parentApp)
|
||||
.then (parentApp) =>
|
||||
Promise.map parentApp?.services ? [], (service) =>
|
||||
@docker.getImageEnv(service.image)
|
||||
.then (imageEnv) ->
|
||||
return imageEnv.RESIN_DEPENDENT_DEVICES_HOOK_ADDRESS
|
||||
.then (imageHookAddresses) ->
|
||||
for addr in imageHookAddresses
|
||||
return addr if addr?
|
||||
return parentApp?.config?.RESIN_DEPENDENT_DEVICES_HOOK_ADDRESS ?
|
||||
"#{constants.proxyvisorHookReceiver}/v1/devices/"
|
||||
|
||||
sendUpdate: (device, timeout, endpoint) =>
|
||||
request.putAsync "#{endpoint}#{device.uuid}", {
|
||||
json: true
|
||||
body: stateToSend
|
||||
body: device.target
|
||||
}
|
||||
.timeout(appConfig.apiTimeout)
|
||||
.spread (response, body) ->
|
||||
.timeout(timeout)
|
||||
.spread (response, body) =>
|
||||
if response.statusCode == 200
|
||||
acknowledgedState[device.uuid] = formatTargetAsState(device)
|
||||
@acknowledgedState[device.uuid] = device.target
|
||||
else
|
||||
acknowledgedState[device.uuid] = null
|
||||
@acknowledgedState[device.uuid] = null
|
||||
throw new Error("Hook returned #{response.statusCode}: #{body}") if response.statusCode != 202
|
||||
.catch (err) ->
|
||||
return console.error("Error updating device #{device.uuid}", err, err.stack)
|
||||
|
||||
sendDeleteHook = (device, endpoint) ->
|
||||
uuid = device.uuid
|
||||
sendDeleteHook: ({ uuid }, timeout, endpoint) =>
|
||||
request.delAsync("#{endpoint}#{uuid}")
|
||||
.timeout(appConfig.apiTimeout)
|
||||
.spread (response, body) ->
|
||||
.timeout(timeout)
|
||||
.spread (response, body) =>
|
||||
if response.statusCode == 200
|
||||
knex('dependentDevice').del().where({ uuid })
|
||||
@db.models('dependentDevice').del().where({ uuid })
|
||||
else
|
||||
throw new Error("Hook returned #{response.statusCode}: #{body}")
|
||||
.catch (err) ->
|
||||
return console.error("Error deleting device #{device.uuid}", err, err.stack)
|
||||
return console.error("Error deleting device #{uuid}", err, err.stack)
|
||||
|
||||
exports.sendUpdates = ->
|
||||
endpoints = {}
|
||||
knex('dependentDevice').select()
|
||||
.map (device) ->
|
||||
currentState = _.pick(device, 'commit', 'environment', 'config')
|
||||
sendUpdates: ({ uuid }) =>
|
||||
Promise.join(
|
||||
@db.models('dependentDevice').where({ uuid }).select()
|
||||
@config.get('apiTimeout')
|
||||
([ dev ], apiTimeout) =>
|
||||
if !dev?
|
||||
console.log("Warning, trying to send update to non-existent device #{uuid}")
|
||||
return
|
||||
@normaliseDependentDeviceFromDB(dev)
|
||||
.then (device) =>
|
||||
currentState = formatCurrentAsState(device)
|
||||
targetState = formatTargetAsState(device)
|
||||
endpoints[device.appId] ?= getHookEndpoint(device.appId)
|
||||
endpoints[device.appId]
|
||||
.then (endpoint) ->
|
||||
@getHookEndpoint(device.appId)
|
||||
.then (endpoint) =>
|
||||
if device.markedForDeletion
|
||||
sendDeleteHook(device, endpoint)
|
||||
else if device.targetCommit? and !_.isEqual(targetState, currentState) and !_.isEqual(targetState, acknowledgedState[device.uuid])
|
||||
sendUpdate(device, endpoint)
|
||||
@sendDeleteHook(device, apiTimeout, endpoint)
|
||||
else if device.targetCommit? and !_.isEqual(targetState, currentState) and !_.isEqual(targetState, @acknowledgedState[device.uuid])
|
||||
@sendUpdate(device, targetState, apiTimeout, endpoint)
|
||||
)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user