+ async def get_services(
+ self, cluster_uuid: str, kdu_instance: str, namespace: str
+ ) -> list:
+ """Return a list of services of a kdu_instance"""
+
+ credentials = self.get_credentials(cluster_uuid=cluster_uuid)
+
+ config_path = "/tmp/{}".format(cluster_uuid)
+ config_file = "{}/config".format(config_path)
+
+ if not os.path.exists(config_path):
+ os.makedirs(config_path)
+ with open(config_file, "w") as f:
+ f.write(credentials)
+
+ kubectl = Kubectl(config_file=config_file)
+ return kubectl.get_services(
+ field_selector="metadata.namespace={}".format(kdu_instance)
+ )
+
+ async def get_service(
+ self, cluster_uuid: str, service_name: str, namespace: str
+ ) -> object:
+ """Return data for a specific service inside a namespace"""
+
+ credentials = self.get_credentials(cluster_uuid=cluster_uuid)
+
+ config_path = "/tmp/{}".format(cluster_uuid)
+ config_file = "{}/config".format(config_path)
+
+ if not os.path.exists(config_path):
+ os.makedirs(config_path)
+ with open(config_file, "w") as f:
+ f.write(credentials)
+
+ kubectl = Kubectl(config_file=config_file)
+
+ return kubectl.get_services(
+ field_selector="metadata.name={},metadata.namespace={}".format(
+ service_name, namespace
+ )
+ )[0]
+