DevOps SRE 实践指南：SLI/SLO、错误预算与可靠性工程

# sre-team-structure.yaml
org:
  name: "Platform Engineering"
  teams:
    - id: csre
      name: "Customer-Facing SRE"
      focus: ["API gateway", "Auth service"]
      embedded: true
      product_team: "Growth"
    - id: psre
      name: "Platform SRE"
      focus: ["K8s cluster", "CI/CD", "Observability"]
      embedded: false
  oncall_rotation:
    primary_shift: 7
    max_pages_per_shift: 2

#!/bin/bash
# pager-rotation.sh
ROSTER=("alice" "bob" "carol" "dave")
WEEK_NUM=$(date +%W)
IDX=$((WEEK_NUM % ${#ROSTER[@]}))
echo "本周值班: ${ROSTER[$IDX]} (主), ${ROSTER[$(( (IDX+1)%4 ))]} (备)"

# slo-rules.yaml
groups:
  - name: http_slo
    interval: 30s
    rules:
      - record: job:http_success_rate:ratio_rate5m
        expr: |
          sum(rate(http_requests_total{status=~"2..|3.."}[5m])) by (job)
          /
          sum(rate(http_requests_total[5m])) by (job)

      - record: job:http_latency_p99:rate5m
        expr: histogram_quantile(0.99,
          sum(rate(http_request_duration_seconds_bucket[5m])) by (job, le))

      - alert: SLOBurnRateHigh
        expr: |
          (sum(rate(http_requests_total{status=~"5..|4.."}[1h])) by (job)
           / sum(rate(http_requests_total[1h])) by (job)) > 0.001 * 14.4
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "{{ $labels.job }} 错误预算消耗过快"

# sli_slo_simulator.py
import dataclasses

@dataclasses.dataclass
class Service:
    name: str
    requests_per_month: int
    target_slo: float

    def monthly_error_budget(self) -> int:
        return int(self.requests_per_month * (1.0 - self.target_slo))

services = [
    Service("payment-api", 1_000_000_000, 0.9999),
    Service("recommendation", 500_000_000, 0.999),
]

for svc in services:
    budget = svc.monthly_error_budget()
    print(f"{svc.name}: 月度错误预算 {budget} 次请求")

// error_budget_gate.go
package main

import ("fmt"; "os")

type BudgetGate struct {
	ServiceName string
	SLO, ObservedBad, TotalRequests float64
}

func (bg *BudgetGate) Remaining() float64 {
	return bg.TotalRequests*(1-bg.SLO) - bg.ObservedBad
}

func (bg *BudgetGate) CanDeploy() (bool, string) {
	r := bg.Remaining()
	if r <= 0 {
		return false, fmt.Sprintf("[%s] 错误预算已耗尽", bg.ServiceName)
	}
	if r < bg.TotalRequests*(1-bg.SLO)*0.1 {
		return false, fmt.Sprintf("[%s] 预算低于安全阈值", bg.ServiceName)
	}
	return true, fmt.Sprintf("[%s] 预算充足 (%.0f)", bg.ServiceName, r)
}

func main() {
	gate := BudgetGate{"order-service", 0.999, 80000, 10_000_000}
	ok, msg := gate.CanDeploy()
	fmt.Println(msg)
	if !ok { os.Exit(1) }
}

# burn-rate-alerts.yaml
groups:
  - name: burn_rate
    rules:
      - alert: ErrorBudgetBurnFast
        expr: |
          (sum(rate(http_requests_total{status=~"5.."}[1h]))
           / sum(rate(http_requests_total[1h]))) > on() (14.4 * 0.001)
        for: 5m
        labels: { severity: p1 }
        annotations:
          summary: "{{ $labels.job }} 快速燃烧告警"

      - alert: ErrorBudgetBurnSlow
        expr: |
          (sum(rate(http_requests_total{status=~"5.."}[6h]))
           / sum(rate(http_requests_total[6h]))) > on() (6 * 0.001)
        for: 30m
        labels: { severity: p2 }
        annotations:
          summary: "{{ $labels.job }} 慢速燃烧告警"

# circuit-breaker-istio.yaml
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: payment-service-cb
spec:
  host: payment-service
  trafficPolicy:
    connectionPool:
      tcp: { maxConnections: 100 }
      http:
        http1MaxPendingRequests: 50
        http2MaxRequests: 100
    outlierDetection:
      consecutive5xxErrors: 5
      interval: 30s
      baseEjectionTime: 60s
      maxEjectionPercent: 50

#!/bin/bash
# chaos-experiment.sh
NAMESPACE="production"; TARGET="inventory-service"; DURATION="300s"

echo "[1/2] 注入 Pod 删除..."
kubectl run chaos-kill --image=litmuschaos/go-runner:3.0.0 --rm -i --restart=Never \
  -- --experiment=generic/pod-delete --app-label="app=$TARGET" \
     --namespace="$NAMESPACE" --duration=$DURATION

echo "[2/2] 验证恢复..."
kubectl rollout status deployment/$TARGET -n $NAMESPACE

# graceful_degradation.py
from functools import wraps
import time

def fallback_on_error(fallback_value, max_ms=500):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            start = time.time() * 1000
            try:
                result = func(*args, **kwargs)
                if time.time() * 1000 - start > max_ms:
                    raise TimeoutError("Latency too high")
                return result
            except Exception as e:
                print(f"[DEGRADE] {func.__name__}: {e}")
                return fallback_value
        return wrapper
    return decorator

class CartService:
    @fallback_on_error(fallback_value=[])
    def get_recommendations(self, uid: str) -> list:
        return ["item-A", "item-B"]

    @fallback_on_error(fallback_value=True)
    def check_inventory(self, sku: str) -> bool:
        return True   # 降级时默认有货

# toil_tracker.py
from dataclasses import dataclass
from collections import Counter

@dataclass
class Task:
    title: str; category: str; hours: float
    repetitive: bool; automatable: bool

class ToilAnalyzer:
    def __init__(self, tasks): self.tasks = tasks
    def toil_ratio(self):
        toil = sum(t.hours for t in self.tasks if t.category == "toil")
        total = sum(t.hours for t in self.tasks)
        return round(toil/total, 3) if total else 0.0
    def top_automatable(self, n=5):
        c = Counter()
        for t in self.tasks:
            if t.repetitive and t.automatable:
                c[t.title.split(":")[0]] += t.hours
        return c.most_common(n)

tasks = [
    Task("扩容 Redis", "toil", 4.0, True, True),
    Task("清理日志", "toil", 2.5, True, True),
    Task("优化 PromQL", "project", 8.0, False, False),
]
analyzer = ToilAnalyzer(tasks)
print(f"Toil 占比: {analyzer.toil_ratio()*100:.1f}%")

# argo-workflow.yaml
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: auto-heal-pods-
spec:
  entrypoint: heal
  templates:
    - name: heal
      steps:
        - - name: check
            template: check-health
        - - name: restart
            template: restart-pods
            when: "{{steps.check.outputs.result}} != healthy"
    - name: check-health
      script:
        image: bitnami/kubectl:latest
        command: [bash]
        source: |
          N=$(kubectl get pods -l app=api-gateway \
            --field-selector=status.phase!=Running -n prod --no-headers | wc -l)
          [[ "$N" -gt 0 ]] && echo "unhealthy" || echo "healthy"
    - name: restart-pods
      script:
        image: bitnami/kubectl:latest
        command: [bash]
        source: |
          kubectl rollout restart deployment/api-gateway -n prod

# main.tf — SRE 标准 GKE 模块
terraform {
  required_providers {
    google = { source = "hashicorp/google", version = ">= 5.0" }
  }
}
variable "cluster_name" { type = string }
variable "region" { type = string }
variable "slo_target" { type = number; default = 0.999 }
variable "project_id" { type = string }

resource "google_container_cluster" "primary" {
  name = var.cluster_name; location = var.region
  remove_default_node_pool = true; initial_node_count = 1
  monitoring_config {
    enable_components = ["SYSTEM_COMPONENTS", "APISERVER"]
    managed_prometheus { enabled = true }
  }
  workload_identity_config {
    workload_pool = "${var.project_id}.svc.id.goog"
  }
}

resource "google_container_node_pool" "nodes" {
  name = "${var.cluster_name}-pool"
  cluster = google_container_cluster.primary.name
  location = var.region; node_count = 3
  node_config {
    machine_type = var.slo_target >= 0.9999 ? "n2-standard-8" : "n2-standard-4"
  }
}

# otel-collector.yaml
receivers:
  otlp:
    protocols:
      grpc: { endpoint: 0.0.0.0:4317 }
      http: { endpoint: 0.0.0.0:4318 }
processors:
  batch: { timeout: 1s, send_batch_size: 1024 }
exporters:
  prometheusremotewrite:
    endpoint: http://thanos-receive.monitoring:19291/api/v1/receive
  loki:
    endpoint: http://loki.monitoring:3100/loki/api/v1/push
  otlp/tempo:
    endpoint: tempo.monitoring:4317
    tls: { insecure: true }
service:
  pipelines:
    metrics:
      receivers: [otlp]; processors: [batch]; exporters: [prometheusremotewrite]
    logs:
      receivers: [otlp]; processors: [batch]; exporters: [loki]
    traces:
      receivers: [otlp]; processors: [batch]; exporters: [otlp/tempo]

# alert_correlator.py
from dataclasses import dataclass
from collections import defaultdict
import re

@dataclass
class Alert:
    name: str; service: str; labels: dict; ts: float

class AlertCorrelator:
    def __init__(self):
        self.patterns = {
            "db_cascade": re.compile(r".*(mysql|postgres|redis).*"),
            "net_cascade": re.compile(r".*(timeout|refused).*"),
        }
    def correlate(self, alerts):
        buckets = defaultdict(list)
        for a in alerts:
            for name, pat in self.patterns.items():
                if pat.match(a.name.lower() + str(a.labels)):
                    buckets[name].append(a); break
            else:
                buckets["other"].append(a)
        return dict(buckets)

alerts = [
    Alert("MySQL_High_Conn", "order", {"i":"db-01"}, 1700000000),
    Alert("MySQL_Slow", "pay", {"i":"db-01"}, 1700000005),
    Alert("MySQL_Lag", "stock", {"i":"db-01"}, 1700000010),
]
buckets = AlertCorrelator().correlate(alerts)
for b, a in buckets.items():
    print(f"{b}: {len(a)} 条告警")

#!/bin/bash
# oncall-schedule.sh
# 采用轮换 + 影子制度（Shadow On-Call）
MEMBERS=("sre1" "sre2" "sre3" "sre4" "sre5" "sre6")
WEEK=$(date +%V)
PRIMARY=$((WEEK % ${#MEMBERS[@]}))
SECONDARY=$(((PRIMARY + 1) % ${#MEMBERS[@]}))
SHADOW=$(((PRIMARY + 2) % ${#MEMBERS[@]}))

echo "本周值班主责: ${MEMBERS[$PRIMARY]}"
echo "本周值班后备: ${MEMBERS[$SECONDARY]}"
echo "本周影子学习: ${MEMBERS[$SHADOW]}"

# user-journey-sli.yaml
groups:
  - name: ecommerce_user_journey
    rules:
      - record: journey:login_success_rate:ratio_5m
        expr: |
          sum(rate(login_attempts_total{result="success"}[5m]))
          /
          sum(rate(login_attempts_total[5m]))

      - record: journey:product_freshness_seconds:p95_5m
        expr: |
          histogram_quantile(0.95,
            sum(rate(product_sync_latency_seconds_bucket[5m])) by (le))

      - record: journey:payment_latency_seconds:p999_5m
        expr: |
          histogram_quantile(0.999,
            sum(rate(payment_callback_duration_seconds_bucket[5m])) by (le))

# slo-lifecycle-alerts.yaml
groups:
  - name: slo_lifecycle
    rules:
      - alert: SLOViolationPredicted
        expr: |
          (
            sum(rate(http_requests_total{status=~"5.."}[24h]))
            /
            sum(rate(http_requests_total[24h]))
          ) > (1 - 0.999) * 1.2
        for: 5m
        labels: { severity: warning, stage: predict }
        annotations:
          summary: "{{ $labels.job }} 预测性 SLO 违约风险"

      - alert: SLOBudgetExhaustedThisQuarter
        expr: |
          (
            sum(increase(http_requests_total{status=~"5.."}[90d]))
            /
            sum(increase(http_requests_total[90d]))
          ) > (1 - 0.999)
        labels: { severity: critical, stage: review }
        annotations:
          summary: "{{ $labels.job }} 本季度错误预算已耗尽"

# error_budget_visualizer.py
import math

class BudgetVisualizer:
    def __init__(self, slo: float, period_requests: int):
        self.slo = slo
        self.budget = period_requests * (1 - slo)

    def burn_rate(self, bad_requests: int, window_hours: int) -> float:
        # 标准化为每小时预算消耗倍数
        expected = (self.budget / (30 * 24)) * window_hours  # 假设 30 天周期
        return bad_requests / expected if expected else math.inf

    def alert_level(self, bad_requests: int, window_hours: int) -> str:
        rate = self.burn_rate(bad_requests, window_hours)
        if rate >= 3.0: return "P1-CRITICAL"
        if rate >= 1.0: return "P2-HIGH"
        if rate >= 0.5: return "P3-WARNING"
        return "OK"

    def remaining_percent(self, consumed: int) -> float:
        return max(0, (self.budget - consumed) / self.budget * 100)

viz = BudgetVisualizer(slo=0.999, period_requests=1_000_000_000)
for consumed in [300_000, 500_000, 700_000, 900_000]:
    pct = viz.remaining_percent(consumed)
    print(f"预算已消耗 {consumed}，剩余 {pct:.2f}%")

# .gitlab-ci.yml — 错误预算门禁
stages:
  - budget-check
  - deploy

budget_check:
  stage: budget-check
  image: python:3.11-slim
  before_script:
    - pip install requests
  script:
    - |
      python -c "
      import sys, requests, json
      resp = requests.get('https://slo-api.internal/api/v1/budget/order-service')
      data = resp.json()
      remaining_pct = data['remaining_percent']
      print(f'错误预算剩余: {remaining_pct:.1f}%')
      if remaining_pct < 10.0:
          print('发布阻断：错误预算已低于安全阈值')
          sys.exit(1)
      if remaining_pct < 30.0:
          print('警告：错误预算消耗过半，需特批')
      print('预算检查通过')
      "
  rules:
    - if: $CI_COMMIT_BRANCH == "main"

deploy_production:
  stage: deploy
  script:
    - helm upgrade --install order-service ./helm
  rules:
    - if: $CI_COMMIT_BRANCH == "main"
      when: on_success
  needs:
    - job: budget_check
      artifacts: false

# automation_roi.py
from dataclasses import dataclass

@dataclass
class ToilItem:
    name: str
    frequency_weekly: float
    duration_minutes: float
    pain_index: int   # 1-5
    stability_impact: int  # 1-5
    auto_dev_weeks: float

    def weekly_cost_hours(self) -> float:
        return (self.frequency_weekly * self.duration_minutes) / 60

    def roi_score(self) -> float:
        # ROI = (年节省工时 × 痛苦加权 × 稳定性加权) / 开发投入
        yearly_hours = self.weekly_cost_hours() * 52
        weight = (self.pain_index * self.stability_impact) / 25
        if self.auto_dev_weeks <= 0:
            return 999.0
        return (yearly_hours * weight) / self.auto_dev_weeks

items = [
    ToilItem("手动扩缩容", 8, 30, 5, 5, 3),
    ToilItem("证书续期", 1, 60, 4, 4, 1),
    ToilItem("日志清理", 5, 15, 2, 2, 0.5),
]

for item in sorted(items, key=lambda x: x.roi_score(), reverse=True):
    print(f"{item.name}: 周耗时 {item.weekly_cost_hours():.1f}h, ROI 得分 {item.roi_score():.1f}")

# risk-register.yaml
risks:
  - id: R001
    title: "数据库主库单点故障"
    category: infrastructure
    probability: 3    # 1-5
    impact: 5         # 1-5
    mitigation:
      - "Q2 完成跨可用区只读副本部署"
      - "自动化故障切换演练每季度 1 次"
    owner: "sre-db-team"
    review_date: "2026-09-30"
    status: active

  - id: R002
    title: "第三方支付网关超时级联"
    category: dependency
    probability: 4
    impact: 4
    mitigation:
      - "增加 2s 超时下熔断降级"
      - "备用支付通道自动切换"
    owner: "sre-platform"
    review_date: "2026-09-30"
    status: active

#!/bin/bash
# dark-launch-validator.sh
# 将生产流量镜像到候选版本，对比响应差异
CANDIDATE_URL="http://candidate.internal/api/v1"
PROD_URL="http://prod.internal/api/v1"
SAMPLE_RATE=0.01  # 1% 流量镜像

echo "启动 Shadow Traffic 对比..."
for i in {1..1000}; do
    PROD_RESP=$(curl -s -o /dev/null -w "%{http_code},%{time_total}" "$PROD_URL/health")
    CAND_RESP=$(curl -s -o /dev/null -w "%{http_code},%{time_total}" "$CANDIDATE_URL/health")
    if [ "$PROD_RESP" != "$CAND_RESP" ]; then
        echo "差异发现: 生产=$PROD_RESP 候选=$CAND_RESP"
    fi
    sleep 0.01
done

echo "Shadow 验证完成"

# argo-rollouts-canary.yaml
apiVersion: argoproj.io/v1alpha1
kind: Rollout
metadata:
  name: order-service
spec:
  replicas: 10
  strategy:
    canary:
      steps:
        - setWeight: 5
        - pause: { duration: 10m }
        - analysis:
            templates:
              - templateName: success-rate
            args:
              - name: service-name
                value: order-service
        - setWeight: 25
        - pause: { duration: 10m }
        - analysis:
            templates:
              - templateName: latency-p99
        - setWeight: 50
        - pause: { duration: 10m }
        - setWeight: 100
      analysis:
        successfulRunHistoryLimit: 3
        templates:
          - templateName: success-rate
---
apiVersion: argoproj.io/v1alpha1
kind: AnalysisTemplate
metadata:
  name: success-rate
spec:
  metrics:
    - name: success-rate
      interval: 1m
      successCondition: result[0] >= 0.99
      provider:
        prometheus:
          address: http://prometheus.monitoring:9090
          query: |
            sum(rate(http_requests_total{service="order-service",status=~"2.."}[2m]))
            /
            sum(rate(http_requests_total{service="order-service"}[2m]))

# dora_metrics_dashboard.py
import json
from dataclasses import dataclass
from datetime import datetime, timedelta

@dataclass
class Deployment:
    deploy_id: str
    timestamp: datetime
    failed: bool
    lead_time_minutes: int
    recovery_minutes: int = 0

class DORAMetrics:
    def __init__(self, deployments):
        self.deployments = deployments

    def deployment_frequency(self, days=7) -> int:
        cutoff = datetime.now() - timedelta(days=days)
        return len([d for d in self.deployments if d.timestamp > cutoff])

    def change_failure_rate(self) -> float:
        if not self.deployments:
            return 0.0
        failed = len([d for d in self.deployments if d.failed])
        return failed / len(self.deployments)

    def median_lead_time(self) -> float:
        times = [d.lead_time_minutes for d in self.deployments]
        if not times:
            return 0.0
        times.sort()
        mid = len(times) // 2
        return times[mid] if len(times) % 2 else (times[mid-1] + times[mid]) / 2

    def median_recovery_time(self) -> float:
        recoveries = [d.recovery_minutes for d in self.deployments if d.failed]
        if not recoveries:
            return 0.0
        recoveries.sort()
        mid = len(recoveries) // 2
        return recoveries[mid] if len(recoveries) % 2 else (recoveries[mid-1] + recoveries[mid]) / 2

deploys = [
    Deployment("d1", datetime.now() - timedelta(days=1), False, 45),
    Deployment("d2", datetime.now() - timedelta(days=2), True, 120, 55),
    Deployment("d3", datetime.now() - timedelta(days=3), False, 30),
]

metrics = DORAMetrics(deploys)
report = {
    "deploy_frequency_7d": metrics.deployment_frequency(),
    "change_failure_rate": f"{metrics.change_failure_rate()*100:.1f}%",
    "median_lead_time_min": metrics.median_lead_time(),
    "median_recovery_time_min": metrics.median_recovery_time(),
}
print(json.dumps(report, indent=2, ensure_ascii=False))

#!/bin/bash
# mttr-mttd-mtbf.sh
# 从 PagerDuty API 拉取事故数据并计算可靠性指标
# MTTD: Mean Time To Detect
# MTTR: Mean Time To Repair
# MTBF: Mean Time Between Failures

START=$(date -v-90d +%Y-%m-%d)
END=$(date +%Y-%m-%d)

echo "拉取 ${START} 至 ${END} 事故数据..."
# curl -s -H "Authorization: Bearer $PD_TOKEN" \
#   "https://api.pagerduty.com/incidents?since=${START}T00:00:00Z&until=${END}T23:59:59Z&service_ids[]=PXXXX"

echo "TODO: 解析 JSON 并计算 MTTD / MTTR / MTBF 趋势"