From 54e11c9aafe9bc9d960c86cfd9b1bd451794c774 Mon Sep 17 00:00:00 2001 From: I515719 Date: Thu, 15 Oct 2026 15:03:13 +0800 Subject: [PATCH 1/4] fix(DM01-6341): catch K8s API timeout in kube_job to prevent crash loop When K8s API is slow (e.g. during NodeOS upgrade), requests.post() in kube_job() raises ReadTimeout. This was uncaught and propagated out of handle() since schedule() was outside the try/except block, crashing the scheduler process (95 restarts observed in 2 days). - Wrap requests.post() in kube_job() with try/except Timeout; return False so the job stays queued and is retried on the next tick - Move assign_cluster() and schedule() inside the existing try/except in handle() so no uncaught exception can escape to main() --- src/scheduler/kubernetes/scheduler.py | 36 +++++++++++++++------------ 1 file changed, 20 insertions(+), 16 deletions(-) diff --git a/src/scheduler/kubernetes/scheduler.py b/src/scheduler/kubernetes/scheduler.py index f0ba37e6..2ecf1815 100644 --- a/src/scheduler/kubernetes/scheduler.py +++ b/src/scheduler/kubernetes/scheduler.py @@ -767,8 +767,12 @@ def kube_job(self, job_id, cpu, mem, services=None): } } - r = requests.post(self.args.api_server + '/apis/core.infrabox.net/v1alpha1/namespaces/%s/ibpipelineinvocations' % self.namespace, - headers=h, json=job, timeout=10) + try: + r = requests.post(self.args.api_server + '/apis/core.infrabox.net/v1alpha1/namespaces/%s/ibpipelineinvocations' % self.namespace, + headers=h, json=job, timeout=10) + except requests.exceptions.Timeout: + self.logger.warn("Timeout creating IBPipelineInvocation for job %s, will retry next tick", job_id) + return False if r.status_code != 201: self.logger.warn(r.text) @@ -1372,6 +1376,9 @@ def _inactive(self): def handle(self): self.update_cluster_state() + cluster_name = os.environ['INFRABOX_CLUSTER_NAME'] + ha_mode = os.environ['INFRABOX_HA_ENABLED'] == "true" + try: self.handle_aborts() self.handle_timeouts() @@ -1380,23 +1387,20 @@ def handle(self): self.handle_pipeline_invocations() self.handle_cron_jobs() self.handle_inactive_cluster_queued_jobs() - except Exception as e: - self.logger.exception(e) - cluster_name = os.environ['INFRABOX_CLUSTER_NAME'] - ha_mode = os.environ['INFRABOX_HA_ENABLED'] == "true" - - if self._inactive(): - self.logger.info('Cluster set to inactive or disabled, sleeping...') - time.sleep(5) - return + if self._inactive(): + self.logger.info('Cluster set to inactive or disabled, sleeping...') + time.sleep(5) + return - if ha_mode: - self.assign_cluster() - elif cluster_name == 'master': - self.assign_cluster() + if ha_mode: + self.assign_cluster() + elif cluster_name == 'master': + self.assign_cluster() - self.schedule() + self.schedule() + except Exception as e: + self.logger.exception(e) def run(self): self.logger.info("Starting scheduler") From 081c979bda7d5bbc56d78d12d0508279c1a8f172 Mon Sep 17 00:00:00 2001 From: I515719 Date: Thu, 15 Oct 2026 15:10:07 +0800 Subject: [PATCH 2/4] fix(DM01-6341): address code review findings - Broaden exception catch from Timeout to RequestException to also handle ConnectionError during K8s API unavailability - Return True on 409 in kube_job() so a job is not stuck queued when a previous POST succeeded but the response timed out - Move update_cluster_state() inside the try/except block in handle() to eliminate a remaining uncaught exception path --- src/scheduler/kubernetes/scheduler.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/src/scheduler/kubernetes/scheduler.py b/src/scheduler/kubernetes/scheduler.py index 2ecf1815..22b99d40 100644 --- a/src/scheduler/kubernetes/scheduler.py +++ b/src/scheduler/kubernetes/scheduler.py @@ -770,10 +770,15 @@ def kube_job(self, job_id, cpu, mem, services=None): try: r = requests.post(self.args.api_server + '/apis/core.infrabox.net/v1alpha1/namespaces/%s/ibpipelineinvocations' % self.namespace, headers=h, json=job, timeout=10) - except requests.exceptions.Timeout: - self.logger.warn("Timeout creating IBPipelineInvocation for job %s, will retry next tick", job_id) + except requests.exceptions.RequestException as e: + self.logger.warn("Failed to create IBPipelineInvocation for job %s: %s, will retry next tick", job_id, e) return False + if r.status_code == 409: + # IBPipelineInvocation already exists — a previous POST succeeded but the response + # timed out on our side. Treat as success so the job can proceed to 'scheduled'. + return True + if r.status_code != 201: self.logger.warn(r.text) return False @@ -1374,12 +1379,11 @@ def _inactive(self): return not (active and enabled) def handle(self): - self.update_cluster_state() - cluster_name = os.environ['INFRABOX_CLUSTER_NAME'] ha_mode = os.environ['INFRABOX_HA_ENABLED'] == "true" try: + self.update_cluster_state() self.handle_aborts() self.handle_timeouts() self.handle_orphaned_jobs() From 1cff295b7b1b17ff6a503a63ff8cd7770e53f54d Mon Sep 17 00:00:00 2001 From: I515719 Date: Thu, 15 Oct 2026 15:18:39 +0800 Subject: [PATCH 3/4] revert: move update_cluster_state() back outside try/except --- src/scheduler/kubernetes/scheduler.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/scheduler/kubernetes/scheduler.py b/src/scheduler/kubernetes/scheduler.py index 22b99d40..fec7daae 100644 --- a/src/scheduler/kubernetes/scheduler.py +++ b/src/scheduler/kubernetes/scheduler.py @@ -1379,11 +1379,12 @@ def _inactive(self): return not (active and enabled) def handle(self): + self.update_cluster_state() + cluster_name = os.environ['INFRABOX_CLUSTER_NAME'] ha_mode = os.environ['INFRABOX_HA_ENABLED'] == "true" try: - self.update_cluster_state() self.handle_aborts() self.handle_timeouts() self.handle_orphaned_jobs() From cbead7c2baea2b50c5fcd4ef98c3a5afb52252d3 Mon Sep 17 00:00:00 2001 From: I515719 Date: Thu, 15 Oct 2026 15:34:59 +0800 Subject: [PATCH 4/4] fix(DM01-6341): move pipeline/function invocation controllers to background threads handle_pipeline_invocations() and handle_function_invocations() make live K8s API calls every tick, blocking handle_orphaned_jobs() and schedule() when K8s API is slow. Move them to dedicated background threads (same pattern as nodes/pipelines cache refresh) so the main scheduling loop is never blocked by CRD controller sync latency. sleep is placed after handle() to prevent concurrent runs if a single iteration takes longer than the sleep interval. --- src/scheduler/kubernetes/scheduler.py | 20 +++++++++++++++++--- 1 file changed, 17 insertions(+), 3 deletions(-) diff --git a/src/scheduler/kubernetes/scheduler.py b/src/scheduler/kubernetes/scheduler.py index fec7daae..155fced4 100644 --- a/src/scheduler/kubernetes/scheduler.py +++ b/src/scheduler/kubernetes/scheduler.py @@ -588,7 +588,23 @@ def pipelines_loop(): except Exception as e: self.logger.exception(e) - for target in (nodes_loop, pipelines_loop): + def pipeline_invocations_loop(): + while True: + try: + self.pipeline_controller.handle() + except Exception as e: + self.logger.exception(e) + time.sleep(5) + + def function_invocations_loop(): + while True: + try: + self.function_controller.handle() + except Exception as e: + self.logger.exception(e) + time.sleep(5) + + for target in (nodes_loop, pipelines_loop, pipeline_invocations_loop, function_invocations_loop): t = threading.Thread(target=target, daemon=True) t.start() @@ -1388,8 +1404,6 @@ def handle(self): self.handle_aborts() self.handle_timeouts() self.handle_orphaned_jobs() - self.handle_function_invocations() - self.handle_pipeline_invocations() self.handle_cron_jobs() self.handle_inactive_cluster_queued_jobs()