本文目录导读：
设计Python脚本的健康检查端点,核心目标是快速、可靠地验证服务是否正常运行，一个优秀的健康检查端点不仅能回答“服务是否活着”，还能回答“服务是否正常工作”。
以下是从基础到最佳实践的完整设计指南,结合了Flask和FastAPI两种主流框架的示例。
核心设计原则
- 简单性：端点应尽可能简单，避免复杂的业务逻辑。
- 独立性：健康检查的代码应尽可能少依赖其他组件，自身要稳定。
- 层次化：一般分为两个层级：
- Liveness Probe (存活检查)：服务进程是否存活，通常只返回
200 OK。 - Readiness Probe (就绪检查)：服务是否准备好接收流量，需要检查数据库、缓存等依赖是否正常。
- （可选）Startup Probe (启动检查)：服务是否成功启动，用于慢启动应用。
- Liveness 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
| 实践 | 描述 |
|---|---|
| 使用独立端口 | 将健康检查端点放在独立端口（如8000用于业务，8080用于健康检查），避免被负载均衡路由干扰。 |
| 限制复杂度 | 避免在健康检查路径中执行数据库写操作、复杂查询或外部API调用（除就绪检查外）。 |
| 缓存结果 | 如果依赖检查开销大，使用短时间缓存（如5秒），防止高并发时打垮下游。 |
| 记录慢健康检查 | 为健康检查端点添加日志，记录耗时超过阈值（如500ms）的检查，定位性能问题。 |
| 分离存活与就绪 | 严格按照k8s的语义设计，存活检查只关心进程，就绪检查关心依赖。 |
| 统一状态码 | 200 = 健康，503 = 不健康 / 未就绪，避免使用500或404。 |
| 包含版本信息 | 返回version或build信息，方便定位问题。 |
完整高级实现示例 (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
常见陷阱
- 将大量业务逻辑放入健康检查：导致端点响应慢，甚至因业务故障导致健康检查失败，进而级联重启。
- 健康检查相互影响：如果健康检查本身消耗大量资源（如内存），反而会影响应用健康。
- 混淆存活与就绪：在存活检查中检查数据库，导致临时网络故障时容器被错误重启。
- 不记录日志：健康检查失败时没有日志，排查困难。
通过以上设计,你可以获得一个稳定、实用且易于维护的健康检查端点，核心是保持简单，分层清晰，并严格遵循Kubernetes的探针语义。
