【网络工程】健康检查与故障转移：主动探测、被动检测与优雅处理

// gRPC 标准健康检查协议
service Health {
    rpc Check(HealthCheckRequest) returns (HealthCheckResponse);
    rpc Watch(HealthCheckRequest) returns (stream HealthCheckResponse);
}

message HealthCheckResponse {
    enum ServingStatus {
        UNKNOWN = 0;
        SERVING = 1;
        NOT_SERVING = 2;
        SERVICE_UNKNOWN = 3;
    }
    ServingStatus status = 1;
}

// Go 示例：分层健康检查
package main

import (
    "encoding/json"
    "net/http"
    "sync/atomic"
    "time"
)

var ready atomic.Bool

type HealthResponse struct {
    Status   string            `json:"status"`
    Checks   map[string]string `json:"checks,omitempty"`
    Uptime   string            `json:"uptime,omitempty"`
}

func livenessHandler(w http.ResponseWriter, r *http.Request) {
    // Liveness: 进程是否存活？
    // 只要进程能响应就返回 200
    // 不检查依赖——依赖故障不应导致进程被杀
    w.WriteHeader(http.StatusOK)
    json.NewEncoder(w).Encode(HealthResponse{Status: "alive"})
}

func readinessHandler(w http.ResponseWriter, r *http.Request) {
    // Readiness: 能否处理请求？
    // 检查所有关键依赖
    checks := make(map[string]string)
    allOK := true

    // 检查数据库
    if err := checkDB(); err != nil {
        checks["database"] = err.Error()
        allOK = false
    } else {
        checks["database"] = "ok"
    }

    // 检查 Redis
    if err := checkRedis(); err != nil {
        checks["redis"] = err.Error()
        allOK = false
    } else {
        checks["redis"] = "ok"
    }

    status := "ready"
    code := http.StatusOK
    if !allOK || !ready.Load() {
        status = "not_ready"
        code = http.StatusServiceUnavailable
    }

    w.WriteHeader(code)
    json.NewEncoder(w).Encode(HealthResponse{
        Status: status,
        Checks: checks,
    })
}

// 启动时标记为未就绪，初始化完成后标记就绪
func main() {
    ready.Store(false)

    // 初始化（加载缓存、建立连接池等）
    go func() {
        initialize()
        ready.Store(true)
    }()

    http.HandleFunc("/healthz", livenessHandler)
    http.HandleFunc("/readyz", readinessHandler)
    http.ListenAndServe(":8080", nil)
}

# HAProxy 健康检查配置
backend app_back
    option httpchk GET /readyz
    http-check expect status 200

    # inter: 检查间隔, fall: 失败阈值, rise: 恢复阈值
    server app1 10.0.2.10:8080 check inter 3s fall 3 rise 2
    server app2 10.0.2.11:8080 check inter 3s fall 3 rise 2

    # 慢启动：恢复后 30 秒内逐渐增加权重
    server app3 10.0.2.12:8080 check inter 3s fall 3 rise 2 slowstart 30s

# Python 示例：分级健康检查
from flask import Flask, jsonify
import redis
import psycopg2
import time

app = Flask(__name__)

# 缓存依赖检查结果（避免每次检查都查数据库）
_cache = {"result": None, "timestamp": 0, "ttl": 5}

def cached_check(func, ttl=5):
    """缓存检查结果，避免频繁探测依赖"""
    key = func.__name__
    now = time.time()
    if _cache.get(key) and now - _cache[key]["ts"] < ttl:
        return _cache[key]["result"]
    result = func()
    _cache[key] = {"result": result, "ts": now}
    return result

def check_database():
    try:
        conn = psycopg2.connect(
            host="db.internal", dbname="app",
            connect_timeout=2
        )
        cur = conn.cursor()
        cur.execute("SELECT 1")
        cur.close()
        conn.close()
        return {"status": "ok"}
    except Exception as e:
        return {"status": "error", "message": str(e)}

def check_redis():
    try:
        r = redis.Redis(host="redis.internal", socket_timeout=2)
        r.ping()
        return {"status": "ok"}
    except Exception as e:
        return {"status": "error", "message": str(e)}

@app.route("/readyz")
def readiness():
    db = cached_check(check_database)
    cache = cached_check(check_redis)

    all_ok = db["status"] == "ok" and cache["status"] == "ok"
    code = 200 if all_ok else 503

    return jsonify({
        "status": "ready" if all_ok else "not_ready",
        "checks": {
            "database": db,
            "redis": cache,
        }
    }), code

@app.route("/healthz")
def liveness():
    # 只检查进程是否存活
    return jsonify({"status": "alive"}), 200

