Docker Compose Health Checker

Problem Statement

In automated deployment pipelines and local development workflows, executing docker compose up -d returns control immediately after containers are created. However, database initialization, schema migrations, and web server startups take time. Running integration tests or routing traffic immediately after bringing up containers often leads to intermittent, race-condition failures.

While Docker Compose supports native healthcheck directives in docker-compose.yml, there is no simple, standardized CLI mechanism to block deployment scripts until all defined services reach a verified healthy state. The Docker Compose Health Checker script addresses this by inspecting the actual runtime health and container state of services defined in a Compose file, dynamically tracking their status until all containers are fully operational or a timeout limit is breached.

How It Works

The script operates through a non-blocking polling architecture built on top of the docker Python SDK and PyYAML parser:

1. Service & Project Discovery: The script parses the specified docker-compose.yml file to extract service names and deduce the Docker Compose project name (from the file metadata or the parent folder name).
2. Container Inspection: It connects to the Docker daemon via Unix socket or TCP environment variables (DOCKER_HOST) and queries containers filtered by the com.docker.compose.project label.
3. Health Status Polling: In a loop, it inspects each service's runtime state (State.Status) and health status (State.Health.Status):
   • healthy: The service passed its defined health check.
   • running (no healthcheck): The service is running normally without an explicit healthcheck block.
   • starting: The service is still executing initial health probes.
   • unhealthy / exited: The service failed its health checks or crashed.
4. Evaluator & Terminal Exit: If all containers achieve a ready state, the script outputs a success report and exits with 0. If a service fails or the polling loop exceeds the configured --timeout, the script prints diagnostic information and exits with 1.

Full Script

#!/usr/bin/env python3
"""
Docker Compose Health Checker

Monitors containers defined in a docker-compose.yml file and blocks execution
until all services report a 'healthy' state (or 'running' if no health check exists).
Exits with 0 on success, or 1 if any service fails or times out.
"""

import argparse
import os
import re
import sys
import time
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple

import docker
from docker.errors import APIError, DockerException
import yaml

# Configuration defaults loaded from environment variables
DEFAULT_COMPOSE_FILE = os.environ.get("COMPOSE_FILE", "docker-compose.yml")
DEFAULT_TIMEOUT = int(os.environ.get("COMPOSE_HEALTH_TIMEOUT", "60"))
DEFAULT_INTERVAL = int(os.environ.get("COMPOSE_HEALTH_INTERVAL", "2"))
DEFAULT_PROJECT_NAME = os.environ.get("COMPOSE_PROJECT_NAME", "")


def sanitize_project_name(name: str) -> str:
    """Sanitize directory or project name to match Docker Compose naming convention."""
    return re.sub(r"[^a-z0-9_-]", "", name.lower())


def parse_compose_file(file_path: Path) -> Tuple[str, List[str]]:
    """
    Parse docker-compose file to extract project name and service list.
    Returns a tuple of (project_name, list_of_service_names).
    """
    if not file_path.is_file():
        raise FileNotFoundError(f"Compose file not found at: {file_path}")

    with open(file_path, "r", encoding="utf-8") as f:
        try:
            data: Dict[str, Any] = yaml.safe_load(f) or {}
        except yaml.YAMLError as exc:
            raise ValueError(f"Failed to parse YAML file: {exc}") from exc

    # Determine project name: explicit 'name' key -> fallback to parent folder name
    project_name = data.get("name")
    if not project_name:
        project_name = file_path.resolve().parent.name

    project_name = sanitize_project_name(project_name)

    services_dict = data.get("services", {})
    if not services_dict:
        raise ValueError(f"No services found in {file_path}")

    return project_name, list(services_dict.keys())


def get_container_health(container: docker.models.containers.Container) -> str:
    """
    Inspect container state and extract normalized health status.
    Possible return values: 'healthy', 'starting', 'unhealthy', 'running', 'exited', 'unknown'
    """
    container.reload()
    state = container.attrs.get("State", {})
    status = state.get("Status", "unknown")

    # Handle containers that have explicit Healthchecks defined
    health_info = state.get("Health", {})
    if health_info:
        return str(health_info.get("Status", "unknown"))

    # Containers without a healthcheck inherit basic status ('running', 'exited', etc.)
    return status


def check_services_health(
    client: docker.DockerClient, project_name: str, expected_services: List[str]
) -> Dict[str, str]:
    """
    Queries current health/status for all target services in the project.
    Returns a dict mapping service_name -> status string.
    """
    status_map: Dict[str, str] = {svc: "missing" for svc in expected_services}

    # Fetch containers filtering by docker-compose project label
    containers = client.containers.list(
        all=True, filters={"label": f"com.docker.compose.project={project_name}"}
    )

    for container in containers:
        labels = container.labels or {}
        service_name = labels.get("com.docker.compose.service")

        if service_name and service_name in status_map:
            status_map[service_name] = get_container_health(container)

    return status_map


