SRE 面试补全内容

[!info] 说明本文是 SRE 面试备战手册的补充内容，覆盖 15 道自测题完整答案、3 道代码题完整实现、以及 SRE 方法论 / K8s / AI 架构的深度展开。

一、15 道自测题完整答案

SRE 方法论（Q1-Q4）

Q1：你负责的业务 SLO 是什么？SLI 怎么定义的？错误预算怎么管理？

[!success] 标准答案模板 SLO 定义：

核心业务（C 端用户访问链路）SLO = 99.9%（月度窗口）

非核心业务（内部管理后台）SLO = 99.5%

SLI 定义：
SLI = 成功请求数 / 总请求数
分子：HTTP 2xx + 3xx 响应数（排除健康检查 /health 和内部探活 / readiness 请求）

分母：所有外部用户请求总数（不含内部组件互调）

5xx 算失败

4xx 不算 SLI 分母（客户端错误，不是服务端问题）

统计窗口：28 天滚动窗口（Google SRE 标准做法，平衡数据稳定性和时效性）

采集方式：全量采集，从 Nginx Ingress access log 统计，不采样

错误预算管理：
99.9% → 每月允许 43.2 分钟不可用（30天 × 24h × 60min × 0.1%）

预算消耗策略：
  < 50%  → 正常发布，可承担风险变更
  50-80% → 只允许低风险变更 + 必须有回滚预案
  > 80%  → 冻结非紧急发布，只允许 bugfix 和稳定性建设
  耗尽   → 完全冻结发布，强制做稳定性建设
落地方式：

Grafana 看板展示每月错误预算消耗趋势

消耗超 80% 自动触发企微通知 + 发布审批升级

每月 SRE 例会复盘预算消耗情况

Q2：MTTR 和 MTBF 你们怎么度量？MTTR 拆成几个阶段？每个阶段怎么优化？

[!success] 标准答案 度量方式：

每个 P0/P1 故障工单记录 5 个时间戳：故障发生时间、告警触发时间、人工响应时间、根因定位时间、服务恢复时间

MTTR = 服务恢复时间 - 故障发生时间

MTBF = 两次 P0/P1 故障之间的间隔时间

MTTR 四段拆解：

阶段计算优化手段我的实践
MTTD 告警触发 - 故障发生告警规则优化、指标采集频率提升 Prometheus 采集间隔从 60s 降到 15s，关键指标配多级告警
MTTA 人工响应 - 告警触发值班机制、告警分级收敛告警按 severity 分路由：critical 打电话、warning 发企微；同类告警 5 分钟内收敛
Diagnose 根因定位 - 人工响应日志/链路/指标关联分析 AI 根因分析系统，Tool Calling 渐进式分析，定位时间从约 1h 降到约 30min
Fix 服务恢复 - 根因定位回滚、扩容、切流、自动化 Runbook 标准化回滚脚本 + Argo Rollouts 灰度回滚；OOM 类故障自动扩容

关键表述： “我的 AI 根因分析项目就是在优化 Diagnose 环节，把诊断时间从约 1 小时压缩到约 30 分钟，降幅 50%。“

阶段	计算	优化手段	我的实践
MTTD	告警触发 - 故障发生	告警规则优化、指标采集频率提升	Prometheus 采集间隔从 60s 降到 15s，关键指标配多级告警
MTTA	人工响应 - 告警触发	值班机制、告警分级收敛	告警按 severity 分路由：critical 打电话、warning 发企微；同类告警 5 分钟内收敛
Diagnose	根因定位 - 人工响应	日志/链路/指标关联分析	AI 根因分析系统，Tool Calling 渐进式分析，定位时间从约 1h 降到约 30min
Fix	服务恢复 - 根因定位	回滚、扩容、切流、自动化 Runbook	标准化回滚脚本 + Argo Rollouts 灰度回滚；OOM 类故障自动扩容

Q3：你们做不做混沌工程？做不做变更冻结？RCA 的 action item 怎么跟踪闭环？

[!success] 标准答案 混沌工程：

目前在非生产环境做有限度演练（ChaosMesh 注入 Pod 故障、网络延迟）

大促前做一次全链路压测 + 故障演练

加分表述： “混沌工程我理解的核心不是制造混乱，而是验证已知假设——你的容灾设计在真实故障下到底能不能work。”

变更冻结：

大促前 3 天 + 大促期间冻结非紧急变更

错误预算消耗 > 80% 时冻结发布

冻结期间只允许 P0 bugfix，需 SRE Leader 审批

RCA 闭环流程：
故障发生 → 24h 内提交 RCA 报告 → 评审会讨论 → 产出 Action Items
  → 分配 Owner + DDL → 周会跟踪进度 → 完成 → 验证 → 关闭
Action Item 分三类：即时修复（24h 内）、短期改进（1-2 周）、长期建设（1-3 月）

用 Jira/TAPD 跟踪，每周 SRE 例会过一遍未关闭的 Action Items

关键指标： 同类故障复现率（目标 < 10%）

Q4：你经历过的最严重的一次线上故障是什么？讲一下从发现到恢复的全过程。

[!tip] 回答框架（STAR 法） Situation（背景）： 什么时间、什么业务、什么场景 Trigger（触发）： 什么变更或什么外部因素导致的 Action（行动）： 你做了什么——发现、响应、排查、恢复 Result（结果）： 影响范围、恢复时间、后续改进

模板示例（可根据实际经历替换）：

背景： 大促当天，核心交易链路 QPS 从 2000 涨到 8000。 触发： 大促流量涌入，下游 Redis 集群连接数打满，导致缓存超时，连锁触发服务雪崩。 发现： Prometheus 告警 3 分钟内触发——接口 5xx 率从 0.1% 飙升到 12%。 响应： 值班 SRE 2 分钟内拉群，拉入业务方 + DBA + 网络组。 排查： 通过 SkyWalking 链路发现大量 Redis 调用超时；查 SLS 日志确认 connection pool exhausted；查 Redis 集群发现连接数已达 maxclients 上限。 恢复： 紧急扩容 Redis 从 3 节点到 6 节点；同时通过 Nginx 限流降低入口 QPS 到 4000；5 分钟内恢复。 后续： RCA 产出 3 个 Action Item：① Redis 连接池参数调优 ② 接入限流中间件 ③ 大促前容量评审流程规范化。

面试加分点：

主动说出自己犯过的错误并讲清后续改进（真诚 > 完美）

强调你的角色：不是旁观者，而是决策者或关键排查者

数据说话：影响多少用户、持续多少分钟、QPS 降到多少

K8s / 云原生（Q5-Q8）

Q5：HPA 的扩容和缩容策略分别是什么？默认值是多少？怎么加速扩容？

[!success] 标准答案 HPA v2 扩容默认策略（scaleUp）：

