Add support for guessing appropriate parameters for clusters + apps

Azimuth/cluster_types.py

@@ -1,7 +1,11 @@
 import typing as t
 
+from robot.api import logger
 from robot.api.deco import keyword
 
+from .external_ips import ExternalIpKeywords
+from .sizes import SizeKeywords
+
 
 class ClusterTypeKeywords:
     """
@@ -20,7 +24,7 @@ def list_cluster_types(self) -> t.List[t.Dict[str, t.Any]]:
         Lists cluster types using the active client.
         """
         return list(self._resource.list())
-    
+
     @keyword
     def fetch_cluster_type(self, name: str) -> t.Dict[str, t.Any]:
         """
@@ -34,3 +38,57 @@ def find_cluster_type_by_name(self, name: str) -> t.Dict[str, t.Any]:
         Fetches a cluster type by name using the active client.
         """
         return self.fetch_cluster_type(name)
+
+    @keyword
+    def get_defaults_for_cluster_type(
+        self,
+        cluster_type: t.Dict[str, t.Any]
+    ) -> t.Dict[str, t.Any]:
+        """
+        Returns the default parameters for a cluster type.
+        """
+        return {
+            param["name"]: param["default"]
+            for param in cluster_type["parameters"]
+            if param.get("default") is not None
+        }
+
+    def _guess_size(self, param):
+        # We want to find the smallest size that fulfils the constraints
+        # We sort by RAM, then CPUs, then disk to find the smallest
+        options = param.get("options", {})
+        try:
+            return SizeKeywords(self._ctx).find_smallest_size_with_resources(**options)
+        except ValueError:
+            return None
+
+    def _guess_ip(self, param):
+        return ExternalIpKeywords(self._ctx).find_free_or_allocate_external_ip().id
+
+    @keyword
+    def guess_parameter_values_for_cluster_type(
+        self,
+        cluster_type: t.Dict[str, t.Any]
+    ) -> t.Dict[str, t.Any]:
+        """
+        Attempts to guess suitable parameter values for a cluster type based on the
+        parameter types and constraints.
+        """
+        params = {}
+        # Try to fill in the remaining parameters by guessing
+        for param in cluster_type["parameters"]:
+            name = param["name"]
+            guess = None
+            # We don't currently handle all the types
+            if param.get("default") is not None:
+                guess = param["default"]
+            elif param["kind"] == "cloud.size":
+                guess = self._guess_size(param)
+            elif param["kind"] == "cloud.ip":
+                guess = self._guess_ip(param)
+            # Emit a warning if we weren't able to make a guess
+            if guess is not None:
+                params[name] = guess
+            else:
+                logger.warn(f"unable to make a guess for param '{name}'")
+        return params

Azimuth/external_ips.py

@@ -0,0 +1,70 @@
+import typing as t
+
+from robot.api.deco import keyword
+
+
+class ExternalIpKeywords:
+    """
+    Keywords for interacting with external IPs.
+    """
+    def __init__(self, ctx):
+        self._ctx = ctx
+
+    @property
+    def _resource(self):
+        return self._ctx.client.external_ips()
+
+    @keyword
+    def list_external_ips(self) -> t.List[t.Dict[str, t.Any]]:
+        """
+        Lists available external IPs using the active client.
+        """
+        return list(self._resource.list())
+
+    @keyword
+    def fetch_external_ip(self, id: str) -> t.Dict[str, t.Any]:
+        """
+        Fetches an external IP by id using the active client.
+        """
+        return self._resource.fetch(id)
+
+    @keyword
+    def find_external_ip_by_address(self, ip_address: str) -> t.Dict[str, t.Any]:
+        """
+        Finds an external IP by IP address using the active client.
+        """
+        try:
+            return next(ip for ip in self._resource.list() if ip.external_ip == ip_address)
+        except StopIteration:
+            raise ValueError(f"no external IP with address '{ip_address}'")
+
+    @keyword
+    def find_free_external_ip(self) -> t.Dict[str, t.Any]:
+        """
+        Searches for an external IP that is allocated but not assigned and returns it.
+
+        If no such IP exists, an exception is raised.
+        """
+        try:
+            return next(ip for ip in self._resource.list() if ip.machine is None)
+        except StopIteration:
+            raise ValueError("unable to find an unassigned external IP address")
+
+    @keyword
+    def allocate_external_ip(self) -> t.Dict[str, t.Any]:
+        """
+        Allocates a new external IP and returns it.
+        """
+        return self._resource.create({})
+
+    @keyword
+    def find_free_or_allocate_external_ip(self) -> t.Dict[str, t.Any]:
+        """
+        Searches for an external IP that is allocated but not assigned and returns it.
+
+        If no such IP exists, a new one is allocated.
+        """
+        try:
+            return self.find_free_external_ip()
+        except ValueError:
+            return self.allocate_external_ip()

