コマンドの実行

exec() を使うと、稼働中の Container 内で別のプロセスを起動できます。次の例は、@cloudflare/containers の Container を継承したクラス内で this.ctx.container.exec() を呼び出します。

exec() は停止中の Container を起動しません。RPC メソッドでは this.ctx.container.running を確認し、必要なら await this.start() を呼び出します。Container 起動時に一連のコマンドを実行したい場合は、onStart() フックも使えます。

起動後にプロセスを実行する

次のフックは、Container が起動するたびに準備用コマンドを実行します。このフックから任意の起動コマンドを実行できます。output() は標準出力と標準エラーを、それぞれ別の ArrayBuffer としてバッファします。

import { Container } from "@cloudflare/containers";

export class MyContainer extends Container {
	async onStart() {
		const process = await this.ctx.container.exec([
			"node",
			"scripts/prepare.js",
		]);
		const output = await process.output();
		const decoder = new TextDecoder();

		if (output.exitCode !== 0) {
			throw new Error(
				`Container preparation failed: ${decoder.decode(output.stderr)}`,
			);
		}

		console.log(decoder.decode(output.stdout));
	}
}

import { Container } from "@cloudflare/containers";

export class MyContainer extends Container {
	override async onStart() {
		const process = await this.ctx.container.exec([
			"node",
			"scripts/prepare.js",
		]);
		const output = await process.output();
		const decoder = new TextDecoder();

	if (output.exitCode !== 0) {
		throw new Error(
			`Container preparation failed: ${decoder.decode(output.stderr)}`,
		);
	}

	console.log(decoder.decode(output.stdout));
    }

}

RPC メソッドでは、exec() を呼ぶ前に Container が稼働していることを確認します。標準出力はデフォルトで読み取り可能なストリームです。

import { Container } from "@cloudflare/containers";

export class MyContainer extends Container {
	async readVersion() {
		if (!this.ctx.container.running) {
			await this.start();
		}

		const process = await this.ctx.container.exec(["node", "--version"]);
		const stdout = process.stdout
			? await new Response(process.stdout).text()
			: "";
		const exitCode = await process.exitCode;

		return { pid: process.pid, stdout, exitCode };
	}
}

import { Container } from "@cloudflare/containers";

export class MyContainer extends Container {
	async readVersion() {
		if (!this.ctx.container.running) {
			await this.start();
		}

		const process = await this.ctx.container.exec(["node", "--version"]);
		const stdout = process.stdout
			? await new Response(process.stdout).text()
			: "";
		const exitCode = await process.exitCode;

		return { pid: process.pid, stdout, exitCode };
	}
}

戻り値の pid は新しいプロセスを識別します。exitCode の Promise は、そのプロセスが終了したときに解決します。

引数を直接渡す

exec() は、渡した引数配列で実行ファイルを直接起動します。先にシェルは起動しません。

配列の各要素が 1 つの引数になります。パイプ、リダイレクト、グロブ、変数展開などのシェル機能は暗黙には動きません。

それらの機能が必要なコマンドでは、シェルを呼び出します。イメージに Bash がある場合は ["bash", "-lc", "<COMMAND>"] を使います。Portable Operating System Interface（POSIX）シェルだけがある場合は ["sh", "-c", "<COMMAND>"] を使います。信頼できない値は、シェルコマンド文字列へ埋め込まず、別引数として渡します。

標準入力を送る

既存データを送るには ReadableStream を渡します。stdout を "ignore" にすると、標準出力は破棄されます。

import { Container } from "@cloudflare/containers";

export class MyContainer extends Container {
	async importData(data) {
		if (!this.ctx.container.running) {
			await this.start();
		}

		const stdin = new ReadableStream({
			start(controller) {
				controller.enqueue(new TextEncoder().encode(data));
				controller.close();
			},
		});
		const process = await this.ctx.container.exec(["cat"], {
			stdin,
			stdout: "ignore",
		});
		const output = await process.output();

		return {
			stdoutBytes: output.stdout.byteLength,
			exitCode: output.exitCode,
		};
	}
}

import { Container } from "@cloudflare/containers";

