-
Notifications
You must be signed in to change notification settings - Fork 0
/
simulation.py
134 lines (112 loc) · 6.13 KB
/
simulation.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
import argparse
import subprocess
from string import Template
def apply(args):
filename = args.name
if args.num != None: # Check if the number of resources is specified
num_resources = args.num_resources
else:
num_resources = 1
with open(filename) as f: # Read the resource file
template = Template(f.read())
try:
for i in range(1, num_resources + 1):
cmd_args = ["kubectl", "apply"]
cmd_args.extend(["-f", "-"])
subprocess.run(cmd_args, input=template.substitute(id=i).encode(), check=True) # Apply the resource
except subprocess.CalledProcessError:
print(f"Failed to apply resource from file '{filename}'.")
def cluster(args):
cluster_action = args.action
cluster_name = args.name
if cluster_action == "create":
try:
subprocess.run(["kwokctl", "create", "cluster", "--name", cluster_name], check=True) # Create a cluster
print(f"Cluster '{cluster_name}' created successfully.")
except subprocess.CalledProcessError:
print(f"Failed to create cluster '{cluster_name}'.")
elif cluster_action == "delete":
try:
subprocess.run(["kwokctl", "delete", "cluster", "--name", cluster_name], check=True) # Delete a cluster
print(f"Cluster '{cluster_name}' deleted successfully.")
except subprocess.CalledProcessError:
print(f"Failed to delete cluster '{cluster_name}'.")
def delete(args):
resource_type = args.type
resource_name = args.name
try:
subprocess.run(["kubectl", "delete", resource_type, resource_name], check=True) # Delete the resource
print(f"Deleted {resource_type} '{resource_name}' successfully.")
except subprocess.CalledProcessError:
print(f"Failed to delete {resource_type} '{resource_name}'.")
def get(args):
resource_type = args.type
try:
subprocess.run(["kubectl", "get", resource_type], check=True) # Get the resource
except subprocess.CalledProcessError:
print(f"Failed to execute 'kubectl get {resource_type}'.")
def show(args):
resource = args.resource
try:
if resource == "assignments":
cmd = "kubectl get pod -o=custom-columns=NODE:.spec.nodeName,NAME:.metadata.name --all-namespaces"
subprocess.run(cmd, shell=True) # Show what pod is on which node
elif resource == "nodes":
cmd = "kubectl describe nodes | awk '/Allocated resources:/,/ephemeral-storage/ {print}'"
subprocess.run(cmd, shell=True) # Show node allocated resources
elif resource == "priority":
subprocess.run(
[
"kubectl",
"get",
"pods",
"-n",
"default",
"-o",
"custom-columns=NAME:.metadata.name,PRIORITY:.spec.priorityClassName",
]
) # See what priority classes are assigned to pods
elif resource == "pods":
subprocess.run(
[
"kubectl",
"get",
"pods",
"--output=custom-columns=NAMESPACE:.metadata.namespace,NAME:.metadata.name,CPU-REQUEST:.spec.containers[].resources.requests.cpu,CPU-LIMIT:.spec.containers[].resources.limits.cpu,MEMORY-REQUEST:.spec.containers[].resources.requests.memory,MEMORY-LIMIT:.spec.containers[].resources.limits.memory,AGE:.metadata.creationTimestamp",
]
) # Show pod requests and limits
except subprocess.CalledProcessError:
if resource == "nodes":
print("Failed to show node allocated resources.")
elif resource == "priority":
print("Failed to get pod priority classes.")
elif resource == "pods":
print("Failed to show pod characteristics.")
def main():
parser = argparse.ArgumentParser(description="Script to setup resources in Kubernetes WithOut Kubelet (KWOK)") # Initialize the command line tool
subparsers = parser.add_subparsers(title="Available Commands") # Initialize subparsers for different commands
apply_parser = subparsers.add_parser("apply", help="apply specified resource") # Add the apply command
apply_parser.add_argument("name", help="Filename") # Add the filename argument
apply_parser.add_argument("--num", type=int, help="Number of resources to deploy (optional)") # Add the optional number of resources argument
apply_parser.set_defaults(func=apply) # Set the function to apply
cluster_parser = subparsers.add_parser("cluster", help="cluster management (create/delete)") # Add the cluster command
cluster_parser.add_argument("action", choices=["create", "delete"], help="Cluster Action") # Add the action argument
cluster_parser.add_argument("name", help="Cluster Name") # Add the cluster name argument
cluster_parser.set_defaults(func=cluster) # Set the function to cluster
delete_parser = subparsers.add_parser("delete", help="delete specified resource") # Add the delete command
delete_parser.add_argument("type", help="Resource Type") # Add the resource type argument
delete_parser.add_argument("name", help="Resource Name") # Add the resource name argument
delete_parser.set_defaults(func=delete) # Set the function to delete
get_parser = subparsers.add_parser("get", help="get specified resource") # Add the get command
get_parser.add_argument("type", help="resource type") # Add the resource type argument
get_parser.set_defaults(func=get) # Set the function to get
show_parser = subparsers.add_parser("show", help="show pod characteristics") # Add the show command
show_parser.add_argument("resource", choices=["assignments", "nodes", "priority", "pods"]) # Add the resource argument
show_parser.set_defaults(func=show) # Set the function to show
args = parser.parse_args() # Parse the arguments
if hasattr(args, "func"): # Check if the function is defined
args.func(args) # Execute the function
else:
parser.print_help() # Print the help message if the function is not defined
if __name__ == "__main__":
main()