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)
class K8sSystem(auto_chaos.base_system.BaseSystem):
 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.
client
namespace
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.