Python脚本健康检查端点如何设计

本文目录导读：

1. 核心设计原则
2. 典型架构与技术栈
3. 分步设计实现
4. 完整高级实现示例 (FastAPI + 异步 + 分层检查)
5. 测试与部署
6. 常见陷阱

设计Python脚本的健康检查端点,核心目标是快速、可靠地验证服务是否正常运行，一个优秀的健康检查端点不仅能回答“服务是否活着”，还能回答“服务是否正常工作”。

以下是从基础到最佳实践的完整设计指南,结合了Flask和FastAPI两种主流框架的示例。

核心设计原则

• 简单性：端点应尽可能简单，避免复杂的业务逻辑。
• 独立性：健康检查的代码应尽可能少依赖其他组件，自身要稳定。
• 层次化：一般分为两个层级：
  • Liveness Probe (存活检查)：服务进程是否存活，通常只返回200 OK。
  • Readiness Probe (就绪检查)：服务是否准备好接收流量，需要检查数据库、缓存等依赖是否正常。
  • （可选）Startup Probe (启动检查)：服务是否成功启动，用于慢启动应用。
• 性能：端点响应必须快（lt; 100ms），否则会干扰监控系统的判断。
• 幂等性：GET请求，且不应修改任何系统状态。

典型架构与技术栈

• Web框架：Flask, FastAPI, Django
• 异步支持：FastAPI (基于 Starlette) 天然支持异步
• 容器编排：Kubernetes (k8s) 是主要消费方
• 监控集成：Prometheus, Grafana

分步设计实现

步骤1：基础实现 (存活检查)

最简单的端点,仅返回一个OK信号。

Flask 示例：

from flask import Flask, jsonify
app = Flask(__name__)
@app.route('/healthz')
def healthz():
    # 最基本的存活检查
    return jsonify({"status": "ok"}), 200

FastAPI 示例：

from fastapi import FastAPI
from fastapi.responses import JSONResponse
app = FastAPI()
@app.get("/healthz")
async def healthz():
    return JSONResponse(content={"status": "ok"}, status_code=200)

步骤2：增强实现 (就绪检查)

检查关键依赖（数据库、缓存、外部API），不健康时返回非200状态码。

FastAPI 带依赖检查示例：

from fastapi import FastAPI
from fastapi.responses import JSONResponse
import asyncpg  # 示例：异步PostgreSQL
import redis.asyncio as redis
app = FastAPI()
async def check_database():
    try:
        # 建立快速连接，不执行复杂查询
        conn = await asyncpg.connect(dsn="your_dsn")
        await conn.close()
        return True, "Connected to primary database"
    except Exception as e:
        return False, f"Database error: {str(e)}"
async def check_redis():
    try:
        r = await redis.Redis.from_url("redis://localhost:6379")
        await r.ping()
        await r.close()
        return True, "Redis OK"
    except Exception as e:
        return False, f"Redis error: {str(e)}"
@app.get("/readyz")
async def readyz():
    checks = {
        "database": await check_database(),
        "cache": await check_redis()
    }
    overall_status = all(result for result, _ in checks.values())
    http_status = 200 if overall_status else 503
    return JSONResponse(
        content={
            "status": "ok" if overall_status else "unhealthy",
            "checks": {
                service: {
                    "status": "ok" if result else "error",
                    "message": message
                }
                for service, (result, message) in checks.items()
            }
        },
        status_code=http_status
    )

步骤3：针对Kubernetes的优化设计

在k8s中,livenessProbe和readinessProbe通常需要不同的行为。

• 存活探针：检查进程是否卡死（例如死锁）。不应该检查外部依赖，否则网络抖动会导致容器被重启。
• 就绪探针：检查服务是否准备好接受流量。应该检查所有关键依赖。

高级模式：隔离内部与外部依赖

# liveness_probe.py
@app.get("/healthz")
async def liveness():
    # 只检查内部健康：内存、CPU、goroutine/线程池
    return {"status": "alive"}
# readiness_probe.py
@app.get("/readyz")
async def readiness():
    # 必须检查外部依赖
    if not await check_database():
        return {"status": "not ready"}, 503
    return {"status": "ready"}

步骤4：包含业务指标的健康检查

有时需要检查业务层面的健康,例如批处理作业的进度。