export class MyContainer extends Container {
	async importData(data: string) {
		if (!this.ctx.container.running) {
			await this.start();
		}

		const stdin = new ReadableStream<Uint8Array>({
			start(controller) {
				controller.enqueue(new TextEncoder().encode(data));
				controller.close();
			},
		});
		const process = await this.ctx.container.exec(["cat"], {
			stdin,
			stdout: "ignore",
		});
		const output = await process.output();

		return {
			stdoutBytes: output.stdout.byteLength,
			exitCode: output.exitCode,
		};
	}
}

無視した標準出力は、output() では空のバッファになります。時間をかけて書き込む場合は、stdin を "pipe" にします。

import { Container } from "@cloudflare/containers";

export class MyContainer extends Container {
	async concatenateInput() {
		if (!this.ctx.container.running) {
			await this.start();
		}

		const process = await this.ctx.container.exec(["cat"], {
			stdin: "pipe",
		});
		const writer = process.stdin?.getWriter();

		if (!writer) {
			throw new Error("Standard input is unavailable");
		}

		const encoder = new TextEncoder();
		await writer.write(encoder.encode("first\n"));
		await writer.write(encoder.encode("second\n"));
		await writer.close();

		const output = await process.output();
		return new TextDecoder().decode(output.stdout);
	}
}

import { Container } from "@cloudflare/containers";

export class MyContainer extends Container {
	async concatenateInput() {
		if (!this.ctx.container.running) {
			await this.start();
		}

		const process = await this.ctx.container.exec(["cat"], {
			stdin: "pipe",
		});
		const writer = process.stdin?.getWriter();

		if (!writer) {
			throw new Error("Standard input is unavailable");
		}

		const encoder = new TextEncoder();
		await writer.write(encoder.encode("first\n"));
		await writer.write(encoder.encode("second\n"));
		await writer.close();

		const output = await process.output();
		return new TextDecoder().decode(output.stdout);
	}
}

writer を閉じると、ファイル終端（EOF）を送ります。stdin を省略すると、exec() は標準入力を閉じて、すぐに EOF を送ります。

標準入力へ RPC ストリームを渡す

RPC メソッドは、元のソースが type: "bytes" のバイト指向 ReadableStream を受け取れます。Request の body はこの条件を満たします。受け取ったストリームは、Durable Object 内で全体をバッファせず、そのまま exec() に渡せます。詳細は RPC 上のストリームを参照してください。

import { Container, getContainer } from "@cloudflare/containers";

export class MyContainer extends Container {
	async writeFile(input) {
		if (!this.ctx.container.running) {
			await this.start();
		}

		const process = await this.ctx.container.exec(["tee", "/tmp/upload.bin"], {
			stdin: input,
			stdout: "ignore",
		});

		return process.exitCode;
	}
}

export default {
	async fetch(request, env) {
		if (!request.body) {
			return new Response("Request body required", { status: 400 });
		}

		const container = getContainer(env.MY_CONTAINER, "upload-worker");
		const exitCode = await container.writeFile(request.body);

		return Response.json({ exitCode });
	},
};

import { Container, getContainer } from "@cloudflare/containers";

export class MyContainer extends Container {
	async writeFile(input: ReadableStream<Uint8Array>) {
		if (!this.ctx.container.running) {
			await this.start();
		}

	const process = await this.ctx.container.exec(
		["tee", "/tmp/upload.bin"],
		{
			stdin: input,
			stdout: "ignore",
		},
	);

	return process.exitCode;
    }

}

export default {
	async fetch(request: Request, env: Env): Promise<Response> {
		if (!request.body) {
			return new Response("Request body required", { status: 400 });
		}

	const container = getContainer(env.MY_CONTAINER, "upload-worker");
	const exitCode = await container.writeFile(request.body);

	return Response.json({ exitCode });
    },

};

RPC はストリームの所有権を Durable Object へ移します。呼び出し元の Worker は、writeFile() に渡したあと読み取れません。

次の cat プロセスは、標準入力を省略しているため終了します。

import { Container } from "@cloudflare/containers";

