执行命令 · Cloudflare Containers docs

使用 exec() 在运行的容器内启动另一个进程。这些示例在扩展了 @cloudflare/containers 中 Container 的类中调用 this.ctx.container.exec()。

exec() 不会启动停止的容器。在远程过程调用 (RPC) 方法中，请检查 this.ctx.container.running，并在需要时调用 await this.start()。您还可以使用 onStart() 钩子在容器启动时运行任何系列的命令。

启动后运行进程

只要容器启动，以下钩子就会运行一条准备命令。您可以从此钩子执行任何系列的启动命令。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() 之前容器正在运行。默认情况下，标准输出使用可读流。

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() 操作使用提供的参数数组直接启动可执行文件。它不会首先启动 shell。

每个数组项成为一个参数。管道、重定向、通配符和变量扩展等 shell 功能不会隐式运行。

当您的命令需要这些功能时，请调用 shell。如果您的镜像中存在 Bash，请使用 ["bash", "-lc", "<COMMAND>"]。如果该镜像仅提供可移植操作系统接口 (POSIX) shell，请使用 ["sh", "-c", "<COMMAND>"]。将不受信任的值作为单独的参数传递，而不是将它们插值到 shell 命令字符串中。

发送标准输入

传递一个 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 主体符合此要求。您可以将接收到的流直接传递给 exec()，而无需在 Durable Object 中缓冲整个流。有关更多信息，请参阅 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 设置的容器环境。每个执行的 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)。容器运行时将解析容器镜像中的用户名。

合并标准错误

将 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 和标准错误文本。其被忽略的标准输出缓冲区的字节数为零。

流式传输大型输出

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() 也会抛出 TypeError。

通过 RPC 返回标准输出

从 RPC 方法返回一个 ReadableStream，以将输出流式传输至调用它的 Worker。合并标准错误为两个输出通道提供了一个流。

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() 调用放置在控制容器的 Durable Object 中。Durable Object 可以协调进程状态和容器生命周期。

一个应用程序 RPC 方法可以执行多个 exec() 操作。每个命令仍是独立的 exec 操作，但调用者只需进行一次 Durable Object RPC 调用。这减少了调用者到 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 契约。