From f6dbb5edca0b1ea5b393037f1c98170d6ecd6149 Mon Sep 17 00:00:00 2001 From: I515719 Date: Tue, 8 Sep 2026 16:58:39 +0800 Subject: [PATCH 1/2] feat(DM01-6251): decouple blocking K8s API calls from scheduler main loop Move GET /api/v1/nodes and GET ibpipelineinvocations out of the main scheduling loop into a daemon thread that refreshes every 10s. The main loop now reads from a lock-protected cache instead of making synchronous HTTP requests on every tick. Root cause: under K8s API load these calls blocked for up to 10s each, causing the 1s scheduler loop to actually run every 10-20s. With a backlog of queued jobs this produced queue-wait times of hours (confirmed: generator/di-embedded-tests waited 2.8h before being scheduled). Also increase main loop sleep from 1s to 3s to further reduce DB and K8s API pressure. abort/timeout response latency increases by at most 3s, which is acceptable. --- src/scheduler/kubernetes/scheduler.py | 69 +++++++++++++++++++++------ 1 file changed, 54 insertions(+), 15 deletions(-) diff --git a/src/scheduler/kubernetes/scheduler.py b/src/scheduler/kubernetes/scheduler.py index 438f540a9..5ac6e46f9 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,49 @@ 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_thread() + + def _start_k8s_refresh_thread(self): + def refresh_loop(): + while True: + 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) + time.sleep(10) + + t = threading.Thread(target=refresh_loop, 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) + 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 + ) + 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 +1113,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 +1298,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 +1384,7 @@ def run(self): self.handle() self.conn.close() - time.sleep(1) + time.sleep(3) def main(): # Arguments From 20021d21c05d359567f9b09348325b12f0f91fef Mon Sep 17 00:00:00 2001 From: I515719 Date: Tue, 8 Sep 2026 17:10:43 +0800 Subject: [PATCH 2/2] fix(DM01-6251): fix three issues found in code review - Add r.raise_for_status() before r.json() in both refresh methods so HTTP error responses (401, 503) raise an exception and leave the previously valid cache intact, instead of silently overwriting it with an empty list - Run initial synchronous K8s fetch in __init__ so caches are populated before the first scheduler tick; prevents update_cluster_state from skipping the DB write during the first 10s after startup - Split nodes and pipelines into independent background threads so a slow/hung K8s nodes API call does not delay the pipelines refresh --- src/scheduler/kubernetes/scheduler.py | 37 ++++++++++++++++++++++----- 1 file changed, 31 insertions(+), 6 deletions(-) diff --git a/src/scheduler/kubernetes/scheduler.py b/src/scheduler/kubernetes/scheduler.py index 5ac6e46f9..f0ba37e60 100644 --- a/src/scheduler/kubernetes/scheduler.py +++ b/src/scheduler/kubernetes/scheduler.py @@ -555,27 +555,49 @@ def __init__(self, args): 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_thread() + self._start_k8s_refresh_threads() - def _start_k8s_refresh_thread(self): - def refresh_loop(): + 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) - time.sleep(10) - t = threading.Thread(target=refresh_loop, daemon=True) - t.start() + 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 @@ -587,6 +609,9 @@ def _refresh_k8s_pipelines(self): 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