From 9307209a80cadf743e1864f8cce682c4f5b602e0 Mon Sep 17 00:00:00 2001 From: shenlan Date: Tue, 30 Sep 2025 16:56:47 +0800 Subject: [PATCH] refactor: replace iac_run helper with python cli --- iac_modules/pulumi/README.md | 21 +- iac_modules/pulumi/cli.py | 390 ++++++++++++++++++++++++++++++++++ iac_modules/pulumi/iac_run.sh | 327 ---------------------------- 3 files changed, 405 insertions(+), 333 deletions(-) create mode 100755 iac_modules/pulumi/cli.py delete mode 100755 iac_modules/pulumi/iac_run.sh diff --git a/iac_modules/pulumi/README.md b/iac_modules/pulumi/README.md index 2179413e..20523041 100644 --- a/iac_modules/pulumi/README.md +++ b/iac_modules/pulumi/README.md @@ -34,7 +34,7 @@ GitHub Actions 与本地调试共享相同的一组环境变量。根据目标 ### 2.1 使用 `~/.iac/credentials` 管理多云凭据 -`iac_run.sh` 会在启动时默认尝试读取 `~/.iac/credentials`（可通过 `IAC_CREDENTIALS_FILE` 覆盖）。 +`cli.py` 会在启动时默认尝试读取 `~/.iac/credentials`（可通过 `IAC_CREDENTIALS_FILE` 或 `--credentials` 覆盖）。 - 为避免泄漏，文件权限需设置为 `0400`： @@ -80,15 +80,24 @@ S3 backend 的 Bucket 需提前创建，并为 Pulumi 访问角色授予读写 export IAC_STATE_BACKEND="s3://my-pulumi-state-bucket/modern-app" ``` -## 3. `iac_run.sh` 辅助脚本 +## 3. `cli.py` 辅助脚本 -为方便本地调试，目录下新增 `iac_run.sh`，与 GitHub Actions 的命令约定保持一致。执行前请确保以上环境变量均已配置。 +为方便本地调试，目录下提供了基于 Python 的 `cli.py`，与 GitHub Actions 的命令约定保持一致。执行前请确保以上环境变量（或凭据文件）均已配置。 ```bash cd iac_modules/pulumi -./iac_run.sh <命令> +python cli.py <命令> +# 或直接执行： ./cli.py <命令> ``` +脚本支持在命令后追加 `--stack`、`--backend`、`--backups-dir` 等参数来临时覆盖对应的环境变量。例如： + +```bash +python cli.py init --stack dev --backend s3://my-state-bucket/dev +``` + +`--credentials` 可用于指定其他凭据文件路径，默认读取 `~/.iac/credentials`。 + 支持的命令如下： | 命令 | 对应操作 | @@ -104,7 +113,7 @@ cd iac_modules/pulumi 查看帮助信息： ```bash -./iac_run.sh --help +python cli.py --help ``` ## 4. 配置目录与多云支持 @@ -121,7 +130,7 @@ Pulumi 入口脚本会根据配置文件中的根节点自动选择部署目标 export CONFIG_PATH="config/vultr/dev" ``` -随后运行 `./iac_run.sh init` 与 `./iac_run.sh create` 即可完成 Vultr 基线的部署与更新。 +随后运行 `python cli.py init` 与 `python cli.py create` 即可完成 Vultr 基线的部署与更新。 ## 5. 常见问题 diff --git a/iac_modules/pulumi/cli.py b/iac_modules/pulumi/cli.py new file mode 100755 index 00000000..f4f9a082 --- /dev/null +++ b/iac_modules/pulumi/cli.py @@ -0,0 +1,390 @@ +#!/usr/bin/env python3 +"""Utility commands for managing Pulumi stacks in this repository.""" +from __future__ import annotations + +import argparse +from datetime import datetime +import os +import stat +import subprocess +import sys +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Callable, Dict, Optional, Union + +PROJECT_DIR = Path(__file__).resolve().parent +DEFAULT_BACKUPS_DIR = Path("backups") +DEFAULT_CREDENTIALS_FILE = Path( + os.environ.get("IAC_CREDENTIALS_FILE", Path.home() / ".iac/credentials") +) + + +class CLIError(RuntimeError): + """Raised when the CLI encounters a user facing error.""" + + +@dataclass +class PulumiContext: + """Holds configuration required for running Pulumi commands.""" + + pulumi_bin: str + stack_name: Optional[str] + backend_url: Optional[str] + backups_dir: Path + + def run( + self, + *args: str, + check: bool = True, + capture_output: bool = False, + stdin: Optional[str] = None, + ) -> subprocess.CompletedProcess[str]: + """Execute a Pulumi command.""" + try: + return subprocess.run( # noqa: S603,S607 (command comes from environment) + [self.pulumi_bin, *args], + check=check, + capture_output=capture_output, + text=True, + input=stdin, + ) + except FileNotFoundError as exc: # pragma: no cover - runtime safeguard + raise CLIError( + f"Unable to locate Pulumi executable '{self.pulumi_bin}'." + ) from exc + + +# --------------------------------------------------------------------------- +# Credential loading helpers +# --------------------------------------------------------------------------- + +def _to_dict(value: Any) -> Dict[str, Any]: + return value if isinstance(value, dict) else {} + + +def _find_section(data: Dict[str, Any], name: str) -> Dict[str, Any]: + lname = name.lower() + for key, value in _to_dict(data).items(): + if str(key).lower() == lname: + return _to_dict(value) + return {} + + +def _find_value(section: Dict[str, Any], *names: str) -> Optional[Any]: + target_names = {name.lower() for name in names} + for key, value in _to_dict(section).items(): + if str(key).lower() in target_names: + return value + return None + + +def _ensure_string(value: Any) -> Optional[str]: + return value.strip() if isinstance(value, str) else None + + +def _select_backend(backends: Any) -> Optional[str]: + if isinstance(backends, str): + return _ensure_string(backends) + + if isinstance(backends, (list, tuple)): + candidates = [candidate for candidate in (_ensure_string(item) for item in backends) if candidate] + for candidate in candidates: + if candidate.lower().startswith("s3://"): + return candidate + return candidates[0] if candidates else None + + if isinstance(backends, dict): + direct = _find_value(backends, "url", "uri", "s3", "backend") + if isinstance(direct, (str, list, tuple, dict)): + selected = _select_backend(direct) + if selected: + return selected + for value in backends.values(): + selected = _select_backend(value) + if selected: + return selected + + return None + + +def _maybe_set_env(key: str, raw_value: Any) -> None: + value = _ensure_string(raw_value) + if value and not os.environ.get(key): + os.environ[key] = value + + +def _warn(message: str) -> None: + print(f"[警告] {message}", file=sys.stderr) + + +def _load_credentials_file(path: Path) -> None: + if not path.exists(): + return + + try: + perms = stat.S_IMODE(path.stat().st_mode) + if perms != 0o400: + _warn(f"{path} 权限建议设置为 0400（当前: {oct(perms)}）。") + except OSError: + pass + + try: + import yaml + except ModuleNotFoundError: # pragma: no cover - depends on runtime + _warn("解析凭据文件需要 PyYAML，请运行 'pip install PyYAML'.") + return + + try: + with path.open("r", encoding="utf-8") as handler: + data: Dict[str, Any] = yaml.safe_load(handler) or {} + except FileNotFoundError: + return + except yaml.YAMLError as exc: # type: ignore[attr-defined] + _warn(f"无法解析凭据文件: {exc}") + return + + iac_state = _find_section(data, "iac_state") + backend_section = _find_section(iac_state, "backend") + _maybe_set_env("IAC_STATE_BACKEND", _select_backend(backend_section)) + backend_region = _find_value(backend_section, "region", "aws_region", "default_region") + _maybe_set_env("AWS_REGION", backend_region) + _maybe_set_env("AWS_DEFAULT_REGION", backend_region) + + state_auth = _find_section(iac_state, "auth") + _maybe_set_env("AWS_ACCESS_KEY_ID", _find_value(state_auth, "ak", "access_key")) + _maybe_set_env("AWS_SECRET_ACCESS_KEY", _find_value(state_auth, "sk", "secret_key")) + + aws_section = ( + _find_section(data, "aws-global") + or _find_section(data, "aws_global") + or _find_section(data, "aws") + ) + _maybe_set_env("AWS_ACCESS_KEY_ID", _find_value(aws_section, "ak", "access_key", "access_key_id")) + _maybe_set_env("AWS_SECRET_ACCESS_KEY", _find_value(aws_section, "sk", "secret_key", "secret_access_key")) + aws_region = _find_value(aws_section, "region", "aws_region", "default_region") + _maybe_set_env("AWS_REGION", aws_region) + _maybe_set_env("AWS_DEFAULT_REGION", aws_region) + + alicloud_section = _find_section(data, "alicloud") + _maybe_set_env("ALICLOUD_ACCESS_KEY", _find_value(alicloud_section, "ak", "access_key", "access_key_id")) + _maybe_set_env("ALICLOUD_SECRET_KEY", _find_value(alicloud_section, "sk", "secret_key", "secret_access_key")) + + vultr_section = _find_section(data, "vultr") + _maybe_set_env("VULTR_API_KEY", _find_value(vultr_section, "api_key", "apikey")) + + +def _ensure_region_harmony() -> None: + if os.environ.get("AWS_REGION") and not os.environ.get("AWS_DEFAULT_REGION"): + os.environ["AWS_DEFAULT_REGION"] = os.environ["AWS_REGION"] + elif os.environ.get("AWS_DEFAULT_REGION") and not os.environ.get("AWS_REGION"): + os.environ["AWS_REGION"] = os.environ["AWS_DEFAULT_REGION"] + + +# --------------------------------------------------------------------------- +# Command helpers +# --------------------------------------------------------------------------- + +def _require_backend(context: PulumiContext) -> str: + backend = ( + context.backend_url + or os.environ.get("IAC_STATE_BACKEND") + or os.environ.get("IAC_State_backend") + ) + if not backend: + raise CLIError("未设置 IAC_STATE_BACKEND 环境变量，无法连接到 S3 backend.") + if not backend.startswith("s3://"): + raise CLIError("IAC_STATE_BACKEND 必须为 s3:// 开头的 Pulumi 后端地址.") + if not os.environ.get("AWS_REGION") and not os.environ.get("AWS_DEFAULT_REGION"): + raise CLIError( + "未设置 AWS_REGION 或 AWS_DEFAULT_REGION 环境变量，无法登录到 S3 backend." + " 请在凭据文件中添加 region 字段，或在运行脚本前导出该环境变量。" + ) + + context.backend_url = backend + context.run("login", backend) + return backend + + +def _require_stack(context: PulumiContext) -> str: + stack_name = context.stack_name or os.environ.get("PULUMI_STACK") + if not stack_name: + raise CLIError("未设置 PULUMI_STACK 环境变量.") + + context.stack_name = stack_name + result = context.run("stack", "select", stack_name, check=False, capture_output=True) + if result.returncode != 0: + context.run("stack", "init", stack_name) + return stack_name + + +def _command_init(context: PulumiContext, _: argparse.Namespace) -> None: + backend = _require_backend(context) + stack_name = _require_stack(context) + print(f"Pulumi backend 已配置: {backend}") + print(f"Pulumi stack 已就绪: {stack_name}") + + +def _command_create(context: PulumiContext, _: argparse.Namespace) -> None: + _require_backend(context) + stack = _require_stack(context) + context.run("up", "--stack", stack, "--yes", "--skip-preview") + + +def _command_migrate(context: PulumiContext, _: argparse.Namespace) -> None: + _require_backend(context) + stack = _require_stack(context) + context.run("refresh", "--stack", stack, "--yes") + + +def _command_upgrade(context: PulumiContext, _: argparse.Namespace) -> None: + _require_backend(context) + stack = _require_stack(context) + context.run("up", "--stack", stack, "--yes") + + +def _command_backup(context: PulumiContext, _: argparse.Namespace) -> None: + _require_backend(context) + stack = _require_stack(context) + + backups_dir = context.backups_dir + backups_dir.mkdir(parents=True, exist_ok=True) + + timestamp = datetime.utcnow().strftime("%Y%m%d%H%M%S") + backup_file = backups_dir / f"{stack}-{timestamp}.json" + export = context.run("stack", "export", "--stack", stack, capture_output=True) + backup_file.write_text(export.stdout, encoding="utf-8") + print(f"Pulumi stack 已备份到 {backup_file}") + + +def _command_restore(context: PulumiContext, args: argparse.Namespace) -> None: + _require_backend(context) + stack = _require_stack(context) + + backup_path = args.file or os.environ.get("BACKUP_FILE", "") + if not backup_path: + raise CLIError("restore 命令需要提供备份文件路径作为参数或通过 BACKUP_FILE 环境变量传入.") + + backup_file = Path(backup_path) + if not backup_file.is_file(): + raise CLIError(f"找不到备份文件 {backup_file}.") + + contents = backup_file.read_text(encoding="utf-8") + context.run("stack", "import", "--stack", stack, stdin=contents) + + +def _command_destroy(context: PulumiContext, _: argparse.Namespace) -> None: + _require_backend(context) + stack = _require_stack(context) + context.run("destroy", "--stack", stack, "--yes") + + +COMMANDS: Dict[str, Callable[[PulumiContext, argparse.Namespace], None]] = { + "init": _command_init, + "create": _command_create, + "migrate": _command_migrate, + "upgrade": _command_upgrade, + "backup": _command_backup, + "restore": _command_restore, + "destroy": _command_destroy, +} + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="cli.py", + description="Pulumi stack helper commands for the Modern Container Application reference architecture.", + ) + parser.add_argument( + "--credentials", + dest="credentials", + type=Path, + help="自定义凭据文件路径，默认读取 ~/.iac/credentials", + ) + + parent = argparse.ArgumentParser(add_help=False) + parent.add_argument( + "--stack", + dest="stack", + help="Pulumi Stack 名称（默认读取 PULUMI_STACK 环境变量）", + ) + parent.add_argument( + "--backend", + dest="backend", + help="Pulumi backend 地址（默认读取 IAC_STATE_BACKEND 环境变量或凭据文件）", + ) + parent.add_argument( + "--backups-dir", + dest="backups_dir", + type=Path, + help="备份文件保存目录，默认为 ./backups", + ) + + subparsers = parser.add_subparsers(dest="command", required=True) + + for command, handler in COMMANDS.items(): + help_text = { + "init": "登录 backend 并准备 Pulumi Stack", + "create": "执行 pulumi up --yes --skip-preview", + "migrate": "执行 pulumi refresh --yes", + "upgrade": "执行 pulumi up --yes", + "backup": "导出 Pulumi stack 状态到备份文件", + "restore": "从备份文件恢复 Pulumi stack", + "destroy": "销毁当前 Pulumi stack 资源", + }[command] + subparser = subparsers.add_parser(command, parents=[parent], help=help_text) + subparser.set_defaults(handler=handler) + if command == "restore": + subparser.add_argument( + "file", + nargs="?", + help="备份文件路径；也可使用 BACKUP_FILE 环境变量", + ) + + return parser + + +def main(argv: Optional[list[str]] = None) -> None: + os.chdir(PROJECT_DIR) + parser = _build_parser() + args = parser.parse_args(argv) + + credentials_path = args.credentials or DEFAULT_CREDENTIALS_FILE + _load_credentials_file(credentials_path) + _ensure_region_harmony() + + backups_dir_value: Optional[Union[Path, str]] + if args.backups_dir is not None: + backups_dir_value = args.backups_dir + else: + backups_dir_value = os.environ.get("PULUMI_BACKUP_DIR") + + backups_dir = Path(backups_dir_value) if backups_dir_value else DEFAULT_BACKUPS_DIR + + context = PulumiContext( + pulumi_bin=os.environ.get("PULUMI_BIN", "pulumi"), + stack_name=args.stack or os.environ.get("PULUMI_STACK") or os.environ.get("STACK_NAME") or os.environ.get("STACK"), + backend_url=args.backend or os.environ.get("IAC_STATE_BACKEND") or os.environ.get("IAC_State_backend"), + backups_dir=backups_dir, + ) + + if args.stack: + os.environ["PULUMI_STACK"] = args.stack + if args.backend: + os.environ["IAC_STATE_BACKEND"] = args.backend + if args.backups_dir: + os.environ["PULUMI_BACKUP_DIR"] = str(args.backups_dir) + + handler = getattr(args, "handler", COMMANDS[args.command]) + + try: + handler(context, args) + except CLIError as exc: + print(f"[错误] {exc}", file=sys.stderr) + raise SystemExit(1) from exc + except subprocess.CalledProcessError as exc: + print(f"[错误] Pulumi 命令执行失败: {' '.join(exc.cmd)}", file=sys.stderr) + raise SystemExit(exc.returncode) from exc + + +if __name__ == "__main__": + main() diff --git a/iac_modules/pulumi/iac_run.sh b/iac_modules/pulumi/iac_run.sh deleted file mode 100755 index 9bdd7669..00000000 --- a/iac_modules/pulumi/iac_run.sh +++ /dev/null @@ -1,327 +0,0 @@ -#!/usr/bin/env bash -set -euo pipefail - -PROJECT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" -cd "$PROJECT_DIR" - -PULUMI_BIN=${PULUMI_BIN:-pulumi} -STACK_NAME=${PULUMI_STACK:-${STACK_NAME:-${STACK:-}}} -BACKEND_URL=${IAC_STATE_BACKEND:-${IAC_State_backend:-}} -BACKUPS_DIR=${PULUMI_BACKUP_DIR:-backups} -CREDENTIALS_FILE=${IAC_CREDENTIALS_FILE:-${HOME}/.iac/credentials} - -load_credentials_file() { - if [[ ! -f "${CREDENTIALS_FILE}" ]]; then - return - fi - - if command -v stat >/dev/null 2>&1; then - local perms - if stat --version >/dev/null 2>&1; then - perms=$(stat -c "%a" "${CREDENTIALS_FILE}") - else - perms=$(stat -f "%OLp" "${CREDENTIALS_FILE}") - fi - if [[ "${perms}" != "400" ]]; then - echo "[警告] ${CREDENTIALS_FILE} 权限建议设置为 0400（当前: ${perms}）。" >&2 - fi - fi - - if ! command -v python3 >/dev/null 2>&1; then - echo "[警告] 未找到 python3，无法解析 ${CREDENTIALS_FILE}。" >&2 - return - fi - - local exports - if ! exports=$(python3 - "${CREDENTIALS_FILE}" <<'PY' -import os -import shlex -import sys - - -def to_dict(value): - return value if isinstance(value, dict) else {} - - -def find_section(data, name): - lname = name.lower() - for key, value in to_dict(data).items(): - if str(key).lower() == lname: - return value - return {} - - -def find_value(section, *names): - section_dict = to_dict(section) - lower_names = {n.lower() for n in names} - for key, value in section_dict.items(): - if str(key).lower() in lower_names: - return value - return None - - -def ensure_string(value): - if isinstance(value, str): - return value.strip() - return None - - -def maybe_export(env_key, raw_value, exports): - value = ensure_string(raw_value) - if value and not os.environ.get(env_key): - exports.append((env_key, value)) - - -def select_backend(backends): - if isinstance(backends, str): - return ensure_string(backends) - - if isinstance(backends, (list, tuple)): - candidates = [ensure_string(item) for item in backends if ensure_string(item)] - for candidate in candidates: - if candidate and candidate.lower().startswith("s3://"): - return candidate - return candidates[0] if candidates else None - - if isinstance(backends, dict): - direct = find_value(backends, "url", "uri", "s3", "backend") - if isinstance(direct, (str, list, tuple, dict)): - selected = select_backend(direct) - if selected: - return selected - - for value in backends.values(): - selected = select_backend(value) - if selected: - return selected - - return None - - -try: - import yaml -except ModuleNotFoundError: - sys.stderr.write("[警告] 解析凭据文件需要 PyYAML，请运行 'pip install PyYAML'.\n") - sys.exit(0) - - -path = sys.argv[1] -try: - with open(path, "r", encoding="utf-8") as fh: - data = yaml.safe_load(fh) or {} -except FileNotFoundError: - sys.exit(0) -except yaml.YAMLError as exc: - sys.stderr.write(f"[警告] 无法解析凭据文件: {exc}\n") - sys.exit(1) - -exports = [] - -iac_state = find_section(data, "iac_state") - backend_section = find_section(iac_state, "backend") - maybe_export("IAC_STATE_BACKEND", select_backend(backend_section), exports) - backend_region = find_value(backend_section, "region", "aws_region", "default_region") - maybe_export("AWS_REGION", backend_region, exports) - maybe_export("AWS_DEFAULT_REGION", backend_region, exports) - -state_auth = find_section(iac_state, "auth") -maybe_export("AWS_ACCESS_KEY_ID", find_value(state_auth, "ak", "access_key"), exports) -maybe_export("AWS_SECRET_ACCESS_KEY", find_value(state_auth, "sk", "secret_key"), exports) - -aws_section = ( - find_section(data, "aws-global") - or find_section(data, "aws_global") - or find_section(data, "aws") -) -maybe_export("AWS_ACCESS_KEY_ID", find_value(aws_section, "ak", "access_key", "access_key_id"), exports) -maybe_export("AWS_SECRET_ACCESS_KEY", find_value(aws_section, "sk", "secret_key", "secret_access_key"), exports) -aws_region = find_value(aws_section, "region", "aws_region", "default_region") -maybe_export("AWS_REGION", aws_region, exports) -maybe_export("AWS_DEFAULT_REGION", aws_region, exports) - -alicloud_section = find_section(data, "alicloud") -maybe_export("ALICLOUD_ACCESS_KEY", find_value(alicloud_section, "ak", "access_key", "access_key_id"), exports) -maybe_export("ALICLOUD_SECRET_KEY", find_value(alicloud_section, "sk", "secret_key", "secret_access_key"), exports) - -vultr_section = find_section(data, "vultr") -maybe_export("VULTR_API_KEY", find_value(vultr_section, "api_key", "apikey"), exports) - -if exports: - for key, value in exports: - print(f"export {key}={shlex.quote(value)}") -PY -); then - echo "[警告] 解析 ${CREDENTIALS_FILE} 失败，已跳过自动加载。" >&2 - return - fi - if [[ -n "${exports}" ]]; then - eval "${exports}" - fi -} - -load_credentials_file - -# Ensure region environment variables are harmonized -if [[ -z "${AWS_REGION:-}" && -n "${AWS_DEFAULT_REGION:-}" ]]; then - export AWS_REGION="${AWS_DEFAULT_REGION}" -elif [[ -z "${AWS_DEFAULT_REGION:-}" && -n "${AWS_REGION:-}" ]]; then - export AWS_DEFAULT_REGION="${AWS_REGION}" -fi - -BACKEND_URL=${IAC_STATE_BACKEND:-${IAC_State_backend:-${BACKEND_URL:-}}} - -usage() { - cat <<'USAGE' -用法: ./iac_run.sh <命令> [参数] - -可用命令: - init 初始化 Pulumi 后端与 Stack（需要 IAC_STATE_BACKEND 和 PULUMI_STACK） - create 创建或更新资源（pulumi up --yes --skip-preview） - migrate 同步当前资源状态到 S3 后端（pulumi refresh --yes） - upgrade 执行常规的基础设施更新（pulumi up --yes） - backup 导出 Stack 状态到本地文件（默认保存到 backups/ 目录） - restore <文件路径> 从指定备份文件恢复 Stack 状态 - destroy 销毁当前 Stack 中的所有资源 - -环境变量: - PULUMI_STACK Pulumi Stack 名称（必需） - IAC_STATE_BACKEND Pulumi backend 地址（必须为 s3:// 前缀） - CONFIG_PATH 自定义配置目录，可选 - PULUMI_BACKUP_DIR 备份输出目录，默认为 ./backups - -若使用 GitHub Actions 对齐流水线，请确保预先设置云厂商访问密钥与 Pulumi 访问令牌。 -USAGE -} - -require_backend() { - if [[ -z "${BACKEND_URL}" ]]; then - echo "[错误] 未设置 IAC_STATE_BACKEND（或 IAC_State_backend）环境变量，无法连接到 S3 backend." >&2 - exit 1 - fi - if [[ "${BACKEND_URL}" != s3://* ]]; then - echo "[错误] IAC_STATE_BACKEND 必须为 s3:// 开头的 Pulumi 后端地址." >&2 - exit 1 - fi - if [[ -z "${AWS_REGION:-}" && -z "${AWS_DEFAULT_REGION:-}" ]]; then - echo "[错误] 未设置 AWS_REGION 或 AWS_DEFAULT_REGION 环境变量，无法登录到 S3 backend." >&2 - echo " 请在凭据文件中添加 region 字段，或在运行脚本前导出该环境变量。" >&2 - exit 1 - fi - if [[ -z "${AWS_REGION:-}" ]]; then - export AWS_REGION="${AWS_DEFAULT_REGION}" - fi - if [[ -z "${AWS_DEFAULT_REGION:-}" ]]; then - export AWS_DEFAULT_REGION="${AWS_REGION}" - fi - "${PULUMI_BIN}" login "${BACKEND_URL}" >/dev/null -} - -require_stack() { - if [[ -z "${STACK_NAME}" ]]; then - echo "[错误] 未设置 PULUMI_STACK 环境变量." >&2 - exit 1 - fi - if ! "${PULUMI_BIN}" stack select "${STACK_NAME}" >/dev/null 2>&1; then - "${PULUMI_BIN}" stack init "${STACK_NAME}" - fi -} - -command_init() { - require_backend - require_stack - echo "Pulumi backend 已配置: ${BACKEND_URL}" - echo "Pulumi stack 已就绪: ${STACK_NAME}" -} - -command_create() { - require_backend - require_stack - "${PULUMI_BIN}" up --stack "${STACK_NAME}" --yes --skip-preview -} - -command_migrate() { - require_backend - require_stack - "${PULUMI_BIN}" refresh --stack "${STACK_NAME}" --yes -} - -command_upgrade() { - require_backend - require_stack - "${PULUMI_BIN}" up --stack "${STACK_NAME}" --yes -} - -command_backup() { - require_backend - require_stack - mkdir -p "${BACKUPS_DIR}" - timestamp=$(date +"%Y%m%d%H%M%S") - backup_file="${BACKUPS_DIR}/${STACK_NAME}-${timestamp}.json" - "${PULUMI_BIN}" stack export --stack "${STACK_NAME}" >"${backup_file}" - echo "Pulumi stack 已备份到 ${backup_file}" -} - -command_restore() { - require_backend - require_stack - local backup_file=${1:-${BACKUP_FILE:-}} - if [[ -z "${backup_file}" ]]; then - echo "[错误] restore 命令需要提供备份文件路径作为参数或通过 BACKUP_FILE 环境变量传入." >&2 - exit 1 - fi - if [[ ! -f "${backup_file}" ]]; then - echo "[错误] 找不到备份文件 ${backup_file}." >&2 - exit 1 - fi - "${PULUMI_BIN}" stack import --stack "${STACK_NAME}" <"${backup_file}" -} - -command_destroy() { - require_backend - require_stack - "${PULUMI_BIN}" destroy --stack "${STACK_NAME}" --yes -} - -main() { - if [[ $# -lt 1 ]]; then - usage - exit 1 - fi - - local command=$1 - shift - - case "${command}" in - init) - command_init "$@" - ;; - create) - command_create "$@" - ;; - migrate) - command_migrate "$@" - ;; - upgrade) - command_upgrade "$@" - ;; - backup) - command_backup "$@" - ;; - restore) - command_restore "$@" - ;; - destroy) - command_destroy "$@" - ;; - -h|--help|help) - usage - ;; - *) - echo "[错误] 未知命令: ${command}" >&2 - usage - exit 1 - ;; - esac -} - -main "$@"