diff --git a/src/scheduler/kubernetes/scheduler.py b/src/scheduler/kubernetes/scheduler.py index 438f540a9..f0ba37e60 100644 --- a/src/scheduler/kubernetes/scheduler.py +++ b/src/scheduler/kubernetes/scheduler.py @@ -5,6 +5,7 @@ import random import json import copy +import threading from datetime import datetime import requests @@ -547,6 +548,74 @@ def __init__(self, args): self.function_controller = FunctionInvocationController(args) self.pipeline_controller = PipelineInvocationController(args) + # K8s state cache — refreshed every 10s by background thread. + # Main scheduling loop reads from cache to avoid blocking on K8s API + # calls (which can stall for up to 10s under load and cause job queuing + # delays of hours when multiplied across many scheduler ticks). + self._k8s_lock = threading.Lock() + self._k8s_nodes = [] # items from GET /api/v1/nodes + self._k8s_pipelines = [] # items from GET ibpipelineinvocations + self._start_k8s_refresh_threads() + + def _start_k8s_refresh_threads(self): + # Fix: run initial synchronous fetch so caches are populated before + # the first scheduler tick (avoids update_cluster_state skipping the + # DB write on startup when the cluster row may not exist yet). + try: + self._refresh_k8s_nodes() + except Exception as e: + self.logger.exception(e) + try: + self._refresh_k8s_pipelines() + except Exception as e: + self.logger.exception(e) + + # Fix: run nodes and pipelines in separate threads so a slow/hung + # nodes API call does not delay the pipelines refresh (and vice versa). + def nodes_loop(): + while True: + time.sleep(10) + try: + self._refresh_k8s_nodes() + except Exception as e: + self.logger.exception(e) + + def pipelines_loop(): + while True: + time.sleep(10) + try: + self._refresh_k8s_pipelines() + except Exception as e: + self.logger.exception(e) + + for target in (nodes_loop, pipelines_loop): + t = threading.Thread(target=target, daemon=True) + t.start() + + def _refresh_k8s_nodes(self): + h = {'Authorization': 'Bearer %s' % self.args.token} + r = requests.get(self.args.api_server + '/api/v1/nodes', headers=h, timeout=10) + # Fix: raise on HTTP error so a non-200 response does not overwrite + # the previously valid cache with an empty list. + r.raise_for_status() + items = r.json().get('items', []) + with self._k8s_lock: + self._k8s_nodes = items + + def _refresh_k8s_pipelines(self): + h = {'Authorization': 'Bearer %s' % self.args.token} + r = requests.get( + self.args.api_server + '/apis/core.infrabox.net/v1alpha1/namespaces/%s/ibpipelineinvocations' % self.namespace, + headers=h, + timeout=10 + ) + # Fix: raise on HTTP error so a non-200 response does not overwrite + # the previously valid cache with an empty list. + r.raise_for_status() + items = r.json().get('items', []) + with self._k8s_lock: + self._k8s_pipelines = items + def handle_function_invocations(self): self.logger.info("handle function invocations") self.function_controller.handle() @@ -1069,16 +1138,13 @@ def handle_cron_jobs(self): def handle_orphaned_jobs(self): self.logger.info("handle orphaned jobs") - h = {'Authorization': 'Bearer %s' % self.args.token} - r = requests.get(self.args.api_server + '/apis/core.infrabox.net/v1alpha1/namespaces/%s/ibpipelineinvocations' % self.namespace, - headers=h, - timeout=10) - data = r.json() + with self._k8s_lock: + items = list(self._k8s_pipelines) - if 'items' not in data: + if not items: return - for j in data['items']: + for j in items: if 'metadata' not in j: continue @@ -1257,18 +1323,16 @@ def update_cluster_state(self): root_url = os.environ['INFRABOX_ROOT_URL'] - h = {'Authorization': 'Bearer %s' % self.args.token} - r = requests.get(self.args.api_server + '/api/v1/nodes', - headers=h, - timeout=10) - data = r.json() + with self._k8s_lock: + items = list(self._k8s_nodes) + + if not items: + return memory = 0 cpu = 0 nodes = 0 - items = data.get('items', []) - for i in items: metadata = i.get('metadata', {}) l = metadata.get('labels', {}) @@ -1345,7 +1409,7 @@ def run(self): self.handle() self.conn.close() - time.sleep(1) + time.sleep(3) def main(): # Arguments