Log max cpu/memory usage of docker container

docker-max-log.py

#!/usr/bin/env python

import argparse
import json
import re
import subprocess
import time
from datetime import datetime
from typing import Any, Dict, List


def parse_docker_stats(stats_output: str) -> List[Dict[str, Any]]:
    lines = stats_output.strip().split("\n")
    headers = ["Container", "CPUPerc", "MemUsage"]
    stats = []

    for line in lines:
        values = line.split(",")
        stats.append(dict(zip(headers, values)))

    return stats


def collect_docker_stats(container_id: str) -> List[Dict[str, Any]]:
    result = subprocess.run(
        [
            "docker",
            "stats",
            "--no-stream",
            "--format",
            "{{.Container}},{{.CPUPerc}},{{.MemUsage}}",
            container_id,
        ],
        capture_output=True,
        text=True,
    )
    return parse_docker_stats(result.stdout)


def parse_cpu_usage(cpu_perc_str: str) -> float:
    return float(cpu_perc_str.rstrip("%"))


def parse_memory_usage(mem_usage_str: str) -> float:
    mem_usage, _, _ = mem_usage_str.partition("/")
    match = re.match(r"([0-9.]+)([MGT]iB)", mem_usage)
    if match is None:
        raise ValueError("Invalid memory usage format.")
    mem_usage_value = float(match.group(1))
    mem_usage_unit = match.group(2)

    if mem_usage_unit == "GiB":
        mem_usage_value *= 1024
    elif mem_usage_unit == "TiB":
        mem_usage_value *= 1024 * 1024

    return mem_usage_value


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Display max CPU and memory usage for a specific Docker container."
    )
    parser.add_argument("container", help="Container ID or name")
    parser.add_argument("--output", help="Output file name (optional)", default=None)
    parser.add_argument(
        "--interval", help="Log interval seconds (optional)", type=float, default=1.0
    )
    args = parser.parse_args()

    max_cpu_usage = 0.0
    max_memory_usage = 0.0
    output_file = None

    if args.output:
        output_file = open(args.output, "a")

    try:
        while True:
            stats = collect_docker_stats(args.container)
            if not stats:
                print("Container not found.")
                break

            stat = stats[0]
            cpu_usage = parse_cpu_usage(stat["CPUPerc"])
            memory_usage = parse_memory_usage(stat["MemUsage"])
            timestamp = datetime.now().isoformat()

            if cpu_usage > max_cpu_usage or memory_usage > max_memory_usage:
                if cpu_usage > max_cpu_usage:
                    max_cpu_usage = cpu_usage
                if memory_usage > max_memory_usage:
                    max_memory_usage = memory_usage

                print(
                    f"{timestamp}: Max CPU: {max_cpu_usage}%, Max Memory: {max_memory_usage}MiB"
                )

                if output_file:
                    stat["MaxCPUPerc"] = max_cpu_usage
                    stat["MaxMemUsage"] = max_memory_usage
                    stat["timestamp"] = timestamp
                    output_file.write(json.dumps(stat) + "\n")
                    output_file.flush()

            time.sleep(args.interval)
    finally:
        if output_file:
            output_file.close()


if __name__ == "__main__":
    main()

Installation

$ curl -sSL -o ~/.local/bin/docker-max-log https://gist.github.com/altescy/eef5593ef2b5a1fdf9b3b107c10784a5/raw && chmod +x ~/.local/bin/docker-max-log
$ docker-max-log --help
usage: docker-max-log [-h] [--output OUTPUT] [--interval INTERVAL] container

Display max CPU and memory usage for a specific Docker container.

positional arguments:
  container            Container ID or name

options:
  -h, --help           show this help message and exit
  --output OUTPUT      Output file name (optional)
  --interval INTERVAL  Log interval seconds (optional)