bot_dev1
97fbbfa7b2
纯标准库实现，对齐 PRD 5.6「⑥ 部署底座」（不真执行 k8s/helm，模拟编排）： - deploy_plan.py：DeployStep（4 类：pre_check/deploy/health_check/post_check） + DeployPlan（顺序约束自动校验，default_helm_release 默认 4 步计划）+ DeployOrchestrator（按序执行，必需步骤失败即中止并标记 needs_rollback， deploy/health_check 为回滚检查点，pre_check 失败无需回滚）+ HealthCheckContract（对齐 deploy/k8s/healthz 99.8% 目标）。 - rollback.py：RollbackPoint（部署前快照，values_hash SHA1 检测漂移）+ RollbackManager（快照栈 + rollback_to_latest/rollback(n)，失败压回栈顶 保持一致性，FIFO 淘汰，支持回滚后健康检查）+ RollbackResult（4 状态）。 - tests：30 用例覆盖计划构造/顺序约束、全成功/各类失败、回滚检查点判定、 action 异常、dry_run、空计划、快照压栈/FIFO、回滚成功/失败/多版本/空栈/ 异常、健康检查契约、values 漂移、报告序列化。 - _sanity_check.py：冒烟验证快照→部署成功→部署失败→回滚全流程。
429 lines
17 KiB
Python
429 lines
17 KiB
Python
# -*- 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
|