From 085112d2193f433c73f5a472e77a317706694833 Mon Sep 17 00:00:00 2001 From: Wei8007 <189627153@qq.com> Date: Mon, 13 Jul 2026 20:57:55 +0800 Subject: [PATCH] Retry starts after concurrent job limits --- CHANGELOG.md | 1 + openeo/extra/job_management/_manager.py | 86 ++++++++++++------- openeo/extra/job_management/_thread_worker.py | 69 +++++++++------ .../test_concurrent_job_limit.py | 31 +++++++ tests/extra/job_management/test_manager.py | 31 +++++++ 5 files changed, 160 insertions(+), 58 deletions(-) create mode 100644 tests/extra/job_management/test_concurrent_job_limit.py diff --git a/CHANGELOG.md b/CHANGELOG.md index e570aba6e..00a405eb3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Retry job starts in a later job manager cycle when a backend reports `ConcurrentJobLimit`. ([#838](https://github.com/Open-EO/openeo-python-client/issues/838)) - `DataCube.resample_spatial()` now supports parameterized `resolution` and `projection` arguments. ([#897](https://github.com/Open-EO/openeo-python-client/issues/897)) - Sanitize asset download filenames (e.g. strip slashes, (semi)colon, hash, ...), instead of blindly using the asset key as filename. ([#820](https://github.com/Open-EO/openeo-python-client/issues/820)) - Support parameters in `DataCube` apply- and band-math operations ([#903](https://github.com/Open-EO/openeo-python-client/issues/903)) diff --git a/openeo/extra/job_management/_manager.py b/openeo/extra/job_management/_manager.py index 5f9c979ca..3081df585 100644 --- a/openeo/extra/job_management/_manager.py +++ b/openeo/extra/job_management/_manager.py @@ -30,9 +30,9 @@ from openeo import BatchJob, Connection from openeo.extra.job_management._interface import JobDatabaseInterface from openeo.extra.job_management._thread_worker import ( + _JobDownloadTask, _JobManagerWorkerThreadPool, _JobStartTask, - _JobDownloadTask ) from openeo.rest import OpenEoApiError from openeo.rest.auth.auth import BearerAuth @@ -512,16 +512,11 @@ def run_jobs( self._worker_pool = _JobManagerWorkerThreadPool() - - while ( - sum( - job_db.count_by_status( - statuses=["not_started", "created", "queued_for_start", "queued", "running"] - ).values()) > 0 - - or (self._worker_pool is not None and self._worker_pool.has_unprocessed_tasks()) - - ): + while sum( + job_db.count_by_status( + statuses=["not_started", "created", "queued_for_start", "queued", "running"] + ).values() + ) > 0 or (self._worker_pool is not None and self._worker_pool.has_unprocessed_tasks()): self._job_update_loop(job_db=job_db, start_job=start_job, stats=stats) stats["run_jobs loop"] += 1 @@ -530,8 +525,6 @@ def run_jobs( time.sleep(self.poll_sleep) stats["sleep"] += 1 - - self._worker_pool.shutdown() self._worker_pool = None @@ -554,6 +547,11 @@ def _job_update_loop( jobs_done, jobs_error, jobs_cancel = self._track_statuses(job_db, stats=stats) stats["track_statuses"] += 1 + # A start task that hits the backend's concurrent-job limit moves the + # job back to "created". Re-submit such jobs here, in a later manager + # cycle, instead of failing them or blocking a worker with sleep/retry. + self._queue_created_jobs(job_db=job_db, stats=stats) + not_started = job_db.get_by_status(statuses=["not_started"], max=200).copy() if len(not_started) > 0: # Check number of jobs queued/running at each backend @@ -581,7 +579,7 @@ def _job_update_loop( if self._worker_pool is not None: self._process_threadworker_updates(worker_pool=self._worker_pool, job_db=job_db, stats=stats) - + # TODO: move this back closer to the `_track_statuses` call above, once job done/error handling is also handled in threads? for job, row in jobs_done: @@ -648,22 +646,10 @@ def _launch_job(self, start_job, df, i, backend_name, stats: Optional[dict] = No if status == "created": # start job if not yet done by callback try: - job_con = job.connection - # Proactively refresh bearer token (because task in thread will not be able to do that) - self._refresh_bearer_token(connection=job_con) - task = _JobStartTask( - root_url=job_con.root_url, - bearer_token=job_con.auth.bearer if isinstance(job_con.auth, BearerAuth) else None, - job_id=job.job_id, - df_idx=i, - ) - _log.info(f"Submitting task {task} to thread pool") - self._worker_pool.submit_task(task=task, pool_name="job_start") - - stats["job_queued_for_start"] += 1 + self._submit_job_start(job=job, df_idx=i, stats=stats) df.loc[i, "status"] = "queued_for_start" except OpenEoApiError as e: - _log.info(f"Failed submitting task {task} to thread pool with error: {e}") + _log.info(f"Failed to submit start task for job {job.job_id!r} with error: {e}") df.loc[i, "status"] = "queued_for_start_failed" stats["job queued for start failed"] += 1 else: @@ -671,6 +657,42 @@ def _launch_job(self, start_job, df, i, backend_name, stats: Optional[dict] = No df.loc[i, "status"] = "skipped" stats["start_job skipped"] += 1 + def _submit_job_start(self, job: BatchJob, df_idx: int, stats: dict) -> None: + """Submit a non-blocking start task for an already created batch job.""" + job_con = job.connection + # Proactively refresh bearer token because the worker cannot refresh it. + self._refresh_bearer_token(connection=job_con) + task = _JobStartTask( + root_url=job_con.root_url, + bearer_token=job_con.auth.bearer if isinstance(job_con.auth, BearerAuth) else None, + job_id=job.job_id, + df_idx=df_idx, + ) + _log.info(f"Submitting task {task} to thread pool") + self._worker_pool.submit_task(task=task, pool_name="job_start") + stats["job_queued_for_start"] += 1 + + def _queue_created_jobs(self, *, job_db: JobDatabaseInterface, stats: dict) -> None: + """Re-submit created jobs whose previous start attempt was deferred.""" + created = job_db.get_by_status(statuses=["created"]).copy() + if created.empty: + return + + for i in created.index: + job_id = created.loc[i, "id"] + backend_name = created.loc[i, "backend_name"] + try: + job = self._get_connection(backend_name).job(job_id) + self._submit_job_start(job=job, df_idx=i, stats=stats) + created.loc[i, "status"] = "queued_for_start" + except OpenEoApiError as e: + _log.info(f"Failed to re-submit start task for job {job_id!r} with error: {e}") + created.loc[i, "status"] = "queued_for_start_failed" + stats["job queued for start failed"] += 1 + + job_db.persist(created) + stats["job_db persist"] += 1 + def _refresh_bearer_token(self, connection: Connection, *, max_age: float = 60) -> None: """ Helper to proactively refresh the bearer (access) token of the connection @@ -758,19 +780,19 @@ def on_job_done(self, job: BatchJob, row): #Proactively refresh bearer token job_con = job.connection self._refresh_bearer_token(connection=job_con) - + task = _JobDownloadTask( job_id=job.job_id, - df_idx=row.name, + df_idx=row.name, root_url=job_con.root_url, bearer_token=job_con.auth.bearer if isinstance(job_con.auth, BearerAuth) else None, download_dir=job_dir, ) _log.info(f"Submitting download task {task} to download thread pool") - + if self._worker_pool is None: self._worker_pool = _JobManagerWorkerThreadPool() - + self._worker_pool.submit_task(task=task, pool_name="job_download") def on_job_error(self, job: BatchJob, row): diff --git a/openeo/extra/job_management/_thread_worker.py b/openeo/extra/job_management/_thread_worker.py index 9e1e93f30..3e83cc649 100644 --- a/openeo/extra/job_management/_thread_worker.py +++ b/openeo/extra/job_management/_thread_worker.py @@ -3,16 +3,17 @@ """ import concurrent.futures +import json import logging from abc import ABC, abstractmethod from dataclasses import dataclass, field -from typing import Any, Dict, List, Optional, Tuple, Union from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple, Union -import json import urllib3.util import openeo +from openeo.rest import OpenEoApiError from openeo.utils.http import HTTP_429_TOO_MANY_REQUESTS, retry_configuration _log = logging.getLogger(__name__) @@ -132,16 +133,34 @@ def execute(self) -> _TaskResult: db_update={"status": "queued"}, stats_update={"job start": 1}, ) + except OpenEoApiError as e: + if e.code == "ConcurrentJobLimit": + _log.warning( + "Backend concurrent job limit reached while starting job %r; " + "scheduling another attempt in a later job manager cycle.", + self.job_id, + ) + return _TaskResult( + job_id=self.job_id, + df_idx=self.df_idx, + db_update={"status": "created"}, + stats_update={"job start retry": 1}, + ) + return self._start_failure_result(e) except Exception as e: - _log.error(f"Failed to start job {self.job_id!r}: {e!r}") - # TODO: more insights about the failure (e.g. the exception) are just logged, but lost from the result - return _TaskResult( - job_id=self.job_id, - df_idx=self.df_idx, - db_update={"status": "start_failed"}, - stats_update={"start_job error": 1}, - ) - + return self._start_failure_result(e) + + def _start_failure_result(self, error: Exception) -> _TaskResult: + """Build the terminal result for a non-retryable job start failure.""" + _log.error(f"Failed to start job {self.job_id!r}: {error!r}") + # TODO: more insights about the failure (e.g. the exception) are just logged, but lost from the result + return _TaskResult( + job_id=self.job_id, + df_idx=self.df_idx, + db_update={"status": "start_failed"}, + stats_update={"start_job error": 1}, + ) + @dataclass(frozen=True) class _JobDownloadTask(ConnectedTask): """ @@ -159,7 +178,7 @@ def execute(self) -> _TaskResult: # Count assets (files to download) file_count = len(job.get_results().get_assets()) - + # Download results job.get_results().download_files(target=self.download_dir) @@ -168,7 +187,7 @@ def execute(self) -> _TaskResult: metadata_path = self.download_dir / f"job_{self.job_id}.json" with metadata_path.open("w", encoding="utf-8") as f: json.dump(job_metadata, f, ensure_ascii=False) - + _log.info(f"Job {self.job_id!r} results downloaded successfully") return _TaskResult( job_id=self.job_id, @@ -184,7 +203,8 @@ def execute(self) -> _TaskResult: db_update={}, stats_update={"job download error": 1, "files downloaded": 0}, ) - + + class _TaskThreadPool: """ Thread pool-based worker that manages the execution of asynchronous tasks. @@ -261,11 +281,11 @@ def process_futures(self, timeout: Union[float, None] = 0) -> Tuple[List[_TaskRe self._total_processed += len(results) return results, len(to_keep) - + def get_unprocessed_count(self) -> int: """Get number of tasks that haven't been processed yet.""" return self._total_submitted - self._total_processed - + def has_unprocessed_tasks(self) -> bool: """Check if there are tasks that haven't been processed yet.""" return self._total_submitted > self._total_processed @@ -280,7 +300,7 @@ class _JobManagerWorkerThreadPool: """ Generic wrapper that manages multiple thread pools with a dict. """ - + def __init__(self, pool_configs: Optional[Dict[str, int]] = None): self._pools: Dict[str, _TaskThreadPool] = {} self._pool_configs = pool_configs or {} @@ -288,23 +308,23 @@ def __init__(self, pool_configs: Optional[Dict[str, int]] = None): def list_pools(self) -> List[str]: """List all active pool names.""" return list(self._pools.keys()) - + def submit_task(self, task: Task, pool_name: str = "default") -> None: if pool_name not in self._pools: max_workers = self._pool_configs.get(pool_name, 2) self._pools[pool_name] = _TaskThreadPool(max_workers=max_workers) _log.info(f"Created pool '{pool_name}' with {max_workers} workers") - + self._pools[pool_name].submit_task(task) def get_unprocessed_counts(self) -> Dict[str, int]: """Get unprocessed (submitted but not processed) task counts per pool.""" return {name: pool.get_unprocessed_count() for name, pool in self._pools.items()} - + def has_unprocessed_tasks(self) -> bool: """Check if any pool has unprocessed (submitted but not processed) tasks.""" return any(pool.has_unprocessed_tasks() for pool in self._pools.values()) - + def process_futures(self, timeout: Union[float, None] = 0) -> Tuple[List[_TaskResult], Dict[str, int]]: """ Process updates from ALL pools. @@ -312,7 +332,7 @@ def process_futures(self, timeout: Union[float, None] = 0) -> Tuple[List[_TaskRe """ all_results = [] to_keep = {} - + for pool_name, pool in self._pools.items(): results, remaining = pool.process_futures(timeout) all_results.extend(results) @@ -320,7 +340,7 @@ def process_futures(self, timeout: Union[float, None] = 0) -> Tuple[List[_TaskRe to_keep[pool_name] = remaining return all_results, to_keep - + def shutdown(self, pool_name: Optional[str] = None) -> None: """ Shutdown pools. @@ -334,6 +354,3 @@ def shutdown(self, pool_name: Optional[str] = None) -> None: for pool_name, pool in list(self._pools.items()): pool.shutdown() del self._pools[pool_name] - - - diff --git a/tests/extra/job_management/test_concurrent_job_limit.py b/tests/extra/job_management/test_concurrent_job_limit.py new file mode 100644 index 000000000..8beaa6fea --- /dev/null +++ b/tests/extra/job_management/test_concurrent_job_limit.py @@ -0,0 +1,31 @@ +import logging + +from openeo.extra.job_management._thread_worker import _JobStartTask, _TaskResult +from openeo.rest._testing import DummyBackend + + +def test_concurrent_job_limit_is_retryable(requests_mock, caplog): + caplog.set_level(logging.WARNING) + backend = DummyBackend.at_url("https://foo.test", requests_mock=requests_mock) + job = backend.connection.create_job(process_graph={}) + backend.setup_job_start_failure( + status_code=400, + response_body={"code": "ConcurrentJobLimit", "message": "Concurrent job limit reached."}, + ) + + result = _JobStartTask( + job_id=job.job_id, + df_idx=0, + root_url=backend.connection.root_url, + ).execute() + + assert result == _TaskResult( + job_id="job-000", + df_idx=0, + db_update={"status": "created"}, + stats_update={"job start retry": 1}, + ) + assert caplog.messages == [ + "Backend concurrent job limit reached while starting job 'job-000'; " + "scheduling another attempt in a later job manager cycle." + ] diff --git a/tests/extra/job_management/test_manager.py b/tests/extra/job_management/test_manager.py index 2c4162974..6719f1fc7 100644 --- a/tests/extra/job_management/test_manager.py +++ b/tests/extra/job_management/test_manager.py @@ -150,6 +150,37 @@ def test_basic_legacy(self, tmp_path, job_manager, job_manager_root_dir, sleep_m for filename in ["job-results.json", f"job_{job_id}.json", "result.data"] } + def test_concurrent_job_limit_is_retried_in_later_cycle( + self, tmp_path, job_manager, dummy_backend_foo, requests_mock, sleep_mock + ): + attempts = [] + + def start_response(request, context): + attempts.append(request.url) + if len(attempts) == 1: + context.status_code = 400 + return {"code": "ConcurrentJobLimit", "message": "At most one job can run."} + + context.status_code = 202 + dummy_backend_foo.batch_jobs["job-2018"]["status"] = "queued" + return {} + + requests_mock.post("https://foo.test/jobs/job-2018/results", json=start_response) + job_db = CsvJobDatabase(tmp_path / "jobs.csv").initialize_from_df(pd.DataFrame({"year": [2018]})) + + stats = job_manager.run_jobs(job_db=job_db, start_job=self._create_year_job) + + assert len(attempts) == 2 + assert stats == dirty_equals.IsPartialDict( + { + "job start retry": 1, + "job_queued_for_start": 2, + "job start": 1, + "job finished": 1, + } + ) + assert job_db.read().loc[0, "status"] == "finished" + def test_basic(self, tmp_path, job_manager, job_manager_root_dir, sleep_mock): """ `run_jobs()` usage with a `CsvJobDatabase`