Docker SDK 容器健康检查

示例说明

蓝绿发布的关键判据是容器是否真正就绪——进程启动不等于服务可用。healthcheck 配置让 Docker 引擎周期执行探针命令并记录状态转换；wait_healthy 轮询容器的 Health.Status，healthy 才把流量切过来，unhealthy 直接失败并拉取最后四十行日志定位原因。SDK 化的意义是发布流水线不需要 shell 拼接。

技术要点

• healthcheck 的时间参数以纳秒表达
• start_period 宽限启动期的抖动
• 轮询区分三态避免误判

代码解读

start_with_healthcheck 启动容器并注入探针：curl 打健康端点、一秒间隔、五次重试、三秒启动宽限。wait_healthy 用单调时钟算截止时间，healthy 与 unhealthy 都立即返回，超时返回假。失败分支读取容器日志尾部帮助定位启动问题。

使用建议

探针命令必须存在于镜像内，curl 不可用时换 wget 或自定义脚本。纳秒单位极易写错，务必核对。生产发布还要配合负载均衡的主动健康探测。

代码示例

import docker

client = docker.from_env()


def start_with_healthcheck(image: str, name: str, port: int) -> dict:
    container = client.containers.run(
        image,
        name=name,
        detach=True,
        ports={f'{port}/tcp': None},
        healthcheck={
            'test': ['CMD-SHELL', f'curl -sf http://localhost:{port}/healthz || exit 1'],
            'interval': 1_000_000_000,   # 纳秒：1 秒
            'timeout': 500_000_000,      # 纳秒：0.5 秒
            'retries': 5,
            'start_period': 3_000_000_000,
        },
        environment={'SERVICE_PORT': str(port)},
    )
    return {'id': container.short_id, 'name': name}


def wait_healthy(name: str, timeout: float = 30.0) -> bool:
    import time
    deadline = time.monotonic() + timeout

    while time.monotonic() < deadline:
        container = client.containers.get(name)
        health = container.attrs['State']['Health']['Status']
        if health == 'healthy':
            return True
        if health == 'unhealthy':
            return False
        time.sleep(0.5)
    return False


info = start_with_healthcheck('registry.example/app:2.4', 'app-blue', 8080)
print('started:', info)

if wait_healthy('app-blue'):
    print('traffic can be routed to app-blue now')
else:
    logs = client.containers.get('app-blue').logs(tail=40).decode()
    print('unhealthy, last logs:')
    print(logs[-600:])