feat(DM01-6251): decouple blocking K8s API calls from scheduler main loop

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