15 SRE 自动化实战

运维自动化首先要安全、可观察、可回滚。脚本不仅要“能执行”，还要说明将执行什么、为什么失败、如何停止，以及留下了什么证据。

1. 设计顺序

1. 定义目标状态和成功证据。
2. 先实现只读巡检。
3. 增加 --dry-run 和结构化输出。
4. 加超时、并发上限和错误分类。
5. 再实现变更动作、幂等判断和回滚。
6. 接入日志、指标、任务 ID 和审计记录。

2. 日志分析器

from __future__ import annotations

import argparse
import json
from collections import Counter
from pathlib import Path
from typing import Any

def parse_line(line: str) -> dict[str, Any] | None:
    try:
        value = json.loads(line)
    except json.JSONDecodeError:
        return None
    return value if isinstance(value, dict) else None

def summarize(path: Path) -> Counter[str]:
    levels: Counter[str] = Counter()
    with path.open("r", encoding="utf-8") as handle:
        for line in handle:
            event = parse_line(line)
            if event is not None:
                levels[str(event.get("level", "UNKNOWN")).upper()] += 1
    return levels

def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("logfile", type=Path)
    args = parser.parse_args()
    print(json.dumps(summarize(args.logfile), ensure_ascii=False, sort_keys=True))
    return 0

if __name__ == "__main__":
    raise SystemExit(main())

生产版本还应记录坏行数量、采样样例、输入字节数和解析耗时，避免“解析成功但数据缺失”不可见。

3. 安全命令执行

from __future__ import annotations

import subprocess
from dataclasses import dataclass

@dataclass(frozen=True, slots=True)
class CommandResult:
    argv: tuple[str, ...]
    returncode: int
    stdout: str
    stderr: str

def run_command(argv: list[str], timeout: float = 10.0) -> CommandResult:
    completed = subprocess.run(
        argv,
        check=False,
        capture_output=True,
        text=True,
        timeout=timeout,
    )
    return CommandResult(
        argv=tuple(argv),
        returncode=completed.returncode,
        stdout=completed.stdout,
        stderr=completed.stderr,
    )

要求：

• 参数列表由程序构造，不拼接用户输入。
• 默认不使用 shell。
• 显式超时。
• 记录 argv 的程序名和参数摘要，但脱敏秘密参数。
• 控制 stdout/stderr 体积；大输出转向临时文件或流式处理。

4. HTTP 健康检查

from __future__ import annotations

import json
import ssl
import urllib.error
import urllib.request
from dataclasses import dataclass

@dataclass(frozen=True, slots=True)
class Health:
    ok: bool
    status: int | None
    detail: str

def check_health(url: str, timeout: float = 5.0) -> Health:
    context = ssl.create_default_context()
    request = urllib.request.Request(url, headers={"User-Agent": "ops-health/1.0"})
    try:
        with urllib.request.urlopen(request, context=context, timeout=timeout) as response:
            body = response.read(64 * 1024).decode("utf-8", errors="replace")
            if response.status != 200:
                return Health(False, response.status, body[:200])
            try:
                payload = json.loads(body)
                return Health(True, response.status, str(payload.get("status", "ok")))
            except json.JSONDecodeError:
                return Health(True, response.status, "non-json response")
    except urllib.error.HTTPError as exc:
        return Health(False, exc.code, exc.reason)
    except (urllib.error.URLError, TimeoutError) as exc:
        return Health(False, None, str(exc))

健康检查应区分“进程存活”“依赖可用”“是否接受新流量”。返回 200 不代表业务依赖一定健康。

5. 批量检查

from concurrent.futures import ThreadPoolExecutor, as_completed

def check_many(urls: list[str], workers: int = 8):
    results = {}
    with ThreadPoolExecutor(max_workers=workers) as executor:
        future_to_url = {executor.submit(check_health, url): url for url in urls}
        for future in as_completed(future_to_url):
            url = future_to_url[future]
            try:
                results[url] = future.result()
            except Exception as exc:
                results[url] = Health(False, None, f"{type(exc).__name__}: {exc}")
    return results

并发上限由目标服务承受能力、本机连接数和监控限流共同决定。批量工具还应支持输出顺序稳定、失败阈值和退出码约定。

6. SSH 与远程执行

优先使用组织已有的配置管理、远程执行平台或受限 API。直接使用 Paramiko 等库时：

• 验证主机密钥，不默认信任未知主机。
• 使用短期凭据或代理，不把私钥写入仓库。
• 限制命令白名单和并发。
• 分别在本地和远端设置超时。
• 记录主机、命令标识、返回码和脱敏后的摘要。
• 明确连接复用、关闭和失败清理。

远程批量变更至少要有 dry-run、分批窗口、停止开关和回滚方案。

7. 幂等与状态

幂等操作可以重复执行而不会产生额外副作用。实现方式：

• 使用资源的期望状态和唯一标识。
• 创建前检查是否存在。
• 使用条件更新和版本号避免覆盖新状态。
• 外部系统提供幂等键时沿用其语义。
• 脚本中断后可安全重跑。

记录本地状态文件不是分布式幂等保证。多执行器和故障切换场景应依赖权威系统状态。

8. 输出契约

机器可读输出推荐 JSON：

{
  "task_id": "check-20260929-001",
  "target": "db-01",
  "status": "failed",
  "error_type": "TimeoutError",
  "duration_ms": 5012,
  "details": "health endpoint timed out"
}

退出码约定：

• 0：全部成功。
• 1：业务检查失败。
• 2：参数或配置错误。
• 3：部分成功。
• 其他：由调用平台定义。

9. 审计与秘密

审计记录应回答：

• 谁发起。
• 何时执行。
• 对哪些目标。
• 使用哪个规则版本。
• 做了什么检查和变更。
• 结果和证据在哪里。

秘密通过环境变量、文件描述符、密钥管理服务或短期凭据获取。命令行参数可能出现在进程列表和历史记录中，敏感值不要直接传参。

10. 运维脚本验收

• 无参数和错误参数有明确帮助。
• --dry-run 不产生变更。
• 超时、并发和重试都有上限。
• 输出同时适合人和机器。
• 关键操作有任务 ID 和审计记录。
• 可以安全中断并重新运行。
• 有最小测试集和故障模拟。

官方参考

• argparse - Parser for command-line options
• subprocess - Subprocess management
• concurrent.futures - Launching parallel tasks
• logging - Logging facility
• json - JSON encoder and decoder