# -*- 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]