Cancel ongoing livepushes when a new change occurs

Also fix livepush logging when a new container is created (previously
the logs of the commands would stop working after this has happened)

Change-type: minor
Signed-off-by: Cameron Diver <cameron@balena.io>

lib/utils/device/deploy.ts

@@ -92,7 +92,9 @@ async function environmentFromInput(
 			// exists
 			if (!(match[1] in ret)) {
 				logger.logDebug(
-					`Warning: Cannot find a service with name ${match[1]}. Treating the string as part of the environment variable name.`,
+					`Warning: Cannot find a service with name ${
+						match[1]
+					}. Treating the string as part of the environment variable name.`,
 				);
 				match[2] = `${match[1]}:${match[2]}`;
 			} else {
@@ -130,7 +132,9 @@ export async function deployToDevice(opts: DeviceDeployOptions): Promise<void> {
 		await api.ping();
 	} catch (e) {
 		exitWithExpectedError(
-			`Could not communicate with local mode device at address ${opts.deviceHost}`,
+			`Could not communicate with local mode device at address ${
+				opts.deviceHost
+			}`,
 		);
 	}
 
@@ -290,8 +294,22 @@ export async function performBuilds(
 	logger.logDebug('Probing remote daemon for cache images');
 	await assignDockerBuildOpts(docker, buildTasks, opts);
 
+	// If we're passed a build logs object make sure to set it
+	// up properly
+	let logHandlers: ((serviceName: string, line: string) => void) | undefined;
+	if (buildLogs != null) {
+		for (const task of buildTasks) {
+			if (!task.external) {
+				buildLogs[task.serviceName] = '';
+			}
+		}
+		logHandlers = (serviceName: string, line: string) => {
+			buildLogs[serviceName] += `${line}\n`;
+		};
+	}
+
 	logger.logDebug('Starting builds...');
-	await assignOutputHandlers(buildTasks, logger, buildLogs);
+	await assignOutputHandlers(buildTasks, logger, logHandlers);
 	const localImages = await multibuild.performBuilds(buildTasks, docker);
 
 	// Check for failures
@@ -324,11 +342,26 @@ export async function rebuildSingleTask(
 	composition: Composition,
 	source: string,
 	opts: DeviceDeployOptions,
+	// To cancel a running build, you must first find the
+	// container id that it's running in. This is printed in
+	// the logs, so any calller who wants to keep track of
+	// this should provide the following callback
+	containerIdCb?: (id: string) => void,
 ): Promise<string> {
 	const { tarDirectory } = await import('../compose');
 	const multibuild = await import('resin-multibuild');
 	// First we run the build task, to get the new image id
-	const buildLogs: Dictionary<string> = {};
+	let buildLogs = '';
+	const logHandler = (_s: string, line: string) => {
+		buildLogs += `${line}\n`;
+
+		if (containerIdCb != null) {
+			const match = line.match(/^\s*--->\s*Running\s*in\s*([a-f0-9]*)\s*$/i);
+			if (match != null) {
+				containerIdCb(match[1]);
+			}
+		}
+	};
 
 	const tarStream = await tarDirectory(source);
 
@@ -342,7 +375,7 @@ export async function rebuildSingleTask(
 	}
 
 	await assignDockerBuildOpts(docker, [task], opts);
-	await assignOutputHandlers([task], logger, buildLogs);
+	await assignOutputHandlers([task], logger, logHandler);
 
 	const [localImage] = await multibuild.performBuilds([task], docker);
 
@@ -355,13 +388,13 @@ export async function rebuildSingleTask(
 		]);
 	}
 
-	return buildLogs[task.serviceName];
+	return buildLogs;
 }
 
 function assignOutputHandlers(
 	buildTasks: BuildTask[],
 	logger: Logger,
-	buildLogs?: Dictionary<string>,
+	logCb?: (serviceName: string, line: string) => void,
 ) {
 	_.each(buildTasks, task => {
 		if (task.external) {
@@ -372,9 +405,6 @@ function assignOutputHandlers(
 				);
 			};
 		} else {
-			if (buildLogs) {
-				buildLogs[task.serviceName] = '';
-			}
 			task.streamHook = stream => {
 				stream.on('data', (buf: Buffer) => {
 					const str = _.trimEnd(buf.toString());
@@ -384,10 +414,8 @@ function assignOutputHandlers(
 							logger,
 						);
 
-						if (buildLogs) {
+						if (logCb) {
-							buildLogs[task.serviceName] = `${
+							logCb(task.serviceName, str);
-								buildLogs[task.serviceName]
-							}\n${str}`;
 						}
 					}
 				});

lib/utils/device/live.ts

@@ -1,10 +1,7 @@
 import * as Bluebird from 'bluebird';
 import * as chokidar from 'chokidar';
 import * as Dockerode from 'dockerode';
-import Livepush, {
+import Livepush, { ContainerNotRunningError } from 'livepush';
-	ContainerNotRunningError,
-	LivepushAlreadyRunningError,
-} from 'livepush';
 import * as _ from 'lodash';
 import * as path from 'path';
 import { Composition } from 'resin-compose-parse';
@@ -67,6 +64,10 @@ export class LivepushManager {
 	private updateEventsWaiting: Dictionary<string[]> = {};
 	private deleteEventsWaiting: Dictionary<string[]> = {};
 
+	private rebuildsRunning: Dictionary<boolean> = {};
+	private rebuildRunningIds: Dictionary<string> = {};
+	private rebuildsCancelled: Dictionary<boolean> = {};
+
 	public constructor(opts: LivepushOpts) {
 		this.buildContext = opts.buildContext;
 		this.composition = opts.composition;
@@ -135,12 +136,6 @@ export class LivepushManager {
 					);
 				}
 
-				const msgString = (msg: string) =>
-					`[${getServiceColourFn(serviceName)(serviceName)}] ${msg}`;
-				const log = (msg: string) => this.logger.logLivepush(msgString(msg));
-				const error = (msg: string) => this.logger.logError(msgString(msg));
-				const debugLog = (msg: string) => this.logger.logDebug(msgString(msg));
-
 				const livepush = await Livepush.init(
 					dockerfile,
 					context,
@@ -149,22 +144,7 @@ export class LivepushManager {
 					this.docker,
 				);
 
-				livepush.on('commandExecute', command =>
+				this.assignLivepushOutputHandlers(serviceName, livepush);
-					log(`Executing command: \`${command.command}\``),
-				);
-				livepush.on('commandOutput', output =>
-					log(`   ${output.output.data.toString()}`),
-				);
-				livepush.on('commandReturn', ({ returnCode, command }) => {
-					if (returnCode !== 0) {
-						error(`  Command ${command} failed with exit code: ${returnCode}`);
-					} else {
-						debugLog(`Command ${command} exited successfully`);
-					}
-				});
-				livepush.on('containerRestart', () => {
-					log('Restarting service...');
-				});
 
 				this.updateEventsWaiting[serviceName] = [];
 				this.deleteEventsWaiting[serviceName] = [];
@@ -196,6 +176,9 @@ export class LivepushManager {
 					monitor,
 					containerId: container.containerId,
 				};
+
+				this.rebuildsRunning[serviceName] = false;
+				this.rebuildsCancelled[serviceName] = false;
 			}
 		}
 
@@ -293,9 +276,7 @@ export class LivepushManager {
 			this.logger.logError(
 				`An error occured whilst trying to perform a livepush: `,
 			);
-			if (e instanceof LivepushAlreadyRunningError) {
+			if (e instanceof ContainerNotRunningError) {
-				this.logger.logError('   Livepush already running');
-			} else if (e instanceof ContainerNotRunningError) {
 				this.logger.logError('   Livepush container not running');
 			} else {
 				this.logger.logError(`   ${e.message}`);
@@ -305,6 +286,17 @@ export class LivepushManager {
 	}
 
 	private async handleServiceRebuild(serviceName: string): Promise<void> {
+		if (this.rebuildsRunning[serviceName]) {
+			this.logger.logLivepush(
+				`Cancelling ongoing rebuild for service ${serviceName}`,
+			);
+			await this.cancelRebuild(serviceName);
+			while (this.rebuildsCancelled[serviceName]) {
+				await Bluebird.delay(1000);
+			}
+		}
+
+		this.rebuildsRunning[serviceName] = true;
 		try {
 			const buildTask = _.find(this.buildTasks, { serviceName });
 			if (buildTask == null) {
@@ -323,18 +315,32 @@ export class LivepushManager {
 					this.composition,
 					this.buildContext,
 					this.deployOpts,
+					id => {
+						this.rebuildRunningIds[serviceName] = id;
+					},
 				);
 			} catch (e) {
 				if (!(e instanceof BuildError)) {
 					throw e;
 				}
 
+				if (this.rebuildsCancelled[serviceName]) {
+					return;
+				}
+
 				this.logger.logError(
 					`Rebuild of service ${serviceName} failed!\n  Error: ${e.getServiceError(
 						serviceName,
 					)}`,
 				);
 				return;
+			} finally {
+				delete this.rebuildRunningIds[serviceName];
+			}
+
+			// If the build has been cancelled, exit early
+			if (this.rebuildsCancelled[serviceName]) {
+				return;
 			}
 
 			// Let's first delete the container from the device
@@ -371,11 +377,63 @@ export class LivepushManager {
 				stageImages[serviceName],
 				this.docker,
 			);
+			this.assignLivepushOutputHandlers(serviceName, instance.livepush);
 		} catch (e) {
 			this.logger.logError(`There was an error rebuilding the service: ${e}`);
+		} finally {
+			this.rebuildsRunning[serviceName] = false;
+			this.rebuildsCancelled[serviceName] = false;
 		}
 	}
 
+	private async cancelRebuild(serviceName: string) {
+		this.rebuildsCancelled[serviceName] = true;
+
+		// If we have a container id of the current build,
+		// attempt to kill it
+		if (this.rebuildRunningIds[serviceName] != null) {
+			try {
+				await this.docker
+					.getContainer(this.rebuildRunningIds[serviceName])
+					.remove({ force: true });
+				await this.containers[serviceName].livepush.cancel();
+			} catch {
+				// No need to do anything here
+			}
+		}
+	}
+
+	private assignLivepushOutputHandlers(
+		serviceName: string,
+		livepush: Livepush,
+	) {
+		const msgString = (msg: string) =>
+			`[${getServiceColourFn(serviceName)(serviceName)}] ${msg}`;
+		const log = (msg: string) => this.logger.logLivepush(msgString(msg));
+		const error = (msg: string) => this.logger.logError(msgString(msg));
+		const debugLog = (msg: string) => this.logger.logDebug(msgString(msg));
+
+		livepush.on('commandExecute', command =>
+			log(`Executing command: \`${command.command}\``),
+		);
+		livepush.on('commandOutput', output =>
+			log(`   ${output.output.data.toString()}`),
+		);
+		livepush.on('commandReturn', ({ returnCode, command }) => {
+			if (returnCode !== 0) {
+				error(`  Command ${command} failed with exit code: ${returnCode}`);
+			} else {
+				debugLog(`Command ${command} exited successfully`);
+			}
+		});
+		livepush.on('containerRestart', () => {
+			log('Restarting service...');
+		});
+		livepush.on('cancel', () => {
+			log('Cancelling current livepush...');
+		});
+	}
+
 	private static extractDockerArrowMessage(
 		outputLine: string,
 	): string | undefined {

package.json

@@ -160,7 +160,7 @@
     "is-root": "^1.0.0",
     "js-yaml": "^3.13.1",
     "klaw": "^3.0.0",
-    "livepush": "^1.2.3",
+    "livepush": "^2.0.0",
     "lodash": "^4.17.4",
     "minimatch": "^3.0.4",
     "mixpanel": "^0.10.1",