class HealthChecker:
    def __init__(self, job_status_file="/tmp/job_status.json"):
        self.job_status_file = job_status_file
    @app.get("/healthz")
    async def detailed_health(self):
        # 读取自动生成的作业状态文件
        try:
            with open(self.job_status_file) as f:
                job_status = json.load(f)
                last_success = job_status.get("last_successful_run")
                if last_success and (datetime.now() - datetime.fromisoformat(last_success)).hours > 1:
                    return {"status": "degraded", "reason": "No successful job run in last hour"}, 200
        except FileNotFoundError:
            return {"status": "unknown", "reason": "Job status file not found"}, 200
        return {"status": "ok"}, 200

完整高级实现示例 (FastAPI + 异步 + 分层检查)

import asyncio
import logging
from typing import Dict, Tuple
from fastapi import FastAPI, Response
from fastapi.responses import JSONResponse
logger = logging.getLogger(__name__)
app = FastAPI()
class HealthStatus:
    """健康状态管理器"""
    def __init__(self):
        self._is_healthy = True
    def set_unhealthy(self):
        self._is_healthy = False
    async def check_all(self) -> Dict[str, Tuple[bool, str]]:
        results = {}
        # 并发执行所有检查
        checks = {
            "database": self._check_db(),
            "cache": self._check_redis(),
            "disk_space": self._check_disk()
        }
        for name, coro in checks.items():
            try:
                results[name] = await coro
            except Exception as e:
                results[name] = (False, str(e))
                logger.error(f"Health check {name} failed: {e}")
        return results
    async def _check_db(self) -> Tuple[bool, str]:
        # 实现数据库检查
        await asyncio.sleep(0.1)  # 模拟检查
        return True, "PostgreSQL connected"
    async def _check_redis(self) -> Tuple[bool, str]:
        await asyncio.sleep(0.05)  # 模拟检查
        return True, "Redis ping OK"
    async def _check_disk(self) -> Tuple[bool, str]:
        import psutil
        disk = psutil.disk_usage('/')
        if disk.percent > 90:
            return False, f"Disk space critical: {disk.percent}% used"
        return True, f"Disk space OK: {disk.percent}% used"
health_status = HealthStatus()
@app.on_event("startup")
async def startup_event():
    # 启动时执行一次检查
    results = await health_status.check_all()
    if not all(r[0] for r in results.values()):
        health_status.set_unhealthy()
        logger.warning("Service started in unhealthy state")
@app.get("/healthz")
async def liveness():
    """存活检查 - 只检查进程是否存活"""
    return Response(status_code=200)
@app.get("/readyz")
async def readiness():
    """就绪检查 - 检查所有依赖"""
    results = await health_status.check_all()
    all_healthy = all(r[0] for r in results.values())
    if not all_healthy:
        return JSONResponse(
            status_code=503,
            content={
                "status": "unhealthy",
                "checks": {
                    name: {
                        "status": "error" if not ok else "ok",
                        "message": msg
                    }
                    for name, (ok, msg) in results.items()
                }
            }
        )
    return JSONResponse(status_code=200, content={"status": "healthy"})

测试与部署

• 单元测试：测试健康检查端点的响应逻辑，可以使用框架的TestClient。

• Kubernetes探针配置：

  apiVersion: v1
  kind: Pod
  spec:
    containers:
    - name: my-app
      livenessProbe:
        httpGet:
          path: /healthz  # 存活检查
          port: 8080
        initialDelaySeconds: 5
        periodSeconds: 10
      readinessProbe:
        httpGet:
          path: /readyz   # 就绪检查
          port: 8080
        initialDelaySeconds: 10
        periodSeconds: 5

常见陷阱

1. 将大量业务逻辑放入健康检查：导致端点响应慢，甚至因业务故障导致健康检查失败，进而级联重启。
2. 健康检查相互影响：如果健康检查本身消耗大量资源（如内存），反而会影响应用健康。
3. 混淆存活与就绪：在存活检查中检查数据库，导致临时网络故障时容器被错误重启。
4. 不记录日志：健康检查失败时没有日志，排查困难。

通过以上设计,你可以获得一个稳定、实用且易于维护的健康检查端点，核心是保持简单，分层清晰，并严格遵循Kubernetes的探针语义。