K8sBench is a python script designed to test and calculate the pass rate for requests made to a kubernetes API server. The 'NoClient` version can be used if encountering issues with the k8s client creation.

K8sBench.py

import kubernetes
from kubernetes.client.rest import ApiException
import time
import os

# Set the path to the kubeconfig file
#kubeconfig_path = "~/#/config"              # "/path/to/your/kubeconfig"
# Set the KUBECONFIG environment variable
#os.environ["KUBECONFIG"] = kubeconfig_path

requests_per_minute = int(input("Enter the requests per minute: "))
net_requests = int(input("Enter the total number of requests to be made: "))
output_file = input("Enter the output file path (leave empty for no file output): ").strip()
requests_made = 0
passed_requests = 0
failed_requests = 0

# returns whether or note the request succeeded and the response
def make_kubectl_request():
    try:
        api = kubernetes.client.CoreV1Api()
        response = api.list_namespaced_pod("default")
        return True, response
    except ApiException as e:
        return False, str(e)

def print_message(message):
    print(message)
    if output_file:
        with open(output_file, "a") as f:
            f.write(message + "\n")

def print_statistics():
    if not output_file:
        print(f"Total Requests: {requests_made}")
        print(f"Passed Requests: {passed_requests}")
        print(f"Failed Requests: {failed_requests}")
        print(f"Pass Percentage: {passed_requests / requests_made * 100:.2f}%")

    else:
        with open(output_file, "a") as f:
            f.write(f"/-----------------------------------/\nTotal Requests: {requests_made}\n")
            f.write(f"Passed Requests: {passed_requests}\n")
            f.write(f"Failed Requests: {failed_requests}\n")
            f.write(f"Pass Percentage: {passed_requests / requests_made * 100:.2f}%\n")

# Calculate the delay between each request based on requests per minute
delay = 60 / requests_per_minute

for x in range(net_requests):
    requests_made += 1
    success, response = make_kubectl_request()
    if success:
        passed_requests += 1
    else:
        failed_requests += 1

    message = f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Request {requests_made} passed. Response: {response}"
    print_message(message) 

    time.sleep(delay)

print_statistics()

K8sBenchNoClient.py

import subprocess
import time


requests_per_minute = int(input("Enter the requests per minute: "))
net_requests = int(input("Enter the total number of requests to be made: "))
output_file = input("Enter the output file path (leave empty for no file output): ").strip()
requests_made = 0
passed_requests = 0
failed_requests = 0

# returns whether or note the request succeeded and the response
# uses current kube context - check with kubectl config current-context
def make_kubectl_request():
    try:
        output = subprocess.check_output(["kubectl", "get", "pods"])                
        return True, output.decode("utf-8") 
    except subprocess.CalledProcessError as e:
        return False, str(e)

def print_message(message):
    print(message)
    if output_file:
        with open(output_file, "a") as f:
            f.write(message + "\n")

def print_statistics():
    if not output_file:
        print(f"Total Requests: {requests_made}")
        print(f"Passed Requests: {passed_requests}")
        print(f"Failed Requests: {failed_requests}")
        print(f"Pass Percentage: {passed_requests / requests_made * 100:.2f}%")

    else:
        with open(output_file, "a") as f:
            f.write(f"/-----------------------------------/\nTotal Requests: {requests_made}\n")
            f.write(f"Passed Requests: {passed_requests}\n")
            f.write(f"Failed Requests: {failed_requests}\n")
            f.write(f"Pass Percentage: {passed_requests / requests_made * 100:.2f}%\n")

# Calculate the delay between each request based on requests per minute
delay = 60 / requests_per_minute

for x in range(net_requests):
    requests_made += 1
    success, response = make_kubectl_request()
    if success:
        passed_requests += 1
    else:
        failed_requests += 1

    message = f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Request {requests_made} passed. Response: {response}"
    print_message(message) 

    time.sleep(delay)

print_statistics()