@app.route("/readyz")
def readiness_with_degradation():
    db = cached_check(check_database)
    cache = cached_check(check_redis)
    search = cached_check(check_elasticsearch)

    # 数据库是硬依赖——挂了就不接流量
    if db["status"] != "ok":
        return jsonify({"status": "not_ready", "reason": "database"}), 503

    # Redis 和 ES 是软依赖——挂了仍可接流量（降级模式）
    status = "ready"
    if cache["status"] != "ok" or search["status"] != "ok":
        status = "degraded"  # 降级但仍可服务

    return jsonify({
        "status": status,
        "checks": {
            "database": db,
            "redis": cache,
            "elasticsearch": search,
        }
    }), 200  # 200 表示仍然可以接收流量

# Envoy Outlier Detection 配置
clusters:
- name: app_cluster
  outlier_detection:
    # 连续失败触发驱逐
    consecutive_5xx: 5                    # 连续 5 个 5xx 响应
    interval: 10s                         # 检测间隔
    base_ejection_time: 30s              # 基础驱逐时间
    max_ejection_percent: 50             # 最多驱逐 50% 的节点
    enforcing_consecutive_5xx: 100       # 100% 执行（不是采样）

    # 成功率触发驱逐
    success_rate_minimum_hosts: 3        # 至少 3 台才启用
    success_rate_request_volume: 100     # 至少 100 个请求
    success_rate_stdev_factor: 1900      # 低于平均值 1.9 个标准差

    # 延迟触发驱逐
    consecutive_local_origin_failure: 5  # 本地错误（连接失败等）

# HAProxy: 主动 + 被动结合
backend app_back
    # 主动检查
    option httpchk GET /readyz
    http-check expect status 200

    # 被动检查 (observe 指令)
    option httpclose
    default-server inter 5s fall 3 rise 2 on-marked-down shutdown-sessions

    server app1 10.0.2.10:8080 check observe layer7
    server app2 10.0.2.11:8080 check observe layer7
    # observe layer7: 观察 HTTP 响应来被动判断健康
    # observe layer4: 观察 TCP 连接来被动判断健康

# Envoy: 限制最大驱逐比例
outlier_detection:
    max_ejection_percent: 30   # 最多驱逐 30%，保证至少 70% 可用

# HAProxy: 至少保留一台
backend app_back
    option allbackups          # 所有 backup 都可以同时使用
    server app1 10.0.2.10:8080 check
    server app2 10.0.2.11:8080 check
    server backup1 10.0.2.20:8080 check backup  # 只在主服务器都不可用时启用

#!/bin/bash
# graceful_shutdown.sh — 优雅下线脚本

SERVER="10.0.2.10:8080"

# 1. 标记为不健康（健康检查端点返回 503）
curl -X POST http://$SERVER/admin/drain

# 2. 通知 LB 停止分配新请求
# HAProxy: 通过 Runtime API
echo "set server app_back/app1 state drain" | socat stdio /var/run/haproxy.sock

# 3. 等待已有连接结束
echo "等待连接排空..."
while true; do
    ACTIVE=$(ss -tn state established | grep "$SERVER" | wc -l)
    echo "剩余活跃连接: $ACTIVE"
    if [ "$ACTIVE" -eq 0 ]; then
        break
    fi
    sleep 5
done

# 4. 停止服务
echo "所有连接已排空，停止服务"
ssh $SERVER "systemctl stop app"

// Go 示例：优雅关闭 HTTP 服务器
package main

import (
    "context"
    "net/http"
    "os"
    "os/signal"
    "sync/atomic"
    "syscall"
    "time"
)

var draining atomic.Bool

func main() {
    mux := http.NewServeMux()
    mux.HandleFunc("/readyz", func(w http.ResponseWriter, r *http.Request) {
        if draining.Load() {
            w.WriteHeader(http.StatusServiceUnavailable)
            return
        }
        w.WriteHeader(http.StatusOK)
    })
    mux.HandleFunc("/api/", apiHandler)

    server := &http.Server{
        Addr:    ":8080",
        Handler: mux,
    }

    // 监听关闭信号
    quit := make(chan os.Signal, 1)
    signal.Notify(quit, syscall.SIGTERM, syscall.SIGINT)

    go func() {
        <-quit
        // 1. 标记为 draining，健康检查返回 503
        draining.Store(true)

        // 2. 等待 LB 检测到并停止发送新请求
        // 这个时间应该 >= LB 的健康检查间隔 × fall 阈值
        time.Sleep(15 * time.Second)

        // 3. 优雅关闭（等待已有请求完成，最多 30 秒）
        ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
        defer cancel()
        server.Shutdown(ctx)
    }()

    server.ListenAndServe()
}