def evaluate_health_results(
    status_map: Dict[str, str]
) -> Tuple[bool, bool]:
    """
    Evaluates current status map.
    Returns (all_ready, has_failures).
    - all_ready: True if all containers are 'healthy' or 'running'.
    - has_failures: True if any container is 'unhealthy' or 'exited'.
    """
    all_ready = True
    has_failures = False

    for service, status in status_map.items():
        if status in ("healthy", "running"):
            continue
        all_ready = False
        if status in ("unhealthy", "exited", "dead"):
            has_failures = True

    return all_ready, has_failures


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Wait for Docker Compose services to become healthy."
    )
    parser.add_argument(
        "-f",
        "--file",
        default=DEFAULT_COMPOSE_FILE,
        help=f"Path to docker-compose file (default: {DEFAULT_COMPOSE_FILE})",
    )
    parser.add_argument(
        "-t",
        "--timeout",
        type=int,
        default=DEFAULT_TIMEOUT,
        help=f"Max wait time in seconds (default: {DEFAULT_TIMEOUT})",
    )
    parser.add_argument(
        "-i",
        "--interval",
        type=int,
        default=DEFAULT_INTERVAL,
        help=f"Poll interval in seconds (default: {DEFAULT_INTERVAL})",
    )
    parser.add_argument(
        "-p",
        "--project-name",
        default=DEFAULT_PROJECT_NAME,
        help="Override Docker Compose project name",
    )

    args = parser.parse_args()
    compose_path = Path(args.file)

    # Load service list and project name
    try:
        parsed_project, expected_services = parse_compose_file(compose_path)
    except (FileNotFoundError, ValueError) as err:
        print(f"[ERROR] Configuration error: {err}", file=sys.stderr)
        sys.exit(1)

    project_name = sanitize_project_name(args.project_name) if args.project_name else parsed_project

    # Initialize Docker Client
    try:
        client = docker.from_env()
        client.ping()
    except (DockerException, APIError) as err:
        print(f"[ERROR] Failed to connect to Docker daemon: {err}", file=sys.stderr)
        sys.exit(1)

    print(f"[*] Monitoring Compose Project: '{project_name}'")
    print(f"[*] Services to verify: {', '.join(expected_services)}")
    print(f"[*] Timeout: {args.timeout}s | Poll Interval: {args.interval}s\n")

    start_time = time.time()
    elapsed = 0

    while elapsed < args.timeout:
        try:
            status_map = check_services_health(client, project_name, expected_services)
        except DockerException as err:
            print(f"[WARN] Error querying Docker API: {err}")
            time.sleep(args.interval)
            elapsed = int(time.time() - start_time)
            continue

        all_ready, has_failures = evaluate_health_results(status_map)

        # Output current snapshot status
        status_line = " | ".join([f"{svc}: {stat}" for svc, stat in status_map.items()])
        print(f"[{int(elapsed):>3}s] {status_line}")

        if all_ready:
            print("\n[SUCCESS] All services are healthy and ready!")
            sys.exit(0)

        if has_failures:
            print(
                "\n[FAILURE] One or more services are unhealthy or exited. Aborting.",
                file=sys.stderr,
            )
            sys.exit(1)

        time.sleep(args.interval)
        elapsed = int(time.time() - start_time)

    print(
        f"\n[TIMEOUT] Reached timeout of {args.timeout} seconds before all services were ready.",
        file=sys.stderr,
    )
    sys.exit(1)


if __name__ == "__main__":
    main()

Configuration

The script reads parameters from command-line arguments, with fallbacks to environment variables or hardcoded sensible defaults:

Usage Examples

1. Standard Run with Custom Timeout

Verify status for containers defined in a local file with a 30-second timeout:

python3 compose_health.py --file docker-compose.staging.yml --timeout 30

Output:

[*] Monitoring Compose Project: 'myapp'
[*] Services to verify: web, db, redis
[*] Timeout: 30s | Poll Interval: 2s

[  0s] web: starting | db: starting | redis: running
[  2s] web: starting | db: healthy | redis: running
[  4s] web: healthy | db: healthy | redis: running

[SUCCESS] All services are healthy and ready!

2. CI/CD Integration with Environment Variables

Use environment variables inside a pipeline step (e.g., GitHub Actions, GitLab CI, or Jenkins) without providing flags:

export COMPOSE_FILE="deploy/compose.yaml"
export COMPOSE_HEALTH_TIMEOUT="120"
export COMPOSE_PROJECT_NAME="ci-pipeline-1042"

docker compose -f $COMPOSE_FILE -p $COMPOSE_PROJECT_NAME up -d
python3 compose_health.py

Output (Failed Service Case):

[*] Monitoring Compose Project: 'ci-pipeline-1042'
[*] Services to verify: api, db
[*] Timeout: 120s | Poll Interval: 2s

[  0s] api: starting | db: starting
[  2s] api: starting | db: unhealthy

[FAILURE] One or more services are unhealthy or exited. Aborting.

Error Handling

• Missing/Corrupt Compose File: Triggers FileNotFoundError or yaml.YAMLError with an explicit diagnostic message and returns exit code 1.
• Docker Socket Unavailable: Catches docker.errors.DockerException (e.g., if Docker daemon is stopped or user lacks socket permissions) and reports the connection issue cleanly without dumping a stack trace.
• Service Failure Fast-Exit: If a container switches to unhealthy, exited, or dead, the script avoids spinning uselessly until timeout and immediately terminates with code 1.
• Transient API Glitches: Errors encountered during individual check_services_health() calls (e.g., high load socket blips) are logged as warnings while polling continues until timeout.

Extending Ideas

• Slack / Discord Webhook Notifications: Add an HTTP callback (urllib.request or requests) upon container failure to alert on-call teams immediately.
• Automatic Container Log Tail: Automatically fetch and print the last 50 lines of logs (container.logs(tail=50)) for any container that transitions into an unhealthy state.
• JSON Output Mode: Add a --json CLI flag to emit machine-readable log lines for integration with structured logging platforms (e.g., Datadog, CloudWatch).
• Restart Retries: Extend logic to automatically execute container.restart() once if a service becomes unhealthy before failing the entire script run.