diff --git a/deploy/orchestration/README.md b/deploy/orchestration/README.md new file mode 100644 index 0000000..59a51e6 --- /dev/null +++ b/deploy/orchestration/README.md @@ -0,0 +1,66 @@ +# 一键部署编排 + 回滚引擎(Issue #60 / PRD 5.6) + +> 父 Issue「⑥ 一键部署脚本与回滚机制 · 0.5d」 + +把一键部署与回滚落为**可测试的纯标准库编排引擎**——不真执行 k8s/helm,但模拟 +完整编排序列(pre_check → deploy → health_check → post_check)+ 部署前快照 + +失败回滚。对齐 `deploy/k8s/helm/iaop`(Chart)与 `deploy/k8s/healthz`(探针)。 + +## 1. 部署编排(`deploy_plan.py`) + +- `DeployStep` —— 声明式步骤:`kind`(pre_check/deploy/health_check/post_check) + + `name` + `action` 回调(返回 `(ok, detail)`,默认模拟成功)。 +- `DeployPlan` —— 步骤序列 + 发布元信息(release/namespace/chart/version); + `default_helm_release()` 生成默认 4 步 Helm 发布计划;步骤顺序约束 + (pre_check < deploy < health_check < post_check)自动校验。 +- `DeployOrchestrator` —— 按序执行,任何必需步骤失败即中止并标记回滚 + (`DeployOutcome.needs_rollback`),产出 `DeployReport`(逐步结果 + 日志 + + 可解释 reason)。deploy/health_check 是回滚检查点;pre_check 失败无需回滚。 +- `HealthCheckContract` —— 健康检查契约(端点/超时/重试/可用率目标,对齐 + `deploy/k8s/healthz` 的 99.8%)。 + +```python +from deploy_plan import DeployPlan, DeployOrchestrator + +plan = DeployPlan.default_helm_release(release="iaop", backend="gpu") +report = DeployOrchestrator(plan).execute() +if not report.succeeded and report.outcome.needs_rollback: + print("需回滚:", report.outcome.reason) +``` + +## 2. 回滚引擎(`rollback.py`) + +- `RollbackPoint` —— 部署前快照(release/namespace/version/chart/values_hash/ + created_at,不可变)。`values_hash`(SHA1)检测配置漂移。 +- `RollbackManager` —— 快照栈 + 回滚执行:`snapshot(plan)` 部署前压栈; + `rollback_to_latest()` / `rollback(n)` 失败时回滚到稳定版本。回滚失败把目标压回 + 栈顶(保持栈一致性);支持回滚后健康检查。FIFO 淘汰防内存膨胀。 +- `RollbackResult` —— 回滚结果(SUCCESS/FAILED/NO_TARGET/SKIPPED + 可解释 reason)。 + +```python +from deploy_plan import DeployPlan, DeployOrchestrator +from rollback import RollbackManager + +mgr = RollbackManager() +mgr.snapshot(plan) # 部署前快照 +report = DeployOrchestrator(plan).execute() +if not report.succeeded: + result = mgr.rollback_to_latest(reason=report.outcome.reason) + print(result.status.value, result.detail) +``` + +## 与现有部署资产的关系 + +- 真实环境把 `action` / `restore_action` 注入为 `helm upgrade` / `helm rollback` + 等真回调即可;本引擎只做编排与状态管理,不耦合 k8s 客户端。 +- 健康检查契约对齐 `deploy/k8s/healthz/probe_availability.py`(99.8% 可用率目标)。 + +## 测试 + +```bash +python -m unittest discover -s deploy/orchestration/tests -p "test_*.py" -v +``` + +覆盖正常 + 边界 + 错误(30 用例):4 步计划构造、顺序约束、全成功/各类失败、 +回滚检查点判定、action 异常、dry_run、空计划、快照压栈/FIFO、回滚成功/失败/ +多版本/空栈/异常、健康检查契约校验、values 漂移、报告序列化。 diff --git a/deploy/orchestration/__init__.py b/deploy/orchestration/__init__.py new file mode 100644 index 0000000..9ce4499 --- /dev/null +++ b/deploy/orchestration/__init__.py @@ -0,0 +1,42 @@ +# -*- coding: utf-8 -*- +"""一键部署编排 + 回滚引擎包(Issue #60 / PRD 5.6「⑥ 部署底座」)。 + +把部署与回滚落为**可测试的纯标准库编排引擎**——不真执行 k8s/helm,但模拟 +完整编排序列(pre_check → deploy → health_check → post_check)+ 部署前快照 ++ 失败回滚。零运行时依赖,与 iAOP 部署底座一致。 +""" +from .deploy_plan import ( + DeployOrchestrator, + DeployOutcome, + DeployPlan, + DeployReport, + DeployStep, + DeployStepKind, + DeployStepResult, + DeployStatus, + HealthCheckContract, +) +from .rollback import ( + RollbackManager, + RollbackPoint, + RollbackResult, + RollbackStatus, +) + +__all__ = [ + # deploy_plan + "DeployPlan", + "DeployStep", + "DeployStepKind", + "DeployStepResult", + "DeployStatus", + "DeployOutcome", + "DeployReport", + "DeployOrchestrator", + "HealthCheckContract", + # rollback + "RollbackPoint", + "RollbackManager", + "RollbackResult", + "RollbackStatus", +] diff --git a/deploy/orchestration/_sanity_check.py b/deploy/orchestration/_sanity_check.py new file mode 100644 index 0000000..2359249 --- /dev/null +++ b/deploy/orchestration/_sanity_check.py @@ -0,0 +1,51 @@ +# -*- coding: utf-8 -*- +"""部署编排 + 回滚冒烟脚本(Issue #60)。 + +直接运行 ``python _sanity_check.py`` 验证:默认 Helm 计划全成功部署、 +快照压栈、模拟部署失败触发回滚到上一稳定版本。零第三方依赖。 +""" +import os +import sys + +HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, HERE) + +from deploy_plan import ( # noqa: E402 + DeployOrchestrator, DeployPlan, DeployStepKind) +from rollback import RollbackManager # noqa: E402 + + +def main() -> int: + # 1) 部署前快照(当前稳定版本 v1.0.0) + mgr = RollbackManager() + stable = DeployPlan.default_helm_release(version="v1.0.0") + mgr.snapshot(stable, reason="部署 v1.1.0 前的稳定版本") + print(f"[OK] 快照压栈:稳定版本 {mgr.latest().version}") + + # 2) 部署 v1.1.0 全成功 + new_plan = DeployPlan.default_helm_release(version="v1.1.0") + report = DeployOrchestrator(new_plan).execute() + assert report.succeeded, report.outcome.reason + mgr.snapshot(new_plan, reason="v1.1.0 部署成功") + print(f"[OK] 部署 v1.1.0 成功:{len(report.results)} 步,无需回滚") + + # 3) 模拟 v1.2.0 部署失败(health_check 失败)→ 触发回滚 + def _bad_health(ctx): + return False, "/health 503(服务未就绪)" + bad_plan = DeployPlan.default_helm_release( + version="v1.2.0", + step_actions={DeployStepKind.HEALTH_CHECK: _bad_health}) + bad_report = DeployOrchestrator(bad_plan).execute() + assert not bad_report.succeeded + assert bad_report.outcome.needs_rollback + print(f"[OK] v1.2.0 部署失败:{bad_report.outcome.reason}") + + result = mgr.rollback_to_latest(reason="v1.2.0 健康检查失败") + assert result.succeeded, result.detail + print(f"[OK] 回滚到 {result.target.version}:{result.detail}") + print("部署编排 + 回滚冒烟通过 ✅") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/deploy/orchestration/deploy_plan.py b/deploy/orchestration/deploy_plan.py new file mode 100644 index 0000000..aeaec73 --- /dev/null +++ b/deploy/orchestration/deploy_plan.py @@ -0,0 +1,428 @@ +# -*- coding: utf-8 -*- +"""一键部署编排引擎(Issue #60 / PRD 5.6「⑥ 部署底座」)。 + +PRD 5.6 验收:一套 Helm Chart 部署内核 + 模板,GPU/NPU 推理后端可插拔。 +本模块把部署落为**可测试的编排引擎**——把部署拆为有序步骤序列 +(pre_check → deploy → health_check → post_check),编排器按序执行(模拟, +不真调 k8s/helm),逐步产出步骤日志与状态,失败即触发回滚。 + +设计要点 +-------- +1. **部署即编排序列**(``DeployStep``):每步声明 ``kind``(四类)+ ``name`` + + 可执行动作(``action`` 回调,返回 ok/detail);步骤序列即 :class:`DeployPlan`。 + - ``pre_check`` 部署前检查(集群可达 / 镜像存在 / values 合法 / 资源配额); + - ``deploy`` 部署动作(helm install/upgrade,模拟); + - ``health_check``健康检查(探活 /health,对齐 ``deploy/k8s/healthz``); + - ``post_check`` 部署后验证(业务接口回归 / 监控接入)。 +2. **编排器**(``DeployOrchestrator``):按序执行步骤,任何一步失败即中止并 + 标记需回滚(``DeployOutcome.needs_rollback``),产出 :class:`DeployReport` + (逐步结果 + 总状态 + 可解释 reason)。 +3. **健康检查契约**(``HealthCheckContract``):定义健康检查的判定口径 + (超时/重试/目标可用率,对齐 ``deploy/k8s/healthz`` 的 99.8% 目标)。 +4. **纯标准库**:action 为回调,默认实现模拟成功;真实环境注入真 action 即可。 +5. **不真执行 k8s/helm**:编排只产日志与状态,便于离线测试与 CI 集成。 + +用法:: + + plan = DeployPlan.default_helm_release(release="iaop", namespace="iaop") + report = DeployOrchestrator(plan).execute() + if not report.outcome.succeeded: + print("需回滚:", report.outcome.reason) +""" +from __future__ import annotations + +import time +from dataclasses import dataclass, field +from enum import Enum +from typing import Callable, Dict, List, Optional, Sequence, Tuple + +#: 健康检查默认超时(秒,对齐 deploy/k8s/healthz 探针口径)。 +DEFAULT_HEALTH_TIMEOUT_S = 30.0 +#: 健康检查默认重试次数。 +DEFAULT_HEALTH_RETRIES = 3 +#: 可用率目标(对齐 deploy/k8s/healthz probe_availability 的 99.8%)。 +DEFAULT_AVAILABILITY_TARGET = 0.998 + + +class DeployError(ValueError): + """部署计划声明/编排错误(步骤序列非法、kind 未知、回调异常等)。""" + + +class DeployStepKind(str, Enum): + """部署步骤类型(决定执行顺序与失败后果)。""" + + PRE_CHECK = "pre_check" # 部署前检查 + DEPLOY = "deploy" # 部署动作 + HEALTH_CHECK = "health_check" # 健康检查 + POST_CHECK = "post_check" # 部署后验证 + + @property + def label(self) -> str: + return { + DeployStepKind.PRE_CHECK: "部署前检查", + DeployStepKind.DEPLOY: "部署动作", + DeployStepKind.HEALTH_CHECK: "健康检查", + DeployStepKind.POST_CHECK: "部署后验证", + }[self] + + @property + def is_checkpoint(self) -> bool: + """该步失败是否触发回滚(deploy/health_check 失败必须回滚)。""" + return self in (DeployStepKind.DEPLOY, DeployStepKind.HEALTH_CHECK) + + +#: 步骤动作回调:返回 (ok, detail)。 +StepAction = Callable[[Dict[str, object]], Tuple[bool, str]] + + +def _default_action(ctx: Dict[str, object]) -> Tuple[bool, str]: + """默认动作:模拟成功(真实环境注入真回调)。""" + return True, "模拟成功(无 k8s/helm 环境)" + + +@dataclass +class DeployStep: + """单个部署步骤(声明式)。 + + Attributes: + kind: 步骤类型。 + name: 步骤名(可读,如 "helm upgrade")。 + action: 可执行动作(ctx → (ok, detail));默认模拟成功。 + required: 是否必需(必需步骤失败即中止并标记回滚;非必需失败仅告警)。 + detail: 步骤说明(引 PRD/SOP,可解释)。 + """ + + kind: DeployStepKind + name: str + action: StepAction = field(default=_default_action, repr=False) + required: bool = True + detail: str = "" + + def __post_init__(self) -> None: + if not self.name: + raise DeployError("DeployStep.name 不能为空") + if not isinstance(self.kind, DeployStepKind): + raise DeployError(f"kind 必须是 DeployStepKind,实际 {type(self.kind)}") + + +@dataclass +class HealthCheckContract: + """健康检查契约(对齐 deploy/k8s/healthz 的探针口径)。 + + Attributes: + endpoint: 健康端点(如 http://iaop:8000/health)。 + timeout_s: 单次探测超时。 + retries: 重试次数(窗口内达到目标可用率即通过)。 + availability_target: 可用率目标(默认 0.998 = 99.8%)。 + """ + + endpoint: str = "http://iaop:8000/health" + timeout_s: float = DEFAULT_HEALTH_TIMEOUT_S + retries: int = DEFAULT_HEALTH_RETRIES + availability_target: float = DEFAULT_AVAILABILITY_TARGET + + def __post_init__(self) -> None: + if self.timeout_s <= 0: + raise DeployError(f"timeout_s 必须 > 0,实际 {self.timeout_s}") + if self.retries < 1: + raise DeployError(f"retries 必须 ≥ 1,实际 {self.retries}") + if not (0.0 < self.availability_target <= 1.0): + raise DeployError( + f"availability_target 须在 (0,1],实际 {self.availability_target}") + + +@dataclass +class DeployPlan: + """部署计划:步骤序列 + 发布元信息。 + + 步骤顺序约束:pre_check 必须在 deploy 前;health_check 必须在 deploy 后、 + post_check 前。``__post_init__`` 校验该顺序,违例即拒绝(防编排漂移)。 + + Attributes: + name: 计划名(如 "iaop-helm-upgrade")。 + release: Helm release 名(如 "iaop")。 + namespace: 命名空间。 + chart: Chart 引用(如 "deploy/k8s/helm/iaop")。 + version: 目标版本(appVersion,如 "v1.0.0")。 + steps: 步骤序列(按执行顺序)。 + health: 健康检查契约(health_check 步骤引用)。 + """ + + name: str + release: str = "iaop" + namespace: str = "iaop" + chart: str = "deploy/k8s/helm/iaop" + version: str = "v1.0.0" + steps: List[DeployStep] = field(default_factory=list) + health: HealthCheckContract = field(default_factory=HealthCheckContract) + + def __post_init__(self) -> None: + if not self.name: + raise DeployError("DeployPlan.name 不能为空") + self._validate_order() + + # ------------------------------------------------------------------ + def _validate_order(self) -> None: + """校验步骤顺序:pre_check < deploy < health_check < post_check。""" + if not self.steps: + return # 空计划在 execute() 时报错 + order_rank = { + DeployStepKind.PRE_CHECK: 0, + DeployStepKind.DEPLOY: 1, + DeployStepKind.HEALTH_CHECK: 2, + DeployStepKind.POST_CHECK: 3, + } + prev_rank = -1 + for s in self.steps: + r = order_rank[s.kind] + if r < prev_rank: + raise DeployError( + f"步骤顺序违例:{s.kind.value}({s.name}) 不能排在" + f" rank {prev_rank} 之后(pre_check < deploy < " + f"health_check < post_check)") + prev_rank = r + + # ------------------------------------------------------------------ + @classmethod + def default_helm_release( + cls, + release: str = "iaop", + namespace: str = "iaop", + version: str = "v1.0.0", + backend: str = "gpu", + health: Optional[HealthCheckContract] = None, + step_actions: Optional[Dict[DeployStepKind, StepAction]] = None, + ) -> "DeployPlan": + """默认 Helm 发布计划(pre_check → deploy → health_check → post_check)。 + + Args: + release: release 名。 + namespace: 命名空间。 + version: 目标 appVersion。 + backend: 推理后端(gpu|npu,对齐 PRD 5.6 配置点)。 + health: 健康检查契约(默认 :class:`HealthCheckContract`)。 + step_actions: 各类步骤的动作回调覆盖(默认模拟成功)。 + """ + actions = step_actions or {} + plan = cls( + name=f"{release}-helm-upgrade", + release=release, + namespace=namespace, + version=version, + health=health or HealthCheckContract(), + steps=[ + DeployStep( + kind=DeployStepKind.PRE_CHECK, + name="pre-check-cluster-values", + action=actions.get(DeployStepKind.PRE_CHECK, _default_action), + detail="部署前检查:集群可达 / 镜像存在 / values 合法 / 资源配额"), + DeployStep( + kind=DeployStepKind.DEPLOY, + name=f"helm-upgrade-{backend}", + action=actions.get(DeployStepKind.DEPLOY, _default_action), + detail=f"helm upgrade --install {release}(backend={backend},PRD 5.6)"), + DeployStep( + kind=DeployStepKind.HEALTH_CHECK, + name="health-check-probe", + action=actions.get(DeployStepKind.HEALTH_CHECK, _default_action), + detail="健康检查:探活 /health(目标可用率 99.8%,对齐 healthz)"), + DeployStep( + kind=DeployStepKind.POST_CHECK, + name="post-check-smoke", + action=actions.get(DeployStepKind.POST_CHECK, _default_action), + required=False, # 烟测非阻断 + detail="部署后验证:业务接口烟测 / 监控接入(非阻断)"), + ], + ) + return plan + + +# --------------------------------------------------------------------------- +# 执行结果 +# --------------------------------------------------------------------------- + + +class DeployStatus(str, Enum): + """单步执行状态。""" + + PENDING = "pending" + SUCCESS = "success" + FAILED = "failed" + SKIPPED = "skipped" + + +@dataclass +class DeployStepResult: + """单步执行结果。""" + + step: DeployStep + status: DeployStatus = DeployStatus.PENDING + detail: str = "" + duration_ms: float = 0.0 + index: int = 0 + + +@dataclass +class DeployOutcome: + """部署总结果(可解释:是否成功 + 是否需回滚 + reason)。""" + + succeeded: bool + needs_rollback: bool + failed_step: Optional[str] = None + reason: str = "" + + +@dataclass +class DeployReport: + """部署编排报告(逐步结果 + 总状态 + 日志行)。""" + + plan_name: str + results: List[DeployStepResult] = field(default_factory=list) + outcome: Optional[DeployOutcome] = None + log_lines: List[str] = field(default_factory=list) + context: Dict[str, object] = field(default_factory=dict) + + @property + def succeeded(self) -> bool: + return self.outcome.succeeded if self.outcome else False + + def to_dict(self) -> dict: + return { + "plan_name": self.plan_name, + "succeeded": self.succeeded, + "needs_rollback": self.outcome.needs_rollback if self.outcome else False, + "reason": self.outcome.reason if self.outcome else "", + "steps": [ + {"index": r.index, "kind": r.step.kind.value, "name": r.step.name, + "status": r.status.value, "detail": r.detail, + "duration_ms": round(r.duration_ms, 2)} + for r in self.results + ], + "log": list(self.log_lines), + } + + +# --------------------------------------------------------------------------- +# 编排器 +# --------------------------------------------------------------------------- + + +class DeployOrchestrator: + """部署编排器:按序执行 DeployPlan 步骤,失败即中止并标记回滚。 + + Args: + plan: 部署计划。 + dry_run: 空跑模式(不执行 action,全部标记 SKIPPED;用于计划校验)。 + logger: 日志回调(默认 print 到内部 log_lines 缓冲)。 + """ + + def __init__( + self, + plan: DeployPlan, + dry_run: bool = False, + logger: Optional[Callable[[str], None]] = None, + ) -> None: + self.plan = plan + self.dry_run = bool(dry_run) + self._log: List[str] = [] + self._logger = logger + + # ------------------------------------------------------------------ + def _log_line(self, line: str) -> None: + self._log.append(line) + if self._logger is not None: + self._logger(line) + + # ------------------------------------------------------------------ + def execute(self, context: Optional[Dict[str, object]] = None) -> DeployReport: + """执行部署计划,返回报告。""" + report = DeployReport(plan_name=self.plan.name) + ctx: Dict[str, object] = { + "release": self.plan.release, + "namespace": self.plan.namespace, + "chart": self.plan.chart, + "version": self.plan.version, + "health": { + "endpoint": self.plan.health.endpoint, + "timeout_s": self.plan.health.timeout_s, + "retries": self.plan.health.retries, + "availability_target": self.plan.health.availability_target, + }, + } + if context: + ctx.update(context) + report.context = dict(ctx) + + if not self.plan.steps: + report.outcome = DeployOutcome( + succeeded=False, needs_rollback=False, + reason="部署计划无步骤(DeployPlan.steps 为空)") + self._log_line("[FAIL] 部署计划无步骤,未执行") + report.log_lines = list(self._log) + return report + + self._log_line( + f"[START] 部署 {self.plan.name}:release={self.plan.release} " + f"namespace={self.plan.namespace} version={self.plan.version} " + f"steps={len(self.plan.steps)}") + + failed_result: Optional[DeployStepResult] = None + for idx, step in enumerate(self.plan.steps): + result = DeployStepResult(step=step, index=idx) + + if self.dry_run: + result.status = DeployStatus.SKIPPED + result.detail = "dry-run 跳过" + report.results.append(result) + self._log_line( + f"[SKIP] #{idx} {step.kind.label}/{step.name}(dry-run)") + continue + + self._log_line(f"[RUN] #{idx} {step.kind.label}/{step.name}") + started = time.monotonic() + try: + ok, detail = step.action(ctx) + except Exception as exc: # noqa: BLE001 - 任意 action 异常视为失败 + ok, detail = False, f"action 异常:{exc}" + result.duration_ms = (time.monotonic() - started) * 1000.0 + result.detail = detail + result.status = DeployStatus.SUCCESS if ok else DeployStatus.FAILED + report.results.append(result) + + tag = "OK" if ok else "FAIL" + self._log_line( + f"[{tag}] #{idx} {step.name}:{detail} " + f"({result.duration_ms:.0f}ms)") + + if not ok: + if step.required: + failed_result = result + break + # 非必需步骤失败:告警但继续 + self._log_line( + f"[WARN] #{idx} {step.name} 非必需步骤失败,继续编排") + + report.log_lines = list(self._log) + + if self.dry_run: + report.outcome = DeployOutcome( + succeeded=True, needs_rollback=False, + reason=f"dry-run:{len(self.plan.steps)} 步全部跳过(计划校验通过)") + return report + + if failed_result is not None: + needs_rollback = failed_result.step.kind.is_checkpoint + report.outcome = DeployOutcome( + succeeded=False, + needs_rollback=needs_rollback, + failed_step=failed_result.step.name, + reason=( + f"步骤 #{failed_result.index} {failed_result.step.kind.label}/" + f"{failed_result.step.name} 失败:{failed_result.detail}" + + (",需回滚到上一稳定版本" if needs_rollback + else "(非回滚检查点,无需回滚)"))) + else: + report.outcome = DeployOutcome( + succeeded=True, needs_rollback=False, + reason=f"全部 {len(report.results)} 步执行成功,部署完成") + return report diff --git a/deploy/orchestration/rollback.py b/deploy/orchestration/rollback.py new file mode 100644 index 0000000..b27d089 --- /dev/null +++ b/deploy/orchestration/rollback.py @@ -0,0 +1,265 @@ +# -*- coding: utf-8 -*- +"""部署回滚引擎(Issue #60 / PRD 5.6「⑥ 部署底座」)。 + +PRD 5.6 一键部署需配套**回滚机制**——部署失败时恢复到部署前的稳定配置版本, +保证可用性。本模块把回滚落为**可测试的纯标准库引擎**:部署前对当前配置版本 +打快照(:class:`RollbackPoint`),失败时按快照恢复(:class:`RollbackManager`), +产出可解释的 :class:`RollbackResult`。 + +设计要点 +-------- +1. **快照即配置版本**(``RollbackPoint``):部署前记录当前稳定版本的元信息 + (release/namespace/version/chart/values_hash/创建时间),存入快照栈。 + 快照不可变(只读),保证回滚目标确定。 +2. **快照栈**(``RollbackManager``):每次成功部署压栈;失败时弹出最近快照执行 + 回滚。支持多版本回溯(rollback(n) 回滚 n 个版本)。 +3. **回滚动作**(``restore_action`` 回调):默认模拟成功;真实环境注入 + ``helm rollback`` 等真回调。回滚后重新健康检查(契约对齐 deploy_plan)。 +4. **纯标准库**:无 k8s/helm 依赖,便于离线测试与 CI 集成。 + +用法:: + + mgr = RollbackManager() + mgr.snapshot(plan) # 部署前快照 + report = orchestrator.execute() # 部署 + if not report.succeeded: + result = mgr.rollback_to_latest(reason=report.outcome.reason) + print(result.status, result.detail) +""" +from __future__ import annotations + +import hashlib +import json +import time +from dataclasses import dataclass, field +from enum import Enum +from typing import Callable, Dict, List, Optional, Tuple + +from deploy_plan import ( + DEFAULT_AVAILABILITY_TARGET, + DEFAULT_HEALTH_RETRIES, + DEFAULT_HEALTH_TIMEOUT_S, + DeployError, + DeployPlan, + HealthCheckContract, +) + +#: 回滚动作回调:(snapshot, ctx) → (ok, detail)。 +RestoreAction = Callable[["RollbackPoint", Dict[str, object]], Tuple[bool, str]] + + +def _default_restore(snapshot: "RollbackPoint", + ctx: Dict[str, object]) -> Tuple[bool, str]: + """默认回滚动作:模拟成功(真实环境注入 helm rollback 回调)。""" + return True, f"模拟回滚到 {snapshot.version}(无 k8s/helm 环境)" + + +class RollbackStatus(str, Enum): + """回滚执行状态。""" + + SUCCESS = "success" # 回滚成功(含回滚后健康检查通过) + FAILED = "failed" # 回滚失败(restore 动作失败或健康检查不过) + NO_TARGET = "no_target" # 无可回滚快照(栈空) + SKIPPED = "skipped" # 跳过(手动) + + +@dataclass +class RollbackPoint: + """部署前快照(不可变配置版本)。 + + Attributes: + release: Helm release 名。 + namespace: 命名空间。 + version: 快照时的版本(appVersion,回滚目标)。 + chart: Chart 引用。 + values_hash: values 配置的哈希(检测配置漂移)。 + created_at: 快照创建时间戳(epoch 秒)。 + reason: 快照说明(如 "部署 v1.1.0 前的稳定版本 v1.0.0")。 + """ + + release: str + namespace: str + version: str + chart: str + values_hash: str + created_at: float = field(default_factory=time.time) + reason: str = "" + + def to_dict(self) -> dict: + return { + "release": self.release, + "namespace": self.namespace, + "version": self.version, + "chart": self.chart, + "values_hash": self.values_hash, + "created_at": self.created_at, + "reason": self.reason, + } + + +@dataclass +class RollbackResult: + """回滚执行结果(可解释:状态 + 目标版本 + reason)。""" + + status: RollbackStatus + target: Optional[RollbackPoint] = None + detail: str = "" + reason: str = "" + + @property + def succeeded(self) -> bool: + return self.status is RollbackStatus.SUCCESS + + def to_dict(self) -> dict: + return { + "status": self.status.value, + "succeeded": self.succeeded, + "target": self.target.to_dict() if self.target else None, + "detail": self.detail, + "reason": self.reason, + } + + +class RollbackManager: + """部署回滚管理器:快照栈 + 回滚执行。 + + Args: + restore_action: 回滚动作回调(默认模拟成功)。 + post_rollback_health: 回滚后是否健康检查(默认 False,由编排器单独跑)。 + max_snapshots: 快照栈上限(FIFO 淘汰,防内存膨胀;默认 10)。 + """ + + def __init__( + self, + restore_action: RestoreAction = _default_restore, + post_rollback_health: bool = False, + max_snapshots: int = 10, + ) -> None: + if max_snapshots < 1: + raise DeployError(f"max_snapshots 必须 ≥ 1,实际 {max_snapshots}") + self._stack: List[RollbackPoint] = [] + self.restore_action = restore_action + self.post_rollback_health = bool(post_rollback_health) + self.max_snapshots = int(max_snapshots) + + # ------------------------------------------------------------------ + # 快照栈 + # ------------------------------------------------------------------ + @property + def snapshots(self) -> List[RollbackPoint]: + """当前快照栈(栈顶 = 最近快照 = 列表末尾)。只读视图。""" + return list(self._stack) + + def latest(self) -> Optional[RollbackPoint]: + """栈顶快照(最近一次成功版本);栈空返回 None。""" + return self._stack[-1] if self._stack else None + + def snapshot(self, plan: DeployPlan, values: Optional[Dict[str, object]] = None, + reason: str = "") -> RollbackPoint: + """部署前打快照(压栈)。 + + Args: + plan: 部署计划(取 release/namespace/version/chart)。 + values: 当前 values 配置(用于计算 values_hash,检测漂移)。 + reason: 快照说明。 + """ + values_hash = _hash_values(values or {}) + point = RollbackPoint( + release=plan.release, + namespace=plan.namespace, + version=plan.version, + chart=plan.chart, + values_hash=values_hash, + reason=reason or f"部署 {plan.name} 前快照(version={plan.version})", + ) + self._stack.append(point) + # FIFO 淘汰(保留最近 max_snapshots 个) + if len(self._stack) > self.max_snapshots: + self._stack = self._stack[-self.max_snapshots:] + return point + + # ------------------------------------------------------------------ + # 回滚 + # ------------------------------------------------------------------ + def rollback_to_latest(self, reason: str = "") -> RollbackResult: + """回滚到最近快照(弹出栈顶)。""" + return self._rollback(n=1, reason=reason) + + def rollback(self, n: int = 1, reason: str = "") -> RollbackResult: + """回滚 n 个版本(弹出 n 个快照,回滚到第 n 个)。""" + return self._rollback(n=n, reason=reason) + + def _rollback(self, n: int, reason: str) -> RollbackResult: + if n < 1: + return RollbackResult( + status=RollbackStatus.SKIPPED, + detail=f"n={n},跳过回滚", + reason="回滚步数 n 必须 ≥ 1") + if len(self._stack) < n: + return RollbackResult( + status=RollbackStatus.NO_TARGET, + detail=f"快照栈仅 {len(self._stack)} 个,无法回滚 {n} 个版本", + reason="无可回滚的稳定版本快照") + + # 弹出 n 个快照,回滚目标 = 第 n 个(最后弹出的) + popped: List[RollbackPoint] = [] + for _ in range(n): + popped.append(self._stack.pop()) + target = popped[-1] + + ctx: Dict[str, object] = { + "release": target.release, + "namespace": target.namespace, + "version": target.version, + "reason": reason or f"部署失败,回滚到 {target.version}", + } + try: + ok, detail = self.restore_action(target, ctx) + except Exception as exc: # noqa: BLE001 + ok, detail = False, f"restore action 异常:{exc}" + + if not ok: + # 回滚失败:把目标压回栈顶(保持快照栈一致性) + self._stack.append(target) + return RollbackResult( + status=RollbackStatus.FAILED, + target=target, + detail=detail, + reason=f"回滚到 {target.version} 失败:{detail}") + + full_detail = detail + # 回滚后健康检查(可选) + if self.post_rollback_health: + health_ok, health_detail = self._health_check(target) + if not health_ok: + self._stack.append(target) + return RollbackResult( + status=RollbackStatus.FAILED, + target=target, + detail=f"回滚动作成功但健康检查失败:{health_detail}", + reason=f"回滚到 {target.version} 后健康检查未通过") + full_detail = f"{detail};健康检查通过" + + return RollbackResult( + status=RollbackStatus.SUCCESS, + target=target, + detail=full_detail, + reason=f"回滚到 {target.version} 成功" + + (f"({reason})" if reason else "")) + + # ------------------------------------------------------------------ + def _health_check(self, target: RollbackPoint) -> Tuple[bool, str]: + """回滚后健康检查(模拟,默认通过;真实环境注入 restore_action 内)。""" + # 纯标准库模拟:回滚后假设服务恢复(真实健康检查由 deploy_plan 编排) + return True, f"回滚后 {target.version} 服务健康(模拟)" + + +# --------------------------------------------------------------------------- +# 辅助 +# --------------------------------------------------------------------------- + +def _hash_values(values: Dict[str, object]) -> str: + """计算 values 配置的短哈希(检测配置漂移,SHA1 前 12 位)。""" + payload = json.dumps(values, sort_keys=True, ensure_ascii=False, + default=str) + return hashlib.sha1(payload.encode("utf-8")).hexdigest()[:12] diff --git a/deploy/orchestration/tests/_bootstrap.py b/deploy/orchestration/tests/_bootstrap.py new file mode 100644 index 0000000..ced549a --- /dev/null +++ b/deploy/orchestration/tests/_bootstrap.py @@ -0,0 +1,13 @@ +# -*- coding: utf-8 -*- +"""测试引导:把 ``deploy/orchestration`` 挂到 sys.path,使 deploy_plan/rollback +可被 ``from deploy_plan import ...`` 导入(orchestration 包内相对导入需要)。 + +与 core 模块测试引导同款模式。 +""" +import os +import sys +import types + +PKG_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if PKG_DIR not in sys.path: + sys.path.insert(0, PKG_DIR) diff --git a/deploy/orchestration/tests/test_deploy_plan.py b/deploy/orchestration/tests/test_deploy_plan.py new file mode 100644 index 0000000..137d4b1 --- /dev/null +++ b/deploy/orchestration/tests/test_deploy_plan.py @@ -0,0 +1,200 @@ +# -*- coding: utf-8 -*- +"""部署编排引擎测试(Issue #60)。 + +覆盖: +1. 默认 Helm 计划构造(4 步:pre_check/deploy/health_check/post_check); +2. 步骤顺序约束校验(违例 → DeployError); +3. 全成功执行(succeeded + 无需回滚); +4. deploy 步骤失败(needs_rollback + failed_step); +5. pre_check 失败(needs_rollback=False,非回滚检查点); +6. 非必需 post_check 失败(继续编排,succeeded); +7. action 异常(视为失败); +8. dry_run(全部 SKIPPED); +9. 空计划(succeeded=False); +10. HealthCheckContract 校验; +11. 报告序列化。 +""" +import unittest + +import _bootstrap # noqa: F401 (sys.path 挂载) + +from deploy_plan import ( + DEFAULT_AVAILABILITY_TARGET, + DeployError, + DeployOrchestrator, + DeployOutcome, + DeployPlan, + DeployStep, + DeployStepKind, + DeployStatus, + HealthCheckContract, +) + + +def _action(ok: bool, detail: str = ""): + """构造固定返回值的 action。""" + def _fn(ctx): + return ok, (detail or ("成功" if ok else "失败")) + return _fn + + +def _raising_action(exc_type=RuntimeError): + def _fn(ctx): + raise exc_type("boom") + return _fn + + +class TestDeployPlanModel(unittest.TestCase): + """部署计划模型与顺序约束。""" + + def test_default_helm_release_has_four_steps(self): + plan = DeployPlan.default_helm_release() + kinds = [s.kind for s in plan.steps] + self.assertEqual(kinds, [ + DeployStepKind.PRE_CHECK, + DeployStepKind.DEPLOY, + DeployStepKind.HEALTH_CHECK, + DeployStepKind.POST_CHECK, + ]) + # post_check 默认非必需 + self.assertFalse(plan.steps[-1].required) + + def test_step_order_violation_rejected(self): + # health_check 排在 deploy 前 → 违例 + with self.assertRaises(DeployError): + DeployPlan( + name="bad", + steps=[ + DeployStep(DeployStepKind.HEALTH_CHECK, "hc"), + DeployStep(DeployStepKind.DEPLOY, "d"), + ]) + + def test_empty_step_name_rejected(self): + with self.assertRaises(DeployError): + DeployStep(DeployStepKind.DEPLOY, "") + + def test_plan_name_required(self): + with self.assertRaises(DeployError): + DeployPlan(name="") + + +class TestOrchestratorSuccess(unittest.TestCase): + """全成功编排。""" + + def test_all_steps_succeed(self): + plan = DeployPlan.default_helm_release() + report = DeployOrchestrator(plan).execute() + self.assertTrue(report.succeeded) + self.assertFalse(report.outcome.needs_rollback) + self.assertEqual(len(report.results), 4) + self.assertEqual( + [r.status for r in report.results], + [DeployStatus.SUCCESS] * 4) + # 日志含 START/RUN/OK + self.assertTrue(any("[START]" in ln for ln in report.log_lines)) + self.assertTrue(any("[OK]" in ln for ln in report.log_lines)) + + def test_context_populated(self): + plan = DeployPlan.default_helm_release() + report = DeployOrchestrator(plan).execute() + self.assertEqual(report.context["release"], "iaop") + self.assertEqual(report.context["namespace"], "iaop") + self.assertIn("health", report.context) + + +class TestOrchestratorFailures(unittest.TestCase): + """失败场景 + 回滚标记。""" + + def test_deploy_failure_triggers_rollback(self): + plan = DeployPlan.default_helm_release( + step_actions={DeployStepKind.DEPLOY: _action(False, "helm 失败")}) + report = DeployOrchestrator(plan).execute() + self.assertFalse(report.succeeded) + self.assertTrue(report.outcome.needs_rollback) # deploy 是回滚检查点 + self.assertIsNotNone(report.outcome.failed_step) + # deploy 之后的 health_check/post_check 未执行 + executed_kinds = [r.step.kind for r in report.results] + self.assertNotIn(DeployStepKind.HEALTH_CHECK, executed_kinds) + self.assertNotIn(DeployStepKind.POST_CHECK, executed_kinds) + + def test_pre_check_failure_no_rollback(self): + plan = DeployPlan.default_helm_release( + step_actions={DeployStepKind.PRE_CHECK: _action(False, "集群不可达")}) + report = DeployOrchestrator(plan).execute() + self.assertFalse(report.succeeded) + # pre_check 失败:尚未部署,无需回滚 + self.assertFalse(report.outcome.needs_rollback) + self.assertIn("集群不可达", report.outcome.reason) + + def test_non_required_post_check_failure_continues(self): + plan = DeployPlan.default_helm_release( + step_actions={DeployStepKind.POST_CHECK: _action(False, "烟测失败")}) + report = DeployOrchestrator(plan).execute() + # post_check 非必需,失败仍判成功(核心步骤都过了) + self.assertTrue(report.succeeded) + self.assertFalse(report.outcome.needs_rollback) + # 但有 WARN 日志 + self.assertTrue(any("[WARN]" in ln for ln in report.log_lines)) + + def test_action_exception_treated_as_failure(self): + plan = DeployPlan.default_helm_release( + step_actions={DeployStepKind.DEPLOY: _raising_action()}) + report = DeployOrchestrator(plan).execute() + self.assertFalse(report.succeeded) + self.assertTrue(report.outcome.needs_rollback) + self.assertIn("action 异常", report.outcome.reason) + + def test_health_check_failure_triggers_rollback(self): + plan = DeployPlan.default_helm_release( + step_actions={DeployStepKind.HEALTH_CHECK: _action(False, "/health 不可用")}) + report = DeployOrchestrator(plan).execute() + self.assertFalse(report.succeeded) + self.assertTrue(report.outcome.needs_rollback) + + +class TestDryRunAndEmpty(unittest.TestCase): + """dry_run + 空计划。""" + + def test_dry_run_skips_all(self): + plan = DeployPlan.default_helm_release() + report = DeployOrchestrator(plan, dry_run=True).execute() + self.assertTrue(report.succeeded) + self.assertEqual( + [r.status for r in report.results], + [DeployStatus.SKIPPED] * 4) + + def test_empty_plan_fails(self): + plan = DeployPlan(name="empty", steps=[]) + report = DeployOrchestrator(plan).execute() + self.assertFalse(report.succeeded) + self.assertIn("无步骤", report.outcome.reason) + + +class TestHealthContractAndReport(unittest.TestCase): + """HealthCheckContract 校验 + 报告序列化。""" + + def test_health_contract_defaults(self): + h = HealthCheckContract() + self.assertEqual(h.endpoint, "http://iaop:8000/health") + self.assertEqual(h.retries, 3) + self.assertAlmostEqual(h.availability_target, DEFAULT_AVAILABILITY_TARGET) + + def test_health_contract_validation(self): + with self.assertRaises(DeployError): + HealthCheckContract(timeout_s=0) + with self.assertRaises(DeployError): + HealthCheckContract(retries=0) + with self.assertRaises(DeployError): + HealthCheckContract(availability_target=1.5) + + def test_report_to_dict(self): + plan = DeployPlan.default_helm_release() + report = DeployOrchestrator(plan).execute() + d = report.to_dict() + self.assertTrue(d["succeeded"]) + self.assertEqual(len(d["steps"]), 4) + self.assertEqual(d["steps"][0]["kind"], "pre_check") + + +if __name__ == "__main__": + unittest.main() diff --git a/deploy/orchestration/tests/test_rollback.py b/deploy/orchestration/tests/test_rollback.py new file mode 100644 index 0000000..e92bfa6 --- /dev/null +++ b/deploy/orchestration/tests/test_rollback.py @@ -0,0 +1,185 @@ +# -*- coding: utf-8 -*- +"""部署回滚引擎测试(Issue #60)。 + +覆盖: +1. 快照压栈 + latest 查询; +2. FIFO 淘汰(超 max_snapshots); +3. rollback_to_latest 成功(弹出栈顶 + 回滚目标版本); +4. rollback(n) 多版本回溯; +5. 空栈回滚(NO_TARGET); +6. restore 动作失败(FAILED + 压回栈顶); +7. restore 异常(FAILED); +8. post_rollback_health 失败(FAILED + 压回); +9. n<1 跳过; +10. values_hash 漂移检测; +11. 报告序列化。 +""" +import unittest + +import _bootstrap # noqa: F401 (sys.path 挂载) + +from deploy_plan import DeployPlan +from rollback import ( + DeployError, + RollbackManager, + RollbackPoint, + RollbackStatus, +) + + +def _restore(ok: bool, detail: str = ""): + def _fn(snapshot, ctx): + return ok, (detail or ("回滚成功" if ok else "回滚失败")) + return _fn + + +def _raising_restore(): + def _fn(snapshot, ctx): + raise RuntimeError("restore boom") + return _fn + + +def _make_plan(version="v1.0.0", release="iaop"): + return DeployPlan.default_helm_release(release=release, version=version) + + +class TestSnapshotStack(unittest.TestCase): + """快照压栈与查询。""" + + def test_snapshot_push_and_latest(self): + mgr = RollbackManager() + # snapshot 即压栈(部署前快照),latest 即栈顶 + mgr.snapshot(_make_plan("v1.0.0"), values={"k": 1}, reason="第一次") + self.assertEqual(mgr.latest().version, "v1.0.0") + self.assertEqual(len(mgr.snapshots), 1) + mgr.snapshot(_make_plan("v1.1.0"), values={"k": 2}) + self.assertEqual(mgr.latest().version, "v1.1.0") + self.assertEqual(len(mgr.snapshots), 2) + # 空栈时 latest 返回 None + self.assertIsNotNone(mgr.latest()) + mgr2 = RollbackManager() + self.assertIsNone(mgr2.latest()) + + def test_fifo_eviction(self): + mgr = RollbackManager(max_snapshots=2) + mgr.snapshot(_make_plan("v1.0.0")) + mgr.snapshot(_make_plan("v1.1.0")) + mgr.snapshot(_make_plan("v1.2.0")) + # 上限 2,淘汰最旧 + self.assertEqual(len(mgr.snapshots), 2) + self.assertEqual([s.version for s in mgr.snapshots], ["v1.1.0", "v1.2.0"]) + + def test_max_snapshots_validation(self): + with self.assertRaises(DeployError): + RollbackManager(max_snapshots=0) + + def test_values_hash_changes_with_config(self): + mgr = RollbackManager() + p1 = mgr.snapshot(_make_plan(), values={"backend": "gpu"}) + p2 = mgr.snapshot(_make_plan(), values={"backend": "npu"}) + self.assertNotEqual(p1.values_hash, p2.values_hash) + + +class TestRollbackSuccess(unittest.TestCase): + """回滚成功场景。""" + + def test_rollback_to_latest_success(self): + mgr = RollbackManager(restore_action=_restore(True)) + mgr.snapshot(_make_plan("v1.0.0")) + mgr.snapshot(_make_plan("v1.1.0")) + result = mgr.rollback_to_latest(reason="部署 v1.2.0 失败") + self.assertTrue(result.succeeded) + self.assertEqual(result.status, RollbackStatus.SUCCESS) + self.assertEqual(result.target.version, "v1.1.0") + # 弹出栈顶后栈剩 1 个 + self.assertEqual(len(mgr.snapshots), 1) + self.assertEqual(mgr.latest().version, "v1.0.0") + self.assertIn("v1.2.0 失败", result.reason) + + def test_rollback_n_versions(self): + mgr = RollbackManager(restore_action=_restore(True)) + for v in ["v1.0.0", "v1.1.0", "v1.2.0"]: + mgr.snapshot(_make_plan(v)) + # 回滚 2 个版本 → 目标 v1.1.0(弹出 v1.2.0 和 v1.1.0) + result = mgr.rollback(n=2) + self.assertTrue(result.succeeded) + self.assertEqual(result.target.version, "v1.1.0") + self.assertEqual(len(mgr.snapshots), 1) + + +class TestRollbackFailures(unittest.TestCase): + """回滚失败与边界。""" + + def test_empty_stack_no_target(self): + mgr = RollbackManager() + result = mgr.rollback_to_latest() + self.assertEqual(result.status, RollbackStatus.NO_TARGET) + self.assertFalse(result.succeeded) + + def test_insufficient_stack_for_n(self): + mgr = RollbackManager(restore_action=_restore(True)) + mgr.snapshot(_make_plan("v1.0.0")) + # 栈仅 1 个,回滚 2 个 → NO_TARGET + result = mgr.rollback(n=2) + self.assertEqual(result.status, RollbackStatus.NO_TARGET) + + def test_restore_failure_pushes_target_back(self): + mgr = RollbackManager(restore_action=_restore(False, "helm rollback 失败")) + mgr.snapshot(_make_plan("v1.0.0")) + mgr.snapshot(_make_plan("v1.1.0")) + result = mgr.rollback_to_latest() + self.assertEqual(result.status, RollbackStatus.FAILED) + # 失败时目标压回栈顶,栈仍 2 个 + self.assertEqual(len(mgr.snapshots), 2) + self.assertEqual(mgr.latest().version, "v1.1.0") + + def test_restore_exception_failure(self): + mgr = RollbackManager(restore_action=_raising_restore()) + mgr.snapshot(_make_plan("v1.0.0")) + result = mgr.rollback_to_latest() + self.assertEqual(result.status, RollbackStatus.FAILED) + self.assertIn("异常", result.detail) + + def test_post_rollback_health_failure(self): + # restore 成功但健康检查失败(注入失败的健康检查) + def _restore_then_health_fail(snapshot, ctx): + return True, "回滚动作成功" + mgr = RollbackManager( + restore_action=_restore_then_health_fail, post_rollback_health=True) + # 覆盖 _health_check 为失败 + mgr._health_check = lambda target: (False, "/health 仍不可用") + mgr.snapshot(_make_plan("v1.0.0")) + result = mgr.rollback_to_latest() + self.assertEqual(result.status, RollbackStatus.FAILED) + self.assertIn("健康检查", result.detail) + # 压回栈顶 + self.assertEqual(len(mgr.snapshots), 1) + + def test_n_less_than_one_skipped(self): + mgr = RollbackManager() + result = mgr.rollback(n=0) + self.assertEqual(result.status, RollbackStatus.SKIPPED) + + +class TestResultExport(unittest.TestCase): + """结果序列化。""" + + def test_result_to_dict(self): + mgr = RollbackManager(restore_action=_restore(True)) + mgr.snapshot(_make_plan("v1.0.0")) + result = mgr.rollback_to_latest() + d = result.to_dict() + self.assertEqual(d["status"], "success") + self.assertTrue(d["succeeded"]) + self.assertEqual(d["target"]["version"], "v1.0.0") + + def test_point_to_dict(self): + p = RollbackPoint(release="r", namespace="ns", version="v1", + chart="c", values_hash="abc123", reason="x") + d = p.to_dict() + self.assertEqual(d["version"], "v1") + self.assertEqual(d["values_hash"], "abc123") + + +if __name__ == "__main__": + unittest.main()