← Containers / containers

Выполнять команды

Используйте exec() чтобы запустить другой процесс внутри работающего Контейнер. В примерах вызывается this.ctx.container.exec() внутри класса, расширяющего Container от @cloudflare/containers.

exec() не запускает остановленный Container. В методах удалённого вызова процедур (RPC) проверяйте this.ctx.container.running и вызвать await this.start() при необходимости. Также можно использовать onStart()-хук для выполнения любой последовательности команд при каждом запуске Container.

Запуск процесса после старта

Следующий хук запускает подготовительную команду при каждом запуске 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 убедитесь, что Container запущен, прежде чем вызывать 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 промис выполняется, когда этот процесс завершается.

Передача аргументов напрямую

exec() операция запускает исполняемый файл напрямую с переданным массивом аргументов. Она не запускает оболочку заранее.

Каждый элемент массива становится отдельным аргументом. Такие возможности командной оболочки, как конвейеры, перенаправление ввода-вывода, globbing и подстановка переменных, при этом не применяются.

Вызывайте оболочку, если вашей команде нужны эти возможности. Используйте ["bash", "-lc", "<COMMAND>"] если Bash присутствует в вашем образе. Используйте ["sh", "-c", "<COMMAND>"] если в образе доступна только POSIX-совместимая оболочка (Portable Operating System Interface). Передавайте недоверенные значения как отдельные аргументы вместо их подстановки в строку команды оболочки.

Отправка стандартного ввода

Передайте 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 могут принимать байт-ориентированные ReadableStream значения, чей исходный источник использует type: "bytes". 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 опция задаёт имя пользователя или числовой идентификатор пользователя (UID) для процесса. Container runtime определяет имена пользователей на основе образа контейнера.

Объедините стандартный поток ошибок

Задайте 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 в обычном режиме. Промис при этом не отклоняется.

В этом примере стандартный вывод ошибок сохраняется, а стандартный вывод игнорируется:

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

Возвращает ReadableStream из метода RPC для потоковой передачи вывода вызывающему 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, который управляет Container. Durable Object может координировать состояние процесса и жизненный цикл Container.

Один метод RPC приложения может выполнять несколько exec() операции. Каждая команда остаётся отдельной операцией exec, но вызывающая сторона делает один RPC вызов к Durable Object. Это уменьшает количество обращений от вызывающей стороны к 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.