behavior:
  scaleUp:
    stabilizationWindowSeconds: 0   # 扩容不做稳定窗口等待
    policies:
      - type: Percent
        value: 100                  # 每轮最多扩 100%（翻倍）
        periodSeconds: 15
      - type: Pods
        value: 4                    # 或最多加 4 个 Pod
        periodSeconds: 15
    selectPolicy: Max               # 取两个策略的较大值

HPA v2 缩容默认策略（scaleDown）：

behavior:
  scaleDown:
    stabilizationWindowSeconds: 300  # 缩容需要 5 分钟稳定窗口
    policies:
      - type: Percent
        value: 100                    # 每轮最多缩 100%
        periodSeconds: 15

缩容比扩容保守得多——快速扩容、缓慢缩容是设计原则。

HPA controller 轮询间隔： 默认 15 秒（--horizontal-pod-autoscaler-sync-period）

metrics-server 采集延迟： 15-30 秒

加速扩容方案：

# 方案 1：激进扩容策略
behavior:
  scaleUp:
    stabilizationWindowSeconds: 0
    policies:
      - type: Percent
        value: 200              # 每轮扩 200%（3倍增长）
        periodSeconds: 15
    selectPolicy: Max

# 方案 2：缩短 HPA 轮询间隔（kube-controller-manager 参数）
--horizontal-pod-autoscaler-sync-period=10s

# 方案 3：大促前预扩到目标副本数（最可靠）
kubectl scale deployment <name> --replicas=20

# 方案 4：预扩节点池（cluster-autoscaler 配合）
# 提前预留节点，Pod 创建后立即可调度

Q6：Pod 状态有哪些？Pending 和 CrashLoopBackOff 分别怎么排查？

[!success] 标准答案 Pod 完整生命周期状态：

状态含义常见原因
Pending 已提交但未调度节点资源不足、调度约束不满足
ContainerCreating 已调度，容器创建中镜像拉取慢、存储挂载失败
Running 正常运行 —
CrashLoopBackOff 容器反复崩溃重启应用启动失败、OOM、配置错误
ImagePullBackOff 镜像拉取失败镜像名错误、仓库认证失败、网络不通
Error 容器退出且非 0 应用异常退出
Completed 正常退出（Job/CronJob）任务执行完毕
Terminating 正在删除优雅退出中，可能卡在 finalizer

Pending 排查：
# 1. 看 Pod 事件
kubectl describe pod <pod-name>
# 关注 Events 区域，常见提示：
# - "0/10 nodes are available: 10 Insufficient cpu" → 节点 CPU 不够
# - "0/10 nodes are available: 10 Insufficient memory" → 节点内存不够
# - "0/10 nodes are available: 3 node(s) had untolerated taint" → 污点调度约束
# - "0/10 nodes are available: 3 node(s) didn't match node selector" → 标签不匹配

# 2. 看节点资源
kubectl top nodes
kubectl describe node <node-name>
# 看 Allocatable vs Allocated
CrashLoopBackOff 排查：
# 1. 看容器日志
kubectl logs <pod-name> --previous
# --previous 看上次崩溃的日志，关键！

# 2. 看 Pod 事件
kubectl describe pod <pod-name>
# 关注 Events 中的 "OOMKilled" / "Error" / "Liveness probe failed"

# 3. 看退出码
# Exit Code 137 → OOMKilled (128 + SIGKILL=9)
# Exit Code 1   → 应用异常
# Exit Code 143 → 正常退出 (128 + SIGTERM=15)

状态	含义	常见原因
Pending	已提交但未调度	节点资源不足、调度约束不满足
ContainerCreating	已调度，容器创建中	镜像拉取慢、存储挂载失败
Running	正常运行	—
CrashLoopBackOff	容器反复崩溃重启	应用启动失败、OOM、配置错误
ImagePullBackOff	镜像拉取失败	镜像名错误、仓库认证失败、网络不通
Error	容器退出且非 0	应用异常退出
Completed	正常退出（Job/CronJob）	任务执行完毕
Terminating	正在删除	优雅退出中，可能卡在 finalizer

Q7：requests 和 limits 的区别？不设 limits 会怎样？HPA 依赖哪个？

[!success] 标准答案 requests vs limits：

requests limits
作用调度依据（kube-scheduler 看这个）运行时上限（cgroup 限制）
调度决定 Pod 能不能调度到某节点不影响调度
运行时 不限制实际使用超了 CPU 会被 throttle，超了 Memory 会被 OOMKill
HPA HPA 依赖 requests HPA 不看 limits

HPA 计算公式（关键！）：
当前CPU利用率 = 实际CPU使用量 / CPU requests
如果 实际利用率 > target utilization → 触发扩容
所以 HPA 是基于 requests 算的，不是 limits。如果 requests 设得太低，HPA 会频繁触发；设得太高，HPA 迟钝。

不设 limits 会怎样：

CPU：不设 limit，Pod 可以使用节点上所有空闲 CPU，不会硬限制

Memory：不设 memory limit，Pod 可以用光节点内存，最终导致节点 OOM，kubelet 会按 QoS 等级杀 Pod

风险：一个 Pod 内存泄漏可以搞垮整个节点

QoS 等级（面试常问）：

QoS 条件优先级
Guaranteed requests == limits（CPU 和 Memory 都设且相等）最后被杀
Burstable requests < limits（至少一个设了 requests）中间被杀
BestEffort 都不设最先被杀

	requests	limits
作用	调度依据（kube-scheduler 看这个）	运行时上限（cgroup 限制）
调度	决定 Pod 能不能调度到某节点	不影响调度
运行时	不限制实际使用	超了 CPU 会被 throttle，超了 Memory 会被 OOMKill
HPA	HPA 依赖 requests	HPA 不看 limits

QoS	条件	优先级
Guaranteed	requests == limits（CPU 和 Memory 都设且相等）	最后被杀
Burstable	requests < limits（至少一个设了 requests）	中间被杀
BestEffort	都不设	最先被杀

Q8：你们 K8s 网络模型是什么？CNI 用的什么？NetworkPolicy 用过吗？

[!tip] 参考答案 K8s 网络模型：

每个 Pod 有独立 IP，Pod 间直接通信，不做 NAT

Pod 看到的自身 IP 和其他 Pod 看到的一致

Node 上的 Pod 可以和所有节点上的 Pod 通信（无 NAT）

CNI（Container Network Interface）：

CNI 模式特点适用场景
Flannel VXLAN/Host-GW 简单，无 NetworkPolicy 小规模集群
Calico BGP/ VXLAN 支持 NetworkPolicy，性能好大规模生产
Cilium eBPF 性能最优，支持 L7 策略高性能场景
Terway 阿里云 ENI Pod 直接分配 ENI 弹性网卡 ACK 环境

NetworkPolicy（如果用过）：
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: allow-web-to-api
  namespace: prod
spec:
  podSelector:
    matchLabels:
      app: api-service      # 作用于 api-service 的 Pod
  policyTypes:
    - Ingress
  ingress:
    - from:
        - podSelector:
            matchLabels:
              app: web-frontend  # 只允许 web-frontend 访问
      ports:
        - protocol: TCP
          port: 8080
