可靠任务与安全变更
“脚本执行成功”不是生产变更的完成标准。生产任务至少要回答：本次计划改变什么？谁批准？同一请求重试会不会重复执行？部分失败如何继续？结果在哪里审计？
1. 将计划与执行分开
先生成机器可读计划，再在明确确认后执行。计划阶段只读、可审阅、可在 CI 产物中保存；执行阶段接收固定计划，不能临时重新发现目标。
# src/ops_tool/service.py
from dataclasses import asdict, dataclass
@dataclass(frozen=True)
class RestartPlan:
change_id: str
host: str
service: str
previous_state: str
action: str = "restart"
def plan_restart(change_id: str, host: str, service: str, inventory: dict) -> RestartPlan:
"""仅根据受管资产与白名单生成计划，不连接目标机执行命令。"""
if host not in inventory:
raise ValueError(f"目标不在 CMDB: {host}")
if service not in inventory[host]["allowed_services"]:
raise ValueError(f"该主机不允许操作服务: {service}")
return RestartPlan(
change_id=change_id,
host=host,
service=service,
previous_state=inventory[host]["service_states"].get(service, "unknown"),
)
def render_plan(plan: RestartPlan) -> dict:
# 序列化后可签名、审批并作为执行输入保存。
return asdict(plan)
执行 API 接受 change_id 与经审批的计划摘要，服务端必须重新校验目标、权限和变更窗口。不要接受一个任意 shell 字符串或由调用方提供的 SSH 命令。
2. 幂等键与任务状态
HTTP 超时、队列重复投递和人工重试都会发生。以 change_id 作为唯一键，持久化状态机可以让重复请求返回原任务而不是再次操作资源。
# 伪代码：数据库应为 change_id 建立唯一索引。
def create_or_get_task(store, change_id: str, requested_by: str, plan: dict) -> dict:
existing = store.find_by_change_id(change_id)
if existing:
# 返回原任务，调用方据此轮询结果；绝不再次执行变更。
return existing
return store.insert({
"change_id": change_id,
"requested_by": requested_by,
"plan": plan,
"status": "queued",
"attempts": 0,
})
def run_task(store, task_id: str) -> None:
task = store.claim(task_id, expected_status="queued")
if task is None:
return # 已被其他 worker 处理或状态已终结。
try:
# 调用受限 adapter；每个目标的结果单独保存。
apply_restart(task["plan"])
except TransientError as exc:
store.retry_or_dead_letter(task_id, error=str(exc))
except Exception as exc:
store.fail(task_id, error=str(exc))
else:
store.succeed(task_id)
幂等不能只依靠内存 set。多副本 API、进程重启和 TTL 到期都要求持久存储或至少使用具备原子条件写入的共享存储。
3. 重试边界
| 情况 | 是否自动重试 | 处理 |
|---|---|---|
| DNS 超时、连接重置、503 | 可以 | 指数退避、抖动、最大次数与任务截止时间 |
| 429 | 可以 | 优先遵循 Retry-After，同时限制全局并发 |
| 401、403、参数校验失败 | 不可以 | 标记失败，修正权限或配置后创建新任务 |
| 删除、重启、扩容等变更 | 仅具备幂等键时 | 先查询资源和已有任务，再决定是否补偿 |
| 部分目标失败 | 按目标重试 | 保存逐目标状态，不重做已经成功的目标 |
重试次数、间隔、执行者和最后错误应成为任务状态的一部分。超过阈值进入死信队列，并向值班人员告警；不能无限循环消耗资源。
4. 并发与限流
并发不是固定常数。应同时限制全局并发、每个环境、每个集群和每台主机的并发度。例如滚动重启 100 台 Web 主机时，可以“全局 10、每个机房 2、同一应用 1”。
# 伪代码：任务调度器取得两个租约才允许执行。
with limiter.acquire("global", limit=10):
with limiter.acquire(f"service:{plan.service}", limit=1):
# 同一服务串行变更，避免同时丢失全部副本。
apply_restart(plan)
对 Kubernetes 操作还要遵守 PDB、maxUnavailable 与应用自身的发布策略；对 SSH 批量任务要考虑堡垒机连接数和远端 MaxSessions。
5. 审计与回滚
每个任务至少记录：change_id、请求人、审批人、环境、目标清单、计划摘要、开始/结束时间、每个目标结果、关联日志和原始错误。敏感参数仅记录摘要或哈希。
回滚不是“再次运行旧脚本”。在计划阶段定义可回滚条件和步骤：配置变更保存旧版本，部署记录先前镜像 digest，扩容记录原副本数。不可逆操作（删除数据、轮换密钥）必须增加备份、双人复核与更短的权限令牌。
6. 变更前检查清单
- 目标来自受管 inventory，不接受裸 IP 或命令字符串；
- 默认只输出计划，真实执行需要显式授权；
- 使用持久
change_id，重复请求会返回同一任务； - 每个外部调用都有超时、退避、限流和终止条件；
- 部分失败可从逐目标状态恢复，死信任务会告警；
- 审计记录与回滚输入在变更结束后仍可查询。