Autoscaling Benchmark Initial

benchmarks/scenarios/autoscaling/README.md

@@ -43,10 +43,10 @@ For example,
 There are two plots that you can plot.
 
 ### Generating report
-`python <aibrix_root_repo>/benchmarks/plot/plot-everything.py <experiment_home_dir>`
+`python plot-everything.py <experiment_home_dir>`
 
 For example, 
-`python <aibrix_root_repo>/benchmarks/plot/plot-everything.py experiment_results/25min_test`
+`python plot-everything.py experiment_results/25min_test`
 
 The directories should look like
 ```bash

benchmarks/scenarios/autoscaling/deepseek-llm-7b-chat/deploy.yaml

@@ -176,5 +176,5 @@ spec:
                   - key: machine.cluster.vke.volcengine.com/gpu-name
                     operator: In
                     values:
-                      # - NVIDIA-L20
-                      - Tesla-V100
+                      - NVIDIA-L20
+                      # - Tesla-V100

benchmarks/scenarios/autoscaling/overnight_run.sh

@@ -1,4 +1,10 @@
 #!/bin/bash
+set -x
+
+export KUBECONFIG=${KUBECONFIG}
+export aibrix_repo=${aibrix_repo}
+export api_key=${api_key}
+export kube_context=${kube_context}
 
 workload_path=$1
 if [ -z "${workload_path}" ]; then
@@ -14,10 +20,11 @@ for autoscaler in ${autoscalers}; do
     echo "started experiment at $(date)"
     echo autoscaler: ${autoscaler}
     echo workload: ${workload_path} 
-    echo "The stdout/stderr is being logged in ./output.txt"
-    ./run-test.sh ${workload_path} ${autoscaler} &> output.txt
+    echo "The stdout/stderr is being logged in output.txt"
+    ./run-test.sh ${workload_path} ${autoscaler} ${aibrix_repo} ${api_key} ${kube_context} > output.txt  2>&1
     end_time=$(date +%s)
     echo "Done: Time taken: $((end_time-start_time)) seconds"
     echo "--------------------------------"
     sleep 10
-done
+done
+python plot-everything.py experiment_results/ 

benchmarks/scenarios/autoscaling/plot-everything.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 import os
 import json
+import re
 import pandas as pd
 import matplotlib.pyplot as plt
 import numpy as np
@@ -24,10 +25,15 @@ def parse_experiment_output(lines):
             continue
         try:
             data = json.loads(line.strip())
-            required_fields = ['status_code', 'start_time', 'end_time', 'latency', 'throughput', 
-                             'prompt_tokens', 'output_tokens', 'total_tokens', 'input', 'output']
+            # required_fields = ['status_code', 'start_time', 'end_time', 'latency', 'throughput', 
+            #                  'prompt_tokens', 'output_tokens', 'total_tokens', 'input', 'output']
+            required_fields = ['status', 'start_time', 'end_time', 'latency', 'throughput', 
+                             'prompt_tokens', 'output_tokens', 'total_tokens', 'input', 'output', 'ttft', 'tpot']
             if any(field not in data for field in required_fields):
+                missingfields = [field not in data for field in required_fields]
+                print(missingfields)
                 continue
+            data['status_code'] = 200
             results.append(data)
         except json.JSONDecodeError:
             continue
@@ -43,15 +49,24 @@ def parse_experiment_output(lines):
     rps_series = df.groupby('second_bucket').size()
     df['rps'] = df['second_bucket'].map(rps_series)
 
-    success_rps = df[df['status_code'] == 200].groupby('second_bucket').size()
-    failed_rps = df[df['status_code'] != 200].groupby('second_bucket').size()
+    success_rps = df[df['status'] == 'success'].groupby('second_bucket').size()
+    failed_rps = df[df['status'] != 'success'].groupby('second_bucket').size()
     df['success_rps'] = df['second_bucket'].map(success_rps).fillna(0)
     df['failed_rps'] = df['second_bucket'].map(failed_rps).fillna(0)
 
     return df, base_time
 
 def get_autoscaler_name(output_dir):
     autoscaling = None
+    print(f"output_dir: {output_dir}")
+    # Extract the last part of the path after the last slash
+    # filename = output_dir.split("/")[-1]
+    # match = re.search(rf"^[^-]+-{workload_type}-([^-]+(?:-[^-]+)*)-\d{{8}}-\d{{6}}$", filename)
+
+    # if match:
+    #     print(match)
+    #     autoscaling = match.group(1)
+
     with open(f"{output_dir}/output.txt", 'r', encoding='utf-8') as f_:
         lines = f_.readlines()
         for line in lines:
@@ -61,6 +76,7 @@ def get_autoscaler_name(output_dir):
     if autoscaling == None:
         print(f"Invalid parsed autoscaling name: {autoscaling}")
         assert False
+    print(autoscaling)
     return autoscaling.upper()
 
 def parse_performance_stats(file_content):
@@ -162,11 +178,11 @@ def analyze_performance(df):
 
 def plot_combined_visualization(experiment_home_dir):
     # Create figure
-    fig = plt.figure(figsize=(12, 12))
+    fig = plt.figure(figsize=(12, 15))
 
     # Create GridSpec for time series plots
-    gs_ts = GridSpec(2, 2, figure=fig)
-    gs_ts.update(top=0.95, bottom=0.45, hspace=0.3, wspace=0.25)
+    gs_ts = GridSpec(4, 2, figure=fig)
+    gs_ts.update(top=0.95, bottom=0.4, hspace=0.5, wspace=0.3)
 
     # Create subplots for time series
     ax_cdf = fig.add_subplot(gs_ts[0, 0])
@@ -180,6 +196,11 @@ def plot_combined_visualization(experiment_home_dir):
 
     ax_cost = fig.add_subplot(gs_ts[1, 1])
 
+    ax_ttft = fig.add_subplot(gs_ts[2, 0])
+    ax_tpot = fig.add_subplot(gs_ts[2, 1])
+
+    ax_throughput = fig.add_subplot(gs_ts[3, 0])
+
     # Create 1x4 grid for bar plots
     gs_bars = GridSpec(1, 4, figure=fig)
     gs_bars.update(top=0.35, bottom=0.1, left=0.1, right=0.9, wspace=0.4)
@@ -232,7 +253,6 @@ def plot_combined_visualization(experiment_home_dir):
         color = colors[autoscaler]
         marker = '.'
         label_name = f'{autoscaler}'
-
         # Read and parse data
         experiment_output_file = os.path.join(output_dir, "output.jsonl")
         parsed_lines = read_experiment_file(experiment_output_file)
@@ -293,7 +313,11 @@ def plot_combined_visualization(experiment_home_dir):
         # ax_latency.scatter(df['end_time'], df['latency'], label=f'{autoscaler}', color=color, marker='.', s=5, alpha=0.3)
         ax_rps.plot(df['start_time'], df['rps'], label=label_name, color=color, linewidth=0.5, alpha=0.7)
         print(f"Max time: {label_name}, {df['start_time'].max()}")
-
+        ax_ttft.plot(df['start_time'], df['ttft'], label=label_name, color=color, linewidth=0.5, alpha=0.7)
+        ax_tpot.plot(df['start_time'], df['tpot'], label=label_name, color=color, linewidth=0.5, alpha=0.7)
+        ax_throughput.plot(df['start_time'], df['throughput'], label=label_name, color=color, linewidth=0.5, alpha=0.7)
+
+
 
         # 3. Running pods over time
         ax_pods.plot(pod_count_df['asyncio_time'], pod_count_df['Running'],
@@ -330,6 +354,21 @@ def plot_combined_visualization(experiment_home_dir):
     ax_rps.grid(True)
     ax_rps.legend()
 
+    ax_ttft.set_title("TTFT Over Time")
+    ax_ttft.set_xlabel("Time (s)")
+    ax_ttft.set_ylabel("TTFT (s)")
+    ax_ttft.legend()
+
+    ax_tpot.set_title("TPOT Over Time")
+    ax_tpot.set_xlabel("Time (s)")
+    ax_tpot.set_ylabel("TPOT (s)")
+    ax_tpot.legend()
+
+    ax_throughput.set_title("Throughput Over Time")
+    ax_throughput.set_xlabel("Time (s)")
+    ax_throughput.set_ylabel("Tokens/sec")
+    ax_throughput.legend()
+
     ax_pods.set_title('Running Pods', fontsize=12)
     ax_pods.set_xlabel('Time (seconds)')
     ax_pods.set_ylabel('# Running Pods')
@@ -362,7 +401,7 @@ def plot_combined_visualization(experiment_home_dir):
         content = read_stats_file(stat_fn)
         if content:
             stats = parse_performance_stats(content)
-            autoscaler = get_autoscaler_name(output_dir)
+            autoscaler = get_autoscaler_name(output_dir, workload_type)
             title = f"{autoscaler}"
             if autoscaler is None or autoscaler == "none" or autoscaler == "NONE":
                 color_list.append(colors[autoscaler])
@@ -400,14 +439,14 @@ def plot_combined_visualization(experiment_home_dir):
     plt.tight_layout()
 
     # Save the combined figure
-    output_path = os.path.join(experiment_home_dir, 'combined_visualization.pdf')
+    output_path = os.path.join(experiment_home_dir, f'combined_visualization-{workload_type}.pdf')
     plt.savefig(output_path, bbox_inches='tight')
     print(f"** Saved combined visualization to: {output_path}")
     plt.close()
 
 def main():
-    if len(sys.argv) != 2:
-        print("Usage: python script.py <experiment_home_dir>")
+    if len(sys.argv) != 3:
+        print("Usage: python script.py <experiment_home_dir> <workload_type>")
         return 1
 
     experiment_home_dir = sys.argv[1]

benchmarks/scenarios/autoscaling/run-test.sh

@@ -2,8 +2,10 @@
 
 input_workload_path=$1
 autoscaler=$2
-aibrix_repo="" # root dir of aibrix repo
-api_key="" # set your api key
+aibrix_repo=$3 # root dir of aibrix repo
+api_key=$4 # set your api key
+kube_context=$5
+
 k8s_yaml_dir="deepseek-llm-7b-chat"
 target_deployment="deepseek-llm-7b-chat" # "aibrix-model-deepseek-llm-7b-chat"
 target_ai_model=deepseek-llm-7b-chat
@@ -66,7 +68,7 @@ kubectl delete -f ${k8s_yaml_dir}/deploy.yaml
 kubectl apply -f ${k8s_yaml_dir}/deploy.yaml
 kubectl apply -f ${k8s_yaml_dir}/${autoscaler}.yaml
 echo "kubectl apply -f ${k8s_yaml_dir}/${autoscaler}.yaml"
-python3 ${aibrix_repo}/benchmarks/utils/set_num_replicas.py --deployment ${target_deployment} --replicas 1
+python3 ${aibrix_repo}/benchmarks/utils/set_num_replicas.py --deployment ${target_deployment} --replicas 1 --context ${kube_context}
 echo "Set number of replicas to \"1\". Autoscaling experiment will start from 1 pod"
 
 echo "Restart aibrix-controller-manager deployment"
@@ -82,9 +84,9 @@ kubectl rollout restart deploy ${target_deployment} -n default
 sleep_before_pod_check=20
 echo "Sleep for ${sleep_before_pod_check} seconds after restarting deployment"
 sleep ${sleep_before_pod_check}
-python3 ${aibrix_repo}/benchmarks/utils/check_k8s_is_ready.py ${target_deployment}
-python3 ${aibrix_repo}/benchmarks/utils/check_k8s_is_ready.py aibrix-controller-manager
-python3 ${aibrix_repo}/benchmarks/utils/check_k8s_is_ready.py aibrix-gateway-plugins
+python3 ${aibrix_repo}/benchmarks/utils/check_k8s_is_ready.py ${target_deployment} ${kube_context}
+python3 ${aibrix_repo}/benchmarks/utils/check_k8s_is_ready.py aibrix-controller-manager ${kube_context}
+python3 ${aibrix_repo}/benchmarks/utils/check_k8s_is_ready.py aibrix-gateway-plugins ${kube_context}
 
 # Start pod log monitoring
 pod_log_dir="${experiment_result_dir}/pod_logs"
@@ -94,7 +96,7 @@ mkdir -p ${pod_log_dir}
 cp ${input_workload_path} ${experiment_result_dir}
 
 # Start pod counter. It will run on background until the end of the experiment.
-python3 ${aibrix_repo}/benchmarks/utils/count_num_pods.py ${target_deployment} ${experiment_result_dir} &
+python3 ${aibrix_repo}/benchmarks/utils/count_num_pods.py ${target_deployment} ${experiment_result_dir} ${kube_context} &
 COUNT_NUM_POD_PID=$!
 echo "started count_num_pods.py with PID: $COUNT_NUM_POD_PID"
 
@@ -105,13 +107,13 @@ python3 ${aibrix_repo}/benchmarks/utils/streaming_pod_log_to_file.py aibrix-gate
 
 # Run experiment!!!
 output_jsonl_path=${experiment_result_dir}/output.jsonl
-python3 ${aibrix_repo}/benchmarks/generator/client.py \
+python3 ${aibrix_repo}/benchmarks/client/client.py \
     --workload-path ${input_workload_path} \
-    --endpoint "localhost:8888" \
+    --endpoint "http://localhost:8888" \
     --model ${target_ai_model} \
     --api-key ${api_key} \
-    --output-dir ${experiment_result_dir} \
-    --output-file-path ${output_jsonl_path}
+    --output-file-path ${output_jsonl_path} \
+    --streaming
 
 echo "Experiment is done. date: $(date)"
 
@@ -123,7 +125,7 @@ sleep 1
 
 # Cleanup
 kubectl delete podautoscaler --all --all-namespaces
-python3 ${aibrix_repo}/benchmarks/utils/set_num_replicas.py --deployment ${target_deployment} --replicas 1
+python3 ${aibrix_repo}/benchmarks/utils/set_num_replicas.py --deployment ${target_deployment} --replicas 1 --context ${kube_context}
 kubectl delete -f ${k8s_yaml_dir}/deploy.yaml
 
 # Stop monitoring processes

benchmarks/scenarios/utils/check_k8s_is_ready.py

@@ -58,6 +58,7 @@ def wait_for_all_podautoscaler_ready(namespace="default"):
 
 if __name__ == "__main__":
     target_deployment = sys.argv[1]
-    config.load_kube_config(context="ccr3aths9g2gqedu8asdg@41073177-kcu0mslcp5mhjsva38rpg")
+    kube_context = sys.argv[2]
+    config.load_kube_config(context=kube_context)
     wait_for_pods_ready(target_deployment)
     print("All pods are ready")

benchmarks/scenarios/utils/count_num_pods.py

@@ -6,8 +6,8 @@
 import os
 import asyncio
 
-def get_pod_status_counts(deployment_name, namespace="default"):
-   config.load_kube_config(context="ccr3aths9g2gqedu8asdg@41073177-kcu0mslcp5mhjsva38rpg")
+def get_pod_status_counts(deployment_name, kube_context, namespace="default"):
+   config.load_kube_config(context=kube_context)
    v1 = client.CoreV1Api()
    pods = v1.list_namespaced_pod(namespace)
    filtered_pods = [pod for pod in pods.items if deployment_name in pod.metadata.name]
@@ -39,12 +39,13 @@ def main():
     parser = argparse.ArgumentParser()
     parser.add_argument("deployment", help="Deployment name")
     parser.add_argument("output_dir", help="Output directory")
+    parser.add_argument("kube_context", help="Kube context")
     args = parser.parse_args()
 
     filename = f"{args.output_dir}/pod_count.csv"
     idx = 0
     while True:
-        status_counts = get_pod_status_counts(args.deployment)
+        status_counts = get_pod_status_counts(args.deployment, args.kube_context)
         write_to_csv(args.deployment, status_counts, filename, idx)
         time.sleep(1)
         idx += 1

benchmarks/scenarios/utils/set_num_replicas.py

@@ -2,9 +2,9 @@
 from kubernetes import client, config
 import sys
 
-def scale_deployment(deployment_name, replicas):
+def scale_deployment(deployment_name, replicas, kube_context):
     try:
-        config.load_kube_config(context="ccr3aths9g2gqedu8asdg@41073177-kcu0mslcp5mhjsva38rpg")
+        config.load_kube_config(context=kube_context)
         apps_v1 = client.AppsV1Api()
         apps_v1.patch_namespaced_deployment(
             name=deployment_name,
@@ -19,7 +19,8 @@ def scale_deployment(deployment_name, replicas):
    parser = argparse.ArgumentParser()
    parser.add_argument("--deployment", required=True)
    parser.add_argument("--replicas", type=int, required=True)
+   parser.add_argument("--context", required=True)
    args = parser.parse_args()
 
-   scale_deployment(args.deployment, args.replicas)
+   scale_deployment(args.deployment, args.replicas, args.context)
    print(f"Deployment {args.deployment} scaled to {args.replicas} replica(s)")

benchmarks/scenarios/utils/streaming_pod_log_to_file.py

@@ -57,12 +57,12 @@ def signal_handler(sig, frame):
     target_deployment = sys.argv[1]
     namespace = sys.argv[2]
     pod_log_dir = sys.argv[3]
-    include = sys.argv[4]
-    exclude = sys.argv[5]
-    if include == "none":
-        include = None
-    if exclude == "none":
-        exclude = None
+    include = None
+    exclude = None
+    if len(sys.argv) > 4:
+        include = sys.argv[4]
+    if len(sys.argv) > 5:
+        exclude = sys.argv[5]
 
     running_processes = []
     signal.signal(signal.SIGINT, signal_handler)