diff --git a/lib/galaxy/jobs/runners/gcp_batch.py b/lib/galaxy/jobs/runners/gcp_batch.py index e03f27d9627..7245f3e8a1d 100644 --- a/lib/galaxy/jobs/runners/gcp_batch.py +++ b/lib/galaxy/jobs/runners/gcp_batch.py @@ -27,12 +27,14 @@ from galaxy.jobs.runners.util.gcp_batch import ( convert_cpu_to_milli, convert_memory_to_mib, DEFAULT_CVMFS_DOCKER_VOLUME, + DEFAULT_MAX_RUN_DURATION, DEFAULT_MEMORY_MIB, DEFAULT_NFS_MOUNT_PATH, DEFAULT_NFS_PATH, DIRECT_SCRIPT_TEMPLATE, parse_docker_volumes_param, parse_volumes_param, + resolve_max_run_duration, sanitize_label_value, ) @@ -63,7 +65,7 @@ class GoogleCloudBatchJobRunner(AsynchronousJobRunner): "boot_disk_size_gb": dict(map=int, default=100), "boot_disk_type": dict(map=str, default="pd-standard"), "max_retry_count": dict(map=int, default=3), - "max_run_duration": dict(map=str, default="3600s"), + "max_run_duration": dict(map=str, default=DEFAULT_MAX_RUN_DURATION), "polling_interval": dict(map=int, default=30), # Volume configuration (generic format: "server:/remote_path:/mount_path[:ro],...") "gcp_batch_volumes": dict(map=str, default=None), @@ -267,6 +269,11 @@ class GoogleCloudBatchJobRunner(AsynchronousJobRunner): # Get compute resources first so we can pass them to script creation cpu_milli, memory_mib = self._get_job_resources(job_wrapper, params) + # Get max run duration (resolves per-job from destination, resource params, or default) + max_run_duration = resolve_max_run_duration( + job_wrapper.job_destination.params, params, job_wrapper.get_resource_parameters() + ) + # Create the execution script based on whether we use containers or not if params.get("use_container", True): execution_script = self._create_container_execution_script( @@ -284,7 +291,7 @@ class GoogleCloudBatchJobRunner(AsynchronousJobRunner): task_spec = batch_v1.TaskSpec() task_spec.runnables = [runnable] task_spec.max_retry_count = params["max_retry_count"] - task_spec.max_run_duration = params["max_run_duration"] + task_spec.max_run_duration = max_run_duration # Set compute resources compute_resource = batch_v1.ComputeResource() @@ -293,10 +300,11 @@ class GoogleCloudBatchJobRunner(AsynchronousJobRunner): task_spec.compute_resource = compute_resource log.debug( - "Configured compute resources for job %s: %d mCPU, %d MiB memory", + "Configured compute resources for job %s: %d mCPU, %d MiB memory, max_run_duration=%s", job_wrapper.get_id_tag(), cpu_milli, memory_mib, + max_run_duration, ) # Configure NFS volumes from gcp_batch_volumes parameter diff --git a/lib/galaxy/jobs/runners/util/gcp_batch/__init__.py b/lib/galaxy/jobs/runners/util/gcp_batch/__init__.py index 2b2cd8cb50e..7bad7c8a5a6 100644 --- a/lib/galaxy/jobs/runners/util/gcp_batch/__init__.py +++ b/lib/galaxy/jobs/runners/util/gcp_batch/__init__.py @@ -6,15 +6,18 @@ from galaxy.util.resources import resource_string from .helpers import ( compute_machine_type, convert_cpu_to_milli, + convert_duration_to_seconds, convert_memory_to_mib, DEFAULT_CPU_MILLI, DEFAULT_CVMFS_DOCKER_VOLUME, + DEFAULT_MAX_RUN_DURATION, DEFAULT_MEMORY_MIB, DEFAULT_NFS_MOUNT_PATH, DEFAULT_NFS_PATH, parse_docker_volumes_param, parse_volume_spec, parse_volumes_param, + resolve_max_run_duration, sanitize_label_value, ) @@ -25,6 +28,7 @@ __all__ = ( "CONTAINER_SCRIPT_TEMPLATE", "DEFAULT_CPU_MILLI", "DEFAULT_CVMFS_DOCKER_VOLUME", + "DEFAULT_MAX_RUN_DURATION", "DEFAULT_MEMORY_MIB", "DEFAULT_NFS_MOUNT_PATH", "DEFAULT_NFS_PATH", @@ -32,7 +36,9 @@ __all__ = ( "compute_machine_type", "convert_cpu_to_milli", "convert_memory_to_mib", + "convert_duration_to_seconds", "parse_docker_volumes_param", + "resolve_max_run_duration", "parse_volume_spec", "parse_volumes_param", "sanitize_label_value", diff --git a/lib/galaxy/jobs/runners/util/gcp_batch/helpers.py b/lib/galaxy/jobs/runners/util/gcp_batch/helpers.py index 8e2cad1da38..ee67b99d80d 100644 --- a/lib/galaxy/jobs/runners/util/gcp_batch/helpers.py +++ b/lib/galaxy/jobs/runners/util/gcp_batch/helpers.py @@ -3,6 +3,8 @@ import logging import re +from humanfriendly import parse_timespan + log = logging.getLogger(__name__) # Default values for GCP Batch runner configuration @@ -178,6 +180,60 @@ def convert_memory_to_mib(memory_str): return int(value) +DEFAULT_MAX_RUN_DURATION = "86400s" + + +def convert_duration_to_seconds(duration_str) -> str: + """ + Convert a duration value to GCP Batch duration format (e.g., '3600s'). + + Accepts: + - Duration string with suffix: '3600s', '60m', '2h', '1d' + - Human-friendly strings: '1.5 hours', '2 days' + - Numeric string or number (treated as seconds): '3600', 3600 + + Returns: + Duration string in GCP Batch format (e.g., '3600s'). + """ + if not duration_str: + return DEFAULT_MAX_RUN_DURATION + try: + return f"{int(parse_timespan(str(duration_str)))}s" + except Exception: + log.warning("Invalid duration format: %s, using default %s", duration_str, DEFAULT_MAX_RUN_DURATION) + return DEFAULT_MAX_RUN_DURATION + + +def resolve_max_run_duration(destination_params, runner_params, resource_params): + """ + Resolve the maximum run duration from multiple configuration sources. + + Resolution order (highest priority first): + 1. Galaxy job resource parameter 'walltime' (user selection in tool form) + 2. Destination param 'max_run_duration' (e.g., from TPV per-tool config) + 3. TPV-style 'walltime' in destination params + 4. Runner-level default from runner_params + + Args: + destination_params: dict of job destination parameters + runner_params: dict of runner-level parameters (fallback defaults) + resource_params: dict of Galaxy job resource parameters (user selections) + + Returns: + Duration string in GCP Batch format (e.g., '86400s') + """ + if resource_params.get("walltime"): + return convert_duration_to_seconds(resource_params["walltime"]) + + if "max_run_duration" in destination_params: + return convert_duration_to_seconds(destination_params["max_run_duration"]) + + if "walltime" in destination_params: + return convert_duration_to_seconds(destination_params["walltime"]) + + return convert_duration_to_seconds(runner_params.get("max_run_duration", DEFAULT_MAX_RUN_DURATION)) + + def compute_machine_type(cpu_milli, memory_mib, machine_type_family="n2"): """ Compute an appropriate GCP machine type based on resource requirements. diff --git a/test/unit/app/jobs/test_gcp_batch_runner.py b/test/unit/app/jobs/test_gcp_batch_runner.py index b53692c7488..22e65225e0d 100644 --- a/test/unit/app/jobs/test_gcp_batch_runner.py +++ b/test/unit/app/jobs/test_gcp_batch_runner.py @@ -4,10 +4,13 @@ import pytest from galaxy.jobs.runners.util.gcp_batch import ( convert_cpu_to_milli, + convert_duration_to_seconds, convert_memory_to_mib, + DEFAULT_MAX_RUN_DURATION, parse_docker_volumes_param, parse_volume_spec, parse_volumes_param, + resolve_max_run_duration, sanitize_label_value, ) @@ -191,3 +194,116 @@ class TestParseDockerVolumesParam: def test_cvmfs_example(self): result = parse_docker_volumes_param("/cvmfs/data.galaxyproject.org:/cvmfs/data.galaxyproject.org:ro") assert result == '-v "/cvmfs/data.galaxyproject.org:/cvmfs/data.galaxyproject.org:ro"' + + +class TestConvertDurationToSeconds: + """Tests for convert_duration_to_seconds helper function.""" + + @pytest.mark.parametrize( + "input_value,expected", + [ + ("3600s", "3600s"), # seconds suffix + ("86400s", "86400s"), # larger seconds value + ("0s", "0s"), # zero seconds + ("30m", "1800s"), # minutes to seconds + ("90m", "5400s"), # larger minutes + ("2h", "7200s"), # hours to seconds + ("24h", "86400s"), # 24 hours + ("1d", "86400s"), # days to seconds + ("7d", "604800s"), # 7 days + ("3600", "3600s"), # plain integer string (seconds assumed) + ("86400", "86400s"), # plain integer string + (3600, "3600s"), # numeric int + (7200, "7200s"), # numeric int + (7200.0, "7200s"), # numeric float + ("1.5h", "5400s"), # fractional hours + ("2.5d", "216000s"), # fractional days + ("0.5h", "1800s"), # half hour + ("", DEFAULT_MAX_RUN_DURATION), # empty string -> default + (None, DEFAULT_MAX_RUN_DURATION), # None -> default + ("invalid", DEFAULT_MAX_RUN_DURATION), # garbage -> default + ("abcs", DEFAULT_MAX_RUN_DURATION), # invalid with s suffix -> default + ("xxh", DEFAULT_MAX_RUN_DURATION), # invalid with h suffix -> default + ("zzm", DEFAULT_MAX_RUN_DURATION), # invalid with m suffix -> default + ("qqd", DEFAULT_MAX_RUN_DURATION), # invalid with d suffix -> default + ], + ) + def test_convert_duration_to_seconds(self, input_value, expected): + result = convert_duration_to_seconds(input_value) + assert result == expected + + +class TestResolveMaxRunDuration: + """Tests for resolve_max_run_duration priority resolution.""" + + def test_resource_param_walltime_highest_priority(self): + """User-specified walltime wins over everything.""" + result = resolve_max_run_duration( + destination_params={"max_run_duration": "2h"}, + runner_params={"max_run_duration": "1h"}, + resource_params={"walltime": "3600"}, + ) + assert result == "3600s" + + def test_destination_max_run_duration_over_dest_walltime(self): + """Destination 'max_run_duration' beats destination 'walltime'.""" + result = resolve_max_run_duration( + destination_params={"max_run_duration": "2h", "walltime": "1h"}, + runner_params={"max_run_duration": "86400s"}, + resource_params={}, + ) + assert result == "7200s" + + def test_destination_walltime_over_runner_default(self): + """Destination 'walltime' beats the runner-level default.""" + result = resolve_max_run_duration( + destination_params={"walltime": "4h"}, + runner_params={"max_run_duration": "86400s"}, + resource_params={}, + ) + assert result == "14400s" + + def test_runner_default_fallback(self): + """Falls back to runner-level max_run_duration when nothing else set.""" + result = resolve_max_run_duration( + destination_params={}, + runner_params={"max_run_duration": "3600s"}, + resource_params={}, + ) + assert result == "3600s" + + def test_global_default_when_nothing_set(self): + """Falls back to DEFAULT_MAX_RUN_DURATION when params dict has no key.""" + result = resolve_max_run_duration( + destination_params={}, + runner_params={}, + resource_params={}, + ) + assert result == DEFAULT_MAX_RUN_DURATION + + def test_resource_walltime_over_destination_max_run_duration(self): + """User walltime overrides destination max_run_duration.""" + result = resolve_max_run_duration( + destination_params={"max_run_duration": "1h"}, + runner_params={"max_run_duration": "86400s"}, + resource_params={"walltime": "2d"}, + ) + assert result == "172800s" + + def test_destination_max_run_duration_normalizes_format(self): + """Duration values are normalized through convert_duration_to_seconds.""" + result = resolve_max_run_duration( + destination_params={"max_run_duration": "48h"}, + runner_params={}, + resource_params={}, + ) + assert result == "172800s" + + def test_empty_walltime_ignored(self): + """Empty walltime in resource params is skipped.""" + result = resolve_max_run_duration( + destination_params={}, + runner_params={"max_run_duration": "7200s"}, + resource_params={"walltime": ""}, + ) + assert result == "7200s"