export class MyContainer extends Container {
	async verifyEndOfFile() {
		if (!this.ctx.container.running) {
			await this.start();
		}

		const process = await this.ctx.container.exec(["cat"]);
		const output = await process.output();

		return {
			stdoutBytes: output.stdout.byteLength,
			exitCode: output.exitCode,
		};
	}
}

import { Container } from "@cloudflare/containers";

export class MyContainer extends Container {
	async verifyEndOfFile() {
		if (!this.ctx.container.running) {
			await this.start();
		}

		const process = await this.ctx.container.exec(["cat"]);
		const output = await process.output();

		return {
			stdoutBytes: output.stdout.byteLength,
			exitCode: output.exitCode,
		};
	}
}

プロセスコンテキストを設定する

cwd、env、user でプロセスコンテキストを設定します。プロセスは envVars で設定した Container 環境を継承します。実行ごとの env は変数を追加するか、同じキーを上書きします。

この例は展開とリダイレクトが必要なため sh を使います。標準出力と標準エラーも別々に取得します。

import { Container } from "@cloudflare/containers";

export class MyContainer extends Container {
	envVars = {
		BASE_VALUE: "inherited",
		MODE: "default",
	};

	async inspectWorkspace() {
		if (!this.ctx.container.running) {
			await this.start();
		}

		const process = await this.ctx.container.exec(
			[
				"sh",
				"-c",
				'printf "%s:%s:%s:%s" "$PWD" "$BASE_VALUE" "$MODE" "$EXTRA_VALUE"; printf "diagnostic" >&2',
			],
			{
				cwd: "/workspace",
				env: {
					MODE: "inspection",
					EXTRA_VALUE: "added",
				},
			},
		);
		const output = await process.output();
		const decoder = new TextDecoder();

		return {
			stdout: decoder.decode(output.stdout),
			stderr: decoder.decode(output.stderr),
		};
	}
}

import { Container } from "@cloudflare/containers";

export class MyContainer extends Container {
	envVars = {
		BASE_VALUE: "inherited",
		MODE: "default",
	};

	async inspectWorkspace() {
		if (!this.ctx.container.running) {
			await this.start();
		}

		const process = await this.ctx.container.exec(
			[
				"sh",
				"-c",
				'printf "%s:%s:%s:%s" "$PWD" "$BASE_VALUE" "$MODE" "$EXTRA_VALUE"; printf "diagnostic" >&2',
			],
			{
				cwd: "/workspace",
				env: {
					MODE: "inspection",
					EXTRA_VALUE: "added",
				},
			},
		);
		const output = await process.output();
		const decoder = new TextDecoder();

		return {
			stdout: decoder.decode(output.stdout),
			stderr: decoder.decode(output.stderr),
		};
	}
}

user オプションは、プロセスのユーザー名または数値のユーザー ID（UID）を設定します。Container ランタイムは、コンテナイメージからユーザー名を解決します。

標準エラーを結合する

stderr を "combined" にすると、標準エラーを標準出力へマージします。結合出力には stdout: "pipe" が必要です。

import { Container } from "@cloudflare/containers";

export class MyContainer extends Container {
	async readCombinedOutput() {
		if (!this.ctx.container.running) {
			await this.start();
		}

		const process = await this.ctx.container.exec(
			[
				"bash",
				"-lc",
				'printf "standard output\n"; printf "standard error\n" >&2',
			],
			{
				stdout: "pipe",
				stderr: "combined",
			},
		);
		const output = await process.output();

		return new TextDecoder().decode(output.stdout);
	}
}

import { Container } from "@cloudflare/containers";

export class MyContainer extends Container {
	async readCombinedOutput() {
		if (!this.ctx.container.running) {
			await this.start();
		}

		const process = await this.ctx.container.exec(
			[
				"bash",
				"-lc",
				'printf "standard output\n"; printf "standard error\n" >&2',
			],
			{
				stdout: "pipe",
				stderr: "combined",
			},
		);
		const output = await process.output();

		return new TextDecoder().decode(output.stdout);
	}
}

マージしたストリームは、元ストリーム間の順序を保証しません。このモードでは process.stderr は null で、output.stderr は空の ArrayBuffer です。この例は、イメージに Bash がある前提です。

