import json import os import argparse import time from datetime import datetime, timezone from clusterloader2.utils import parse_xml_to_json, run_cl2_command, get_measurement, str2bool from clients.kubernetes_client import KubernetesClient DEFAULT_PODS_PER_NODE = 40 DEFAULT_NODES_PER_NAMESPACE = 100 CPU_REQUEST_LIMIT_MILLI = 1 DAEMONSETS_PER_NODE = { "aws": 2, "azure": 6, "aks": 6 } CPU_CAPACITY = { "aws": 0.94, "azure": 0.87, "aks": 0.87 } # TODO: Remove aks once CL2 update provider name to be azure def calculate_config(cpu_per_node, node_count, max_pods, provider, service_test, cnp_test, ccnp_test): throughput = 100 nodes_per_namespace = min(node_count, DEFAULT_NODES_PER_NAMESPACE) pods_per_node = DEFAULT_PODS_PER_NODE if service_test: pods_per_node = max_pods if cnp_test or ccnp_test: pods_per_node = max_pods # Different cloud has different reserved values and number of daemonsets # Using the same percentage will lead to incorrect nodes number as the number of nodes grow # For AWS, see: https://github.com/awslabs/amazon-eks-ami/blob/main/templates/al2/runtime/bootstrap.sh#L290 # For Azure, see: https://learn.microsoft.com/en-us/azure/aks/node-resource-reservations#cpu-reservations capacity = CPU_CAPACITY[provider] cpu_request = (cpu_per_node * 1000 * capacity) // pods_per_node cpu_request = max(cpu_request, CPU_REQUEST_LIMIT_MILLI) return throughput, nodes_per_namespace, pods_per_node, cpu_request def configure_clusterloader2( cpu_per_node, node_count, node_per_step, max_pods, repeats, operation_timeout, provider, cilium_enabled, scrape_containerd, service_test, cnp_test, ccnp_test, num_cnps, num_ccnps, dualstack, override_file): steps = node_count // node_per_step throughput, nodes_per_namespace, pods_per_node, cpu_request = calculate_config(cpu_per_node, node_per_step, max_pods, provider, service_test, cnp_test, ccnp_test) with open(override_file, 'w', encoding='utf-8') as file: file.write(f"CL2_NODES: {node_count}\n") file.write(f"CL2_LOAD_TEST_THROUGHPUT: {throughput}\n") file.write(f"CL2_NODES_PER_NAMESPACE: {nodes_per_namespace}\n") file.write(f"CL2_NODES_PER_STEP: {node_per_step}\n") file.write(f"CL2_PODS_PER_NODE: {pods_per_node}\n") file.write(f"CL2_DEPLOYMENT_SIZE: {pods_per_node}\n") file.write(f"CL2_LATENCY_POD_CPU: {cpu_request}\n") file.write(f"CL2_REPEATS: {repeats}\n") file.write(f"CL2_STEPS: {steps}\n") file.write(f"CL2_OPERATION_TIMEOUT: {operation_timeout}\n") file.write("CL2_PROMETHEUS_TOLERATE_MASTER: true\n") file.write("CL2_PROMETHEUS_MEMORY_LIMIT_FACTOR: 100.0\n") file.write("CL2_PROMETHEUS_MEMORY_SCALE_FACTOR: 100.0\n") file.write("CL2_PROMETHEUS_CPU_SCALE_FACTOR: 30.0\n") file.write("CL2_PROMETHEUS_NODE_SELECTOR: \"prometheus: \\\"true\\\"\"\n") file.write("CL2_POD_STARTUP_LATENCY_THRESHOLD: 3m\n") if scrape_containerd: file.write(f"CL2_SCRAPE_CONTAINERD: {str(scrape_containerd).lower()}\n") file.write("CONTAINERD_SCRAPE_INTERVAL: 5m\n") if cilium_enabled: file.write("CL2_CILIUM_METRICS_ENABLED: true\n") file.write("CL2_PROMETHEUS_SCRAPE_CILIUM_OPERATOR: true\n") file.write("CL2_PROMETHEUS_SCRAPE_CILIUM_AGENT: true\n") file.write("CL2_PROMETHEUS_SCRAPE_CILIUM_AGENT_INTERVAL: 30s\n") if service_test: file.write("CL2_SERVICE_TEST: true\n") else: file.write("CL2_SERVICE_TEST: false\n") if cnp_test: file.write("CL2_CNP_TEST: true\n") file.write(f"CL2_CNPS_PER_NAMESPACE: {num_cnps}\n") file.write(f"CL2_DUALSTACK: {dualstack}\n") file.write("CL2_GROUP_NAME: cnp-ccnp\n") if ccnp_test: file.write("CL2_CCNP_TEST: true\n") file.write(f"CL2_CCNPS: {num_ccnps}\n") file.write(f"CL2_DUALSTACK: {dualstack}\n") file.write("CL2_GROUP_NAME: cnp-ccnp\n") with open(override_file, 'r', encoding='utf-8') as file: print(f"Content of file {override_file}:\n{file.read()}") file.close() def validate_clusterloader2(node_count, operation_timeout_in_minutes=10): kube_client = KubernetesClient() ready_node_count = 0 timeout = time.time() + (operation_timeout_in_minutes * 60) while time.time() < timeout: ready_nodes = kube_client.get_ready_nodes() ready_node_count = len(ready_nodes) print(f"Currently {ready_node_count} nodes are ready.") if ready_node_count == node_count: break print(f"Waiting for {node_count} nodes to be ready.") time.sleep(10) if ready_node_count != node_count: raise Exception(f"Only {ready_node_count} nodes are ready, expected {node_count} nodes!") def execute_clusterloader2( cl2_image, cl2_config_dir, cl2_report_dir, cl2_config_file, kubeconfig, provider, scrape_containerd ): run_cl2_command(kubeconfig, cl2_image, cl2_config_dir, cl2_report_dir, provider, cl2_config_file=cl2_config_file, overrides=True, enable_prometheus=True, scrape_containerd=scrape_containerd) def collect_clusterloader2( cpu_per_node, node_count, max_pods, repeats, cl2_report_dir, cloud_info, run_id, run_url, service_test, cnp_test, ccnp_test, result_file, test_type, start_timestamp, ): details = parse_xml_to_json(os.path.join(cl2_report_dir, "junit.xml"), indent = 2) json_data = json.loads(details) testsuites = json_data["testsuites"] provider = json.loads(cloud_info)["cloud"] if testsuites: status = "success" if testsuites[0]["failures"] == 0 else "failure" else: raise Exception(f"No testsuites found in the report! Raw data: {details}") _, _, pods_per_node, _ = calculate_config(cpu_per_node, node_count, max_pods, provider, service_test, cnp_test, ccnp_test) pod_count = node_count * pods_per_node # TODO: Expose optional parameter to include test details template = { "timestamp": datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ'), "cpu_per_node": cpu_per_node, "node_count": node_count, "pod_count": pod_count, "churn_rate": repeats, "status": status, "group": None, "measurement": None, "result": None, # "test_details": details, "cloud_info": cloud_info, "run_id": run_id, "run_url": run_url, "test_type": test_type, "start_timestamp": start_timestamp, } content = "" for f in os.listdir(cl2_report_dir): file_path = os.path.join(cl2_report_dir, f) with open(file_path, 'r', encoding='utf-8') as file: print(f"Processing {file_path}") measurement, group_name = get_measurement(file_path) if not measurement: continue print(measurement, group_name) data = json.loads(file.read()) if "dataItems" in data: items = data["dataItems"] if not items: print(f"No data items found in {file_path}") print(f"Data:\n{data}") continue for item in items: result = template.copy() result["group"] = group_name result["measurement"] = measurement result["result"] = item content += json.dumps(result) + "\n" else: result = template.copy() result["group"] = group_name result["measurement"] = measurement result["result"] = data content += json.dumps(result) + "\n" os.makedirs(os.path.dirname(result_file), exist_ok=True) with open(result_file, 'w', encoding='utf-8') as file: file.write(content) def main(): parser = argparse.ArgumentParser(description="SLO Kubernetes resources.") subparsers = parser.add_subparsers(dest="command") # Sub-command for configure_clusterloader2 parser_configure = subparsers.add_parser("configure", help="Override CL2 config file") parser_configure.add_argument("cpu_per_node", type=int, help="CPU per node") parser_configure.add_argument("node_count", type=int, help="Number of nodes") parser_configure.add_argument("node_per_step", type=int, help="Number of nodes per scaling step") parser_configure.add_argument("max_pods", type=int, nargs='?', default=0, help="Maximum number of pods per node") parser_configure.add_argument("repeats", type=int, help="Number of times to repeat the deployment churn") parser_configure.add_argument("operation_timeout", type=str, help="Timeout before failing the scale up test") parser_configure.add_argument("provider", type=str, help="Cloud provider name") parser_configure.add_argument("cilium_enabled", type=str2bool, choices=[True, False], default=False, help="Whether cilium is enabled. Must be either True or False") parser_configure.add_argument("scrape_containerd", type=str2bool, choices=[True, False], default=False, help="Whether to scrape containerd metrics. Must be either True or False") parser_configure.add_argument("service_test", type=str2bool, choices=[True, False], default=False, help="Whether service test is running. Must be either True or False") parser_configure.add_argument("cnp_test", type=str2bool, choices=[True, False], nargs='?', default=False, help="Whether cnp test is running. Must be either True or False") parser_configure.add_argument("ccnp_test", type=str2bool, choices=[True, False], nargs='?', default=False, help="Whether ccnp test is running. Must be either True or False") parser_configure.add_argument("num_cnps", type=int, nargs='?', default=0, help="Number of cnps") parser_configure.add_argument("num_ccnps", type=int, nargs='?', default=0, help="Number of ccnps") parser_configure.add_argument("dualstack", type=str2bool, choices=[True, False], nargs='?', default=False, help="Whether cluster is dualstack. Must be either True or False") parser_configure.add_argument("cl2_override_file", type=str, help="Path to the overrides of CL2 config file") # Sub-command for validate_clusterloader2 parser_validate = subparsers.add_parser("validate", help="Validate cluster setup") parser_validate.add_argument("node_count", type=int, help="Number of desired nodes") parser_validate.add_argument("operation_timeout", type=int, default=600, help="Operation timeout to wait for nodes to be ready") # Sub-command for execute_clusterloader2 parser_execute = subparsers.add_parser("execute", help="Execute scale up operation") parser_execute.add_argument("cl2_image", type=str, help="Name of the CL2 image") parser_execute.add_argument("cl2_config_dir", type=str, help="Path to the CL2 config directory") parser_execute.add_argument("cl2_report_dir", type=str, help="Path to the CL2 report directory") parser_execute.add_argument("cl2_config_file", type=str, help="Path to the CL2 config file") parser_execute.add_argument("kubeconfig", type=str, help="Path to the kubeconfig file") parser_execute.add_argument("provider", type=str, help="Cloud provider name") parser_execute.add_argument("scrape_containerd", type=str2bool, choices=[True, False], default=False, help="Whether to scrape containerd metrics. Must be either True or False") # Sub-command for collect_clusterloader2 parser_collect = subparsers.add_parser("collect", help="Collect scale up data") parser_collect.add_argument("cpu_per_node", type=int, help="CPU per node") parser_collect.add_argument("node_count", type=int, help="Number of nodes") parser_collect.add_argument("max_pods", type=int, nargs='?', default=0, help="Maximum number of pods per node") parser_collect.add_argument("repeats", type=int, help="Number of times to repeat the deployment churn") parser_collect.add_argument("cl2_report_dir", type=str, help="Path to the CL2 report directory") parser_collect.add_argument("cloud_info", type=str, help="Cloud information") parser_collect.add_argument("run_id", type=str, help="Run ID") parser_collect.add_argument("run_url", type=str, help="Run URL") parser_collect.add_argument("service_test", type=str2bool, choices=[True, False], default=False, help="Whether service test is running. Must be either True or False") parser_collect.add_argument("cnp_test", type=str2bool, choices=[True, False], nargs='?', default=False, help="Whether cnp test is running. Must be either True or False") parser_collect.add_argument("ccnp_test", type=str2bool, choices=[True, False], nargs='?', default=False, help="Whether ccnp test is running. Must be either True or False") parser_collect.add_argument("result_file", type=str, help="Path to the result file") parser_collect.add_argument("test_type", type=str, nargs='?', default="default-config", help="Description of test type") parser_collect.add_argument("start_timestamp", type=str, help="Test start timestamp") args = parser.parse_args() if args.command == "configure": configure_clusterloader2(args.cpu_per_node, args.node_count, args.node_per_step, args.max_pods, args.repeats, args.operation_timeout, args.provider, args.cilium_enabled, args.scrape_containerd, args.service_test, args.cnp_test, args.ccnp_test, args.num_cnps, args.num_ccnps, args.dualstack, args.cl2_override_file) elif args.command == "validate": validate_clusterloader2(args.node_count, args.operation_timeout) elif args.command == "execute": execute_clusterloader2(args.cl2_image, args.cl2_config_dir, args.cl2_report_dir, args.cl2_config_file, args.kubeconfig, args.provider, args.scrape_containerd) elif args.command == "collect": collect_clusterloader2(args.cpu_per_node, args.node_count, args.max_pods, args.repeats, args.cl2_report_dir, args.cloud_info, args.run_id, args.run_url, args.service_test, args.cnp_test, args.ccnp_test, args.result_file, args.test_type, args.start_timestamp) if __name__ == "__main__": main()