run_multi_concurrency_benchmark_local_v2.py

#!/usr/bin/env python3
"""
Run GuideLLM benchmarks with multiple concurrency rates from LOCAL machine.
This script runs on your laptop/workstation and executes commands in the pod remotely.
Captures console output for each run.

Usage:
  python3 run_multi_concurrency_benchmark_local_v2.py
"""

import subprocess
import time
import json
import sys
from datetime import datetime
from typing import List, Tuple

# Configuration
NAMESPACE = "kserve-e2e-perf"
GUIDELLM_POD = "guidellm-052-mehulm"
POD_LABEL = "serving.kserve.io/inferenceservice=qwen3-vl-30b-a3b-inst-10ce087b"
TARGET_URL = "http://qwen3-vl-30b-a3b-inst-10ce087b-predictor.kserve-e2e-perf.svc.cluster.local:8080/v1"
HEALTH_URL = "http://qwen3-vl-30b-a3b-inst-10ce087b-predictor.kserve-e2e-perf.svc.cluster.local:8080/health"
DATA_PATH = "/datasets/gpt_oss_perf.parquet"
OUTPUT_DIR = "/results"
LOCAL_OUTPUT_DIR = "./benchmark-results"  # Local directory to copy results

# Concurrency rates to test
CONCURRENCY_RATES = [1, 10, 50, 100, 200, 300]

# Timeouts
POD_RESTART_TIMEOUT = 600  # 10 minutes
HEALTH_CHECK_TIMEOUT = 300  # 5 minutes
HEALTH_CHECK_INTERVAL = 10  # 10 seconds


def run_command(cmd: List[str], check: bool = True, capture_output: bool = True) -> Tuple[int, str, str]:
    """Run a shell command and return exit code, stdout, stderr."""
    print(f"\n🔧 Running: {' '.join(cmd)}")

    result = subprocess.run(
        cmd,
        capture_output=capture_output,
        text=True,
        check=False
    )

    if check and result.returncode != 0:
        print(f"❌ Command failed with exit code {result.returncode}")
        print(f"STDERR: {result.stderr}")
        if not capture_output:
            sys.exit(1)

    return result.returncode, result.stdout, result.stderr


def restart_pod():
    """Delete the pod to trigger restart."""
    print("\n" + "=" * 80)
    print(f"🔄 Restarting pod with label: {POD_LABEL}")
    print("=" * 80)

    # Get current pod name
    cmd_get_pod = [
        "oc", "get", "pods",
        "-l", POD_LABEL,
        "-n", NAMESPACE,
        "-o", "jsonpath={.items[0].metadata.name}"
    ]
    _, current_pod, _ = run_command(cmd_get_pod, check=True)
    current_pod = current_pod.strip()

    if not current_pod:
        print("❌ No pod found with the label")
        sys.exit(1)

    print(f"Current pod: {current_pod}")

    # Delete the pod
    cmd = ["oc", "delete", "pod", current_pod, "-n", NAMESPACE, "--wait=false"]
    returncode, stdout, stderr = run_command(cmd, check=False)

    if returncode != 0:
        print(f"⚠️  Pod delete returned code {returncode}, continuing anyway...")
    else:
        print(f"✓ Pod deletion initiated")

    # Wait a bit for deletion to start
    time.sleep(5)