如果没用过也要会答： “NetworkPolicy 是 K8s 原生的网络隔离机制，通过 label selector 控制 Pod 间的通信白名单。需要 CNI 支持（Calico/Cilium 支持，Flannel 默认不支持）。“

CNI	模式	特点	适用场景
Flannel	VXLAN/Host-GW	简单，无 NetworkPolicy	小规模集群
Calico	BGP/ VXLAN	支持 NetworkPolicy，性能好	大规模生产
Cilium	eBPF	性能最优，支持 L7 策略	高性能场景
Terway	阿里云 ENI	Pod 直接分配 ENI 弹性网卡	ACK 环境

AI / AIOps（Q9-Q12）

Q9：你的 AI 根因分析系统，LLM 具体做了什么决策？Tool Calling 的流程画出来。

[!success] 标准答案 LLM 的核心角色：推理决策者

不是把所有数据丢给 LLM 让它总结
LLM 每轮根据当前 context（告警信息 + 已有 Tool 返回结果），推理”下一步该查什么”，调用对应 Tool
这是一个 ReAct（Reasoning + Acting）循环

Tool Calling 流程：

[告警Context]
      │
      ▼
┌──────────────┐
│  LLM Round 1 │ ──→ 推理：502是网关错误，先查链路
│  (分析Context) │ ──→ 调用：query_skywalking_trace(...)
└──────┬───────┘            │
       │              ┌─────▼─────┐
       │              │ Tool 执行  │ ──→ SkyWalking API
       │              └─────┬─────┘
       │                    │ 返回链路数据
       ▼                    │
┌──────────────┐            │
│  LLM Round 2 │ ◄─────────┘
│  (分析结果)   │ ──→ 推理：service-B 异常，查日志
│               │ ──→ 调用：query_sls_logs(...)
└──────┬───────┘
       │ ... 继续循环 ...
       ▼
┌──────────────┐
│  LLM Round N │ ──→ 推理：已定位根因，输出结论
│  (输出根因)   │ ──→ 生成根因报告 + 处理建议
└──────────────┘

Tool 的 schema 示例（OpenAI Function Calling 格式）：

{
  "name": "query_sls_logs",
  "description": "查询阿里云SLS日志服务，获取指定服务在指定时间范围内的日志。用于排查服务异常、错误日志、OOM等问题。",
  "parameters": {
    "type": "object",
    "properties": {
      "service": {
        "type": "string",
        "description": "服务名称，如 'service-A', 'service-B'"
      },
      "keyword": {
        "type": "string",
        "description": "日志关键词，如 'error', 'OOM', 'panic', 'timeout'"
      },
      "time_range": {
        "type": "string",
        "description": "时间范围，如 '最近10分钟', '最近1小时'"
      },
      "limit": {
        "type": "integer",
        "description": "返回日志条数，默认50"
      }
    },
    "required": ["service", "time_range"]
  }
}

关键设计点：

Tool 的 description 要写得足够清晰，让 LLM 知道什么时候该调这个 Tool
每轮把上一步的 Tool 返回结果 append 到 context 中，LLM 看到累积的信息做推理
设置最大轮次（如 10 轮）防止 LLM 陷入死循环
每轮加超时控制（如 30 秒），超时则强制结束

Q10：知识库是怎么建设的？RAG？向量库？图结构？相似度怎么算的？

[!success] 标准答案 知识库分三层（不是单一方案）：

层级技术内容用途
L1：案例库 结构化存储（DB） {症状, 根因, 服务, 方案, 触发条件, 时间} 精确匹配历史案例
L2：文档库 RAG + 向量检索公司链路梳理文档、SOP、RCA 报告语义检索相关知识
L3：拓扑图 图结构（可选）服务依赖关系（A→B→C）快速定位影响面和上下游

L1 案例库查询（精确匹配）：
# 告警触发后，先查案例库
def search_rca_cases(symptom: str, service: str) -> list[dict]:
    """从历史 RCA 案例库中匹配相似故障"""
    # 精确匹配：症状 + 服务名
    exact = db.query(
        "SELECT * FROM rca_cases WHERE symptom = ? AND service = ? ORDER BY created_at DESC",
        [symptom, service]
    )
    if exact:
        return exact[:5]  # 返回最近5个匹配案例
    # 模糊匹配：症状关键词 LIKE
    fuzzy = db.query(
        "SELECT * FROM rca_cases WHERE symptom LIKE ? ORDER BY created_at DESC",
        [f"%{symptom}%"]
    )
    return fuzzy[:3]
L2 向量检索（语义匹配）：
# 用 embedding 模型把文档和告警描述向量化
from openai import OpenAI
client = OpenAI()

def search_rag_docs(query: str, top_k: int = 5) -> list[str]:
    """向量检索相关文档"""
    query_vec = client.embeddings.create(
        input=query, model="text-embedding-3-small"
    ).data[0].embedding
    # 在向量库（Milvus/Pgvector）中做余弦相似度搜索
    results = vector_db.search(
        query_vec, top_k=top_k, metric="cosine"
    )
    return [r.text for r in results if r.score > 0.7]
相似度计算：

L1 案例库：结构化字段精确匹配 + 关键词 LIKE 模糊匹配

L2 向量库：余弦相似度（cosine similarity），阈值 > 0.7 算相关

如果有人问”知识图谱”： 可以说 L3 拓扑图是图结构，用 Neo4j 存服务依赖关系，但核心分析还是靠 L1+L2

面试表述： “知识库不是单一方案，而是三层结构——结构化案例库做精确匹配，RAG 向量检索做语义关联，拓扑图做影响面分析。三者协同，加速收敛。“

层级	技术	内容	用途
L1：案例库	结构化存储（DB）	`{症状, 根因, 服务, 方案, 触发条件, 时间}`	精确匹配历史案例
L2：文档库	RAG + 向量检索	公司链路梳理文档、SOP、RCA 报告	语义检索相关知识
L3：拓扑图	图结构（可选）	服务依赖关系（A→B→C）	快速定位影响面和上下游

Q11：LLM 分析出错了怎么办？有没有护栏机制？结果怎么校验？

[!success] 标准答案 护栏机制（Guardrails）四层设计：

┌─────────────────────────────────────────────────────┐
│ Layer 1: 轮次控制                                     │
│   max_rounds = 10，超过强制结束                       │
│   每轮超时 30 秒，超时强制下一步                       │
├─────────────────────────────────────────────────────┤
│ Layer 2: Tool 调用校验                                │
│   - Tool 参数白名单校验（防注入）                       │
│   - 只读操作，不允许写操作（不做回滚/扩容等变更）        │
│   - 返回结果格式校验（JSON schema validate）           │
├─────────────────────────────────────────────────────┤
│ Layer 3: 结果置信度评估                                │
│   - LLM 输出时要求附带 confidence score（0-1）        │
│   - confidence < 0.6 → 标记为"需人工确认"              │
│   - confidence > 0.8 → 自动推送到工单系统              │
├─────────────────────────────────────────────────────┤
│ Layer 4: 人工兜底                                     │
│   - 所有分析结果推送企微通知值班 SRE                    │
│   - SRE 确认后才会写入知识库（防止错误案例污染）        │
│   - 如果 LLM 分析错误，SRE 手动纠正 → 反馈给系统       │
└─────────────────────────────────────────────────────┘

