Discord 用 Scylla Control Plane 把 ScyllaDB 运维从"人肉"变成自动化

预计阅读时间：12 分钟

Discord 的基础设施团队规模不大，却要管着支撑数亿用户消息的 ScyllaDB 集群。以前做一次节点替换或集群扩容，几个人手动操作要耗好几天。他们干脆造了一个内部编排框架——Scylla Control Plane（SCP），把大规模 ScyllaDB 管理变成可编程的自动化流程。

手动运维撑不住的规模

Discord 从 Cassandra 迁移到 ScyllaDB 之后，性能和延迟确实大幅改善，但运维复杂度并没有消失。ScyllaDB 集群节点数量庞大，日常要处理的操作包括：

节点替换（硬件故障、升级）
集群扩缩容
数据再平衡与修复
版本升级滚动部署

这些操作每一步都涉及 token range 调整、流控参数设置、健康检查等待。手动执行时，工程师要反复在终端确认状态、敲命令、等响应，一个节点替换可能就要数小时，多节点滚动升级更是以天计。

人肉运维的另一个隐患是步骤遗漏——忘记设置流控上限导致集群被压垮，或者没等 repair 完成就继续下一步，数据一致性出问题。

SCP 的核心思路：把运维流程变成可编排的状态机

Scylla Control Plane 的设计理念并不复杂：把每个运维操作定义成一个有明确阶段和状态转换的工作流，由控制器自动推进，人只做审批和观察。

关键设计要素：

操作模板化：每种运维场景（节点替换、扩容、升级）都有预定义的 workflow，包含顺序执行的 stage 和每个 stage 的前置条件。
状态驱动：每个 stage 完成后自动检查目标状态是否达成，达标才进入下一 stage，否则回退或等待。
安全护栏：流控参数、并发度上限、超时阈值都硬编码在 workflow 定义里，操作者无法绕过。
观测集成：每步操作都写事件日志，与 Discord 的监控体系打通，异常随时可见。

这本质上就是 Kubernetes Operator 模式在 ScyllaDB 上的定制实现——用一个控制循环持续把集群的实际状态向期望状态收敛，只不过 SCP 的 workflow 更偏"一次性任务编排"而非持续 reconciliation。

一个节点替换的自动化流程长什么样

以节点替换为例，手动操作的典型步骤和 SCP 的自动化对应：

手动步骤	SCP 自动化
确认故障节点、选好替换节点	SCP 从 inventory 自动选取备机
新节点加入集群、分配 token range	SCP 调用 ScyllaDB API 完成 bootstrap
设置流控避免压垮集群	SCP 自动注入 `streaming_concurrency` 等参数
等待数据流完成	SCP 葎询 nodetool 状态直到 streaming 完成
旧节点 decommission	SCP 执行 decommission 并等待确认
验证数据一致性	SCP 触发 repair 并检查结果

整个过程从"工程师盯着终端敲命令"变成"工程师在 SCP 界面点一下 Replace Node，然后去喝咖啡"。

可落地的实践：用 Python 写一个迷你 ScyllaDB 操作编排器

SCP 是 Discord 内部系统，不开源，但它的核心模式可以复现。下面是一个用 Python 实现的简化版 ScyllaDB 节点替换编排器，你可以改造后用在自己的 ScyllaDB 或 Cassandra 集群上。

#!/usr/bin/env python3
"""mini_scylla_orchestrator.py — 简化版 ScyllaDB 节点替换编排器

依赖: pip install requests
运行前修改 CLUSTER_API_BASE 和节点信息。
仅用于学习与原型验证，生产环境需加入错误恢复、并发控制、审计日志等。
"""

import requests
import time
import sys

CLUSTER_API_BASE = "http://scylla-manager.internal:5080/api/v1"  # 改成你的 Scylla Manager API
STREAMING_CONCURRENCY = 4  # 流控上限，防止压垮集群
POLL_INTERVAL = 30  # 状态检查间隔秒数
MAX_WAIT_MINUTES = 120  # 单阶段最大等待时间


def api_get(path: str) -> dict:
    r = requests.get(f"{CLUSTER_API_BASE}{path}", timeout=10)
    r.raise_for_status()
    return r.json()


def api_post(path: str, body: dict = None) -> dict:
    r = requests.post(f"{CLUSTER_API_BASE}{path}", json=body, timeout=10)
    r.raise_for_status()
    return r.json()


def wait_for_condition(check_fn, description: str, max_minutes: int = MAX_WAIT_MINUTES):
    """轮询直到 check_fn 返回 True，超时则抛异常。"""
    deadline = time.time() + max_minutes * 60
    while time.time() < deadline:
        try:
            if check_fn():
                print(f"  ✓ {description} 完成")
                return
        except Exception as e:
            print(f"  ⚠ 检查异常: {e}")
        print(f"  … 等待 {description}，{POLL_INTERVAL}s 后重试")
        time.sleep(POLL_INTERVAL)
    raise TimeoutError(f"{description} 超时 ({max_minutes}min)")