def wait_for_pod_ready():
    """Wait for the pod to be ready."""
    print("\n" + "=" * 80)
    print(f"⏳ Waiting for pod to be ready...")
    print("=" * 80)

    start_time = time.time()

    while time.time() - start_time < POD_RESTART_TIMEOUT:
        # Get pod status
        cmd = [
            "oc", "get", "pods",
            "-l", POD_LABEL,
            "-n", NAMESPACE,
            "-o", "jsonpath={.items[0].status.phase}"
        ]
        returncode, stdout, stderr = run_command(cmd, check=False)

        if returncode == 0 and stdout.strip() == "Running":
            # Check if ready
            cmd_ready = [
                "oc", "get", "pods",
                "-l", POD_LABEL,
                "-n", NAMESPACE,
                "-o", "jsonpath={.items[0].status.conditions[?(@.type=='Ready')].status}"
            ]
            returncode_ready, stdout_ready, _ = run_command(cmd_ready, check=False)

            if returncode_ready == 0 and stdout_ready.strip() == "True":
                # Get the new pod name
                cmd_name = [
                    "oc", "get", "pods",
                    "-l", POD_LABEL,
                    "-n", NAMESPACE,
                    "-o", "jsonpath={.items[0].metadata.name}"
                ]
                _, pod_name, _ = run_command(cmd_name, check=False)

                elapsed = time.time() - start_time
                print(f"✓ Pod is ready: {pod_name.strip()}")
                print(f"  Elapsed time: {elapsed:.1f} seconds")
                return pod_name.strip()

        # Show progress
        elapsed = time.time() - start_time
        print(f"  ⏳ Waiting... ({elapsed:.0f}s / {POD_RESTART_TIMEOUT}s) - Status: {stdout.strip()}")
        time.sleep(10)

    print(f"❌ Pod did not become ready within {POD_RESTART_TIMEOUT} seconds")
    sys.exit(1)


def check_health():
    """Check if the health endpoint is responding from inside the guidellm pod."""
    print("\n" + "=" * 80)
    print(f"🏥 Checking health endpoint: {HEALTH_URL}")
    print("=" * 80)

    start_time = time.time()

    while time.time() - start_time < HEALTH_CHECK_TIMEOUT:
        # Run curl from inside the guidellm pod
        cmd = [
            "oc", "exec", "-n", NAMESPACE, GUIDELLM_POD, "--",
            "curl", "-s", "-o", "/dev/null", "-w", "%{http_code}", HEALTH_URL
        ]
        returncode, stdout, stderr = run_command(cmd, check=False)

        if returncode == 0 and stdout.strip() == "200":
            elapsed = time.time() - start_time
            print(f"✓ Health check passed!")
            print(f"  Elapsed time: {elapsed:.1f} seconds")
            return True

        elapsed = time.time() - start_time
        print(f"  ⏳ Waiting for health... ({elapsed:.0f}s / {HEALTH_CHECK_TIMEOUT}s) - HTTP: {stdout.strip()}")
        time.sleep(HEALTH_CHECK_INTERVAL)

    print(f"❌ Health check did not pass within {HEALTH_CHECK_TIMEOUT} seconds")
    sys.exit(1)


def run_benchmark(concurrency: int) -> Tuple[str, str]:
    """Run GuideLLM benchmark with specified concurrency rate.

    Returns:
        Tuple of (output_file, console_log_file) paths in the pod
    """
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    output_file = f"{OUTPUT_DIR}/gpt-oss-parquet-concurrency{concurrency}_{timestamp}.json"
    console_log_file = f"{OUTPUT_DIR}/gpt-oss-parquet-concurrency{concurrency}_{timestamp}_console.log"

    print("\n" + "=" * 80)
    print(f"🚀 Running benchmark with concurrency: {concurrency}")
    print(f"📝 Output file: {output_file}")
    print(f"📝 Console log: {console_log_file}")
    print("=" * 80)

    # Build the guidellm command
    guidellm_cmd = (
        f"guidellm benchmark "
        f"--target {TARGET_URL} "
        f"--data {DATA_PATH} "
        f"--data-column-mapper '{{\"text_column\":\"text_input\"}}' "
        f"--data-samples 500 "
        f"--processor gpt2 "
        f"--rate-type concurrent "
        f"--rate {concurrency} "
        f"--output-path {output_file}"
    )

    # Build command that captures output to log file AND displays it
    full_cmd = f"{guidellm_cmd} 2>&1 | tee {console_log_file}"

    # Run guidellm inside the pod
    cmd = [
        "oc", "exec", "-n", NAMESPACE, GUIDELLM_POD, "--",
        "bash", "-c", full_cmd
    ]

    start_time = time.time()

    print(f"\n🔧 Executing in pod: {guidellm_cmd}\n")
    print(f"📋 Console output will be saved to: {console_log_file}\n")

    # Run without capturing output so we see live progress
    returncode, _, _ = run_command(cmd, check=False, capture_output=False)

    elapsed = time.time() - start_time

    if returncode == 0:
        print(f"\n✓ Benchmark completed successfully!")
        print(f"  Elapsed time: {elapsed:.1f} seconds ({elapsed/60:.1f} minutes)")
        print(f"  Results saved to: {output_file}")
        print(f"  Console log saved to: {console_log_file}")
        return output_file, console_log_file
    else:
        print(f"\n❌ Benchmark failed with exit code {returncode}")
        print(f"  Elapsed time: {elapsed:.1f} seconds")
        return None, None