关键设计原则：

AI 只做分析，不做执行 — 根因分析结果给出建议，但回滚/扩容等操作需要人工确认
错误可追溯 — 每轮 Tool 调用记录日志，出错可以回放分析过程
持续学习 — SRE 纠正的结果作为 negative case 存入知识库，优化下次分析

如果面试官问”LLM 幻觉怎么办”：

“LLM 幻觉是真实风险。我的应对是：① Tool 返回的都是真实数据（API 实时查询），不是 LLM 生成的，LLM 只做推理不做数据生成；② 要求 LLM 每个结论都标注数据来源（引用哪轮 Tool 的哪个返回值）；③ 最终结果人工确认后才入库。“

Q12：你觉得 AI 在 SRE 领域还有哪些应用场景？SRE Agent 应该具备什么能力？

[!tip] 开放题，展示对 AI+SRE 的深度思考 AI 在 SRE 的应用场景矩阵：

场景 AI 能力成熟度价值
故障根因分析 LLM + Tool Calling 渐进式分析 ✅ 我已落地缩短 MTTR Diagnose
告警降噪与分组 LLM 语义理解合并同类告警 ✅ 可落地减少 MTTA
日志异常检测 LLM 总结异常模式 / 时序模型 ✅ 可落地缩短 MTTD
容量预测 时序预测 + LLM 生成建议 ⚠️ 中等预防性扩容
自动化 Runbook 执行 Agent 自主执行恢复操作 ⚠️ 探索中缩短 Fix 环节
变更风险评估 LLM 分析变更内容 + 历史故障关联 ⚠️ 探索中提升 MTBF
RCA 报告生成 LLM 从故障时间线自动生成 RCA ✅ 可落地效率提升
值班 Copilot 值班时实时辅助排查 + 知识检索 ✅ 可落地降门槛

SRE Agent 的核心能力（面试加分点）：
┌──────────────────────────────────────────────────┐
│            SRE Agent 能力金字塔                     │
├──────────────────────────────────────────────────┤
│                                                   │
│            [自主执行恢复操作]                        │
│           [风险评估 + 决策建议]                      │
│          [根因分析 + 知识检索]                       │
│        [日志/指标/链路 查询能力]                     │
│      [告警理解 + 上下文构建]                         │
│    [K8s/云平台 API 调用能力]                        │
│  [安全边界：只读优先，写操作需确认]                   │
└──────────────────────────────────────────────────┘
关键表述：

“SRE Agent 的终局不是替代人，而是把 SRE 从重复劳动中解放出来。短期是 Copilot（辅助排查），中期是 Autopilot（自主分析+建议），长期才可能走向闭环自主恢复。安全边界很重要——Agent 可以做分析、做建议，但执行变更操作必须有 human-in-the-loop。“

场景	AI 能力	成熟度	价值
故障根因分析	LLM + Tool Calling 渐进式分析	✅ 我已落地	缩短 MTTR Diagnose
告警降噪与分组	LLM 语义理解合并同类告警	✅ 可落地	减少 MTTA
日志异常检测	LLM 总结异常模式 / 时序模型	✅ 可落地	缩短 MTTD
容量预测	时序预测 + LLM 生成建议	⚠️ 中等	预防性扩容
自动化 Runbook 执行	Agent 自主执行恢复操作	⚠️ 探索中	缩短 Fix 环节
变更风险评估	LLM 分析变更内容 + 历史故障关联	⚠️ 探索中	提升 MTBF
RCA 报告生成	LLM 从故障时间线自动生成 RCA	✅ 可落地	效率提升
值班 Copilot	值班时实时辅助排查 + 知识检索	✅ 可落地	降门槛

代码能力（Q13-Q15）

Q13：写一个令牌桶限流器（支持并发安全）

[!example] Python 实现

import threading
import time


class TokenBucketRateLimiter:
    """令牌桶限流器

    原理：
    - 以固定速率往桶里放令牌（rate: 每秒放多少令牌）
    - 桶有容量上限（capacity: 桶最多放多少令牌）
    - 请求来了取一个令牌，有令牌就放行，没有就等待/拒绝
    - 支持突发流量：桶满时可以一次性消费多个令牌
    """

    def __init__(self, rate: float, capacity: int):
        self._rate = rate            # 每秒生成令牌数
        self._capacity = capacity     # 桶容量
        self._tokens = float(capacity)  # 当前令牌数，初始化为满
        self._last_time = time.time()  # 上次补充令牌的时间
        self._lock = threading.Lock()

    def acquire(self, tokens: int = 1, timeout: float = 0) -> bool:
        """获取令牌，返回是否成功

        Args:
            tokens: 需要的令牌数
            timeout: 超时时间，0=不等待直接返回
        Returns:
            True: 获取成功，False: 获取失败
        """
        deadline = time.time() + timeout if timeout > 0 else None
        while True:
            with self._lock:
                self._refill()
                if self._tokens >= tokens:
                    self._tokens -= tokens
                    return True
                if deadline is None:
                    return False  # 不等待，直接拒绝
                # 计算需要等待的时间
                deficit = tokens - self._tokens
                wait_time = deficit / self._rate
            # 释放锁后等待（不持有锁等待！）
            remaining = deadline - time.time()
            if remaining <= 0 or wait_time > remaining:
                return False
            time.sleep(min(wait_time, 0.1))  # 最多等 100ms 再重试

    def _refill(self):
        """补充令牌（必须在锁内调用）"""
        now = time.time()
        elapsed = now - self._last_time
        # 经过的时间 × 速率 = 新增令牌
        new_tokens = elapsed * self._rate
        self._tokens = min(self._capacity, self._tokens + new_tokens)
        self._last_time = now


# 使用示例
if __name__ == "__main__":
    limiter = TokenBucketRateLimiter(rate=10, capacity=20)
    # rate=10: 每秒10个令牌, capacity=20: 桶最多20个
    for i in range(30):
        if limiter.acquire(timeout=0.5):
            print(f"请求 {i}: 通过")
        else:
            print(f"请求 {i}: 被限流")

[!example] Go 实现

package main

import (
    "sync"
    "time"
)

type TokenBucketLimiter struct {
    mu        sync.Mutex
    rate      float64       // 每秒生成令牌数
    capacity  float64       // 桶容量
    tokens    float64       // 当前令牌数
    lastTime  time.Time     // 上次补充时间
}

func NewTokenBucketLimiter(rate float64, capacity float64) *TokenBucketLimiter {
    return &TokenBucketLimiter{
        rate:     rate,
        capacity: capacity,
        tokens:   capacity,  // 初始化为满
        lastTime: time.Now(),
    }
}

