diff --git a/kernelci/runtime/kubernetes.py b/kernelci/runtime/kubernetes.py index 4548304b78..aa79a57c2f 100644 --- a/kernelci/runtime/kubernetes.py +++ b/kernelci/runtime/kubernetes.py @@ -10,6 +10,7 @@ import random import re import string +import time import kubernetes @@ -29,6 +30,12 @@ class Kubernetes(Runtime): # Default timeout for Kubernetes API calls in seconds # This prevents indefinite hangs on network issues DEFAULT_API_TIMEOUT = 30 + # Seconds to skip a cluster after it failed, before probing it again + DEAD_CONTEXT_TTL = 600 + # HTTP statuses that mean the cluster is unusable, not just busy + CLUSTER_ERROR_STATUSES = (401, 403, 500, 502, 503, 504) + # Recently failed clusters, shared by all instances: name -> expiry + _dead_contexts = {} def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) @@ -58,8 +65,47 @@ def generate(self, job, params): params["k8s_job_name"] = k8s_job_name return template.render(params) + @classmethod + def _is_dead(cls, ctxname): + expiry = cls._dead_contexts.get(ctxname) + if expiry is None: + return False + if time.monotonic() >= expiry: + del cls._dead_contexts[ctxname] + logger.info("k8s cluster %s: probing again", ctxname) + return False + return True + + @classmethod + def _mark_dead(cls, ctxname): + if ctxname not in cls._dead_contexts: + logger.error( + "k8s cluster %s: disabled for %ds", + ctxname, + cls.DEAD_CONTEXT_TTL, + ) + cls._dead_contexts[ctxname] = time.monotonic() + cls.DEAD_CONTEXT_TTL + + @classmethod + def _is_cluster_error(cls, error): + """Tell apart an unusable cluster from a problem with the job""" + if isinstance(error, kubernetes.utils.FailToCreateError): + return any( + cls._is_cluster_error(api_error) + for api_error in error.api_exceptions + ) + if isinstance(error, kubernetes.client.rest.ApiException): + return error.status in cls.CLUSTER_ERROR_STATUSES + # Connection errors, timeouts, credential plugin failures + return True + def _fetch_load(self, ctxname): - """Fetch load with retry and workaround due repeating errors""" + """Fetch load with retry and workaround due repeating errors + + Returns None if the cluster is unreachable or recently was. + """ + if self._is_dead(ctxname): + return None kubernetes.config.load_kube_config(context=ctxname) core_v1 = kubernetes.client.CoreV1Api() pods = None @@ -83,6 +129,9 @@ def _fetch_load(self, ctxname): attempt + 1, error, ) + # Credentials are rejected, retrying will not help + if error.status in (401, 403): + break continue except Exception as error: last_error = error @@ -97,11 +146,12 @@ def _fetch_load(self, ctxname): if not pods: logger.error( - "k8s cluster %s: failed to list pods after 3 attempts, last error: %s", + "k8s cluster %s: failed to list pods, last error: %s", ctxname, last_error, ) - return 1000 + self._mark_dead(ctxname) + return None load = len([pod for pod in pods.items if pod.status.phase == "Pending"]) return load @@ -111,26 +161,57 @@ def _get_clusters_load(self): state)""" load = {} for ctxname in self.config.context: - load[ctxname] = self._fetch_load(ctxname) + ctx_load = self._fetch_load(ctxname) + if ctx_load is None: + continue + load[ctxname] = ctx_load # We found free cluster, return it, no need to check others - if load[ctxname] == 0: + if ctx_load == 0: break return load + def _get_candidates(self): + """Clusters to try submitting to, best first""" + if not isinstance(self.config.context, list): + return [self.config.context] + load = self._get_clusters_load() + candidates = sorted(load, key=load.get) + # Live clusters skipped by the early exit, as fallback + candidates += [ + ctxname + for ctxname in self.config.context + if ctxname not in load and not self._is_dead(ctxname) + ] + if not candidates: + # Try them all anyway rather than drop the job + logger.error( + "All k8s clusters unreachable: %s", + ", ".join(self.config.context), + ) + candidates = list(self.config.context) + return candidates + def submit(self, job_path): # if context is array, we have multiple k8s build clusters # TBD: Implement caching to not check load for each job? - if isinstance(self.config.context, list): - # get the cluster with the least load - load = self._get_clusters_load() - self.kcontext = min(load, key=load.get) - if load[self.kcontext] >= 1000: - logger.error("All k8s clusters unreachable, loads: %s", load) - else: - self.kcontext = self.config.context - kubernetes.config.load_kube_config(context=self.kcontext) - client = kubernetes.client.ApiClient() - return kubernetes.utils.create_from_yaml(client, job_path) + candidates = self._get_candidates() + last_error = None + for ctxname in candidates: + self.kcontext = ctxname + try: + kubernetes.config.load_kube_config(context=ctxname) + client = kubernetes.client.ApiClient() + return kubernetes.utils.create_from_yaml(client, job_path) + except Exception as error: # pylint: disable=broad-except + if not self._is_cluster_error(error): + raise + last_error = error + logger.warning( + "k8s cluster %s: submit failed: %s", ctxname, error + ) + if len(candidates) > 1: + self._mark_dead(ctxname) + raise last_error def get_job_id(self, job_object): return job_object[0][0].metadata.labels["job-name"]