docker-healthcheck/docker-healthcheck.py

#! /usr/bin/env python3

import requests
import yaml
import docker
import docker.errors
import sys


def read_checks() -> dict:
    if len(sys.argv) < 2:
        print("You need to specify a config by path")
        sys.exit(1)
    with open(sys.argv[1]) as f:
        checks = yaml.safe_load(f)
    return checks


def main() -> None:
    checks = read_checks()
    if "self" in checks:
        requests.get(checks["self"])

    client = docker.from_env()
    for container_id in checks:
        if container_id == "self":
            continue
        try:
            container = client.containers.get(container_id)
            container_status = container.attrs["State"]["Status"]
            if container_status in ["healthy", "running"]:
                retry_count = 0
                while requests.post(checks[container_id]).content != b'OK':
                    if retry_count > 10:
                        print(f"error posting to '{container_id}' with url '{checks[container_id]}'")
                    retry_count += 1
            else:
                requests.post(
                    checks[container_id] + "/fail",
                    f"Container has status '{container_status}' which is not one of 'healthy' or 'running'"
                )
        except docker.errors.NotFound:
            requests.post(checks[container_id] + "/fail", f"Container '{container_id}' not found.")


if __name__ == "__main__":
    main()