func (l *TokenBucketLimiter) Acquire(tokens float64, timeout time.Duration) bool {
    deadline := time.Now().Add(timeout)
    for {
        l.mu.Lock()
        l.refill()
        if l.tokens >= tokens {
            l.tokens -= tokens
            l.mu.Unlock()
            return true
        }
        deficit := tokens - l.tokens
        waitTime := time.Duration(float64(time.Second) * deficit / l.rate)
        l.mu.Unlock()

        if timeout == 0 {
            return false // 不等待
        }
        remaining := time.Until(deadline)
        if remaining <= 0 || waitTime > remaining {
            return false
        }
        time.Sleep(minDuration(waitTime, 100*time.Millisecond))
    }
}

func (l *TokenBucketLimiter) refill() {
    now := time.Now()
    elapsed := now.Sub(l.lastTime).Seconds()
    newTokens := elapsed * l.rate
    l.tokens = minFloat(l.capacity, l.tokens+newTokens)
    l.lastTime = now
}

func minFloat(a, b float64) float64 {
    if a < b {
        return a
    }
    return b
}

func minDuration(a, b time.Duration) time.Duration {
    if a < b {
        return a
    }
    return b
}

设计要点：

要点	说明
令牌补充是惰性的	不是真的有个线程在放令牌，而是请求来了时算”从上次到现在该补多少”
不持有锁等待	获取不到令牌时释放锁、sleep 后重试，避免阻塞其他请求
capacity 决定突发能力	桶容量越大，能承受的瞬时突发流量越大
rate 决定平均速率	每秒生成多少令牌 = 平均允许的 QPS

[!question] 面试追问：令牌桶 vs 漏桶？

令牌桶：允许突发（桶满时一次性消费多个令牌），适合有波峰的场景

漏桶：匀速流出，不允许突发，适合需要严格匀速的场景

生产实践：API 限流一般用令牌桶（允许短暂突发），消息队列消费一般用漏桶（匀速消费）

Q14：写一个简单的 HTTP 健康检查器（并发检查多个 endpoint，超时控制）

[!example] Go 实现（Go 最适合这类并发场景）

package main

import (
    "context"
    "fmt"
    "net/http"
    "sync"
    "time"
)

type CheckResult struct {
    Name     string
    URL      string
    Status   int
    Latency  time.Duration
    Error    error
    Healthy  bool
}

type HealthChecker struct {
    client    *http.Client
    endpoints map[string]string  // name → URL
}

func NewHealthChecker(timeout time.Duration) *HealthChecker {
    return &HealthChecker{
        client: &http.Client{
            Timeout: timeout,  // 每个 HTTP 请求的超时
        },
        endpoints: make(map[string]string),
    }
}

func (hc *HealthChecker) AddEndpoint(name, url string) {
    hc.endpoints[name] = url
}

// CheckAll 并发检查所有 endpoint
func (hc *HealthChecker) CheckAll(ctx context.Context) []CheckResult {
    var wg sync.WaitGroup
    results := make([]CheckResult, len(hc.endpoints))
    resultChan := make(chan CheckResult, len(hc.endpoints))

    for name, url := range hc.endpoints {
        wg.Add(1)
        go func(name, url string) {
            defer wg.Done()
            result := hc.checkOne(ctx, name, url)
            resultChan <- result
        }(name, url)
    }

    // 等所有 goroutine 完成
    go func() {
        wg.Wait()
        close(resultChan)
    }()

    // 收集结果
    var i int
    for r := range resultChan {
        results[i] = r
        i++
    }
    return results
}

func (hc *HealthChecker) checkOne(ctx context.Context, name, url string) CheckResult {
    start := time.Now()
    result := CheckResult{Name: name, URL: url}

    req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
    if err != nil {
        result.Error = fmt.Errorf("create request: %w", err)
        return result
    }

    resp, err := hc.client.Do(req)
    if err != nil {
        result.Error = fmt.Errorf("request failed: %w", err)
        result.Latency = time.Since(start)
        return result
    }
    defer resp.Body.Close()

    result.Status = resp.StatusCode
    result.Latency = time.Since(start)
    result.Healthy = resp.StatusCode >= 200 && resp.StatusCode < 400
    return result
}

// 使用示例
func main() {
    checker := NewHealthChecker(5 * time.Second)
    checker.AddEndpoint("api-gateway", "http://api.example.com/health")
    checker.AddEndpoint("user-service", "http://user.example.com/health")
    checker.AddEndpoint("order-service", "http://order.example.com/health")

    // 总超时 10 秒
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()

    results := checker.CheckAll(ctx)
    for _, r := range results {
        status := "✓"
        if !r.Healthy {
            status = "✗"
        }
        fmt.Printf("%s %s: status=%d latency=%v err=%v\n",
            status, r.Name, r.Status, r.Latency, r.Error)
    }
}

设计要点：

要点	说明
`context.WithTimeout`	总超时控制，所有请求共享一个 context
`http.Client.Timeout`	单个请求超时
`sync.WaitGroup` + channel	并发执行 + 收集结果
`NewRequestWithContext`	请求绑定 context，超时自动取消
`defer resp.Body.Close()`	防止连接泄漏

Q15：写一个日志 tail 工具（监听文件变化，实时输出新增内容）

[!example] Go 实现

package main

import (
    "bufio"
    "flag"
    "fmt"
    "os"
    "time"
)

// LogTail 实时跟踪文件新增内容
// 原理：
// 1. 打开文件，定位到文件末尾
2. 循环读取新内容
// 3. 文件没有新内容时 sleep 后重试
// 4. 支持文件被 rotate（重命名+重建）
type LogTail struct {
    filename string
    file     *os.File
    reader   *bufio.Reader
}

func NewLogTail(filename string) (*LogTail, error) {
    f, err := os.Open(filename)
    if err != nil {
        return nil, fmt.Errorf("open file: %w", err)
    }
    // 定位到文件末尾（只看新增内容）
    if _, err := f.Seek(0, 2); err != nil {
        f.Close()
        return nil, fmt.Errorf("seek end: %w", err)
    }
    return &LogTail{
        filename: filename,
        file:     f,
        reader:   bufio.NewReader(f),
    }, nil
}

// Follow 持续读取新内容，通过 line channel 输出
func (lt *LogTail) Follow(lines chan<- string) {
    for {
        line, err := lt.reader.ReadString('\n')
        if err == nil {
            // 正常读取到一行
            lines <- line
            continue
        }
        // 读到 EOF，没有新内容
        // 检查文件是否被 rotate（inode 变化）
        if lt.isRotated() {
            lt.file.Close()
            f, err := os.Open(lt.filename)
            if err != nil {
                time.Sleep(1 * time.Second)
                continue
            }
            lt.file = f
            lt.reader = bufio.NewReader(f)
            continue
        }
        // 没有新内容，sleep 后重试
        time.Sleep(200 * time.Millisecond)
    }
}