apiVersion: apps/v1
kind: Deployment
spec:
  replicas: 4
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxUnavailable: 1     # 最多 1 个 Pod 不可用
      maxSurge: 1           # 最多多出 1 个 Pod
  template:
    spec:
      terminationGracePeriodSeconds: 60  # 优雅关闭时间
      containers:
      - name: app
        livenessProbe:
          httpGet:
            path: /healthz
            port: 8080
          initialDelaySeconds: 10  # 启动后等 10 秒再检查
          periodSeconds: 5
          failureThreshold: 3
          timeoutSeconds: 2

        readinessProbe:
          httpGet:
            path: /readyz
            port: 8080
          initialDelaySeconds: 5
          periodSeconds: 3         # Readiness 检查更频繁
          failureThreshold: 2      # 更快摘除
          successThreshold: 2      # 需要连续成功 2 次才标记就绪
          timeoutSeconds: 2

        # 启动探针（1.20+）
        startupProbe:
          httpGet:
            path: /healthz
            port: 8080
          initialDelaySeconds: 0
          periodSeconds: 2
          failureThreshold: 30     # 最多等 60 秒启动

lifecycle:
  preStop:
    exec:
      command: ["sh", "-c", "sleep 10"]
      # 等待 10 秒让 Endpoints 更新传播到所有节点
      # 然后 SIGTERM 触发应用的优雅关闭

#!/bin/bash
# test_failover.sh — 故障转移测试脚本

VIP="10.0.1.100"
BACKEND="10.0.2.10"

echo "=== 故障转移测试 ==="

# 1. 基准：确认所有后端正常
echo "Step 1: 确认基准状态"
curl -s http://$VIP/health | jq .
echo "当前后端分布:"
for i in $(seq 1 100); do
    curl -s http://$VIP/api/whoami
done | sort | uniq -c

# 2. 模拟故障：停止一台后端
echo "Step 2: 停止 $BACKEND"
ssh $BACKEND "systemctl stop app"

# 3. 观察故障检测时间
echo "Step 3: 等待 LB 检测到故障..."
START=$(date +%s)
while true; do
    RESP=$(curl -s -o /dev/null -w "%{http_code}" http://$VIP/api/test)
    if [ "$RESP" = "200" ]; then
        NOW=$(date +%s)
        ELAPSED=$((NOW - START))
        echo "  ${ELAPSED}s: 仍在返回 200"
    else
        NOW=$(date +%s)
        ELAPSED=$((NOW - START))
        echo "  ${ELAPSED}s: 检测到 $RESP，故障检测完成"
        break
    fi
    sleep 1
done

# 4. 验证流量已切走
echo "Step 4: 验证流量分布"
for i in $(seq 1 50); do
    curl -s http://$VIP/api/whoami
done | sort | uniq -c
# 应该不再出现故障后端

# 5. 恢复
echo "Step 5: 恢复 $BACKEND"
ssh $BACKEND "systemctl start app"

# 6. 等待恢复检测
echo "Step 6: 等待 LB 检测到恢复..."
sleep 15
for i in $(seq 1 50); do
    curl -s http://$VIP/api/whoami
done | sort | uniq -c

echo "=== 测试完成 ==="

# HAProxy 统计接口
echo "show stat" | socat stdio /var/run/haproxy.sock | \
    awk -F, '{print $2, $18, $19, $20}'
# server_name  chkfail  chkdown  lastchg
# app1         12       3        1234

# 关键指标：
# chkfail: 健康检查失败总次数
# chkdown: 被标记为 down 的总次数
# lastchg: 最后一次状态变化的时间（秒）
# check_duration: 健康检查响应时间

# 告警规则
# 1. 健康检查失败率 > 10%/分钟 → Warning
# 2. 任何 RS 被标记 down → Alert
# 3. 可用 RS 数量 < 总数的 50% → Critical
# 4. 健康检查响应时间 > 正常值 3 倍 → Warning（故障前兆）

# Prometheus 告警规则
groups:
- name: health_check_alerts
  rules:
  - alert: BackendDown
    expr: haproxy_server_up == 0
    for: 1m
    labels:
      severity: critical
    annotations:
      summary: "后端 {{ $labels.server }} 不健康超过 1 分钟"

  - alert: TooFewHealthyBackends
    expr: |
      count(haproxy_server_up == 1) by (backend)
      / count(haproxy_server_up) by (backend) < 0.5
    for: 30s
    labels:
      severity: critical
    annotations:
      summary: "后端组 {{ $labels.backend }} 健康节点不足 50%"

  - alert: HealthCheckSlow
    expr: haproxy_server_check_duration_seconds > 2
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "{{ $labels.server }} 健康检查响应变慢"