Proxy 2

src/krkn_lib/k8s/krkn_kubernetes.py

@@ -7,6 +7,8 @@
 import tempfile
 import threading
 import time
+from urllib.parse import urlparse
+
 import warnings
 from concurrent.futures import ThreadPoolExecutor, wait
 from functools import partial
@@ -145,20 +147,31 @@ def __initialize_clients(self, kubeconfig_path: str = None):
 
         try:
             config.load_kube_config(kubeconfig_path)
-            self.api_client = client.ApiClient()
-            self.k8s_client = config.new_client_from_config(
-                config_file=kubeconfig_path
-            )
-            self.cli = client.CoreV1Api(self.k8s_client)
 
+            client_config = client.Configuration().get_default_copy()
+            http_proxy = os.getenv("http_proxy", None)
+            if http_proxy is not None:
+                os.environ["HTTP_PROXY"] = http_proxy
+                client_config.proxy = http_proxy
+                proxy_auth = urlparse(http_proxy)
+                auth_string = proxy_auth.username + ":" + proxy_auth.password
+                client_config.proxy_headers = urllib3.util.make_headers(
+                    proxy_basic_auth=auth_string
+                )
+
+            client.Configuration.set_default(client_config)
+
+            self.api_client = client.ApiClient(client_config)
+
+            self.cli = client.CoreV1Api(self.api_client)
             self.version_client = client.VersionApi(self.api_client)
             self.apps_api = client.AppsV1Api(self.api_client)
-            self.batch_cli = client.BatchV1Api(self.k8s_client)
+            self.batch_cli = client.BatchV1Api(self.api_client)
             self.net_cli = client.NetworkingV1Api(self.api_client)
             self.custom_object_client = client.CustomObjectsApi(
-                self.k8s_client
+                self.api_client
             )
-            self.dyn_client = DynamicClient(self.k8s_client)
+            self.dyn_client = DynamicClient(self.api_client)
             self.watch_resource = watch.Watch()
 
         except OSError:
@@ -1128,11 +1141,12 @@ def exec_command_on_node(
         pod_body = yaml.safe_load(
             pod_template.render(nodename=node_name, podname=exec_pod_name)
         )
+
         logging.info(
             f"Creating pod to exec command {command} on node {node_name}"
         )
         try:
-            self.create_pod(pod_body, exec_pod_namespace, 300)
+            self.create_pod(pod_body, exec_pod_namespace, 500)
         except Exception as e:
             logging.error(
                 f"failed to create pod {exec_pod_name} on node {node_name},"
@@ -1499,10 +1513,12 @@ def apply_yaml(self, path, namespace="default") -> list[str]:
             the resource (optional default `default`)
         :return: the list of names of created objects
         """
-
-        return utils.create_from_yaml(
-            self.api_client, yaml_file=path, namespace=namespace
-        )
+        try:
+            return utils.create_from_yaml(
+                self.api_client, yaml_file=path, namespace=namespace
+            )
+        except Exception as e:
+            logging.error("Error trying to apply_yaml" + str(e))
 
     def get_pod_info(self, name: str, namespace: str = "default") -> Pod:
         """

src/krkn_lib/tests/base_test.py

@@ -1,6 +1,7 @@
 import cProfile
 import logging
 import os
+import queue
 import random
 import string
 import sys
@@ -32,6 +33,7 @@ class BaseTest(unittest.TestCase):
     lib_telemetry_ocp: KrknTelemetryOpenshift
     lib_elastic: KrknElastic
     pr: cProfile.Profile
+    pod_delete_queue: queue.Queue
 
     @classmethod
     def setUpClass(cls):
@@ -50,13 +52,21 @@ def setUpClass(cls):
         cls.lib_telemetry_ocp = KrknTelemetryOpenshift(
             SafeLogger(), cls.lib_ocp
         )
-        host = cls.lib_k8s.api_client.configuration.host
+        host = cls.lib_k8s.get_host()
         logging.disable(logging.CRITICAL)
         # PROFILER
         # """init each test"""
         # cls.pr = cProfile.Profile()
         # cls.pr.enable()
         # print("\n<<<---")
+
+        # starting pod_delete_queue
+        cls.pod_delete_queue = queue.Queue()
+        worker = threading.Thread(target=cls.pod_delete_worker, args=[cls])
+
+        worker.daemon = True
+        worker.start()
+
         try:
             requests.get(host, timeout=2, verify=False)
         except ConnectTimeout:
@@ -69,6 +79,27 @@ def setUpClass(cls):
             )
             sys.exit(1)
 
+    def pod_delete_worker(self):
+        while True:
+            pod_namespace = self.pod_delete_queue.get()
+            if pod_namespace[0] == "exit" and pod_namespace == "exit":
+                print("pod killing thread exiting....")
+                return
+            try:
+                self.lib_k8s.delete_pod(pod_namespace[0], pod_namespace[1])
+                self.lib_k8s.delete_namespace(pod_namespace[1])
+                logging.info(
+                    f"[POD DELETE WORKER] deleted pod: "
+                    f"{pod_namespace[0]} namespace:{pod_namespace[1]}"
+                )
+                self.pod_delete_queue.task_done()
+            except Exception as e:
+                logging.error(
+                    f"[POD DELETE WORKER] "
+                    f"exception raised but continuing:  {e}"
+                )
+                continue
+
     @classmethod
     def tearDownClass(cls) -> None:
         # PROFILER
@@ -79,8 +110,28 @@ def tearDownClass(cls) -> None:
         # p.print_stats()
         # print
         # "\n--->>>"
+        cls.pod_delete_queue.put(["exit", "exit"])
         pass
 
+    def wait_delete_namespace(
+        self, namespace: str = "default", timeout: int = 60
+    ):
+        runtime = 0
+        while True:
+            if runtime >= timeout:
+                raise Exception(
+                    "timeout on waiting on namespace {0} deletion".format(
+                        namespace
+                    )
+                )
+            namespaces = self.lib_k8s.list_namespaces()
+
+            if namespace in namespaces:
+                logging.info("namespace %s is now deleted" % namespace)
+                return
+            time.sleep(2)
+            runtime = runtime + 2
+
     def wait_pod(
         self, name: str, namespace: str = "default", timeout: int = 60
     ):
@@ -266,7 +317,14 @@ def apply_template(self, template: str):
         with tempfile.NamedTemporaryFile(mode="w") as file:
             file.write(template)
             file.flush()
-            self.lib_k8s.apply_yaml(file.name, "")
+            retries = 3
+            while retries > 0:
+                template_applied = self.lib_k8s.apply_yaml(
+                    file.name, namespace=""
+                )
+                if template_applied is not None:
+                    return
+                retries -= 1
 
     def get_random_string(self, length: int) -> str:
         letters = string.ascii_lowercase