// isRotated 检测文件是否被 rotate（logrotate 场景）
func (lt *LogTail) isRotated() bool {
    fi1, err := os.Stat(lt.filename)
    if err != nil {
        return false
    }
    fi2, err := lt.file.Stat()
    if err != nil {
        return true // 文件句柄失效，重新打开
    }
    // 比较 inode（Linux）/ 体积突然变小（跨平台）
    if fi1.Size() < fi2.Size() {
        return true // 文件变小了，说明被 rotate
    }
    return false
}

func (lt *LogTail) Close() {
    if lt.file != nil {
        lt.file.Close()
    }
}

// 使用示例
func main() {
    filename := flag.String("f", "", "log file to tail")
    flag.Parse()
    if *filename == "" {
        fmt.Fprintln(os.Stderr, "Usage: logtail -f <file>")
        os.Exit(1)
    }

    tail, err := NewLogTail(*filename)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Error: %v\n", err)
        os.Exit(1)
    }
    defer tail.Close()

    lines := make(chan string, 100)
    go tail.Follow(lines)

    for line := range lines {
        fmt.Print(line)
    }
}

设计要点：

要点	说明
`Seek(0, 2)`	打开文件后定位到末尾，只看新增内容
`ReadString('\n')`	按行读取，读到 `\n` 为一行
轮询 + sleep	没有新内容时 sleep 200ms 重试，避免 CPU 空转
文件 rotate 检测	文件大小突然变小 → 被重命名重建了 → 重新打开
channel 缓冲	100 行缓冲，防止消费方慢导致阻塞

[!question] 面试追问：为什么不用 inotify/fsnotify？

“inotify 是更高效的方案（事件驱动，不需要轮询），但跨平台兼容性不如轮询。生产环境可以用 fsnotify 库。这个实现用轮询是为了简单可移植。如果面试官问，说知道 fsnotify 但手写轮询更理解底层原理。“

二、SRE 方法论深度补充

2.1 RCA 报告模板

[!example] RCA 模板（面试时可以说”我们每次故障后都会写 RCA”）

# RCA：[故障标题]

## 1. 基本信息
| 项目 | 内容 |
|------|------|
| 故障时间 | 2026-08-03 14:00 - 14:15 (15min) |
| 影响范围 | C端用户登录接口 5xx 率 15%，约 50 万用户受影响 |
| 严重等级 | P0 |
| 故障负责人 | 徐航 |

## 2. 故障时间线
| 时间 | 事件 |
|------|------|
| 14:00:00 | 流量突增，Redis 连接数开始上升 |
| 14:01:30 | Prometheus 告警触发：Redis连接池使用率 > 90% |
| 14:02:00 | 值班 SRE 响应，拉群 |
| 14:03:00 | 通过链路分析定位到 Redis 超时 |
| 14:05:00 | 确认根因：Redis 连接数达 maxclients 上限 |
| 14:08:00 | 紧急扩容 Redis 3→6 节点 |
| 14:10:00 | Nginx 限流降到 4000 QPS |
| 14:13:00 | 5xx 率恢复到 0.1% |
| 14:15:00 | 确认恢复，关闭故障工单 |

## 3. 根因分析
### 直接原因
Redis 集群连接数达到 maxclients (10000) 上限，新连接被拒绝，
导致缓存查询超时，上层服务雪崩。

### 根本原因
1. Redis 连接池参数未随流量增长调优（maxclients 仍为大促前配置）
2. 缺少自动限流保护机制
3. 大促前容量评审未覆盖 Redis 连接数维度

## 4. 影响评估
- 用户影响：约 50 万用户登录失败/超时
- 业务影响：约 15 分钟服务降级
- 错误预算消耗：15/43.2 = 34.7%

## 5. Action Items
| 类型 | 内容 | Owner | DDL |
|------|------|-------|-----|
| 即时修复 | Redis maxclients 调至 20000 | DBA | 24h |
| 短期改进 | 接入限流中间件（Sentinel） | SRE | 1周 |
| 短期改进 | 大促前容量评审增加连接数维度 | SRE | 1周 |
| 长期建设 | Redis 集群自动扩容方案 | DBA+SRE | 1月 |

2.2 错误预算实战计算

[!example] 错误预算计算（面试可能会考）
SLO = 99.9%
统计窗口 = 30天

总分钟数 = 30 × 24 × 60 = 43,200 分钟
允许不可用时间 = 43,200 × (1 - 0.999) = 43.2 分钟

不同 SLO 对应的每月允许不可用时间：
| SLO | 每月允许不可用 | 每天 |
|-----|-------------|------|
| 99%    | 432 分钟 (7.2h) | 14.4 分钟 |
| 99.5%  | 216 分钟 (3.6h) | 7.2 分钟 |
| 99.9%  | 43.2 分钟       | 1.44 分钟 |
| 99.95% | 21.6 分钟       | 0.72 分钟 |
| 99.99% | 4.32 分钟       | 0.14 分钟 |
| 99.999%| 0.43 分钟 (26秒) | 0.86 秒 |
面试表述： “99.9% 看似很高，但每月只有 43 分钟的容错空间。一次 P0 故障如果持续 15 分钟，就消耗了 35% 的错误预算。如果同月再来一次类似故障，错误预算就耗尽了，需要冻结发布做稳定性建设。“

2.3 Runbook 示例

[!example] 标准化 Runbook（面试可以说”我们有标准化运维手册”）

# Runbook：Redis 连接数告警

## 告警信息
- 告警名称：Redis_Connection_Pool_High
- 触发条件：连接池使用率 > 90% 持续 1 分钟
- 严重等级：Warning

## 影响判断
1. 确认告警真实性：Grafana 查看 Redis 连接数趋势
2. 判断影响面：是否已导致上层服务 5xx？

## 处理步骤
### Step 1：确认当前连接数
redis-cli -h <host> -p <port> info clients
# 看 connected_clients 和 maxclients

### Step 2：排查连接来源
# 看哪个服务连接最多
redis-cli -h <host> -p <port> client list | awk '{print $2}' | sort | uniq -c | sort -rn | head

### Step 3：决策
- 连接数 < 80% maxclients → 临时波动，观察
- 连接数 80-95% → 检查应用连接池配置，是否泄漏
- 连接数 > 95% → 执行扩容

### Step 4：扩容操作（需审批）
# 1. 增加 Redis 节点
# 2. 更新应用配置中的 Redis 地址
# 3. 滚动重启应用
# 4. 确认连接数下降

## 回滚方案
- 扩容前的 Redis 节点保留 24h，可随时切回
- 应用配置回滚：git revert + 重新发布

三、K8s 深度补充

3.1 Pod 故障排查速查表

[!info] 常见 Pod 问题排查

现象第一步第二步第三步
Pending kubectl describe pod 看 Events 检查节点资源 kubectl top nodes 检查 nodeSelector/toleration
CrashLoopBackOff kubectl logs --previous 检查退出码检查资源 limits（OOM）
ImagePullBackOff kubectl describe pod 看 Events 检查镜像名拼写检查 imagePullSecrets
Pod 被驱逐 kubectl describe node 看 Events 检查节点压力状况调整 requests 或加节点
服务不通 kubectl get endpoints kubectl describe svc 检查 readiness probe
DNS 不解析 kubectl exec -- nslookup 检查 CoreDNS 检查 DNS policy