def create_summary(results: List[dict]):
    """Create a summary file in the pod and copy it locally."""
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    summary_file = f"{OUTPUT_DIR}/benchmark_summary_{timestamp}.json"
    local_summary_file = f"{LOCAL_OUTPUT_DIR}/benchmark_summary_{timestamp}.json"

    summary = {
        "timestamp": timestamp,
        "namespace": NAMESPACE,
        "target": TARGET_URL,
        "data": DATA_PATH,
        "data_samples": 500,
        "concurrency_rates": CONCURRENCY_RATES,
        "results": results
    }

    # Write summary to a temp file locally
    import tempfile
    with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as tmp:
        json.dump(summary, tmp, indent=2)
        tmp_path = tmp.name

    # Copy to pod
    cmd_copy = [
        "oc", "cp", tmp_path,
        f"{NAMESPACE}/{GUIDELLM_POD}:{summary_file}"
    ]
    run_command(cmd_copy, check=True)

    # Copy from pod to local directory
    subprocess.run(["mkdir", "-p", LOCAL_OUTPUT_DIR], check=True)
    cmd_copy_local = [
        "oc", "cp",
        f"{NAMESPACE}/{GUIDELLM_POD}:{summary_file}",
        local_summary_file
    ]
    run_command(cmd_copy_local, check=True)

    print(f"\n📊 Summary saved to:")
    print(f"  Pod:   {summary_file}")
    print(f"  Local: {local_summary_file}")

    return summary_file, local_summary_file


def copy_results_to_local(results: List[dict]):
    """Copy all benchmark results from pod to local machine."""
    print("\n" + "=" * 80)
    print("📥 Copying results to local machine")
    print("=" * 80)

    subprocess.run(["mkdir", "-p", LOCAL_OUTPUT_DIR], check=True)

    for result in results:
        if result["status"] == "success":
            # Copy JSON result file
            if result.get("output_file"):
                remote_file = result["output_file"]
                local_file = f"{LOCAL_OUTPUT_DIR}/{remote_file.split('/')[-1]}"

                cmd = [
                    "oc", "cp",
                    f"{NAMESPACE}/{GUIDELLM_POD}:{remote_file}",
                    local_file
                ]

                returncode, _, _ = run_command(cmd, check=False)
                if returncode == 0:
                    print(f"  ✓ Copied JSON: {local_file}")
                else:
                    print(f"  ✗ Failed to copy: {remote_file}")

            # Copy console log file
            if result.get("console_log_file"):
                remote_log = result["console_log_file"]
                local_log = f"{LOCAL_OUTPUT_DIR}/{remote_log.split('/')[-1]}"

                cmd_log = [
                    "oc", "cp",
                    f"{NAMESPACE}/{GUIDELLM_POD}:{remote_log}",
                    local_log
                ]

                returncode_log, _, _ = run_command(cmd_log, check=False)
                if returncode_log == 0:
                    print(f"  ✓ Copied LOG:  {local_log}")
                else:
                    print(f"  ✗ Failed to copy log: {remote_log}")


