Python应用Pod健康检查怎么配置

wen python案例 2026-07-10 177

本文目录导读：

Python应用Pod健康检查怎么配置

核心概念
常用探测方式
参数调优建议
Python 最佳实践
完整示例配置
常见问题与诊断
不同框架的推荐配置
总结建议

配置 Python 应用的 Kubernetes Pod 健康检查（包括存活探针 livenessProbe 和就绪探针 readinessProbe）是保障服务高可用的关键，以下是针对 Python 应用的最佳实践配置指南。

核心概念

livenessProbe (存活探针)：判断容器是否存活，如果失败，Kubernetes 会重启容器。
readinessProbe (就绪探针)：判断容器是否准备好接受流量，如果失败，Pod 会被从 Service 的端点中移除。
startupProbe (启动探针)：用于解决启动慢的应用，防止存活探针在启动期间误杀容器。

常用探测方式

HTTP GET 探测（最推荐）

在 Python 应用中暴露一个轻量级的健康检查端点。

# FastAPI/Flask 应用示例
from fastapi import FastAPI
from flask import Flask
# FastAPI
app = FastAPI()
@app.get("/health")
async def health():
    return {"status": "healthy"}
# Flask
app = Flask(__name__)
@app.route("/health")
def health():
    return "OK", 200

# Kubernetes Deployment YAML
apiVersion: apps/v1
kind: Deployment
metadata:
  name: python-app
spec:
  replicas: 3
  selector:
    matchLabels:
      app: python-app
  template:
    metadata:
      labels:
        app: python-app
    spec:
      containers:
      - name: app
        image: python-app:latest
        ports:
        - containerPort: 8000
        # 就绪探针 - 检查流量是否就绪
        readinessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 10
          timeoutSeconds: 3
          successThreshold: 1
          failureThreshold: 3
        # 存活探针 - 检查应用是否活着
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 15
          periodSeconds: 20
          timeoutSeconds: 3
          failureThreshold: 3
        # 启动探针（可选）- 给慢启动应用缓冲时间
        startupProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 0
          periodSeconds: 5
          failureThreshold: 30  # 最多等待 30*5=150秒

TCP 探测（仅检查端口）

适用于你的应用没有 HTTP 服务器，或不想暴露 HTTP 端点。

readinessProbe:
  tcpSocket:
    port: 8000
  initialDelaySeconds: 5
  periodSeconds: 10
livenessProbe:
  tcpSocket:
    port: 8000
  initialDelaySeconds: 15
  periodSeconds: 20

Exec 命令探测

在容器内执行一条命令,返回 0 表示成功。

livenessProbe:
  exec:
    command:
    - python
    - -c
    - "import requests; requests.get('http://localhost:8000/health')"
  initialDelaySeconds: 15
  periodSeconds: 20

参数调优建议

参数	说明	建议值
`initialDelaySeconds`	容器启动后多久开始探测	5-15秒（取决于应用启动时间）
`periodSeconds`	探测频率	10-30秒
`timeoutSeconds`	探测超时时间	1-3秒
`successThreshold`	成功阈值	1
`failureThreshold`	失败阈值	3-5

Python 最佳实践

健康检查端点实现要点

# health.py
import psutil
import os
def health_check():
    status = {
        "status": "healthy",
        "version": os.environ.get("APP_VERSION", "unknown"),
        "memory": psutil.virtual_memory().percent,
        "cpu": psutil.cpu_percent(interval=0.5)
    }
    # 检查数据库连接
    try:
        db_connected = check_database_connection()
        status["database"] = "connected" if db_connected else "disconnected"
    except Exception as e:
        status["database"] = str(e)
        status["status"] = "degraded"
    # 检查其他依赖（Redis、消息队列等）
    try:
        redis_connected = check_redis_connection()
        status["redis"] = "connected" if redis_connected else "disconnected"
    except Exception as e:
        status["redis"] = str(e)
        status["status"] = "degraded"
    return status
# FastAPI 路由
@app.get("/health")
async def health():
    result = health_check()
    if result["status"] == "degraded":
        return result, 500  # 存活探针失败
    return result, 200

完整示例配置

apiVersion: apps/v1
kind: Deployment
metadata:
  name: python-api
  labels:
    app: python-api
spec:
  replicas: 3
  selector:
    matchLabels:
      app: python-api
  template:
    metadata:
      labels:
        app: python-api
    spec:
      containers:
      - name: api
        image: registry.example.com/python-api:1.0.0
        ports:
        - containerPort: 8000
          name: http
        env:
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: db-secret
              key: url
        resources:
          limits:
            cpu: "500m"
            memory: "512Mi"
          requests:
            cpu: "200m"
            memory: "256Mi"
        # 就绪探针 - 决定是否将流量路由到Pod
        readinessProbe:
          httpGet:
            path: /ready
            port: http
          initialDelaySeconds: 5
          periodSeconds: 5
          timeoutSeconds: 2
          successThreshold: 1
          failureThreshold: 3
        # 存活探针 - 决定是否重启Pod
        livenessProbe:
          httpGet:
            path: /live
            port: http
          initialDelaySeconds: 15
          periodSeconds: 15
          timeoutSeconds: 3
          successThreshold: 1
          failureThreshold: 3
        # 启动探针 - 给慢应用缓冲时间
        startupProbe:
          httpGet:
            path: /startup
            port: http
          initialDelaySeconds: 0
          periodSeconds: 5
          failureThreshold: 30

常见问题与诊断

查看探针日志

# 查看Pod状态
kubectl get pods -o wide
# 查看Pod事件（包括探针失败原因）
kubectl describe pod <pod-name>
# 查看容器日志
kubectl logs <pod-name>
# 查看前一个容器的日志（被重启后）
kubectl logs <pod-name> --previous

探针失败常见原因

端口配置错误：容器端口与探针配置的端口不匹配
路径错误：健康检查端点的路径不存在
启动时间不足：initialDelaySeconds 设置太短
资源限制：CPU/内存不足导致应用无法及时响应
依赖服务未就绪：数据库、Redis等外部服务未准备好

调试技巧

# 临时关闭探针进行调试（不推荐生产环境使用）
kubectl edit deployment python-app
# 移除 livenessProbe 或修改为 TCP探测

不同框架的推荐配置

Django + Gunicorn

# views.py
from django.http import HttpResponse
from django.db import connection
def health_check(request):
    # 检查数据库
    try:
        connection.ensure_connection()
        return HttpResponse("OK", status=200)
    except Exception:
        return HttpResponse("Error", status=500)

FastAPI + Uvicorn

from fastapi import FastAPI, Response
from pydantic import BaseModel
class HealthResponse(BaseModel):
    status: str
    database: str = "unknown"
@app.get("/health", response_model=HealthResponse)
async def health():
    # 检查依赖
    db_status = "connected" if await check_db() else "disconnected"
    if db_status == "disconnected":
        return HealthResponse(status="degraded", database=db_status), 500
    return HealthResponse(status="healthy", database=db_status)