退出码速查：

Exit Code 含义
0 正常退出
1 应用错误
137 OOMKilled (128 + SIGKILL 9)
143 收到 SIGTERM 正常退出 (128 + 15)
126 权限不足
127 命令未找到

现象	第一步	第二步	第三步
Pending	`kubectl describe pod` 看 Events	检查节点资源 `kubectl top nodes`	检查 nodeSelector/toleration
CrashLoopBackOff	`kubectl logs --previous`	检查退出码	检查资源 limits（OOM）
ImagePullBackOff	`kubectl describe pod` 看 Events	检查镜像名拼写	检查 imagePullSecrets
Pod 被驱逐	`kubectl describe node` 看 Events	检查节点压力状况	调整 requests 或加节点
服务不通	`kubectl get endpoints`	`kubectl describe svc`	检查 readiness probe
DNS 不解析	`kubectl exec -- nslookup`	检查 CoreDNS	检查 DNS policy

Exit Code	含义
0	正常退出
1	应用错误
137	OOMKilled (128 + SIGKILL 9)
143	收到 SIGTERM 正常退出 (128 + 15)
126	权限不足
127	命令未找到

3.2 requests / limits / QoS 最佳实践

[!tip] 生产环境配置建议 核心业务 Pod：Guaranteed QoS
resources:
  requests:
    cpu: 2           # requests = limits
    memory: 2Gi
  limits:
    cpu: 2
    memory: 2Gi
优点：QoS 等级最高，最后被驱逐；HPA 基于 requests 计算最稳定。

非核心业务 Pod：Burstable QoS
resources:
  requests:
    cpu: 500m
    memory: 512Mi
  limits:
    cpu: 1           # limits > requests，允许突发
    memory: 1Gi
优点：允许突发 CPU 使用，内存有上限保护。

CPU 不要设 limit（推荐做法）：
resources:
  requests:
    cpu: 2
    memory: 2Gi
  # 不设 limits
原因：CPU 是可压缩资源，设 limit 会导致 CPU throttle，应用延迟突增但很难排查。只设 requests，让 Pod 使用空闲 CPU。

3.3 常用 kubectl 排查命令速查

[!example] 面试时能随口说出的命令

# === Pod 级别 ===
kubectl get pods -n <ns> -o wide              # Pod 状态和所在节点
kubectl describe pod <name> -n <ns>           # 事件、探针、资源、调度
kubectl logs <name> -n <ns> --previous         # 上次崩溃的日志
kubectl logs <name> -n <ns> -c <container>    # 指定容器
kubectl exec -it <name> -n <ns> -- sh         # 进入 Pod

# === Service / Endpoint ===
kubectl get svc -n <ns> -o wide                # Service 列表
kubectl get endpoints <name> -n <ns>           # Endpoints（关键！服务不通先看这个）
kubectl describe svc <name> -n <ns>            # selector 和端口映射

# === HPA ===
kubectl get hpa -n <ns>                        # HPA 状态
kubectl describe hpa <name> -n <ns>           # 扩缩容事件和 Conditions

# === Node ===
kubectl get nodes -o wide                      # 节点状态
kubectl describe node <name>                   # 资源分配、污点、事件
kubectl top nodes                              # 节点实时资源使用
kubectl top pods -n <ns>                       # Pod 实时资源使用

# === 集群级别 ===
kubectl get events -n <ns> --sort-by=.metadata.creationTimestamp  # 按时间排序事件
kubectl get all -n <ns>                        # 所有资源

四、AI 架构深度补充

4.1 完整 Tool 注册表

[!example] 面试时能说出的完整 Tool 列表