非ゼロ終了を扱う

終了コードが非ゼロでも、exitCode は通常どおり解決します。Promise は拒否されません。

この例は標準エラーを残し、標準出力は無視します。

import { Container } from "@cloudflare/containers";

export class MyContainer extends Container {
	async runCheck() {
		if (!this.ctx.container.running) {
			await this.start();
		}

		const process = await this.ctx.container.exec(
			[
				"sh",
				"-c",
				'printf "not captured"; printf "check failed\n" >&2; exit 7',
			],
			{ stdout: "ignore" },
		);
		const output = await process.output();

		return {
			exitCode: output.exitCode,
			stdoutBytes: output.stdout.byteLength,
			stderr: new TextDecoder().decode(output.stderr),
		};
	}
}

import { Container } from "@cloudflare/containers";

export class MyContainer extends Container {
	async runCheck() {
		if (!this.ctx.container.running) {
			await this.start();
		}

	const process = await this.ctx.container.exec(
		[
			"sh",
			"-c",
			'printf "not captured"; printf "check failed\n" >&2; exit 7',
		],
		{ stdout: "ignore" },
	);
	const output = await process.output();

	return {
		exitCode: output.exitCode,
		stdoutBytes: output.stdout.byteLength,
		stderr: new TextDecoder().decode(output.stderr),
	};
    }

}

結果には終了コード 7 と標準エラーのテキストが含まれます。無視した標準出力バッファのバイト数は 0 です。

大きな出力をストリームする

output() は両方のストリームをメモリにバッファします。大きな出力では、代わりに stdout と stderr を同時に読み切ってください。

import { Container } from "@cloudflare/containers";

async function countBytes(stream) {
	if (!stream) {
		return 0;
	}

	let bytes = 0;
	for await (const chunk of stream) {
		bytes += chunk.byteLength;
	}
	return bytes;
}

export class MyContainer extends Container {
	async generateLargeOutput() {
		if (!this.ctx.container.running) {
			await this.start();
		}

		const process = await this.ctx.container.exec([
			"sh",
			"-c",
			'i=0; while [ "$i" -lt 100000 ]; do printf "output %s\n" "$i"; printf "error %s\n" "$i" >&2; i=$((i + 1)); done',
		]);

		const [stdoutBytes, stderrBytes, exitCode] = await Promise.all([
			countBytes(process.stdout),
			countBytes(process.stderr),
			process.exitCode,
		]);

		return { stdoutBytes, stderrBytes, exitCode };
	}
}

import { Container } from "@cloudflare/containers";

async function countBytes(stream: ReadableStream<Uint8Array> | null) {
	if (!stream) {
		return 0;
	}

	let bytes = 0;
	for await (const chunk of stream) {
		bytes += chunk.byteLength;
	}
	return bytes;
}

export class MyContainer extends Container {
	async generateLargeOutput() {
		if (!this.ctx.container.running) {
			await this.start();
		}

		const process = await this.ctx.container.exec([
			"sh",
			"-c",
			'i=0; while [ "$i" -lt 100000 ]; do printf "output %s\n" "$i"; printf "error %s\n" "$i" >&2; i=$((i + 1)); done',
		]);

		const [stdoutBytes, stderrBytes, exitCode] = await Promise.all([
			countBytes(process.stdout),
			countBytes(process.stderr),
			process.exitCode,
		]);

		return { stdoutBytes, stderrBytes, exitCode };
	}
}

ストリーミングと output() は、どちらか一方で消費する方法です。どちらかのストリームの消費が始まっていると、output() は TypeError を投げます。output() の 2 回目の呼び出しも TypeError を投げます。

RPC 経由で標準出力を返す

RPC メソッドから ReadableStream を返すと、呼び出し元の Worker へ出力をストリームできます。標準エラーを結合すると、両方の出力チャネルを 1 つのストリームにできます。

import { Container } from "@cloudflare/containers";