def replace_node(faulty_host: str, new_host: str):
    """编排一次节点替换：bootstrap 新节点 → 流控 → decommission 旧节点 → repair"""

    print(f"=== 开始替换节点: {faulty_host} → {new_host} ===")

    # Stage 1: Bootstrap 新节点
    print("[Stage 1] Bootstrap 新节点")
    api_post("/cluster/nodes", {"host": new_host, "replace_host": faulty_host})
    wait_for_condition(
        lambda: api_get(f"/cluster/nodes/{new_host}/status")["status"] == "UN",
        "新节点加入集群并达到 UP+NORMAL"
    )

    # Stage 2: 设置流控参数
    print("[Stage 2] 设置流控")
    api_post(f"/cluster/nodes/{new_host}/config", {"streaming_concurrency": STREAMING_CONCENCY})
    print(f"  ✓ streaming_concurrency 设为 {STREAMING_CONCURRENCY}")

    # Stage 3: 等待数据流完成
    print("[Stage 3] 等待数据流完成")
    wait_for_condition(
        lambda: api_get(f"/cluster/nodes/{new_host}/streaming")["progress"] == 1.0,
        "数据流 100% 完成"
    )

    # Stage 4: Decommission 旧节点
    print("[Stage 4] Decommission 旧节点")
    api_post(f"/cluster/nodes/{faulty_host}/decommission")
    wait_for_condition(
        lambda: api_get(f"/cluster/nodes/{faulty_host}/status")["status"] == "LEFT",
        "旧节点完成 decommission"
    )

    # Stage 5: 触发 repair 确保一致性
    print("[Stage 5] Repair 数据一致性")
    repair_task = api_post("/repair", {"cluster_id": api_get("/cluster")["id"]})
    wait_for_condition(
        lambda: api_get(f"/repair/{repair_task['id']}")["status"] == "DONE",
        "Repair 完成"
    )

    print(f"=== 节点替换完成: {faulty_host} → {new_host} ===")


if __name__ == "__main__":
    if len(sys.argv) != 3:
        print("用法: python mini_scylla_orchestrator.py <故障节点IP> <新节点IP>")
        sys.exit(1)
    replace_node(sys.argv[1], sys.argv[2])

运行前你需要改的东西：

CLUSTER_API_BASE 指向你的 Scylla Manager 或自建的管理 API
流控参数 STREAMING_CONCURRENCY 根据集群规模调整，通常 2-8
如果没有 Scylla Manager，可以把 api_get/api_post 替换成直接调用 nodetool 的 subprocess 封装

这个脚本的核心价值不在代码本身，而在编排思路：每个阶段有明确的完成条件，流控参数硬编码不可绕过，超时自动失败而非无限等待。

如果你想更进一步：用 Kubernetes Operator 模式

SCP 的思路天然适合用 Kubernetes Operator 来实现。下面是一个最小化的 CRD 定义和 Operator 框架，适合已经在 K8s 上跑 ScyllaDB 的团队：

# scyllareplace.yaml — 节点替换任务 CRD 示例
apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
  name: scyllareplaces.db.internal
spec:
  group: db.internal
  versions:
    - name: v1
      served: true
      storage: true
      schema:
        openAPIV3Schema:
          type: object
          properties:
            spec:
              type: object
              properties:
                faultyNode:
                  type: string
                newNode:
                  type: string
                streamingConcurrency:
                  type: integer
                  default: 4
                maxWaitMinutes:
                  type: integer
                  default: 120
            status:
              type: object
              properties:
                phase:
                  type: string
                  enum: [Bootstrap, Streaming, Decommission, Repair, Done, Failed]
                message:
                  type: string
  scope: Namespaced
  names:
    plural: scyllareplaces
    singular: scyllareplace
    kind: ScyllaReplace
    shortNames: [sr]

提交一个替换任务：

# my-replace-task.yaml
apiVersion: db.internal/v1
kind: ScyllaReplace
metadata:
  name: replace-node-2024-06
spec:
  faultyNode: "10.0.1.42"
  newNode: "10.0.1.99"
  streamingConcurrency: 4
  maxWaitMinutes: 90

# 应用 CRD 和任务
kubectl apply -f scyllareplace.yaml
kubectl apply -f my-replace-task.yaml

# 查看任务状态
kubectl get scyllareplace replace-node-2024-06 -o wide
# 期望输出类似:
# NAME                PHASE        MESSAGE               AGE
# replace-node-2024-06   Streaming    数据流 67% 进行中    8m

Operator 控制循环的伪代码逻辑与上面的 Python 编排器一致：watch CRD 变更 → 根据 status.phase 决定下一步 → 更新 status → 重复直到 Done 或 Failed。

从 Discord 的做法里能带走什么

1. 运维流程先文档化，再代码化

Discord 在造 SCP 之前，手动运维步骤已经有完整文档。这很关键——如果你连手动怎么做都没写清楚，自动化就是空中楼阁。先把操作手册写出来，再逐段翻译成代码。

2. 流控护栏不可绕过

大规模数据集群最怕的不是慢，而是一个操作把整个集群压垮。SCP 把流控参数写进 workflow 定义，操作者无法"为了快点完成"而调高并发。你的编排系统也应该有类似的硬约束。

3. 小团队管大集群的关键是减少人在回路中的时间

Discord 的基础设施团队人数不多，SCP 让他们从"每步操作都要人盯着"变成"人只做决策和异常处理"。如果你的团队也在 5-10 人规模管着数十上百节点，这个投入方向值得认真考虑。

4. 不要一步到位造平台

SCP 也是逐步演进的——先自动化最频繁的节点替换，再覆盖扩容和升级。建议你也从最高频、最易出错的那个操作开始，写一个脚本跑稳了，再逐步扩展成框架。

5. 状态检查是自动化的命脉

每个 stage 的完成条件必须可观测、可程序化检查。如果你的集群缺少状态暴露的 API，先补这一层，否则编排器就是瞎子开车。

Discord 的 SCP 证明了一件事：当集群规模大到一定程度，运维自动化的投入不再是"锦上添花"，而是"没有就会死人"的基础设施。思路不复杂——把人做的事变成状态机驱动的工作流，加上护栏和观测。复杂的是坚持把它做完整、做可靠。