# Tool 注册表 — 每个工具的 schema 清晰描述用途和参数
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "query_skywalking_trace",
            "description": "查询 SkyWalking 分布式调用链路。输入服务名和时间范围，返回该服务的调用链路拓扑和失败节点。用于定位故障发生在哪一跳。",
            "parameters": {
                "type": "object",
                "properties": {
                    "service": {"type": "string", "description": "服务名称"},
                    "time_range": {"type": "string", "description": "时间范围，如 '最近10分钟'"},
                    "endpoint": {"type": "string", "description": "可选，指定接口路径"}
                },
                "required": ["service", "time_range"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "query_sls_logs",
            "description": "查询阿里云 SLS 日志服务。输入服务名、关键词和时间范围，返回匹配的日志条目。用于查看应用错误日志、异常堆栈、OOM 信息等。",
            "parameters": {
                "type": "object",
                "properties": {
                    "service": {"type": "string", "description": "服务名称"},
                    "keyword": {"type": "string", "description": "日志关键词，如 'error', 'OOM', 'timeout'"},
                    "time_range": {"type": "string", "description": "时间范围"},
                    "limit": {"type": "integer", "description": "返回条数，默认50"}
                },
                "required": ["service", "time_range"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "query_prometheus_metrics",
            "description": "查询 Prometheus 监控指标。输入指标名、服务名和时间范围，返回时序数据。用于确认 CPU/内存/QPS/延迟 等指标趋势。",
            "parameters": {
                "type": "object",
                "properties": {
                    "metric": {"type": "string", "description": "PromQL 指标名，如 'container_memory_working_set_bytes'"},
                    "service": {"type": "string", "description": "服务名（label过滤）"},
                    "time_range": {"type": "string", "description": "时间范围"}
                },
                "required": ["metric", "time_range"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "query_k8s_status",
            "description": "查询 K8s 资源状态。输入 namespace 和资源类型，返回 Pod 的 Ready 状态、Phase、重启次数等。用于确认 Pod 是否健康。",
            "parameters": {
                "type": "object",
                "properties": {
                    "namespace": {"type": "string", "description": "K8s namespace"},
                    "resource": {"type": "string", "description": "资源名，如 'service-B'"}
                },
                "required": ["namespace", "resource"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "query_k8s_events",
            "description": "查询 K8s Events。返回最近的 Pod 事件，如 OOMKilled、FailedScheduling、Unhealthy 等。用于确认 K8s 层面的异常。",
            "parameters": {
                "type": "object",
                "properties": {
                    "namespace": {"type": "string", "description": "K8s namespace"},
                    "resource": {"type": "string", "description": "资源名"}
                },
                "required": ["namespace", "resource"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_rca_knowledge",
            "description": "从历史 RCA 案例库中检索相似故障。输入症状描述和服务名，返回历史相似故障的根因和解决方案。用于加速同类故障的收敛。",
            "parameters": {
                "type": "object",
                "properties": {
                    "symptom": {"type": "string", "description": "症状描述，如 '502错误率飙升'"},
                    "service": {"type": "string", "description": "服务名"}
                },
                "required": ["symptom"]
            }
        }
    }
]

4.2 LLM Agent 循环核心代码

[!example] 面试时可以说”核心循环大概 50 行代码”

import json
from openai import OpenAI

client = OpenAI()

def run_rca_agent(alert_context: dict, max_rounds: int = 10) -> dict:
    """AI 根因分析 Agent 主循环

    ReAct 模式：Reasoning + Acting
    每轮：LLM 推理 → 调用 Tool → 获取结果 → 下一轮推理
    """
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": json.dumps(alert_context, ensure_ascii=False)}
    ]

    for round_num in range(max_rounds):
        # 1. LLM 推理：分析当前 context，决定下一步
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=TOOLS,
            tool_choice="auto",
            timeout=30
        )
        msg = response.choices[0].message
        messages.append(msg)

        # 2. 如果 LLM 决定调用 Tool
        if msg.tool_calls:
            for tool_call in msg.tool_calls:
                func_name = tool_call.function.name
                func_args = json.loads(tool_call.function.arguments)

                # 3. 执行 Tool（实际调用后端 API）
                try:
                    result = execute_tool(func_name, func_args)
                except Exception as e:
                    result = {"error": str(e)}

                # 4. 把 Tool 结果加入 context
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": json.dumps(result, ensure_ascii=False)[:4000]  # 截断防止 context 过长
                })
        else:
            # 5. LLM 没有调用 Tool，说明已得出结论
            return {
                "root_cause": msg.content,
                "rounds": round_num + 1,
                "confidence": parse_confidence(msg.content)
            }

    # 6. 超过最大轮次，强制结束
    return {
        "root_cause": "分析超时，未能在限定轮次内定位根因",
        "rounds": max_rounds,
        "confidence": 0.0
    }


SYSTEM_PROMPT = """你是一个 SRE 根因分析助手。你的任务是通过渐进式分析定位线上故障的根因。

工作方式：

分析当前告警信息和已有数据

决定下一步查询什么（调用对应的 Tool）

根据返回结果继续推理，逐步收敛

定位根因后输出结论和处理建议

规则：

每轮只调用一个最相关的 Tool

如果已有数据足够定位根因，直接输出结论

结论必须附带置信度（0-1）和数据来源

不要编造数据，所有数据必须来自 Tool 返回结果 """

def execute_tool(name: str, args: dict) -> dict: """Tool 执行路由""" handlers = { “query_skywalking_trace”: lambda a: skywalking_api.query(**a), “query_sls_logs”: lambda a: sls_api.query(**a), “query_prometheus_metrics”: lambda a: prometheus_api.query(**a), “query_k8s_status”: lambda a: k8s_api.get_status(**a), “query_k8s_events”: lambda a: k8s_api.get_events(**a), “search_rca_knowledge”: lambda a: rca_db.search(**a), } handler = handlers.get(name) if not handler: return {“error”: f”unknown tool: {name}”} return handler(args)

4.3 知识库表结构

[!example] 面试时能说出的知识库设计

-- L1：结构化案例库（PostgreSQL）
CREATE TABLE rca_cases (
    id          SERIAL PRIMARY KEY,
    symptom     VARCHAR(500) NOT NULL,        -- 症状描述
    root_cause  VARCHAR(500) NOT NULL,        -- 根因
    service     VARCHAR(100) NOT NULL,        -- 受影响服务
    solution    TEXT,                          -- 解决方案
    trigger_condition VARCHAR(200),           -- 触发条件
    severity    VARCHAR(20),                  -- 严重等级
    created_at  TIMESTAMP DEFAULT NOW(),
    updated_at  TIMESTAMP DEFAULT NOW()
);
CREATE INDEX idx_rca_symptom ON rca_cases(symptom);
CREATE INDEX idx_rca_service ON rca_cases(service);

-- L2：向量库（pgvector 扩展，复用 PostgreSQL）
CREATE TABLE rca_embeddings (
    id          SERIAL PRIMARY KEY,
    case_id     INTEGER REFERENCES rca_cases(id),
    content     TEXT,                          -- 用于生成 embedding 的文本
    embedding   VECTOR(1536),                  -- OpenAI text-embedding-3-small 维度
    created_at  TIMESTAMP DEFAULT NOW()
);
-- 向量相似度索引（HNSW，加速近邻搜索）
CREATE INDEX idx_embedding_hnsw ON rca_embeddings
    USING hnsw (embedding vector_cosine_ops);

五、面试回答话术模板

[!tip] 高分回答的结构面试中回答技术问题的通用框架：

1. 先给结论（1 句话）
2. 再讲机制/原理（2-3 句话）
3. 结合自己的实践（1-2 句话，给具体数据）
4. 如果有不足，主动说出来 + 改进方向

示例：回答”你负责的业务 SLO 是什么？”

结论： “我们核心业务 SLO 是 99.9%。”

原理： “SLI 定义为成功请求数除以总请求数，5xx 算失败，4xx 不算分母，28 天滚动窗口统计。每月允许 43 分钟不可用，作为错误预算。”

实践： “我们用 Grafana 看板跟踪预算消耗，消耗超 80% 自动触发企微告警和发布审批升级。上个月一次 Redis 故障消耗了 15 分钟，占预算的 35%。”

不足： “目前 SLI 定义还不够细，没有按接口维度拆分 SLO，这是下一步要做的。“

示例：回答”你的 AI 根因分析怎么做的？”

结论： “通过 LLM Tool Calling 做渐进式分析，故障定位时间从约 1 小时降到约 30 分钟。”

原理： “告警触发后，构建初始 context 传给 LLM。LLM 每轮推理决定下一步查什么——链路、日志、指标、K8s 状态，通过调用对应 Tool 获取真实数据。每轮结果 append 到 context，逐步收敛。”

实践： “注册了 6 个 Tool：SkyWalking 链路查询、SLS 日志查询、Prometheus 指标查询、K8s 状态和事件查询、RCA 案例库检索。设置最大 10 轮 + 每轮 30 秒超时作为护栏。”

不足： “目前 LLM 分析结果还需要人工确认才入库，下一步想做 confidence score 自动评估，高置信度自动入库。“

关联知识

SRE 面试备战手册 — 主文档，本文是其补充
MTTR 与 MTBF 体系详解 — 四段模型详解
根因定位方法论 — Diagnose 环节方法论
弹性伸缩策略 — HPA 策略
SRE Agent 运维落地实践 — AI Agent 落地
LLM 辅助 RCA 与日志分析 — AI 根因分析
无指责复盘与故障分析方法论 — RCA 方法论
AIOps 实践与智能运维 — AIOps 整体

状态

15 道自测题完整答案
3 道代码题完整实现（令牌桶限流器、HTTP 健康检查器、日志 tail）
SRE 方法论深度补充（RCA 模板、错误预算、Runbook）
K8s 深度补充（Pod 故障排查、QoS、kubectl 速查）
AI 架构深度补充（Tool 注册表、Agent 循环代码、知识库表结构）
面试回答话术模板