Merge pull request #22880 from ksuderman/gcp-batch-bug-fixes

[26.1] Fix memory-unit conversion and polling_interval bugs in the GCP Batch runner
This commit is contained in:
Marius van den Beek
2026-06-11 08:54:42 +02:00
committed by GitHub
3 changed files with 77 additions and 5 deletions
+12
View File
@@ -136,6 +136,18 @@ class GoogleCloudBatchJobRunner(AsynchronousJobRunner):
log.info("Google Cloud Batch client initialized successfully")
@property
def monitor_sleep_time(self):
"""Throttle the monitor loop so it polls the GCP Batch API no more often
than the configured ``polling_interval``.
The base ``AsynchronousJobRunner.monitor()`` loop sleeps this long between
sweeps; each runner has its own monitor thread, so this only affects the
GCP Batch runner. Item access (not ``.get()``) is used so the spec default
of 30 resolves through ``ParamsWithSpecs.__missing__``.
"""
return max(self.app.config.job_runner_monitor_sleep, int(self.runner_params["polling_interval"]))
def queue_job(self, job_wrapper):
"""Queue a job for execution on Google Cloud Batch."""
log.debug("Starting queue_job for Galaxy job %s", job_wrapper.get_id_tag())
@@ -12,6 +12,16 @@ DEFAULT_NFS_MOUNT_PATH = "/mnt/nfs"
DEFAULT_NFS_PATH = "/"
DEFAULT_MEMORY_MIB = 2048
DEFAULT_CPU_MILLI = 1000
# Memory-unit conversion factors to MiB (1 MiB = 1024 * 1024 bytes). Decimal units
# (KB/MB/GB) scale by powers of 1000; binary units (KiB/GiB) by powers of 1024.
# Precomputed so the constant arithmetic isn't re-evaluated on every conversion.
BYTES_PER_MIB = 1024 * 1024
KIB_TO_MIB = 1 / 1024
GIB_TO_MIB = 1024
KB_TO_MIB = 1000 / BYTES_PER_MIB
MB_TO_MIB = 1000 * 1000 / BYTES_PER_MIB
GB_TO_MIB = 1000 * 1000 * 1000 / BYTES_PER_MIB
DEFAULT_CVMFS_DOCKER_VOLUME = (
'-v "/cvmfs/data.galaxyproject.org:/cvmfs/data.galaxyproject.org:ro" '
'-v "/cvmfs/cloud.galaxyproject.org:/cvmfs/cloud.galaxyproject.org:ro"'
@@ -166,15 +176,15 @@ def convert_memory_to_mib(memory_str):
if unit in ["", "mib", "mi"]:
return int(value)
elif unit in ["gib", "gi"]:
return int(value * 1024) # GiB to MiB
return int(value * GIB_TO_MIB) # GiB to MiB
elif unit in ["mb", "m"]:
return int(value * 1000 / 1024) # MB to MiB (decimal to binary)
return int(value * MB_TO_MIB) # MB to MiB (decimal to binary)
elif unit in ["gb", "g"]:
return int(value * 1000 * 1000 / 1024 / 1024) # GB to MiB
return int(value * GB_TO_MIB) # GB to MiB (decimal to binary)
elif unit in ["kib", "ki"]:
return int(value / 1024) # KiB to MiB
return int(value * KIB_TO_MIB) # KiB to MiB
elif unit in ["kb", "k"]:
return int(value * 1000 / 1024 / 1024) # KB to MiB
return int(value * KB_TO_MIB) # KB to MiB (decimal to binary)
else:
log.warning("Unknown memory unit: %s, treating as MiB", unit)
return int(value)
@@ -1,7 +1,15 @@
"""Unit tests for Google Cloud Batch job runner utility methods."""
from types import SimpleNamespace
from typing import (
Any,
cast,
)
import pytest
from galaxy.jobs.runners import RunnerParams
from galaxy.jobs.runners.gcp_batch import GoogleCloudBatchJobRunner
from galaxy.jobs.runners.util.gcp_batch import (
convert_cpu_to_milli,
convert_duration_to_seconds,
@@ -81,6 +89,19 @@ class TestConvertMemoryToMib:
("1.5Gi", 1536), # decimal GiB
("256Mi", 256), # small MiB value
("4Gi", 4096), # larger GiB value
# Decimal units (KB/MB/GB use powers of 1000, converted to MiB)
("1gb", 953), # 1 GB -> MiB (10^9 / 1024^2)
("4gb", 3814), # larger GB value
("2g", 1907), # GB short form
("1000mb", 953), # 1000 MB == 1 GB == 953 MiB (consistency check)
("1024mb", 976), # MB -> MiB
("1000m", 953), # MB short form
("1048576kb", 1000), # KB -> MiB
# Binary KiB
("1024kib", 1), # KiB -> MiB (value / 1024)
("2048ki", 2), # KiB short form
# Unknown unit falls back to treating the value as MiB
("5xyz", 5),
],
)
def test_convert_memory_to_mib(self, input_value, expected):
@@ -307,3 +328,32 @@ class TestResolveMaxRunDuration:
resource_params={"walltime": ""},
)
assert result == "7200s"
def _sleep_runner(runner_params, job_runner_monitor_sleep=1.0):
"""A runner instance with __init__ skipped, for exercising monitor_sleep_time."""
runner = object.__new__(GoogleCloudBatchJobRunner)
runner.runner_params = runner_params
config = SimpleNamespace(job_runner_monitor_sleep=job_runner_monitor_sleep)
runner.app = cast(Any, SimpleNamespace(config=config))
return runner
class TestMonitorSleepTime:
"""The runner throttles its monitor loop to the configured polling_interval so
the base loop does not poll the GCP Batch API more often than necessary."""
def test_uses_polling_interval_over_default_monitor_sleep(self):
runner = _sleep_runner({"polling_interval": 30}, job_runner_monitor_sleep=1.0)
assert runner.monitor_sleep_time == 30
def test_respects_larger_monitor_sleep(self):
# max(global, polling_interval): a larger global sleep wins.
runner = _sleep_runner({"polling_interval": 30}, job_runner_monitor_sleep=60)
assert runner.monitor_sleep_time == 60
def test_interval_default_resolves_via_spec(self):
# polling_interval unset -> must fall back to the spec default (30) through
# ParamsWithSpecs.__missing__; .get() would yield None and raise in max().
runner = _sleep_runner(RunnerParams(specs={"polling_interval": dict(map=int, default=30)}, params={}))
assert runner.monitor_sleep_time == 30