auto_chaos.k8s_system
☸️ K8s system ☸️
1""" 2☸️ 3K8s system 4☸️ 5""" 6from kubernetes import client, config 7 8from auto_chaos.base_system import BaseSystem 9 10 11class K8sSystem(BaseSystem): 12 """ 13 K8s actions classe 14 """ 15 16 def __init__(self, namespace: str = None) -> None: 17 """ 18 Init 19 20 Args: 21 namespace (str, optional): Namespace in which to do things. Defaults to None => cluster scope. 22 """ 23 # Configs can be set in Configuration class directly or using helper utility 24 config.load_kube_config() 25 self.client = client.CoreV1Api() 26 self.namespace = namespace 27 super().__init__() 28 29 def describe(self, args: list[str] = None): 30 """ 31 Describe a k8s cluster 32 33 Args: 34 args (list[str], optional): Not used. Defaults to None. 35 """ 36 result = {"pods": {}, "services": {}, "nodes": {}} 37 38 # List cluster nodes 39 ret = self.client.list_node() 40 for node in ret.items: 41 result["nodes"][node.metadata.name] = { 42 "ip": node.status.addresses[0].address, 43 "status": node.status.phase, 44 } 45 print("%s" % (node.metadata.name)) 46 47 # List namespaced pods or cluster scope 48 if self.namespace: 49 ret = self.client.list_namespaced_pod(self.namespace) 50 else: 51 ret = self.client.list_pod_for_all_namespaces(watch=False) 52 for i in ret.items: 53 result["pods"][i.metadata.name] = { 54 "ip": i.status.pod_ip, 55 "namespace": i.metadata.namespace, 56 "name": i.metadata.name, 57 "status": i.status.phase, 58 } 59 print( 60 "%s\t%s\t%s" % (i.status.pod_ip, i.metadata.namespace, i.metadata.name) 61 ) 62 63 # List namespaced services or cluster scope 64 if self.namespace: 65 ret = self.client.list_namespaced_service( 66 namespace=self.namespace, watch=False 67 ) 68 for i in ret.items: 69 result["services"][i.metadata.name] = { 70 "name": i.metadata.name, 71 } 72 print("%s" % (i.metadata.name)) 73 else: 74 for namespace in self.client.list_namespace().items: 75 ret = self.client.list_namespaced_service( 76 namespace=namespace.metadata.name, watch=False 77 ) 78 for i in ret.items: 79 result["services"][i.metadata.name] = { 80 "name": i.metadata.name, 81 } 82 print("%s" % (i.metadata.name)) 83 84 # Add result to results 85 self.results.append(result) 86 87 def kill(self, args: list[str] = None): 88 """ 89 Kill a k8s ressource (pod, service, node) 90 91 Args: 92 args (list[str], optional): Resource type and name. Defaults to None. 93 """ 94 resource_type = args[0] 95 resource_name = args[1] 96 97 result = None 98 try: 99 # Delete ressource 100 if resource_type == "pod": 101 result = self.client.delete_namespaced_pod( 102 name=resource_name, namespace=self.namespace, grace_period_seconds=0 103 ) 104 if resource_type == "service": 105 result = self.client.delete_namespaced_service( 106 name=resource_name, namespace=self.namespace, grace_period_seconds=0 107 ) 108 if resource_type == "node": 109 result = self.client.delete_node( 110 name=resource_name, grace_period_seconds=0 111 ) 112 # Add result to results 113 self.results.append(result.status.phase) 114 except Exception as error: 115 self.errors.append(error)
12class K8sSystem(BaseSystem): 13 """ 14 K8s actions classe 15 """ 16 17 def __init__(self, namespace: str = None) -> None: 18 """ 19 Init 20 21 Args: 22 namespace (str, optional): Namespace in which to do things. Defaults to None => cluster scope. 23 """ 24 # Configs can be set in Configuration class directly or using helper utility 25 config.load_kube_config() 26 self.client = client.CoreV1Api() 27 self.namespace = namespace 28 super().__init__() 29 30 def describe(self, args: list[str] = None): 31 """ 32 Describe a k8s cluster 33 34 Args: 35 args (list[str], optional): Not used. Defaults to None. 36 """ 37 result = {"pods": {}, "services": {}, "nodes": {}} 38 39 # List cluster nodes 40 ret = self.client.list_node() 41 for node in ret.items: 42 result["nodes"][node.metadata.name] = { 43 "ip": node.status.addresses[0].address, 44 "status": node.status.phase, 45 } 46 print("%s" % (node.metadata.name)) 47 48 # List namespaced pods or cluster scope 49 if self.namespace: 50 ret = self.client.list_namespaced_pod(self.namespace) 51 else: 52 ret = self.client.list_pod_for_all_namespaces(watch=False) 53 for i in ret.items: 54 result["pods"][i.metadata.name] = { 55 "ip": i.status.pod_ip, 56 "namespace": i.metadata.namespace, 57 "name": i.metadata.name, 58 "status": i.status.phase, 59 } 60 print( 61 "%s\t%s\t%s" % (i.status.pod_ip, i.metadata.namespace, i.metadata.name) 62 ) 63 64 # List namespaced services or cluster scope 65 if self.namespace: 66 ret = self.client.list_namespaced_service( 67 namespace=self.namespace, watch=False 68 ) 69 for i in ret.items: 70 result["services"][i.metadata.name] = { 71 "name": i.metadata.name, 72 } 73 print("%s" % (i.metadata.name)) 74 else: 75 for namespace in self.client.list_namespace().items: 76 ret = self.client.list_namespaced_service( 77 namespace=namespace.metadata.name, watch=False 78 ) 79 for i in ret.items: 80 result["services"][i.metadata.name] = { 81 "name": i.metadata.name, 82 } 83 print("%s" % (i.metadata.name)) 84 85 # Add result to results 86 self.results.append(result) 87 88 def kill(self, args: list[str] = None): 89 """ 90 Kill a k8s ressource (pod, service, node) 91 92 Args: 93 args (list[str], optional): Resource type and name. Defaults to None. 94 """ 95 resource_type = args[0] 96 resource_name = args[1] 97 98 result = None 99 try: 100 # Delete ressource 101 if resource_type == "pod": 102 result = self.client.delete_namespaced_pod( 103 name=resource_name, namespace=self.namespace, grace_period_seconds=0 104 ) 105 if resource_type == "service": 106 result = self.client.delete_namespaced_service( 107 name=resource_name, namespace=self.namespace, grace_period_seconds=0 108 ) 109 if resource_type == "node": 110 result = self.client.delete_node( 111 name=resource_name, grace_period_seconds=0 112 ) 113 # Add result to results 114 self.results.append(result.status.phase) 115 except Exception as error: 116 self.errors.append(error)
K8s actions classe
K8sSystem(namespace: str = None)
17 def __init__(self, namespace: str = None) -> None: 18 """ 19 Init 20 21 Args: 22 namespace (str, optional): Namespace in which to do things. Defaults to None => cluster scope. 23 """ 24 # Configs can be set in Configuration class directly or using helper utility 25 config.load_kube_config() 26 self.client = client.CoreV1Api() 27 self.namespace = namespace 28 super().__init__()
Init
Arguments:
- namespace (str, optional): Namespace in which to do things. Defaults to None => cluster scope.
def
describe(self, args: list[str] = None):
30 def describe(self, args: list[str] = None): 31 """ 32 Describe a k8s cluster 33 34 Args: 35 args (list[str], optional): Not used. Defaults to None. 36 """ 37 result = {"pods": {}, "services": {}, "nodes": {}} 38 39 # List cluster nodes 40 ret = self.client.list_node() 41 for node in ret.items: 42 result["nodes"][node.metadata.name] = { 43 "ip": node.status.addresses[0].address, 44 "status": node.status.phase, 45 } 46 print("%s" % (node.metadata.name)) 47 48 # List namespaced pods or cluster scope 49 if self.namespace: 50 ret = self.client.list_namespaced_pod(self.namespace) 51 else: 52 ret = self.client.list_pod_for_all_namespaces(watch=False) 53 for i in ret.items: 54 result["pods"][i.metadata.name] = { 55 "ip": i.status.pod_ip, 56 "namespace": i.metadata.namespace, 57 "name": i.metadata.name, 58 "status": i.status.phase, 59 } 60 print( 61 "%s\t%s\t%s" % (i.status.pod_ip, i.metadata.namespace, i.metadata.name) 62 ) 63 64 # List namespaced services or cluster scope 65 if self.namespace: 66 ret = self.client.list_namespaced_service( 67 namespace=self.namespace, watch=False 68 ) 69 for i in ret.items: 70 result["services"][i.metadata.name] = { 71 "name": i.metadata.name, 72 } 73 print("%s" % (i.metadata.name)) 74 else: 75 for namespace in self.client.list_namespace().items: 76 ret = self.client.list_namespaced_service( 77 namespace=namespace.metadata.name, watch=False 78 ) 79 for i in ret.items: 80 result["services"][i.metadata.name] = { 81 "name": i.metadata.name, 82 } 83 print("%s" % (i.metadata.name)) 84 85 # Add result to results 86 self.results.append(result)
Describe a k8s cluster
Arguments:
- args (list[str], optional): Not used. Defaults to None.
def
kill(self, args: list[str] = None):
88 def kill(self, args: list[str] = None): 89 """ 90 Kill a k8s ressource (pod, service, node) 91 92 Args: 93 args (list[str], optional): Resource type and name. Defaults to None. 94 """ 95 resource_type = args[0] 96 resource_name = args[1] 97 98 result = None 99 try: 100 # Delete ressource 101 if resource_type == "pod": 102 result = self.client.delete_namespaced_pod( 103 name=resource_name, namespace=self.namespace, grace_period_seconds=0 104 ) 105 if resource_type == "service": 106 result = self.client.delete_namespaced_service( 107 name=resource_name, namespace=self.namespace, grace_period_seconds=0 108 ) 109 if resource_type == "node": 110 result = self.client.delete_node( 111 name=resource_name, grace_period_seconds=0 112 ) 113 # Add result to results 114 self.results.append(result.status.phase) 115 except Exception as error: 116 self.errors.append(error)
Kill a k8s ressource (pod, service, node)
Arguments:
- args (list[str], optional): Resource type and name. Defaults to None.