diff --git a/CHANGELOG.md b/CHANGELOG.md index 9c24e88c..df16542c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,14 @@ ## lifebit-ai/cloudos-cli: changelog +## v2.95.1 (2026-07-16) + +### Fix: + +- `cloudos job list --filter-queue` appeared to hang forever on large workspaces. Queue filtering is client-side (the jobs API has no queue parameter), so all workspace jobs must be scanned; the scan now uses the API maximum page size (100 jobs per request instead of `--page-size`, typically 10), reuses a single HTTP connection and prints scan progress +- The queue-filtered scan now always starts from the first API page, so results are no longer silently missing when combined with `--page` +- `--page` is now honoured as the initially displayed page of the filtered results when combined with `--filter-queue` +- Documentation now correctly states that `--filter-queue` returns only batch jobs running on the specified queue (non-batch jobs were never preserved) + ## v2.95.0 (2026-06-30) ### Breaking: diff --git a/README.md b/README.md index c95f8872..98abddee 100644 --- a/README.md +++ b/README.md @@ -231,9 +231,8 @@ cloudos job list --help │ workflow (by date). │ │ --filter-job-id TEXT Filter jobs by specific job ID. │ │ --filter-only-mine Filter to show only jobs belonging to the current user. │ -│ --filter-queue TEXT Filter jobs by queue name. Only applies to jobs running │ -│ in batch environment. Non-batch jobs are preserved in │ -│ results. │ +│ --filter-queue TEXT Filter jobs by queue name. Only batch jobs running on │ +│ the specified queue are returned. │ │ --filter-owner TEXT Filter jobs by owner username. │ │ --verbose Whether to print information messages or not. │ │ --disable-ssl-verification Disable SSL certificate verification. Please, remember │ @@ -912,7 +911,7 @@ You can find specific jobs within your workspace using the filtering options. Fi - **`--filter-job-id`**: Filter jobs by specific job ID (exact match required) - **`--filter-only-mine`**: Show only jobs belonging to the current user - **`--filter-owner`**: Show only jobs for the specified owner (exact match required, e.g., "John Doe") -- **`--filter-queue`**: Filter jobs by queue name (works with both regular and system queues; only applies to batch jobs) +- **`--filter-queue`**: Filter jobs by queue name (works with both regular and system queues; only applies to batch jobs). Queue filtering is performed client-side, so all the workspace jobs are scanned; a progress message is displayed during the scan, which can take a while on workspaces with many jobs **Filtering Examples** diff --git a/cloudos_cli/_version.py b/cloudos_cli/_version.py index d72db282..fa202cff 100644 --- a/cloudos_cli/_version.py +++ b/cloudos_cli/_version.py @@ -1 +1 @@ -__version__ = '2.95.0' +__version__ = '2.95.1' diff --git a/cloudos_cli/clos.py b/cloudos_cli/clos.py index 73f6f5c2..a732556a 100644 --- a/cloudos_cli/clos.py +++ b/cloudos_cli/clos.py @@ -3,12 +3,14 @@ """ import requests +import sys import time import json from dataclasses import dataclass from cloudos_cli.utils.cloud import find_cloud from cloudos_cli.utils.errors import BadRequestException, JoBNotCompletedException, NotAuthorisedException, JobAccessDeniedException -from cloudos_cli.utils.requests import retry_requests_get, retry_requests_post, retry_requests_put +from cloudos_cli.utils.requests import (retry_requests_get, retry_requests_post, + retry_requests_put, create_retry_session) import pandas as pd from cloudos_cli.utils.last_wf import youngest_workflow_id_by_name from datetime import datetime, timezone @@ -1049,8 +1051,10 @@ def get_job_list(self, workspace_id, last_n_jobs=None, page=None, page_size=None Filter jobs by owner username (will be resolved to user ID). filter_queue : string, optional Filter jobs by queue name (will be resolved to queue ID). - Only applies to jobs running in batch environment. - Non-batch jobs are preserved in results as they don't use queues. + Only batch jobs running on the specified queue are returned; + jobs without a batch queue are excluded. + Note: the API does not support server-side queue filtering, so + all the workspace jobs are scanned and filtered client-side. last : bool, optional When workflows are duplicated, use the latest imported workflow (by date). @@ -1199,57 +1203,85 @@ def get_job_list(self, workspace_id, last_n_jobs=None, page=None, page_size=None # --- Fetch jobs page by page --- all_jobs = [] - params["limit"] = current_page_size + # The queue filter is applied client-side, so all the workspace jobs must + # be scanned. Scan from the first page using the API maximum page size and + # a reusable HTTP session to minimise the number of requests, and report + # progress as the scan can take a while on large workspaces. + scan_all_pages = bool(filter_queue) + if scan_all_pages: + current_page = 1 + params["limit"] = 100 + else: + params["limit"] = current_page_size last_pagination_metadata = None # Track the last pagination metadata + scanned_job_count = 0 - while True: - params["page"] = current_page - - r = retry_requests_get(f"{self.cloudos_url}/api/v2/jobs", params=params, headers=headers, verify=verify) - if r.status_code >= 400: - raise BadRequestException(r) + with create_retry_session() as session: + while True: + params["page"] = current_page - content = r.json() - page_jobs = content.get('jobs', []) + r = retry_requests_get(f"{self.cloudos_url}/api/v2/jobs", params=params, + headers=headers, verify=verify, session=session) + if r.status_code >= 400: + raise BadRequestException(r) - # Capture pagination metadata - last_pagination_metadata = content.get('paginationMetadata', None) + content = r.json() + page_jobs = content.get('jobs', []) + raw_page_job_count = len(page_jobs) - # No jobs returned, we've reached the end - if not page_jobs: - break + # Capture pagination metadata + last_pagination_metadata = content.get('paginationMetadata', None) - # Apply queue filter during pagination (if specified) - # jobQueue is a dict with "id" and "name" keys, extract the id for comparison - if filter_queue and queue_id: - filtered_jobs = [] - for job in page_jobs: - job_queue = job.get("batch", {}).get("jobQueue", {}) - # jobQueue is a dict like {"id": "...", "name": "...", ...} - job_queue_id = job_queue.get("id") if isinstance(job_queue, dict) else job_queue - if job_queue_id == queue_id: - filtered_jobs.append(job) - page_jobs = filtered_jobs - - all_jobs.extend(page_jobs) - - # Check stopping conditions based on mode - if use_pagination_mode: - # In pagination mode (last_n_jobs), continue until we have enough jobs (after filtering) - if target_job_count != 'all' and len(all_jobs) >= target_job_count: - break - else: - if not filter_queue and len(all_jobs) >= current_page_size: + # No jobs returned, we've reached the end + if not page_jobs: break - # Check if we reached the last page (fewer jobs than requested page size) - # Note: For queue filter, we check the unfiltered page_jobs count from the API - # This ensures we stop when the API has exhausted results - raw_page_jobs = content.get('jobs', []) - if len(raw_page_jobs) < params["limit"]: - break # Last page + # Apply queue filter during pagination (if specified) + # jobQueue is a dict with "id" and "name" keys, extract the id for comparison + if filter_queue and queue_id: + filtered_jobs = [] + for job in page_jobs: + job_queue = job.get("batch", {}).get("jobQueue", {}) + # jobQueue is a dict like {"id": "...", "name": "...", ...} + job_queue_id = job_queue.get("id") if isinstance(job_queue, dict) else job_queue + if job_queue_id == queue_id: + filtered_jobs.append(job) + page_jobs = filtered_jobs + + all_jobs.extend(page_jobs) + + if scan_all_pages: + scanned_job_count += raw_page_job_count + total_job_count = (last_pagination_metadata or {}).get('Pagination-Count', '?') + progress_msg = (f"\tScanning workspace jobs for queue '{filter_queue}': " + f"{scanned_job_count}/{total_job_count} jobs scanned, " + f"{len(all_jobs)} matching...") + if sys.stderr.isatty(): + # Self-updating single line on interactive terminals + print(f"\r{progress_msg}", end='', flush=True, file=sys.stderr) + else: + # Newline-terminated lines when redirected (e.g. CI logs) + print(progress_msg, flush=True, file=sys.stderr) + + # Check stopping conditions based on mode + if use_pagination_mode: + # In pagination mode (last_n_jobs), continue until we have enough jobs (after filtering) + if target_job_count != 'all' and len(all_jobs) >= target_job_count: + break + else: + if not filter_queue and len(all_jobs) >= current_page_size: + break + + # Check if we reached the last page (fewer jobs than requested page size) + # Note: For queue filter, we check the unfiltered job count from the API + # This ensures we stop when the API has exhausted results + if raw_page_job_count < params["limit"]: + break # Last page + + current_page += 1 - current_page += 1 + if scan_all_pages and scanned_job_count and sys.stderr.isatty(): + print(file=sys.stderr) # End the self-updating progress line # --- Apply limit after all filtering --- if use_pagination_mode and target_job_count != 'all' and isinstance(target_job_count, int) and target_job_count > 0: diff --git a/cloudos_cli/jobs/cli.py b/cloudos_cli/jobs/cli.py index d86767ad..07cf1ed6 100644 --- a/cloudos_cli/jobs/cli.py +++ b/cloudos_cli/jobs/cli.py @@ -1185,7 +1185,7 @@ def job_details(ctx, help='Filter to show only jobs belonging to the current user.', is_flag=True) @click.option('--filter-queue', - help='Filter jobs by queue name. Only applies to jobs running in batch environment. Non-batch jobs are preserved in results.') + help='Filter jobs by queue name. Only batch jobs running on the specified queue are returned.') @click.option('--filter-owner', help='Filter jobs by owner username.') @click.option('--verbose', @@ -1326,15 +1326,11 @@ def list_jobs(ctx, # For client-filtered results, we have all jobs already # Create a callback that paginates them client-side using helper function fetch_page = create_client_pagination_callback(my_jobs_r, page_size) - - # Show first page of filtered results - first_page_jobs = my_jobs_r[:page_size] - first_page_metadata = { - 'Pagination-Count': len(my_jobs_r), - 'Pagination-Page': 1, - 'Pagination-Limit': page_size - } - create_job_list_table(first_page_jobs, cloudos_url, first_page_metadata, selected_columns, fetch_page_callback=fetch_page) + + # Show the requested page of filtered results + initial_page = fetch_page(page) + create_job_list_table(initial_page['jobs'], cloudos_url, initial_page['pagination_metadata'], + selected_columns, fetch_page_callback=fetch_page) else: # For normal (non-filtered) results, use API pagination with helper function fetch_page = create_api_pagination_callback( diff --git a/cloudos_cli/utils/details.py b/cloudos_cli/utils/details.py index f55c3e2d..059cd315 100644 --- a/cloudos_cli/utils/details.py +++ b/cloudos_cli/utils/details.py @@ -756,7 +756,7 @@ def create_job_list_table(jobs, cloudos_url, pagination_metadata=None, selected_ # Show pagination controls only if there are multiple pages if total_pages > 1: if not sys.stdin.isatty(): - console.print("\n[yellow]Note: Pagination not available in non-interactive mode. Showing page 1 of {0}.[/yellow]".format(total_pages)) + console.print("\n[yellow]Note: Pagination not available in non-interactive mode. Showing page {0} of {1}.[/yellow]".format(current_page, total_pages)) console.print("[yellow]Run in an interactive terminal to navigate through all pages.[/yellow]") break diff --git a/cloudos_cli/utils/requests.py b/cloudos_cli/utils/requests.py index 16f156a7..2e82d1d2 100644 --- a/cloudos_cli/utils/requests.py +++ b/cloudos_cli/utils/requests.py @@ -7,42 +7,42 @@ from urllib3.util import Retry -def retry_requests_get(url, total=5, status_forcelist=[429, 500, 502, 503, 504], **kwargs): - """Wrap normal requests get with an error strategy. +def create_retry_session(total=5, status_forcelist=(429, 500, 502, 503, 504), + allowed_methods=None): + """Create a requests.Session with a retry strategy mounted. + + Reusing the returned session across several requests to the same host + avoids re-establishing a TCP/TLS connection for every request. Parameters ---------- - url : string - The request URL total : int Total number of retries - status_forcelist : list - A list of ints with the status codes to trigger the retries + status_forcelist : iterable of int + The status codes to trigger the retries + allowed_methods : iterable of str, optional + HTTP methods allowed to be retried. When None, the urllib3 + default set of idempotent methods is used. - Return - ------ - response : requests.Response - The Response object returned by the API server + Returns + ------- + session : requests.Session + A session object with the retry strategy mounted. """ - retry_strategy = Retry( - total=total, - status_forcelist=status_forcelist - ) - # Create an HTTP adapter with the retry strategy and mount it to session + retry_kwargs = dict(total=total, status_forcelist=status_forcelist) + if allowed_methods is not None: + retry_kwargs['allowed_methods'] = allowed_methods + retry_strategy = Retry(**retry_kwargs) adapter = HTTPAdapter(max_retries=retry_strategy) - - # Create a new session object session = requests.Session() session.mount('http://', adapter) session.mount('https://', adapter) + return session - # Make a request using the session object - response = session.get(url, **kwargs) - return response - -def retry_requests_post(url, total=5, status_forcelist=[429, 500, 502, 503, 504], **kwargs): - """Wrap normal requests post with an error strategy. +def retry_requests_get(url, total=5, status_forcelist=(429, 500, 502, 503, 504), + session=None, **kwargs): + """Wrap normal requests get with an error strategy. Parameters ---------- @@ -50,32 +50,46 @@ def retry_requests_post(url, total=5, status_forcelist=[429, 500, 502, 503, 504] The request URL total : int Total number of retries - status_forcelist : list - A list of ints with the status codes to trigger the retries + status_forcelist : iterable of int + HTTP status codes to trigger the retries + session : requests.Session, optional + An existing session to reuse (e.g. created with + `create_retry_session`). When None, a new session is created + for this single request. - Return - ------ + Returns + ------- response : requests.Response The Response object returned by the API server """ - retry_strategy = Retry( - total=total, - status_forcelist=status_forcelist - ) - # Create an HTTP adapter with the retry strategy and mount it to session - adapter = HTTPAdapter(max_retries=retry_strategy) + if session is not None: + return session.get(url, **kwargs) + with create_retry_session(total, status_forcelist) as single_use_session: + return single_use_session.get(url, **kwargs) - # Create a new session object - session = requests.Session() - session.mount('http://', adapter) - session.mount('https://', adapter) - # Make a request using the session object - response = session.post(url, **kwargs) - return response +def retry_requests_post(url, total=5, status_forcelist=(429, 500, 502, 503, 504), **kwargs): + """Wrap normal requests post with an error strategy. + + Parameters + ---------- + url : string + The request URL + total : int + Total number of retries + status_forcelist : iterable of int + HTTP status codes to trigger the retries + + Returns + ------- + response : requests.Response + The Response object returned by the API server + """ + with create_retry_session(total, status_forcelist) as session: + return session.post(url, **kwargs) -def retry_requests_put(url, total=5, status_forcelist=[429, 500, 502, 503, 504], **kwargs): +def retry_requests_put(url, total=5, status_forcelist=(429, 500, 502, 503, 504), **kwargs): """Wrap normal requests put with an error strategy. Parameters @@ -84,32 +98,19 @@ def retry_requests_put(url, total=5, status_forcelist=[429, 500, 502, 503, 504], The request URL total : int Total number of retries - status_forcelist : list - A list of ints with the status codes to trigger the retries + status_forcelist : iterable of int + HTTP status codes to trigger the retries - Return - ------ + Returns + ------- response : requests.Response The Response object returned by the API server """ - retry_strategy = Retry( - total=total, - status_forcelist=status_forcelist - ) - # Create an HTTP adapter with the retry strategy and mount it to session - adapter = HTTPAdapter(max_retries=retry_strategy) - - # Create a new session object - session = requests.Session() - session.mount('http://', adapter) - session.mount('https://', adapter) + with create_retry_session(total, status_forcelist) as session: + return session.put(url, **kwargs) - # Make a request using the session object - response = session.put(url, **kwargs) - return response - -def retry_requests_delete(url, total=5, status_forcelist=[429, 500, 502, 503, 504], **kwargs): +def retry_requests_delete(url, total=5, status_forcelist=(429, 500, 502, 503, 504), **kwargs): """ Wrap normal requests DELETE with an error retry strategy. @@ -119,7 +120,7 @@ def retry_requests_delete(url, total=5, status_forcelist=[429, 500, 502, 503, 50 The request URL. total : int Total number of retry attempts. - status_forcelist : list of int + status_forcelist : iterable of int HTTP status codes that should trigger a retry. **kwargs : Additional keyword arguments passed to `requests.delete`. @@ -129,16 +130,5 @@ def retry_requests_delete(url, total=5, status_forcelist=[429, 500, 502, 503, 50 requests.Response The Response object returned by the API server. """ - retry_strategy = Retry( - total=total, - status_forcelist=status_forcelist, - allowed_methods=["DELETE"] - ) - adapter = HTTPAdapter(max_retries=retry_strategy) - - session = requests.Session() - session.mount("http://", adapter) - session.mount("https://", adapter) - - response = session.delete(url, **kwargs) - return response \ No newline at end of file + with create_retry_session(total, status_forcelist, allowed_methods=["DELETE"]) as session: + return session.delete(url, **kwargs) diff --git a/tests/test_clos/test_get_job_list_filtering.py b/tests/test_clos/test_get_job_list_filtering.py index 26747f8b..c43373a2 100644 --- a/tests/test_clos/test_get_job_list_filtering.py +++ b/tests/test_clos/test_get_job_list_filtering.py @@ -212,3 +212,76 @@ def test_filter_by_system_queue(mock_get_queues): mock_get_queues.assert_called_once() +def make_job(job_id, queue_id): + """Build a minimal job dict assigned to the given queue.""" + return { + "_id": job_id, + "name": job_id, + "status": "completed", + "batch": {"jobQueue": {"id": queue_id, "name": "queue-name"}} + } + + +@responses.activate +@mock.patch('cloudos_cli.queue.queue.Queue.get_job_queues') +def test_filter_by_queue_scans_with_max_page_size(mock_get_queues): + """Queue filtering must scan from page 1 using the API maximum page size (100), + regardless of the requested page and page_size (pagination is client-side).""" + mock_get_queues.return_value = MOCK_QUEUE_LIST + expected_params = { + "teamId": WORKSPACE_ID, + "archived.status": "false", + "limit": 100, + "page": 1 + } + responses.add( + responses.GET, + url=f"{CLOUDOS_URL}/api/v2/jobs", + json=MOCK_JOB_LIST, + match=[matchers.query_param_matcher(expected_params)], + status=200 + ) + clos = setup_clos() + result = clos.get_job_list(WORKSPACE_ID, filter_queue="v41", page=2, page_size=10) + jobs = result['jobs'] + assert len(jobs) == 1 + assert jobs[0]["_id"] == "job1" + + +@responses.activate +@mock.patch('cloudos_cli.queue.queue.Queue.get_job_queues') +def test_filter_by_queue_aggregates_all_pages(mock_get_queues): + """Queue filtering must aggregate matching jobs across every API page and + return pagination metadata flagged for client-side pagination.""" + mock_get_queues.return_value = MOCK_QUEUE_LIST + # First page is full (100 jobs, 2 matching), second page is the last (1 matching) + page_1_jobs = [make_job(f"job_{i}", QUEUE_ID if i < 2 else "other_queue_id") + for i in range(100)] + page_2_jobs = [make_job("job_100", QUEUE_ID)] + common_params = {"teamId": WORKSPACE_ID, "archived.status": "false", "limit": 100} + responses.add( + responses.GET, + url=f"{CLOUDOS_URL}/api/v2/jobs", + json={"jobs": page_1_jobs, + "paginationMetadata": {"Pagination-Count": 101, "Pagination-Page": 1, + "Pagination-Limit": 100}}, + match=[matchers.query_param_matcher({**common_params, "page": 1})], + status=200 + ) + responses.add( + responses.GET, + url=f"{CLOUDOS_URL}/api/v2/jobs", + json={"jobs": page_2_jobs, + "paginationMetadata": {"Pagination-Count": 101, "Pagination-Page": 2, + "Pagination-Limit": 100}}, + match=[matchers.query_param_matcher({**common_params, "page": 2})], + status=200 + ) + clos = setup_clos() + result = clos.get_job_list(WORKSPACE_ID, filter_queue="v41", page=1, page_size=10) + jobs = result['jobs'] + assert [job["_id"] for job in jobs] == ["job_0", "job_1", "job_100"] + metadata = result['pagination_metadata'] + assert metadata['_client_filtered'] is True + assert metadata['Pagination-Count'] == 3 + assert metadata['Pagination-Limit'] == 10 diff --git a/tests/test_utils/test_requests.py b/tests/test_utils/test_requests.py new file mode 100644 index 00000000..f9523ff6 --- /dev/null +++ b/tests/test_utils/test_requests.py @@ -0,0 +1,29 @@ +import mock +import requests +from requests.adapters import HTTPAdapter + +from cloudos_cli.utils.requests import create_retry_session, retry_requests_get + + +def test_create_retry_session_mounts_retry_adapters(): + session = create_retry_session(total=3, status_forcelist=[500]) + assert isinstance(session, requests.Session) + for prefix in ('http://', 'https://'): + adapter = session.get_adapter(f'{prefix}example.com') + assert isinstance(adapter, HTTPAdapter) + assert adapter.max_retries.total == 3 + assert set(adapter.max_retries.status_forcelist) == {500} + + +def test_create_retry_session_allowed_methods(): + session = create_retry_session(allowed_methods=["DELETE"]) + adapter = session.get_adapter('https://example.com') + assert set(adapter.max_retries.allowed_methods) == {"DELETE"} + + +def test_retry_requests_get_reuses_given_session(): + session = mock.Mock() + session.get.return_value = 'response' + response = retry_requests_get('https://example.com', session=session, timeout=5) + assert response == 'response' + session.get.assert_called_once_with('https://example.com', timeout=5)