Merge pull request #123 from svc-design/codex/rename-iac_run.sh-to-cli.py
refactor: replace shell-based Pulumi helper with Python CLI
This commit is contained in:
shenlan
GitHub
commit
902da7f62e
@ -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. 常见问题
|
||||
|
||||
|
||||
390
iac_modules/pulumi/cli.py
Executable file
390
iac_modules/pulumi/cli.py
Executable file
@ -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()
|
||||
@ -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 "$@"
|
||||
Loading…
Reference in New Issue
Block a user