diff --git a/src/application-manager.d.ts b/src/application-manager.d.ts index a5e2f19d..ee77c42d 100644 --- a/src/application-manager.d.ts +++ b/src/application-manager.d.ts @@ -7,7 +7,6 @@ import { ServiceAction } from './device-api/common'; import { DeviceStatus, InstancedAppState } from './types/state'; import type { Image } from './compose/images'; -import ServiceManager from './compose/service-manager'; import DeviceState from './device-state'; import { APIBinder } from './api-binder'; @@ -45,8 +44,6 @@ class ApplicationManager extends EventEmitter { public deviceState: DeviceState; public apiBinder: APIBinder; - public services: ServiceManager; - public proxyvisor: any; public timeSpentFetching: number; public fetchesInProgress: number; diff --git a/src/application-manager.js b/src/application-manager.js index 20d340a2..25a97ec2 100644 --- a/src/application-manager.js +++ b/src/application-manager.js @@ -21,11 +21,11 @@ import { import * as dbFormat from './device-state/db-format'; -import { ServiceManager } from './compose/service-manager'; import * as Images from './compose/images'; import { Network } from './compose/network'; import * as networkManager from './compose/network-manager'; import * as volumeManager from './compose/volume-manager'; +import * as serviceManager from './compose/service-manager'; import * as compositionSteps from './compose/composition-steps'; import { Proxyvisor } from './proxyvisor'; @@ -159,7 +159,6 @@ export class ApplicationManager extends EventEmitter { this.deviceState = deviceState; this.apiBinder = apiBinder; - this.services = new ServiceManager(); this.proxyvisor = new Proxyvisor({ applications: this, }); @@ -171,7 +170,6 @@ export class ApplicationManager extends EventEmitter { this.actionExecutors = compositionSteps.getExecutors({ lockFn: this._lockingIfNecessary, - services: this.services, applications: this, callbacks: { containerStarted: (id) => { @@ -198,7 +196,7 @@ export class ApplicationManager extends EventEmitter { ); this.router = createApplicationManagerRouter(this); Images.on('change', this.reportCurrentState); - this.services.on('change', this.reportCurrentState); + serviceManager.on('change', this.reportCurrentState); } reportCurrentState(data) { @@ -223,14 +221,14 @@ export class ApplicationManager extends EventEmitter { // But also run it in on startup await cleanup(); await this.localModeManager.init(); - await this.services.attachToRunning(); - await this.services.listenToEvents(); + await serviceManager.attachToRunning(); + await serviceManager.listenToEvents(); } // Returns the status of applications and their services getStatus() { return Promise.join( - this.services.getStatus(), + serviceManager.getStatus(), Images.getStatus(), config.get('currentCommit'), function (services, images, currentCommit) { @@ -362,7 +360,7 @@ export class ApplicationManager extends EventEmitter { getCurrentForComparison() { return Promise.join( - this.services.getAll(), + serviceManager.getAll(), networkManager.getAll(), volumeManager.getAll(), config.get('currentCommit'), @@ -372,7 +370,7 @@ export class ApplicationManager extends EventEmitter { getCurrentApp(appId) { return Promise.join( - this.services.getAllByAppId(appId), + serviceManager.getAllByAppId(appId), networkManager.getAllByAppId(appId), volumeManager.getAllByAppId(appId), config.get('currentCommit'), @@ -1339,13 +1337,13 @@ export class ApplicationManager extends EventEmitter { } stopAll({ force = false, skipLock = false } = {}) { - return Promise.resolve(this.services.getAll()) + return Promise.resolve(serviceManager.getAll()) .map((service) => { return this._lockingIfNecessary( service.appId, { force, skipLock }, () => { - return this.services + return serviceManager .kill(service, { removeContainer: false, wait: true }) .then(() => { delete this._containerStarted[service.containerId]; @@ -1391,7 +1389,7 @@ export class ApplicationManager extends EventEmitter { if (intId == null) { throw new Error(`Invalid id: ${id}`); } - containerIdsByAppId[intId] = this.services.getContainerIdMap(intId); + containerIdsByAppId[intId] = serviceManager.getContainerIdMap(intId); }); return config.get('localMode').then((localMode) => { diff --git a/src/compose/composition-steps.ts b/src/compose/composition-steps.ts index de517a96..d20d86f8 100644 --- a/src/compose/composition-steps.ts +++ b/src/compose/composition-steps.ts @@ -7,7 +7,7 @@ import type { Image } from './images'; import * as images from './images'; import Network from './network'; import Service from './service'; -import ServiceManager from './service-manager'; +import * as serviceManager from './service-manager'; import Volume from './volume'; import { checkTruthy } from '../lib/validation'; @@ -136,7 +136,6 @@ interface CompositionCallbacks { export function getExecutors(app: { lockFn: LockingFn; - services: ServiceManager; applications: ApplicationManager; callbacks: CompositionCallbacks; }) { @@ -150,7 +149,7 @@ export function getExecutors(app: { }, async () => { const wait = _.get(step, ['options', 'wait'], false); - await app.services.kill(step.current, { + await serviceManager.kill(step.current, { removeContainer: false, wait, }); @@ -166,7 +165,7 @@ export function getExecutors(app: { skipLock: step.skipLock || _.get(step, ['options', 'skipLock']), }, async () => { - await app.services.kill(step.current); + await serviceManager.kill(step.current); app.callbacks.containerKilled(step.current.containerId); if (_.get(step, ['options', 'removeImage'])) { await images.removeByDockerId(step.current.config.image); @@ -177,7 +176,7 @@ export function getExecutors(app: { remove: async (step) => { // Only called for dead containers, so no need to // take locks - await app.services.remove(step.current); + await serviceManager.remove(step.current); }, updateMetadata: (step) => { const skipLock = @@ -190,7 +189,7 @@ export function getExecutors(app: { skipLock: skipLock || _.get(step, ['options', 'skipLock']), }, async () => { - await app.services.updateMetadata(step.current, step.target); + await serviceManager.updateMetadata(step.current, step.target); }, ); }, @@ -202,9 +201,9 @@ export function getExecutors(app: { skipLock: step.skipLock || _.get(step, ['options', 'skipLock']), }, async () => { - await app.services.kill(step.current, { wait: true }); + await serviceManager.kill(step.current, { wait: true }); app.callbacks.containerKilled(step.current.containerId); - const container = await app.services.start(step.target); + const container = await serviceManager.start(step.target); app.callbacks.containerStarted(container.id); }, ); @@ -216,7 +215,7 @@ export function getExecutors(app: { }); }, start: async (step) => { - const container = await app.services.start(step.target); + const container = await serviceManager.start(step.target); app.callbacks.containerStarted(container.id); }, updateCommit: async (step) => { @@ -230,7 +229,7 @@ export function getExecutors(app: { skipLock: step.skipLock || _.get(step, ['options', 'skipLock']), }, async () => { - await app.services.handover(step.current, step.target); + await serviceManager.handover(step.current, step.target); }, ); }, diff --git a/src/compose/service-manager.ts b/src/compose/service-manager.ts index 814fd452..4784cce1 100644 --- a/src/compose/service-manager.ts +++ b/src/compose/service-manager.ts @@ -32,322 +32,428 @@ type ServiceManagerEventEmitter = StrictEventEmitter< EventEmitter, ServiceManagerEvents >; +const events: ServiceManagerEventEmitter = new EventEmitter(); interface KillOpts { removeContainer?: boolean; wait?: boolean; } -export class ServiceManager extends (EventEmitter as new () => ServiceManagerEventEmitter) { - // Whether a container has died, indexed by ID - private containerHasDied: Dictionary = {}; - private listening = false; - // Volatile state of containers, indexed by containerId (or random strings if - // we don't yet have an id) - private volatileState: Dictionary> = {}; +export const on: typeof events['on'] = events.on.bind(events); +export const once: typeof events['once'] = events.once.bind(events); +export const removeListener: typeof events['removeListener'] = events.removeListener.bind( + events, +); +export const removeAllListeners: typeof events['removeAllListeners'] = events.removeAllListeners.bind( + events, +); - public constructor() { - super(); - } +// Whether a container has died, indexed by ID +const containerHasDied: Dictionary = {}; +let listening = false; +// Volatile state of containers, indexed by containerId (or random strings if +// we don't yet have an id) +const volatileState: Dictionary> = {}; - public async getAll( - extraLabelFilters: string | string[] = [], - ): Promise { - const filterLabels = ['supervised'].concat(extraLabelFilters); - const containers = await this.listWithBothLabels(filterLabels); +export async function getAll( + extraLabelFilters: string | string[] = [], +): Promise { + const filterLabels = ['supervised'].concat(extraLabelFilters); + const containers = await listWithBothLabels(filterLabels); - const services = await Bluebird.map(containers, async (container) => { - try { - const serviceInspect = await docker - .getContainer(container.Id) - .inspect(); - const service = Service.fromDockerContainer(serviceInspect); - // We know that the containerId is set below, because `fromDockerContainer` - // always sets it - const vState = this.volatileState[service.containerId!]; - if (vState != null && vState.status != null) { - service.status = vState.status; - } - return service; - } catch (e) { - if (NotFoundError(e)) { - return null; - } - throw e; + const services = await Bluebird.map(containers, async (container) => { + try { + const serviceInspect = await docker.getContainer(container.Id).inspect(); + const service = Service.fromDockerContainer(serviceInspect); + // We know that the containerId is set below, because `fromDockerContainer` + // always sets it + const vState = volatileState[service.containerId!]; + if (vState != null && vState.status != null) { + service.status = vState.status; + } + return service; + } catch (e) { + if (NotFoundError(e)) { + return null; } - }); - - return services.filter((s) => s != null) as Service[]; - } - - public async get(service: Service) { - // Get the container ids for special network handling - const containerIds = await this.getContainerIdMap(service.appId!); - const services = ( - await this.getAll(`service-id=${service.serviceId}`) - ).filter((currentService) => - currentService.isEqualConfig(service, containerIds), - ); - - if (services.length === 0) { - const e: StatusCodeError = new Error( - 'Could not find a container matching this service definition', - ); - e.statusCode = 404; throw e; } - return services[0]; - } + }); - public async getStatus() { - const services = await this.getAll(); - const status = _.clone(this.volatileState); + return services.filter((s) => s != null) as Service[]; +} - for (const service of services) { - if (service.containerId == null) { - throw new InternalInconsistencyError( - `containerId not defined in ServiceManager.getStatus: ${service}`, - ); - } - if (status[service.containerId] == null) { - status[service.containerId] = _.pick(service, [ - 'appId', - 'imageId', - 'status', - 'releaseId', - 'commit', - 'createdAt', - 'serviceName', - ]) as Partial; - } - } +export async function get(service: Service) { + // Get the container ids for special network handling + const containerIds = await getContainerIdMap(service.appId!); + const services = ( + await getAll(`service-id=${service.serviceId}`) + ).filter((currentService) => + currentService.isEqualConfig(service, containerIds), + ); - return _.values(status); - } - - public async getByDockerContainerId( - containerId: string, - ): Promise { - const container = await docker.getContainer(containerId).inspect(); - if ( - container.Config.Labels['io.balena.supervised'] == null && - container.Config.Labels['io.resin.supervised'] == null - ) { - return null; - } - return Service.fromDockerContainer(container); - } - - public async updateMetadata( - service: Service, - metadata: { imageId: number; releaseId: number }, - ) { - const svc = await this.get(service); - if (svc.containerId == null) { - throw new InternalInconsistencyError( - `No containerId provided for service ${service.serviceName} in ServiceManager.updateMetadata. Service: ${service}`, - ); - } - - await docker.getContainer(svc.containerId).rename({ - name: `${service.serviceName}_${metadata.imageId}_${metadata.releaseId}`, - }); - } - - public async handover(current: Service, target: Service) { - // We set the running container to not restart so that in case of a poweroff - // it doesn't come back after boot. - await this.prepareForHandover(current); - await this.start(target); - await this.waitToKill( - current, - target.config.labels['io.balena.update.handover-timeout'], + if (services.length === 0) { + const e: StatusCodeError = new Error( + 'Could not find a container matching this service definition', ); - await this.kill(current); + e.statusCode = 404; + throw e; } + return services[0]; +} - public async killAllLegacy(): Promise { - // Containers haven't been normalized (this is an updated supervisor) - // so we need to stop and remove them - const supervisorImageId = ( - await docker.getImage(constants.supervisorImage).inspect() - ).Id; +export async function getStatus() { + const services = await getAll(); + const status = _.clone(volatileState); - for (const container of await docker.listContainers({ all: true })) { - if (container.ImageID !== supervisorImageId) { - await this.killContainer(container.Id, { - serviceName: 'legacy', - }); - } - } - } - - public kill(service: Service, opts: KillOpts = {}) { + for (const service of services) { if (service.containerId == null) { throw new InternalInconsistencyError( - `Attempt to kill container without containerId! Service :${service}`, + `containerId not defined in ServiceManager.getStatus: ${service}`, ); } - return this.killContainer(service.containerId, service, opts); + if (status[service.containerId] == null) { + status[service.containerId] = _.pick(service, [ + 'appId', + 'imageId', + 'status', + 'releaseId', + 'commit', + 'createdAt', + 'serviceName', + ]) as Partial; + } } - public async remove(service: Service) { - logger.logSystemEvent(LogTypes.removeDeadService, { service }); - const existingService = await this.get(service); + return _.values(status); +} - if (existingService.containerId == null) { +export async function getByDockerContainerId( + containerId: string, +): Promise { + const container = await docker.getContainer(containerId).inspect(); + if ( + container.Config.Labels['io.balena.supervised'] == null && + container.Config.Labels['io.resin.supervised'] == null + ) { + return null; + } + return Service.fromDockerContainer(container); +} + +export async function updateMetadata( + service: Service, + metadata: { imageId: number; releaseId: number }, +) { + const svc = await get(service); + if (svc.containerId == null) { + throw new InternalInconsistencyError( + `No containerId provided for service ${service.serviceName} in ServiceManager.updateMetadata. Service: ${service}`, + ); + } + + await docker.getContainer(svc.containerId).rename({ + name: `${service.serviceName}_${metadata.imageId}_${metadata.releaseId}`, + }); +} + +export async function handover(current: Service, target: Service) { + // We set the running container to not restart so that in case of a poweroff + // it doesn't come back after boot. + await prepareForHandover(current); + await start(target); + await waitToKill( + current, + target.config.labels['io.balena.update.handover-timeout'], + ); + await kill(current); +} + +export async function killAllLegacy(): Promise { + // Containers haven't been normalized (this is an updated supervisor) + const supervisorImageId = ( + await docker.getImage(constants.supervisorImage).inspect() + ).Id; + + for (const container of await docker.listContainers({ all: true })) { + if (container.ImageID !== supervisorImageId) { + await killContainer(container.Id, { + serviceName: 'legacy', + }); + } + } +} + +export function kill(service: Service, opts: KillOpts = {}) { + if (service.containerId == null) { + throw new InternalInconsistencyError( + `Attempt to kill container without containerId! Service :${service}`, + ); + } + return killContainer(service.containerId, service, opts); +} + +export async function remove(service: Service) { + logger.logSystemEvent(LogTypes.removeDeadService, { service }); + const existingService = await get(service); + + if (existingService.containerId == null) { + throw new InternalInconsistencyError( + `No containerId provided for service ${service.serviceName} in ServiceManager.updateMetadata. Service: ${service}`, + ); + } + + try { + await docker.getContainer(existingService.containerId).remove({ v: true }); + } catch (e) { + if (!NotFoundError(e)) { + logger.logSystemEvent(LogTypes.removeDeadServiceError, { + service, + error: e, + }); + throw e; + } + } +} +export function getAllByAppId(appId: number) { + return getAll(`app-id=${appId}`); +} + +export async function stopAllByAppId(appId: number) { + for (const app of await getAllByAppId(appId)) { + await kill(app, { removeContainer: false }); + } +} + +export async function create(service: Service) { + const mockContainerId = config.newUniqueKey(); + try { + const existing = await get(service); + if (existing.containerId == null) { throw new InternalInconsistencyError( `No containerId provided for service ${service.serviceName} in ServiceManager.updateMetadata. Service: ${service}`, ); } - - try { - await docker - .getContainer(existingService.containerId) - .remove({ v: true }); - } catch (e) { - if (!NotFoundError(e)) { - logger.logSystemEvent(LogTypes.removeDeadServiceError, { - service, - error: e, - }); - throw e; - } - } - } - public getAllByAppId(appId: number) { - return this.getAll(`app-id=${appId}`); - } - - public async stopAllByAppId(appId: number) { - for (const app of await this.getAllByAppId(appId)) { - await this.kill(app, { removeContainer: false }); - } - } - - public async create(service: Service) { - const mockContainerId = config.newUniqueKey(); - try { - const existing = await this.get(service); - if (existing.containerId == null) { - throw new InternalInconsistencyError( - `No containerId provided for service ${service.serviceName} in ServiceManager.updateMetadata. Service: ${service}`, - ); - } - return docker.getContainer(existing.containerId); - } catch (e) { - if (!NotFoundError(e)) { - logger.logSystemEvent(LogTypes.installServiceError, { - service, - error: e, - }); - throw e; - } - - const deviceName = await config.get('name'); - if (!isValidDeviceName(deviceName)) { - throw new Error( - 'The device name contains a newline, which is unsupported by balena. ' + - 'Please fix the device name', - ); - } - - // Get all created services so far - if (service.appId == null) { - throw new InternalInconsistencyError( - 'Attempt to start a service without an existing application ID', - ); - } - const serviceContainerIds = await this.getContainerIdMap(service.appId); - const conf = service.toDockerContainer({ - deviceName, - containerIds: serviceContainerIds, + return docker.getContainer(existing.containerId); + } catch (e) { + if (!NotFoundError(e)) { + logger.logSystemEvent(LogTypes.installServiceError, { + service, + error: e, }); - const nets = serviceNetworksToDockerNetworks( - service.extraNetworksToJoin(), + throw e; + } + + const deviceName = await config.get('name'); + if (!isValidDeviceName(deviceName)) { + throw new Error( + 'The device name contains a newline, which is unsupported by balena. ' + + 'Please fix the device name', ); + } - logger.logSystemEvent(LogTypes.installService, { service }); - this.reportNewStatus(mockContainerId, service, 'Installing'); - - const container = await docker.createContainer(conf); - service.containerId = container.id; - - await Promise.all( - _.map((nets || {}).EndpointsConfig, (endpointConfig, name) => - docker.getNetwork(name).connect({ - Container: container.id, - EndpointConfig: endpointConfig, - }), - ), + // Get all created services so far + if (service.appId == null) { + throw new InternalInconsistencyError( + 'Attempt to start a service without an existing application ID', ); + } + const serviceContainerIds = await getContainerIdMap(service.appId); + const conf = service.toDockerContainer({ + deviceName, + containerIds: serviceContainerIds, + }); + const nets = serviceNetworksToDockerNetworks(service.extraNetworksToJoin()); - logger.logSystemEvent(LogTypes.installServiceSuccess, { service }); - return container; + logger.logSystemEvent(LogTypes.installService, { service }); + reportNewStatus(mockContainerId, service, 'Installing'); + + const container = await docker.createContainer(conf); + service.containerId = container.id; + + await Promise.all( + _.map((nets || {}).EndpointsConfig, (endpointConfig, name) => + docker.getNetwork(name).connect({ + Container: container.id, + EndpointConfig: endpointConfig, + }), + ), + ); + + logger.logSystemEvent(LogTypes.installServiceSuccess, { service }); + return container; + } finally { + reportChange(mockContainerId); + } +} + +export async function start(service: Service) { + let alreadyStarted = false; + let containerId: string | null = null; + + try { + const container = await create(service); + containerId = container.id; + logger.logSystemEvent(LogTypes.startService, { service }); + + reportNewStatus(containerId, service, 'Starting'); + + let shouldRemove = false; + let err: Error | undefined; + try { + await container.start(); + } catch (e) { + // Get the statusCode from the original cause and make sure it's + // definitely an int for comparison reasons + const maybeStatusCode = PermissiveNumber.decode(e.statusCode); + if (isLeft(maybeStatusCode)) { + shouldRemove = true; + err = new Error(`Could not parse status code from docker error: ${e}`); + throw err; + } + const statusCode = maybeStatusCode.right; + const message = e.message; + + // 304 means the container was already started, precisely what we want + if (statusCode === 304) { + alreadyStarted = true; + } else if ( + statusCode === 500 && + _.isString(message) && + message.trim().match(/exec format error$/) + ) { + // Provide a friendlier error message for "exec format error" + const deviceType = await config.get('deviceType'); + err = new Error( + `Application architecture incompatible with ${deviceType}: exec format error`, + ); + throw err; + } else { + // rethrow the same error + err = e; + throw e; + } } finally { - this.reportChange(mockContainerId); + if (shouldRemove) { + // If starting the container fialed, we remove it so that it doesn't litter + await container.remove({ v: true }).catch(_.noop); + logger.logSystemEvent(LogTypes.startServiceError, { + service, + error: err, + }); + } + } + + const serviceId = service.serviceId; + const imageId = service.imageId; + if (serviceId == null || imageId == null) { + throw new InternalInconsistencyError( + `serviceId and imageId not defined for service: ${service.serviceName} in ServiceManager.start`, + ); + } + + logger.attach(container.id, { serviceId, imageId }); + + if (!alreadyStarted) { + logger.logSystemEvent(LogTypes.startServiceSuccess, { service }); + } + + service.config.running = true; + return container; + } finally { + if (containerId != null) { + reportChange(containerId); } } +} - public async start(service: Service) { - let alreadyStarted = false; - let containerId: string | null = null; +export function listenToEvents() { + if (listening) { + return; + } - try { - const container = await this.create(service); - containerId = container.id; - logger.logSystemEvent(LogTypes.startService, { service }); + listening = true; - this.reportNewStatus(containerId, service, 'Starting'); + const listen = async () => { + const stream = await docker.getEvents({ + filters: { type: ['container'] } as any, + }); - let remove = false; - let err: Error | undefined; - try { - await container.start(); - } catch (e) { - // Get the statusCode from the original cause and make sure it's - // definitely an int for comparison reasons - const maybeStatusCode = PermissiveNumber.decode(e.statusCode); - if (isLeft(maybeStatusCode)) { - remove = true; - err = new Error( - `Could not parse status code from docker error: ${e}`, - ); - throw err; - } - const statusCode = maybeStatusCode.right; - const message = e.message; + stream.on('error', (e) => { + log.error(`Error on docker events stream:`, e); + }); + const parser = JSONStream.parse(); + parser.on('data', async (data: { status: string; id: string }) => { + if (data != null) { + const status = data.status; + if (status === 'die' || status === 'start') { + try { + let service: Service | null = null; + try { + service = await getByDockerContainerId(data.id); + } catch (e) { + if (!NotFoundError(e)) { + throw e; + } + } + if (service != null) { + events.emit('change'); + if (status === 'die') { + logger.logSystemEvent(LogTypes.serviceExit, { service }); + containerHasDied[data.id] = true; + } else if (status === 'start' && containerHasDied[data.id]) { + delete containerHasDied[data.id]; + logger.logSystemEvent(LogTypes.serviceRestart, { + service, + }); - // 304 means the container was already started, precisely what we want - if (statusCode === 304) { - alreadyStarted = true; - } else if ( - statusCode === 500 && - _.isString(message) && - message.trim().match(/exec format error$/) - ) { - // Provide a friendlier error message for "exec format error" - const deviceType = await config.get('deviceType'); - err = new Error( - `Application architecture incompatible with ${deviceType}: exec format error`, - ); - throw err; - } else { - // rethrow the same error - err = e; - throw e; - } - } finally { - if (remove) { - // If starting the container fialed, we remove it so that it doesn't litter - await container.remove({ v: true }).catch(_.noop); - logger.logSystemEvent(LogTypes.startServiceError, { - service, - error: err, - }); + const serviceId = service.serviceId; + const imageId = service.imageId; + if (serviceId == null || imageId == null) { + throw new InternalInconsistencyError( + `serviceId and imageId not defined for service: ${service.serviceName} in ServiceManager.listenToEvents`, + ); + } + logger.attach(data.id, { + serviceId, + imageId, + }); + } + } + } catch (e) { + log.error('Error on docker event:', e, e.stack); + } } } + }); + return new Promise((resolve, reject) => { + parser + .on('error', (e: Error) => { + log.error('Error on docker events stream:', e); + reject(e); + }) + .on('end', resolve); + stream.pipe(parser); + }); + }; + + Bluebird.resolve(listen()) + .catch((e) => { + log.error('Error listening to events:', e, e.stack); + }) + .finally(() => { + listening = false; + setTimeout(listenToEvents, 1000); + }); + + return; +} + +export async function attachToRunning() { + const services = await getAll(); + for (const service of services) { + if (service.status === 'Running') { const serviceId = service.serviceId; const imageId = service.imageId; if (serviceId == null || imageId == null) { @@ -356,302 +462,187 @@ export class ServiceManager extends (EventEmitter as new () => ServiceManagerEve ); } - logger.attach(container.id, { serviceId, imageId }); - - if (!alreadyStarted) { - logger.logSystemEvent(LogTypes.startServiceSuccess, { service }); - } - - service.config.running = true; - return container; - } finally { - if (containerId != null) { - this.reportChange(containerId); + if (service.containerId == null) { + throw new InternalInconsistencyError( + `containerId not defined for service: ${service.serviceName} in ServiceManager.attachToRunning`, + ); } + logger.attach(service.containerId, { + serviceId, + imageId, + }); } } - - public listenToEvents() { - if (this.listening) { - return; - } - - this.listening = true; - - const listen = async () => { - const stream = await docker.getEvents({ - filters: { type: ['container'] } as any, - }); - - stream.on('error', (e) => { - log.error(`Error on docker events stream:`, e); - }); - const parser = JSONStream.parse(); - parser.on('data', async (data: { status: string; id: string }) => { - if (data != null) { - const status = data.status; - if (status === 'die' || status === 'start') { - try { - let service: Service | null = null; - try { - service = await this.getByDockerContainerId(data.id); - } catch (e) { - if (!NotFoundError(e)) { - throw e; - } - } - if (service != null) { - this.emit('change'); - if (status === 'die') { - logger.logSystemEvent(LogTypes.serviceExit, { service }); - this.containerHasDied[data.id] = true; - } else if ( - status === 'start' && - this.containerHasDied[data.id] - ) { - delete this.containerHasDied[data.id]; - logger.logSystemEvent(LogTypes.serviceRestart, { - service, - }); - - const serviceId = service.serviceId; - const imageId = service.imageId; - if (serviceId == null || imageId == null) { - throw new InternalInconsistencyError( - `serviceId and imageId not defined for service: ${service.serviceName} in ServiceManager.listenToEvents`, - ); - } - logger.attach(data.id, { - serviceId, - imageId, - }); - } - } - } catch (e) { - log.error('Error on docker event:', e, e.stack); - } - } - } - }); - - return new Promise((resolve, reject) => { - parser - .on('error', (e: Error) => { - log.error('Error on docker events stream:', e); - reject(e); - }) - .on('end', resolve); - stream.pipe(parser); - }); - }; - - Bluebird.resolve(listen()) - .catch((e) => { - log.error('Error listening to events:', e, e.stack); - }) - .finally(() => { - this.listening = false; - setTimeout(() => this.listenToEvents(), 1000); - }); - - return; - } - - public async attachToRunning() { - const services = await this.getAll(); - for (const service of services) { - if (service.status === 'Running') { - const serviceId = service.serviceId; - const imageId = service.imageId; - if (serviceId == null || imageId == null) { - throw new InternalInconsistencyError( - `serviceId and imageId not defined for service: ${service.serviceName} in ServiceManager.start`, - ); - } - - if (service.containerId == null) { - throw new InternalInconsistencyError( - `containerId not defined for service: ${service.serviceName} in ServiceManager.attachToRunning`, - ); - } - logger.attach(service.containerId, { - serviceId, - imageId, - }); - } - } - } - - public async getContainerIdMap(appId: number): Promise> { - return _(await this.getAllByAppId(appId)) - .keyBy('serviceName') - .mapValues('containerId') - .value() as Dictionary; - } - - private reportChange(containerId?: string, status?: Partial) { - if (containerId != null) { - if (status != null) { - this.volatileState[containerId] = {}; - _.merge(this.volatileState[containerId], status); - } else if (this.volatileState[containerId] != null) { - delete this.volatileState[containerId]; - } - } - this.emit('change'); - } - - private reportNewStatus( - containerId: string, - service: Partial, - status: string, - ) { - this.reportChange( - containerId, - _.merge( - { status }, - _.pick(service, ['imageId', 'appId', 'releaseId', 'commit']), - ), - ); - } - - private killContainer( - containerId: string, - service: Partial = {}, - { removeContainer = true, wait = false }: KillOpts = {}, - ): Bluebird { - // To maintain compatibility of the `wait` flag, this function is not - // async, but it feels like whether or not the promise should be waited on - // should performed by the caller - // TODO: Remove the need for the wait flag - - return Bluebird.try(() => { - logger.logSystemEvent(LogTypes.stopService, { service }); - if (service.imageId != null) { - this.reportNewStatus(containerId, service, 'Stopping'); - } - - const containerObj = docker.getContainer(containerId); - const killPromise = Bluebird.resolve(containerObj.stop()) - .then(() => { - if (removeContainer) { - return containerObj.remove({ v: true }); - } - }) - .catch((e) => { - // Get the statusCode from the original cause and make sure it's - // definitely an int for comparison reasons - const maybeStatusCode = PermissiveNumber.decode(e.statusCode); - if (isLeft(maybeStatusCode)) { - throw new Error( - `Could not parse status code from docker error: ${e}`, - ); - } - const statusCode = maybeStatusCode.right; - - // 304 means the container was already stopped, so we can just remove it - if (statusCode === 304) { - logger.logSystemEvent(LogTypes.stopServiceNoop, { service }); - // Why do we attempt to remove the container again? - if (removeContainer) { - return containerObj.remove({ v: true }); - } - } else if (statusCode === 404) { - // 404 means the container doesn't exist, precisely what we want! - logger.logSystemEvent(LogTypes.stopRemoveServiceNoop, { - service, - }); - } else { - throw e; - } - }) - .tap(() => { - delete this.containerHasDied[containerId]; - logger.logSystemEvent(LogTypes.stopServiceSuccess, { service }); - }) - .catch((e) => { - logger.logSystemEvent(LogTypes.stopServiceError, { - service, - error: e, - }); - }) - .finally(() => { - if (service.imageId != null) { - this.reportChange(containerId); - } - }); - - if (wait) { - return killPromise; - } - return; - }); - } - - private async listWithBothLabels( - labelList: string[], - ): Promise { - const listWithPrefix = (prefix: string) => - docker.listContainers({ - all: true, - filters: { - label: _.map(labelList, (v) => `${prefix}${v}`), - }, - }); - - const [legacy, current] = await Promise.all([ - listWithPrefix('io.resin.'), - listWithPrefix('io.balena.'), - ]); - - return _.unionBy(legacy, current, 'Id'); - } - - private async prepareForHandover(service: Service) { - const svc = await this.get(service); - if (svc.containerId == null) { - throw new InternalInconsistencyError( - `No containerId provided for service ${service.serviceName} in ServiceManager.prepareForHandover. Service: ${service}`, - ); - } - const container = docker.getContainer(svc.containerId); - await container.update({ RestartPolicy: {} }); - return await container.rename({ - name: `old_${service.serviceName}_${service.imageId}_${service.imageId}_${service.releaseId}`, - }); - } - - private waitToKill(service: Service, timeout: number | string) { - const pollInterval = 100; - timeout = checkInt(timeout, { positive: true }) || 60000; - const deadline = Date.now() + timeout; - - const handoverCompletePaths = service.handoverCompleteFullPathsOnHost(); - - const wait = (): Bluebird => - Bluebird.any( - handoverCompletePaths.map((file) => - fs.stat(file).then(() => fs.unlink(file).catch(_.noop)), - ), - ).catch(async () => { - if (Date.now() < deadline) { - await Bluebird.delay(pollInterval); - return wait(); - } else { - log.info( - `Handover timeout has passed, assuming handover was completed for service ${service.serviceName}`, - ); - } - }); - - log.info( - `Waiting for handover to be completed for service: ${service.serviceName}`, - ); - - return wait().then(() => { - log.success(`Handover complete for service ${service.serviceName}`); - }); - } } -export default ServiceManager; +export async function getContainerIdMap( + appId: number, +): Promise> { + return _(await getAllByAppId(appId)) + .keyBy('serviceName') + .mapValues('containerId') + .value() as Dictionary; +} + +function reportChange(containerId?: string, status?: Partial) { + if (containerId != null) { + if (status != null) { + volatileState[containerId] = { ...status }; + } else if (volatileState[containerId] != null) { + delete volatileState[containerId]; + } + } + events.emit('change'); +} + +function reportNewStatus( + containerId: string, + service: Partial, + status: string, +) { + reportChange( + containerId, + _.merge( + { status }, + _.pick(service, ['imageId', 'appId', 'releaseId', 'commit']), + ), + ); +} + +function killContainer( + containerId: string, + service: Partial = {}, + { removeContainer = true, wait = false }: KillOpts = {}, +): Bluebird { + // To maintain compatibility of the `wait` flag, this function is not + // async, but it feels like whether or not the promise should be waited on + // should performed by the caller + // TODO: Remove the need for the wait flag + + return Bluebird.try(() => { + logger.logSystemEvent(LogTypes.stopService, { service }); + if (service.imageId != null) { + reportNewStatus(containerId, service, 'Stopping'); + } + + const containerObj = docker.getContainer(containerId); + const killPromise = Bluebird.resolve(containerObj.stop()) + .then(() => { + if (removeContainer) { + return containerObj.remove({ v: true }); + } + }) + .catch((e) => { + // Get the statusCode from the original cause and make sure it's + // definitely an int for comparison reasons + const maybeStatusCode = PermissiveNumber.decode(e.statusCode); + if (isLeft(maybeStatusCode)) { + throw new Error( + `Could not parse status code from docker error: ${e}`, + ); + } + const statusCode = maybeStatusCode.right; + + // 304 means the container was already stopped, so we can just remove it + if (statusCode === 304) { + logger.logSystemEvent(LogTypes.stopServiceNoop, { service }); + // Why do we attempt to remove the container again? + if (removeContainer) { + return containerObj.remove({ v: true }); + } + } else if (statusCode === 404) { + // 404 means the container doesn't exist, precisely what we want! + logger.logSystemEvent(LogTypes.stopRemoveServiceNoop, { + service, + }); + } else { + throw e; + } + }) + .tap(() => { + delete containerHasDied[containerId]; + logger.logSystemEvent(LogTypes.stopServiceSuccess, { service }); + }) + .catch((e) => { + logger.logSystemEvent(LogTypes.stopServiceError, { + service, + error: e, + }); + }) + .finally(() => { + if (service.imageId != null) { + reportChange(containerId); + } + }); + + if (wait) { + return killPromise; + } + return; + }); +} + +async function listWithBothLabels( + labelList: string[], +): Promise { + const listWithPrefix = (prefix: string) => + docker.listContainers({ + all: true, + filters: { + label: _.map(labelList, (v) => `${prefix}${v}`), + }, + }); + + const [legacy, current] = await Promise.all([ + listWithPrefix('io.resin.'), + listWithPrefix('io.balena.'), + ]); + + return _.unionBy(legacy, current, 'Id'); +} + +async function prepareForHandover(service: Service) { + const svc = await get(service); + if (svc.containerId == null) { + throw new InternalInconsistencyError( + `No containerId provided for service ${service.serviceName} in ServiceManager.prepareForHandover. Service: ${service}`, + ); + } + const container = docker.getContainer(svc.containerId); + await container.update({ RestartPolicy: {} }); + return await container.rename({ + name: `old_${service.serviceName}_${service.imageId}_${service.imageId}_${service.releaseId}`, + }); +} + +function waitToKill(service: Service, timeout: number | string) { + const pollInterval = 100; + timeout = checkInt(timeout, { positive: true }) || 60000; + const deadline = Date.now() + timeout; + + const handoverCompletePaths = service.handoverCompleteFullPathsOnHost(); + + const wait = (): Bluebird => + Bluebird.any( + handoverCompletePaths.map((file) => + fs.stat(file).then(() => fs.unlink(file).catch(_.noop)), + ), + ).catch(async () => { + if (Date.now() < deadline) { + await Bluebird.delay(pollInterval); + return wait(); + } else { + log.info( + `Handover timeout has passed, assuming handover was completed for service ${service.serviceName}`, + ); + } + }); + + log.info( + `Waiting for handover to be completed for service: ${service.serviceName}`, + ); + + return wait().then(() => { + log.success(`Handover complete for service ${service.serviceName}`); + }); +} diff --git a/src/device-api/v2.ts b/src/device-api/v2.ts index 5b5d33d5..d07ba145 100644 --- a/src/device-api/v2.ts +++ b/src/device-api/v2.ts @@ -10,6 +10,7 @@ import * as db from '../db'; import * as logger from '../logger'; import * as images from '../compose/images'; import * as volumeManager from '../compose/volume-manager'; +import * as serviceManager from '../compose/service-manager'; import { spawnJournalctl } from '../lib/journald'; import { appNotFoundMessage, @@ -153,7 +154,7 @@ export function createV2Api(router: Router, applications: ApplicationManager) { // It's kinda hacky to access the services and db via the application manager // maybe refactor this code Bluebird.join( - applications.services.getStatus(), + serviceManager.getStatus(), images.getStatus(), db.models('app').select(['appId', 'commit', 'name']), ( @@ -359,7 +360,7 @@ export function createV2Api(router: Router, applications: ApplicationManager) { }); router.get('/v2/containerId', async (req, res) => { - const services = await applications.services.getAll(); + const services = await serviceManager.getAll(); if (req.query.serviceName != null || req.query.service != null) { const serviceName = req.query.serviceName || req.query.service; @@ -393,7 +394,7 @@ export function createV2Api(router: Router, applications: ApplicationManager) { const currentRelease = await config.get('currentCommit'); const pending = applications.deviceState.applyInProgress; - const containerStates = (await applications.services.getAll()).map((svc) => + const containerStates = (await serviceManager.getAll()).map((svc) => _.pick( svc, 'status', diff --git a/src/lib/migration.ts b/src/lib/migration.ts index 9dfc3cdf..b4b7f968 100644 --- a/src/lib/migration.ts +++ b/src/lib/migration.ts @@ -13,6 +13,7 @@ import { ApplicationManager } from '../application-manager'; import * as config from '../config'; import * as db from '../db'; import * as volumeManager from '../compose/volume-manager'; +import * as serviceManager from '../compose/service-manager'; import DeviceState from '../device-state'; import * as constants from '../lib/constants'; import { BackupError, DatabaseParseError, NotFoundError } from '../lib/errors'; @@ -244,7 +245,7 @@ export async function normaliseLegacyDatabase( } log.debug('Killing legacy containers'); - await application.services.killAllLegacy(); + await serviceManager.killAllLegacy(); log.debug('Migrating legacy app volumes'); const targetApps = await application.getTargetApps(); diff --git a/test/lib/mocked-device-api.ts b/test/lib/mocked-device-api.ts index 2353c8c0..8380e082 100644 --- a/test/lib/mocked-device-api.ts +++ b/test/lib/mocked-device-api.ts @@ -1,10 +1,9 @@ import { Router } from 'express'; import { fs } from 'mz'; -import { stub } from 'sinon'; import { ApplicationManager } from '../../src/application-manager'; import * as networkManager from '../../src/compose/network-manager'; -import { ServiceManager } from '../../src/compose/service-manager'; +import * as serviceManager from '../../src/compose/service-manager'; import * as volumeManager from '../../src/compose/volume-manager'; import * as config from '../../src/config'; import * as db from '../../src/db'; @@ -135,22 +134,23 @@ function buildRoutes(appManager: ApplicationManager): Router { const originalNetGetAll = networkManager.getAllByAppId; const originalVolGetAll = volumeManager.getAllByAppId; +const originalSvcGetStatus = serviceManager.getStatus; function setupStubs() { - stub(ServiceManager.prototype, 'getStatus').resolves(STUBBED_VALUES.services); - // @ts-expect-error Assigning to a RO property - networkManager.getAllByAppId = () => Promise.resolve(STUBBED_VALUES.networks); + networkManager.getAllByAppId = async () => STUBBED_VALUES.networks; // @ts-expect-error Assigning to a RO property - volumeManager.getAllByAppId = () => Promise.resolve(STUBBED_VALUES.volumes); + volumeManager.getAllByAppId = async () => STUBBED_VALUES.volumes; + // @ts-expect-error Assigning to a RO property + serviceManager.getStatus = async () => STUBBED_VALUES.services; } function restoreStubs() { - (ServiceManager.prototype as any).getStatus.restore(); - // @ts-expect-error Assigning to a RO property networkManager.getAllByAppId = originalNetGetAll; // @ts-expect-error Assigning to a RO property volumeManager.getAllByAppId = originalVolGetAll; + // @ts-expect-error Assigning to a RO property + serviceManager.getStatus = originalSvcGetStatus; } interface SupervisorAPIOpts {