CLOUDP-64640: fix deploy operator (#55)

Author: Nikolas De Giorgis

scripts/dev/build_and_deploy_operator.py

@@ -1,16 +1,16 @@
 import io
 import os
-import time
 from typing import Dict, Optional
 
 import yaml
 from kubernetes import client, config
-from kubernetes.client.rest import ApiException
 
 from dev_config import DevConfig, load_config
 from dockerfile_generator import render
 from dockerutil import build_and_push_image
 
+from k8sutil import wait_for_condition, ignore_if_already_exists, ignore_if_doesnt_exist
+
 
 def _load_operator_service_account() -> Optional[Dict]:
     return load_yaml_from_file("deploy/service_account.yaml")
@@ -51,6 +51,17 @@ def _ensure_crds():
         lambda: crdv1.delete_custom_resource_definition("mongodb.mongodb.com")
     )
 
+    # Make sure that the CRD has being deleted before trying to create it again
+    if not wait_for_condition(
+        lambda: crdv1.list_custom_resource_definition(
+            field_selector="metadata.name==mongodb.mongodb.com"
+        ),
+        lambda crd_list: len(crd_list.items) == 0,
+        timeout=5,
+        sleep_time=0.5,
+    ):
+        raise Exception("Execution timed out while waiting for the CRD to be deleted")
+
     # TODO: fix this, when calling create_custom_resource_definition, we get the error
     # ValueError("Invalid value for `conditions`, must not be `None`")
     # but the crd is still successfully created
@@ -70,31 +81,6 @@ def build_and_push_operator(repo_url: str, tag: str, path: str):
     return build_and_push_image(repo_url, tag, path, "operator")
 
 
-def _ignore_error_codes(fn, codes):
-    try:
-        fn()
-    except ApiException as e:
-        if e.status not in codes:
-            raise
-
-
-def ignore_if_already_exists(fn):
-    """
-    ignore_if_already_exists accepts a function and calls it,
-    ignoring an Kubernetes API conflict errors
-    """
-
-    return _ignore_error_codes(fn, [409])
-
-
-def ignore_if_doesnt_exist(fn):
-    """
-    ignore_if_doesnt_exist accepts a function and calls it,
-    ignoring an Kubernetes API not found errors
-    """
-    return _ignore_error_codes(fn, [404])
-
-
 def deploy_operator():
     """
     deploy_operator ensures the CRDs are created, and als creates all the required ServiceAccounts, Roles

scripts/dev/e2e.py

@@ -4,10 +4,13 @@
 from build_and_deploy_operator import (
     build_and_push_operator,
     deploy_operator,
+    load_yaml_from_file,
+)
+from k8sutil import (
+    wait_for_condition,
     ignore_if_doesnt_exist,
     ignore_if_already_exists,
-    load_yaml_from_file,
-)  # TODO: put these function somewhere else
+)
 from dockerutil import build_and_push_image
 from typing import Dict, Optional
 from dev_config import load_config
@@ -140,9 +143,35 @@ def create_test_runner_pod(test: str):
     dev_config = load_config()
     corev1 = client.CoreV1Api()
     pod_body = _get_testrunner_pod_body(test)
+
+    if not wait_for_condition(
+        lambda: corev1.list_namespaced_pod(
+            dev_config.namespace, field_selector=f"metadata.name=={TEST_RUNNER_NAME}"
+        ),
+        lambda pod_list: len(pod_list.items) == 0,
+        timeout=10,
+        sleep_time=0.5,
+    ):
+
+        raise Exception(
+            "Execution timed out while waiting for the existing pod to be deleted"
+        )
+
     return corev1.create_namespaced_pod(dev_config.namespace, body=pod_body)
 
 
+def wait_for_pod_to_be_running(corev1, name, namespace):
+    print("Waiting for pod to be running")
+    if not wait_for_condition(
+        lambda: corev1.read_namespaced_pod(name, namespace),
+        lambda pod: pod.status.phase == "Running",
+        sleep_time=5,
+        timeout=50,
+        exceptions_to_ignore=ApiException,
+    ):
+        raise Exception("Pod never got into Running state!")
+
+
 def _get_testrunner_pod_body(test: str) -> Dict:
     dev_config = load_config()
     return {
@@ -173,19 +202,6 @@ def _get_testrunner_pod_body(test: str) -> Dict:
     }
 
 
-def wait_for_pod_to_be_running(corev1, name, namespace):
-    print("Waiting for pod to be running")
-    for i in range(10):
-        try:
-            pod = corev1.read_namespaced_pod(name, namespace)
-            if pod.status.phase == "Running":
-                return True
-        except ApiException as e:
-            pass
-        time.sleep(5)
-    raise Exception("Pod never got into Running state!")
-
-
 def parse_args():
     parser = argparse.ArgumentParser()
     parser.add_argument("--test", help="Name of the test to run")
@@ -222,6 +238,7 @@ def main():
 
     pod = create_test_runner_pod(args.test)
     corev1 = client.CoreV1Api()
+
     wait_for_pod_to_be_running(corev1, TEST_RUNNER_NAME, dev_config.namespace)
 
     # stream all of the pod output as the pod is running

scripts/dev/k8sutil.py

@@ -0,0 +1,70 @@
+import time
+
+from kubernetes.client.rest import ApiException
+
+# time to sleep between retries
+SLEEP_TIME = 2
+# no timeout (loop forever)
+INFINITY = -1
+
+
+def _current_milliseconds() -> int:
+    return int(round(time.time() * 1000))
+
+
+def wait_for_condition(
+    fn,
+    condition,
+    exceptions_to_ignore=None,
+    codes_to_ignore=None,
+    sleep_time=SLEEP_TIME,
+    timeout=INFINITY,
+) -> bool:
+    """
+    wait_for_condition accepts a function fn and a function condition,
+    it periodically calls the function fn and then applies the condition function on the result
+    until it returns True or we reach timeout
+
+    exceptions_to_ignore is a tuple of Exceptions to ignore is raised by the call to fn
+    If ApiException is not ignored, if raised by the call to fn codes in codes_to_ignore are ignored
+    """
+    start_time = _current_milliseconds()
+    end = start_time + (timeout * 1000)
+
+    while _current_milliseconds() < end or timeout <= 0:
+        res = None
+        try:
+            res = _ignore_error_codes(fn, codes_to_ignore)
+        except exceptions_to_ignore:
+            pass
+        if res is not None and condition(res):
+            return True
+
+        time.sleep(sleep_time)
+
+    return False
+
+
+def _ignore_error_codes(fn, codes):
+    try:
+        return fn()
+    except ApiException as e:
+        if e.status not in codes:
+            raise
+
+
+def ignore_if_already_exists(fn):
+    """
+    ignore_if_already_exists accepts a function and calls it,
+    ignoring an Kubernetes API conflict errors
+    """
+
+    return _ignore_error_codes(fn, [409])
+
+
+def ignore_if_doesnt_exist(fn):
+    """
+    ignore_if_doesnt_exist accepts a function and calls it,
+    ignoring an Kubernetes API not found errors
+    """
+    return _ignore_error_codes(fn, [404])