def main():
    """Main execution loop."""
    print("\n" + "=" * 80)
    print("🎯 Multi-Concurrency Benchmark Runner (Local Execution)")
    print("=" * 80)
    print(f"\nConfiguration:")
    print(f"  Namespace:        {NAMESPACE}")
    print(f"  GuideLLM Pod:     {GUIDELLM_POD}")
    print(f"  Model Pod Label:  {POD_LABEL}")
    print(f"  Target URL:       {TARGET_URL}")
    print(f"  Data:             {DATA_PATH}")
    print(f"  Concurrency:      {CONCURRENCY_RATES}")
    print(f"  Output dir (pod): {OUTPUT_DIR}")
    print(f"  Output dir (local): {LOCAL_OUTPUT_DIR}")

    # Verify oc is available
    returncode, _, _ = run_command(["which", "oc"], check=False)
    if returncode != 0:
        print("\n❌ 'oc' command not found. Please install OpenShift CLI.")
        sys.exit(1)

    # Verify guidellm pod exists
    cmd = ["oc", "get", "pod", GUIDELLM_POD, "-n", NAMESPACE]
    returncode, _, _ = run_command(cmd, check=False)
    if returncode != 0:
        print(f"\n❌ GuideLLM pod '{GUIDELLM_POD}' not found in namespace '{NAMESPACE}'")
        sys.exit(1)

    results = []

    for i, concurrency in enumerate(CONCURRENCY_RATES, 1):
        print("\n" + "=" * 80)
        print(f"📍 BENCHMARK {i}/{len(CONCURRENCY_RATES)} - Concurrency: {concurrency}")
        print("=" * 80)

        try:
            # Step 1: Restart pod (skip for first iteration)
            if i > 1:
                restart_pod()

                # Step 2: Wait for pod to be ready
                new_pod_name = wait_for_pod_ready()

                # Step 3: Health check
                check_health()

                # Give it a bit more time to stabilize
                print(f"\n⏸️  Waiting 30 seconds for model to fully load...")
                time.sleep(30)
            else:
                print(f"\n⏭️  Skipping restart for first iteration")
                # Still do health check
                check_health()

            # Step 4: Run benchmark
            output_file, console_log_file = run_benchmark(concurrency)

            if output_file and console_log_file:
                results.append({
                    "concurrency": concurrency,
                    "output_file": output_file,
                    "console_log_file": console_log_file,
                    "status": "success",
                    "timestamp": datetime.now().isoformat()
                })
            else:
                results.append({
                    "concurrency": concurrency,
                    "output_file": None,
                    "console_log_file": None,
                    "status": "failed",
                    "timestamp": datetime.now().isoformat()
                })

        except KeyboardInterrupt:
            print("\n\n⚠️  Interrupted by user")
            break
        except Exception as e:
            print(f"\n❌ Error during concurrency {concurrency}: {e}")
            results.append({
                "concurrency": concurrency,
                "output_file": None,
                "console_log_file": None,
                "status": "error",
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            })

    # Create summary
    print("\n" + "=" * 80)
    print("📊 Creating Summary Report")
    print("=" * 80)

    summary_file, local_summary_file = create_summary(results)

    # Copy results to local machine
    copy_results_to_local(results)

    # Final report
    print("\n" + "=" * 80)
    print("✅ BENCHMARK SUITE COMPLETE")
    print("=" * 80)

    print(f"\nResults:")
    for result in results:
        status_icon = "✓" if result["status"] == "success" else "✗"
        print(f"  {status_icon} Concurrency {result['concurrency']:3d}: {result['status']}")
        if result.get("output_file"):
            print(f"      → JSON: {result['output_file']}")
        if result.get("console_log_file"):
            print(f"      → LOG:  {result['console_log_file']}")

    print(f"\n📊 Summary:")
    print(f"  Pod:   {summary_file}")
    print(f"  Local: {local_summary_file}")

    print(f"\n📁 Local results directory: {LOCAL_OUTPUT_DIR}")
    print(f"   Contents:")

    import os
    if os.path.exists(LOCAL_OUTPUT_DIR):
        files = sorted(os.listdir(LOCAL_OUTPUT_DIR))
        for f in files:
            size = os.path.getsize(os.path.join(LOCAL_OUTPUT_DIR, f))
            size_mb = size / (1024 * 1024)
            print(f"   - {f} ({size_mb:.2f} MB)")

    # Exit with error if any failed
    failed = [r for r in results if r["status"] != "success"]
    if failed:
        print(f"\n⚠️  {len(failed)} benchmark(s) failed")
        sys.exit(1)
    else:
        print(f"\n🎉 All {len(results)} benchmarks completed successfully!")
        sys.exit(0)


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\n\n⚠️  Script interrupted by user")
        sys.exit(130)
    except Exception as e:
        print(f"\n❌ Fatal error: {e}")
        import traceback
        traceback.print_exc()
        sys.exit(1)