Skip to content

Commit 03e41d2

Browse files
liuwei08liuwei08
authored
fix(DM01-6341): catch K8s API timeout in kube_job to prevent crash loop (#651)
* 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() * 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 * revert: move update_cluster_state() back outside try/except * 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.
1 parent 06aa9f0 commit 03e41d2

1 file changed

Lines changed: 42 additions & 19 deletions

File tree

‎src/scheduler/kubernetes/scheduler.py‎

Lines changed: 42 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -588,7 +588,23 @@ def pipelines_loop():
588588
except Exception as e:
589589
self.logger.exception(e)
590590

591-
for target in (nodes_loop, pipelines_loop):
591+
def pipeline_invocations_loop():
592+
while True:
593+
try:
594+
self.pipeline_controller.handle()
595+
except Exception as e:
596+
self.logger.exception(e)
597+
time.sleep(5)
598+
599+
def function_invocations_loop():
600+
while True:
601+
try:
602+
self.function_controller.handle()
603+
except Exception as e:
604+
self.logger.exception(e)
605+
time.sleep(5)
606+
607+
for target in (nodes_loop, pipelines_loop, pipeline_invocations_loop, function_invocations_loop):
592608
t = threading.Thread(target=target, daemon=True)
593609
t.start()
594610

@@ -767,8 +783,17 @@ def kube_job(self, job_id, cpu, mem, services=None):
767783
}
768784
}
769785

770-
r = requests.post(self.args.api_server + '/apis/core.infrabox.net/v1alpha1/namespaces/%s/ibpipelineinvocations' % self.namespace,
771-
headers=h, json=job, timeout=10)
786+
try:
787+
r = requests.post(self.args.api_server + '/apis/core.infrabox.net/v1alpha1/namespaces/%s/ibpipelineinvocations' % self.namespace,
788+
headers=h, json=job, timeout=10)
789+
except requests.exceptions.RequestException as e:
790+
self.logger.warn("Failed to create IBPipelineInvocation for job %s: %s, will retry next tick", job_id, e)
791+
return False
792+
793+
if r.status_code == 409:
794+
# IBPipelineInvocation already exists — a previous POST succeeded but the response
795+
# timed out on our side. Treat as success so the job can proceed to 'scheduled'.
796+
return True
772797

773798
if r.status_code != 201:
774799
self.logger.warn(r.text)
@@ -1372,31 +1397,29 @@ def _inactive(self):
13721397
def handle(self):
13731398
self.update_cluster_state()
13741399

1400+
cluster_name = os.environ['INFRABOX_CLUSTER_NAME']
1401+
ha_mode = os.environ['INFRABOX_HA_ENABLED'] == "true"
1402+
13751403
try:
13761404
self.handle_aborts()
13771405
self.handle_timeouts()
13781406
self.handle_orphaned_jobs()
1379-
self.handle_function_invocations()
1380-
self.handle_pipeline_invocations()
13811407
self.handle_cron_jobs()
13821408
self.handle_inactive_cluster_queued_jobs()
1383-
except Exception as e:
1384-
self.logger.exception(e)
13851409

1386-
cluster_name = os.environ['INFRABOX_CLUSTER_NAME']
1387-
ha_mode = os.environ['INFRABOX_HA_ENABLED'] == "true"
1388-
1389-
if self._inactive():
1390-
self.logger.info('Cluster set to inactive or disabled, sleeping...')
1391-
time.sleep(5)
1392-
return
1410+
if self._inactive():
1411+
self.logger.info('Cluster set to inactive or disabled, sleeping...')
1412+
time.sleep(5)
1413+
return
13931414

1394-
if ha_mode:
1395-
self.assign_cluster()
1396-
elif cluster_name == 'master':
1397-
self.assign_cluster()
1415+
if ha_mode:
1416+
self.assign_cluster()
1417+
elif cluster_name == 'master':
1418+
self.assign_cluster()
13981419

1399-
self.schedule()
1420+
self.schedule()
1421+
except Exception as e:
1422+
self.logger.exception(e)
14001423

14011424
def run(self):
14021425
self.logger.info("Starting scheduler")

0 commit comments

Comments
 (0)