Times "kubectl apply" to "pod is ready".

podbenchmark.py

#!/usr/bin/env python3

"""
Benchmark Pod Startup Time

This script is designed to measure how long it takes for a Kubernetes pod to become
fully available, from the moment the pod manifest is applied until the pod is in a Ready
state—and optionally, until a specified health check endpoint inside the pod returns HTTP 200.

Usage:
    benchmark_pod_startup.py <manifest> [--pod-name POD_NAME] [--health-check HEALTH_CHECK_URL]

Arguments:
    manifest          Path to the pod manifest YAML file.
    --pod-name        (Optional) Name of the pod. If not provided, the script will parse
                      the pod name from the manifest (using the metadata.name field).
    --health-check    (Optional) URL for an in-pod health check (e.g., http://localhost:8080/health).
                      If specified, the script will poll this endpoint until it returns HTTP 200.
                      If omitted, the script only waits for the pod to be in the Ready state.

This tool uses the kubectl CLI to apply the pod manifest and waits for the pod to become
ready. It then (optionally) uses `kubectl exec` to poll a health endpoint inside the pod.
The total elapsed time (in milliseconds) is printed when the pod is fully available.
"""

import argparse
import subprocess
import time
import yaml
import sys

def run_cmd(cmd, capture_output=True):
    """Run a shell command and return its output.

    Args:
        cmd (str): The shell command to run.
        capture_output (bool): Whether to capture the output (default is True).

    Returns:
        str: The trimmed output of the command.

    Raises:
        SystemExit: Exits if the command returns a non-zero exit status.
    """
    result = subprocess.run(cmd, shell=True, capture_output=capture_output, text=True)
    if result.returncode != 0:
        print(f"Error running command: {cmd}", file=sys.stderr)
        print(result.stderr, file=sys.stderr)
        sys.exit(result.returncode)
    return result.stdout.strip()

def parse_pod_name(manifest):
    """Parse the pod name from a YAML manifest file.

    Args:
        manifest (str): The path to the YAML manifest file.

    Returns:
        str: The pod name extracted from the manifest.

    Raises:
        SystemExit: If metadata.name is not found in the manifest.
    """
    with open(manifest, "r") as f:
        data = yaml.safe_load(f)
    try:
        return data["metadata"]["name"]
    except KeyError:
        print("Error: Could not find metadata.name in the manifest.", file=sys.stderr)
        sys.exit(1)

def wait_for_pod_ready(pod_name, timeout=300):
    """Wait until the specified pod reaches the Ready condition.

    This function uses 'kubectl wait' to block until the pod is ready.

    Args:
        pod_name (str): The name of the pod to wait for.
        timeout (int): Maximum time to wait in seconds (default is 300).
    """
    print(f"Waiting for pod '{pod_name}' to become Ready...")
    cmd = f"kubectl wait --for=condition=Ready pod/{pod_name} --timeout={timeout}s"
    run_cmd(cmd)

def wait_for_health_check(pod_name, url, poll_interval=0.1):
    """Poll a health endpoint inside the pod until it returns HTTP 200.

    This function continuously executes a curl command inside the pod until the endpoint responds with status code 200.

    Args:
        pod_name (str): The name of the pod in which to run the health check.
        url (str): The health check URL inside the pod (e.g., http://localhost:8080/health).
        poll_interval (float): Time in seconds between polling attempts (default is 0.1).
    """
    print(f"Polling health endpoint '{url}' in pod '{pod_name}'...")
    cmd_template = f"kubectl exec {pod_name} -- curl -s -o /dev/null -w '%{{http_code}}' {url}"
    while True:
        output = run_cmd(cmd_template)
        if output == "200":
            break
        time.sleep(poll_interval)

def main():
    """Benchmark pod startup time from applying a manifest to readiness (and optionally passing a health check).

    The script:
    1. Parses command-line arguments for the pod manifest, optional pod name, and optional health check endpoint.
    2. If no pod name is provided, it extracts the pod name from the manifest.
    3. Applies the pod manifest using kubectl.
    4. Waits until the pod is in the Ready state.
    5. If a health check URL is provided, polls the endpoint until a 200 response is received.
    6. Prints the elapsed time in milliseconds.
    """
    parser = argparse.ArgumentParser(
        description="Benchmark pod startup time from kubectl apply to readiness (and optional health check)."
    )
    parser.add_argument("manifest", help="Path to the pod manifest YAML file.")
    parser.add_argument("--pod-name", help="Pod name. If not provided, it will be parsed from the manifest.")
    parser.add_argument("--health-check", help="Health check endpoint inside the pod (e.g., http://localhost:8080/health). If omitted, health check is skipped.")
    args = parser.parse_args()

    # Determine pod name
    pod_name = args.pod_name if args.pod_name else parse_pod_name(args.manifest)
    print(f"Using pod name: {pod_name}")

    # Record start time in milliseconds
    start_time = int(time.time() * 1000)

    # Apply the pod manifest
    print("Applying pod manifest...")
    run_cmd(f"kubectl apply -f {args.manifest}")

    # Wait for the pod to be ready
    wait_for_pod_ready(pod_name)

    # If a health check endpoint is provided, poll it until it returns HTTP 200
    if args.health_check:
        wait_for_health_check(pod_name, args.health_check)
    else:
        print("No health check endpoint specified; skipping health check step.")

    # Record the end time and calculate elapsed time in milliseconds
    end_time = int(time.time() * 1000)
    elapsed_ms = end_time - start_time
    print(f"Pod became available after {elapsed_ms} ms.")

    # Delete the pod
    print("Deleting pod...")
    run_cmd(f"kubectl delete -f {args.manifest} --grace-period=0 --force")

if __name__ == "__main__":
    main()