Merge pull request #1301 from balena-io/cancellable-livepushes

Add cancellable livepushes, when a file changes during a current livepush

automation/capitanodoc/utils.ts

@@ -137,7 +137,9 @@ export class MarkdownFileParser {
 				} else {
 					reject(
 						new Error(
-							`Markdown section not found: title="${title}" file="${this.mdFilePath}"`,
+							`Markdown section not found: title="${title}" file="${
+								this.mdFilePath
+							}"`,
 						),
 					);
 				}

bin/balena-dev

@@ -11,13 +11,19 @@ process.env.UV_THREADPOOL_SIZE = '64';
 
 // Use fast-boot to cache require lookups, speeding up startup
 require('fast-boot2').start({
-	cacheFile: '.fast-boot.json'
+	cacheFile: '.fast-boot.json',
 });
 require('coffeescript/register');
+
+const path = require('path');
+const rootDir = path.join(__dirname, '..');
 // Note: before ts-node v6.0.0, 'transpile-only' (no type checking) was the
 // default option. We upgraded ts-node and found that adding 'transpile-only'
 // was necessary to avoid a mysterious 'null' error message. On the plus side,
 // it is supposed to run faster. We still benefit from type checking when
 // running 'npm run build'.
-require('ts-node/register/transpile-only');
+require('ts-node').register({
+	project: path.join(rootDir, 'tsconfig.json'),
+	transpileOnly: true,
+});
 require('../lib/app').run();

lib/actions/push.ts

@@ -88,7 +88,9 @@ async function getAppOwner(sdk: BalenaSDK, appName: string) {
 	});
 
 	const selected = await selectFromList(
-		`${entries.length} applications found with that name, please select the application you would like to push to`,
+		`${
+			entries.length
+		} applications found with that name, please select the application you would like to push to`,
 		entries,
 	);
 

lib/actions/ssh.ts

@@ -58,13 +58,17 @@ async function getContainerId(
 		});
 		if (request.status !== 200) {
 			throw new Error(
-				`There was an error connecting to device ${uuid}, HTTP response code: ${request.status}.`,
+				`There was an error connecting to device ${uuid}, HTTP response code: ${
+					request.status
+				}.`,
 			);
 		}
 		const body = request.body;
 		if (body.status !== 'success') {
 			throw new Error(
-				`There was an error communicating with device ${uuid}.\n\tError: ${body.message}`,
+				`There was an error communicating with device ${uuid}.\n\tError: ${
+					body.message
+				}`,
 			);
 		}
 		containerId = body.services[serviceName];

lib/actions/tunnel.ts

@@ -201,7 +201,9 @@ export const tunnel: CommandDefinition<Args, Options> = {
 					)
 					.then(() => {
 						logger.logInfo(
-							` - tunnelling ${localAddress}:${localPort} to ${device.uuid}:${remotePort}`,
+							` - tunnelling ${localAddress}:${localPort} to ${
+								device.uuid
+							}:${remotePort}`,
 						);
 
 						return true;

lib/app-common.ts

@@ -45,7 +45,9 @@ function checkNodeVersion() {
 		const { stripIndent } = require('common-tags');
 		console.warn(stripIndent`
 			------------------------------------------------------------------------------
-			Warning: Node version "${process.version}" does not match required versions "${validNodeVersions}".
+			Warning: Node version "${
+				process.version
+			}" does not match required versions "${validNodeVersions}".
 			This may cause unexpected behaviour. To upgrade Node, visit:
 			https://nodejs.org/en/download/
 			------------------------------------------------------------------------------

lib/app.ts

@@ -25,7 +25,9 @@ import { exitWithExpectedError } from './utils/patterns';
 function routeCliFramework(argv: string[]): void {
 	if (process.env.DEBUG) {
 		console.log(
-			`Debug: original argv0="${process.argv0}" argv=[${argv}] length=${argv.length}`,
+			`Debug: original argv0="${process.argv0}" argv=[${argv}] length=${
+				argv.length
+			}`,
 		);
 	}
 	const cmdSlice = argv.slice(2);

lib/utils/compose_ts.ts

@@ -51,7 +51,9 @@ export async function parseRegistrySecrets(
 		return registrySecrets;
 	} catch (error) {
 		return exitWithExpectedError(
-			`Error validating registry secrets file "${secretsFilename}":\n${error.message}`,
+			`Error validating registry secrets file "${secretsFilename}":\n${
+				error.message
+			}`,
 		);
 	}
 }
@@ -142,7 +144,9 @@ async function performResolution(
 					buildTask.buildStream = clonedStream;
 					if (!buildTask.external && !buildTask.resolved) {
 						throw new Error(
-							`Project type for service "${buildTask.serviceName}" could not be determined. Missing a Dockerfile?`,
+							`Project type for service "${
+								buildTask.serviceName
+							}" could not be determined. Missing a Dockerfile?`,
 						);
 					}
 					return buildTask;

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 {

lib/utils/helpers.ts

@@ -45,9 +45,13 @@ export function stateToString(state: OperationState) {
 
 	switch (state.operation.command) {
 		case 'copy':
-			return `${result} ${state.operation.from.path} -> ${state.operation.to.path}`;
+			return `${result} ${state.operation.from.path} -> ${
+				state.operation.to.path
+			}`;
 		case 'replace':
-			return `${result} ${state.operation.file.path}, ${state.operation.copy} -> ${state.operation.replace}`;
+			return `${result} ${state.operation.file.path}, ${
+				state.operation.copy
+			} -> ${state.operation.replace}`;
 		case 'run-script':
 			return `${result} ${state.operation.script}`;
 		default:

lib/utils/remote-build.ts

@@ -279,7 +279,9 @@ function createRemoteBuildRequest(
 			if (response.statusCode >= 100 && response.statusCode < 400) {
 				if (DEBUG_MODE) {
 					console.log(
-						`[debug] received HTTP ${response.statusCode} ${response.statusMessage}`,
+						`[debug] received HTTP ${response.statusCode} ${
+							response.statusMessage
+						}`,
 					);
 				}
 			} else {

package.json

@@ -106,7 +106,7 @@
     "gulp-shell": "^0.5.2",
     "mochainon": "^2.0.0",
     "pkg": "~4.3.8",
-    "prettier": "^1.17.0",
+    "prettier": "1.17.0",
     "publish-release": "^1.6.0",
     "resin-lint": "^3.0.1",
     "rewire": "^3.0.2",
@@ -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",