Azimuth/kubernetes_cluster_templates.py

@@ -20,7 +20,7 @@ def _resource(self):
     def list_kubernetes_cluster_templates(
         self,
         *,
-        tags: t.Optional[t.List[str]],
+        tags: t.Optional[t.List[str]] = None,
         include_deprecated: bool = True
     ) -> t.List[t.Dict[str, t.Any]]:
         """
@@ -30,7 +30,7 @@ def list_kubernetes_cluster_templates(
             template
             for template in self._resource.list()
             if (
-                set(tags).issubset(template.tags) and
+                set(tags or []).issubset(template.tags) and
                 (include_deprecated or not template.deprecated)
             )
         )
@@ -57,22 +57,20 @@ def find_latest_kubernetes_cluster_template(
         self,
         *,
         constraints: str = ">=0.0.0",
-        tags: t.Optional[t.List[str]],
+        tags: t.Optional[t.List[str]] = None,
         include_deprecated: bool = False
     ) -> t.Dict[str, t.Any]:
         """
-        Finds the Kubernetes template with the most recent version that matches the given tags.
+        Finds the Kubernetes template with the most recent version that matches the constraints.
         """
+        templates = self.list_kubernetes_cluster_templates(
+            tags = tags,
+            include_deprecated = include_deprecated
+        )
         version_range = easysemver.Range(constraints)
         latest_version = None
         latest_template = None
-        for template in self._resource.list():
-            # If the template is deprecated, skip it unless explicitly included
-            if not include_deprecated and template.deprecated:
-                continue
-            # All the given tags must be in the template tags
-            if not set(tags).issubset(template.tags):
-                continue
+        for template in templates:
             current_version = easysemver.Version(template.kubernetes_version)
             # The version must match the constraints
             if current_version not in version_range:
@@ -83,5 +81,5 @@ def find_latest_kubernetes_cluster_template(
             latest_version = current_version
             latest_template = template
         if not latest_template:
-            raise ValueError("No template matching conditions")
+            raise ValueError("no template matching conditions")
         return latest_template

Azimuth/main.py

@@ -8,6 +8,7 @@
 
 from .cluster_types import ClusterTypeKeywords
 from .clusters import ClusterKeywords
+from .external_ips import ExternalIpKeywords
 from .kubernetes_app_templates import KubernetesAppTemplateKeywords
 from .kubernetes_apps import KubernetesAppKeywords
 from .kubernetes_cluster_templates import KubernetesClusterTemplateKeywords
@@ -30,6 +31,7 @@ def __init__(self):
         super().__init__([
             ClusterTypeKeywords(self),
             ClusterKeywords(self),
+            ExternalIpKeywords(self),
             KubernetesAppTemplateKeywords(self),
             KubernetesAppKeywords(self),
             KubernetesClusterTemplateKeywords(self),

Azimuth/sizes.py

@@ -37,3 +37,32 @@ def find_size_by_name(self, name: str) -> t.Dict[str, t.Any]:
             return next(c for c in self._resource.list() if c.name == name)
         except StopIteration:
             raise ValueError(f"no size with name '{name}'")
+
+    @keyword
+    def find_smallest_size_with_resources(
+        self,
+        *,
+        min_cpus: int = 0,
+        min_ram: int = 0,
+        min_disk: int = 0,
+        min_ephemeral_disk: int = 0,
+        sort_by: str = "ram,cpus,disk,ephemeral_disk"
+    ) -> t.Dict[str, t.Any]:
+        """
+        Finds the smallest size that fulfils the specified resource requirements.
+        """
+        candidates = (
+            size
+            for size in self._resource.list()
+            if (
+                size.cpus >= min_cpus and
+                size.ram >= min_ram and
+                size.disk >= min_disk and
+                size.ephemeral_disk >= min_ephemeral_disk
+            )
+        )
+        key_func = lambda size: tuple(getattr(size, attr) for attr in sort_by.split(","))
+        try:
+            return next(iter(sorted(candidates, key = key_func)))
+        except StopIteration:
+            raise ValueError("no available sizes fulfilling resource requirements")

Azimuth/zenith.py

@@ -1,5 +1,7 @@
 import time
 
+import httpx
+
 from robot.api.deco import keyword
 
 from selenium import webdriver
@@ -22,7 +24,7 @@ def __call__(self, driver):
         current_url = driver.current_url
         # If the page is not fully loaded, we are done
         ready_state = driver.execute_script("return document.readyState")
-        if ready_state.lower() != "complete":
+        if not ready_state or ready_state.lower() != "complete":
             return False
         # If the URL has changed, then it is not stable
         if not self._last_url or self._last_url != current_url:
@@ -43,7 +45,7 @@ def __init__(self, expected_title):
     def __call__(self, driver):
         # Wait until the page is fully loaded
         ready_state = driver.execute_script("return document.readyState")
-        if ready_state.lower() != "complete":
+        if not ready_state or ready_state.lower() != "complete":
             return False
         return self._expected_title in driver.title
 
@@ -57,17 +59,15 @@ def __init__(self, ctx):
         self._driver = None
 
     @keyword
-    def open_browser(self, browser: str = "firefox", headless: bool = True):
+    def open_browser(self, browser: str = "firefox"):
         """
         Opens the specified browser.
         """
         if self._driver is not None:
             self.close_browser()
 
         if browser == "firefox":
-            options = webdriver.FirefoxOptions()
-            options.headless = headless
-            self._driver = webdriver.Firefox(options = options)
+            self._driver = webdriver.Firefox()
         else:
             raise AssertionError(f"browser is not supported - {browser}")
 
@@ -110,8 +110,19 @@ def open_zenith_service(self, fqdn: str, authenticate: bool = True):
             self.authenticate_browser()
         # Use the scheme from the Azimuth base URL
         scheme = self._ctx.client.base_url.scheme
+        zenith_url = f"{scheme}://{fqdn}?kc_idp_hint=azimuth"
+        # Wait for the Zenith URL to return something other than a 404, 502 or 503
+        # These statuses could occur while the Zenith tunnel is establishing
+        while True:
+            response = httpx.get(zenith_url)
+            if response.status_code < 400:
+                break
+            elif response.status_code in [404, 502, 503]:
+                continue
+            else:
+                response.raise_for_status()
         # Visit the Zenith URL and wait for it to stabilise
-        self._driver.get(f"{scheme}://{fqdn}?kc_idp_hint=azimuth")
+        self._driver.get(zenith_url)
         WebDriverWait(self._driver, 86400).until(url_is_stable())
 
     @keyword

setup.cfg

@@ -15,6 +15,7 @@ packages = find:
 install_requires =
     azimuth-sdk
     easysemver
+    httpx
     jsonschema-default
     robotframework-pythonlibcore
     selenium