Use portainer API to query for container health, with support for stacks, needs python3, no external deps

portainer_status.py

#!/usr/bin/python3

from typing import Union, List, Any, Dict, Optional
import argparse
import os
import json
import ssl
from urllib.request import urlopen, Request

JSON = Union[List[Any], Dict[str, Any]]
OK_STATES = ('running', 'starting', 'healthy')


def setup(api_key: str, base_url: str):
    """
        Setup the checker, sets API key and base URL
    """

    global API_KEY, BASE_URL
    API_KEY = api_key
    BASE_URL = base_url


def make_request(endpoint: str, data: Optional[JSON] = None, method: str = "GET") -> JSON:
    """
        Run a HTTP(s) request against an API endpoint, returns decoded JSON data
        or None if decoding failed.
    """
    global API_KEY, BASE_URL

    headers = {
        "X-API-Key": API_KEY,
        "Accept": "application/json",
        "Content-Type": "application/json"
    }
    request = Request(BASE_URL + endpoint, headers=headers, data=data, method=method)

    # Disable certificate verification
    sslcontext = ssl.create_default_context()
    sslcontext.check_hostname = False
    sslcontext.verify_mode = ssl.VerifyMode.CERT_NONE

    # Call endpoint request
    with urlopen(request, context=sslcontext) as fp:
        try:
            data = json.load(fp)
        except json.decoder.JSONDecodeError:
            data = []

    return data

def check_state(container: Dict[str, Any]) -> bool:
    """
        Check various fields in container struct to determine
        runtime state of a container
    """
    if container['State'] == 'running':
        # check if we have a health check
        if "(healthy)" in container['Status']:
            return "healthy"
        if "(health: starting)" in container['Status']:
            return "starting"
        if "(unhealthy)" in container['Status']:
            return "unhealthy"
        return "running"
    return container['State']


def perform_check(cluster: str, container_image: Optional[str] = None, stack: Optional[str] = None, container_name: Optional[str] = None) -> bool:
    """
        Check if the defined image is running. Search order is:
        - Container name if not None
        - Stack if not None
            - Image in stack if not none
            - If no image given report on all containers in stack
        - Summary of all containers running an image if all other tests are
          None.

        Prints container states as they are parsed, returns True if everything
        is ok, False otherwise
    """
    data = make_request("/api/endpoints/snapshot", method="POST")
    data = make_request("/api/endpoints")

    # find cluster
    for machine in data:
        if machine['Name'] == cluster:
            break

    # find container
    ok: Optional[bool] = None
    for snapshot in machine['Snapshots']:
        for container in snapshot['DockerSnapshotRaw']['Containers']:
            state = check_state(container)
            name = container['Names'][0][1:]

            # Search for container name first
            if container_name is not None:
                if ('/' + container_name) in container['Names']:
                    print(f"Container {name} is {state}.")
                    ok = state in OK_STATES

            # Search for stack name next
            elif stack is not None:
                if container['Labels'].get('com.docker.compose.project', None) == stack:
                    if container_image is not None: # Report on single container
                        if container['Image'] == container_image:
                            print(f"Container {name} is {state}.")
                            ok = state in OK_STATES
                    else: # Report on the complete stack
                        print(f"Container {name} is {state}.")
                        if state not in OK_STATES:
                            ok = False
                        elif ok is None:
                            ok = True
            else:  
                # Just find all containers running an image image last
                if container_image is not None:
                    if container['Image'] == container_image:
                        print(f"Container {name} is {state}.")
                        if state not in OK_STATES:
                            ok = False
                        elif ok is None:
                            ok = True

    if ok is None:
        # container not found, so status unknown
        print(f"Container for query not found")
        return False
    return ok


if __name__ == '__main__':
    parser = argparse.ArgumentParser(prog='portainer_check', description='check status of containers with portainer API')
    parser.add_argument('-k', '--api-key', dest='key', type=str, default=os.environ.get('API_KEY', None), help="API Key to use, defaults to environment variable API_KEY")
    parser.add_argument('-u', '--url', dest='base_url', type=str, default='https://localhost:9443', help="URL to query, defaults to localhost")
    parser.add_argument('-c', '--cluster', dest='cluster', type=str, default='local', help="Cluster to query, defaults to 'local'")
    parser.add_argument('-i', '--image', dest='container_image', type=str, help="Container image to search")
    parser.add_argument('-s', '--stack', dest='stack', type=str, help="Docker compose stack to search")
    parser.add_argument('-n', '--name', dest='name', type=str, help="Docker container name to query")

    args=parser.parse_args()
    if args.key is None:
        print("Please provide an API key by either using -k or setting the API_KEY environment")
        exit(1)

    if args.container_image is None and args.stack is None and args.name is None:
        print("Specify at least one of -i -s -n!")
        exit(1)

    setup(args.key, args.base_url)

    ok = perform_check(args.cluster, container_image=args.container_image, stack=args.stack, container_name=args.name)
    if not ok:
        exit(1)

    exit(0)