export class MyContainer extends Container {
	async streamCommandOutput() {
		if (!this.ctx.container.running) {
			await this.start();
		}

		const process = await this.ctx.container.exec(
			["sh", "-c", 'printf "starting\n"; run-report'],
			{ stderr: "combined" },
		);

		return process.stdout;
	}
}

import { Container } from "@cloudflare/containers";

export class MyContainer extends Container {
	async streamCommandOutput(): Promise<ReadableStream<Uint8Array>> {
		if (!this.ctx.container.running) {
			await this.start();
		}

	const process = await this.ctx.container.exec(
		["sh", "-c", 'printf "starting\n"; run-report'],
		{ stderr: "combined" },
	);

	return process.stdout!;
    }

}

RPC はストリームの所有権を呼び出し元へ移し、フロー制御も維持します。呼び出し元はストリームを消費するかキャンセルする必要があります。呼び出し元が読み取りを止めると、バックプレッシャーで書き込みを続けるプロセスが一時停止することがあります。

このメソッドが渡すのは出力であり、ExecProcess ハンドルではありません。呼び出し元が完了メタデータやプロセス制御を必要とする場合は、別のアプリケーションプロトコルを定義してください。

プロセスを停止する

exec() に組み込みのタイムアウトはありません。遅延後に終了させたい場合は kill() で終了を要求し、その後 exitCode を待ちます。

import { Container } from "@cloudflare/containers";

export class MyContainer extends Container {
	async runWithTimeout() {
		if (!this.ctx.container.running) {
			await this.start();
		}

		const process = await this.ctx.container.exec(["sleep", "120"]);
		const timer = setTimeout(() => process.kill(), 30_000);

		try {
			return await process.exitCode;
		} finally {
			clearTimeout(timer);
		}
	}
}

import { Container } from "@cloudflare/containers";

export class MyContainer extends Container {
	async runWithTimeout() {
		if (!this.ctx.container.running) {
			await this.start();
		}

		const process = await this.ctx.container.exec(["sleep", "120"]);
		const timer = setTimeout(() => process.kill(), 30_000);

		try {
			return await process.exitCode;
		} finally {
			clearTimeout(timer);
		}
	}
}

引数なしの kill() は SIGTERM（シグナル 15）をキューします。プロセスが必要とする場合は、別のシグナルを渡せます。プロセスはシグナルを処理したり無視したりできるため、これは厳密な実行期限ではありません。完了は exitCode で確認し、シグナルから特定の終了コードを推測しないでください。

操作を協調させる

exec() の呼び出しは、Container を制御する Durable Object に置きます。Durable Object がプロセス状態と Container のライフサイクルを調整できます。

1 つのアプリケーション RPC メソッドで、複数の exec() 操作を実行できます。各コマンドは別々の exec 操作のままですが、呼び出し元の Durable Object RPC は 1 回です。呼び出し元と Durable Object の往復を減らしつつ、ライフサイクルの判断をまとめられます。

import { Container } from "@cloudflare/containers";

export class MyContainer extends Container {
	async runDiagnostics() {
		if (!this.ctx.container.running) {
			await this.start();
		}

		const commands = [
			["uname", "-a"],
			["node", "--version"],
		];
		const decoder = new TextDecoder();
		const results = [];

		for (const command of commands) {
			const process = await this.ctx.container.exec(command);
			const output = await process.output();
			results.push({
				command,
				exitCode: output.exitCode,
				stdout: decoder.decode(output.stdout),
				stderr: decoder.decode(output.stderr),
			});
		}

		return results;
	}
}

import { Container } from "@cloudflare/containers";

export class MyContainer extends Container {
	async runDiagnostics() {
		if (!this.ctx.container.running) {
			await this.start();
		}

		const commands = [
			["uname", "-a"],
			["node", "--version"],
		];
		const decoder = new TextDecoder();
		const results = [];

		for (const command of commands) {
			const process = await this.ctx.container.exec(command);
			const output = await process.output();
			results.push({
				command,
				exitCode: output.exitCode,
				stdout: decoder.decode(output.stdout),
				stderr: decoder.decode(output.stderr),
			});
		}

		return results;
	}
}

フィールドと戻り値の型はすべて exec() の API 契約を参照してください。