Merge pull request #21928 from ksuderman/set-max-run-duration

Allow per-job max_run_duration for GCP Batch runner

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

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",

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.

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"