opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-09-24 16:30:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #22871 from ksuderman/batch-job-id-prefix

Browse Source 
[26.1] Use the correct default value for job_id_prefix
This commit is contained in:
Marius van den Beek
2026-06-12 14:33:27 +02:00
committed by GitHub
parent e1539031f9 cacaebb283
commit 0f86c3c823
2 changed files with 127 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/jobs/runners/gcp_batch.py
+56 -50
View File
@@ -43,6 +43,56 @@ log = logging.getLogger(__name__)
__all__ = ("GoogleCloudBatchJobRunner",)
# Runner parameter specifications. The defaults defined here are served lazily by
# the ParamsWithSpecs defaultdict via __missing__, which is only triggered by
# subscript access (runner_params[key]) -- not by .get(). See _get_job_params.
RUNNER_PARAM_SPECS: dict[str, dict[str, Any]] = {
"project_id": dict(map=str, default=None),
"region": dict(map=str, default="us-central1"),
"zone": dict(map=str, default=None),
"service_account_file": dict(map=str, default=None),
"service_account_email": dict(map=str, default=None),
"machine_type": dict(map=str, default="n2-standard-4"),
"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=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),
# Extra docker volume mounts (format: "/host/path:/container/path[:ro],...")
"docker_extra_volumes": dict(map=str, default=None),
# Network configuration for NFS access
"network": dict(map=str, default="default"),
"subnet": dict(map=str, default="default"),
# Compute resource configuration (defaults - will be overridden by job requirements)
"vcpu": dict(map=float, default=1.0),
"memory_mib": dict(map=int, default=DEFAULT_MEMORY_MIB),
# Job-specific resource requests (same as Kubernetes runner)
"requests_cpu": dict(map=str, default=None),
"requests_memory": dict(map=str, default=None),
"limits_cpu": dict(map=str, default=None),
"limits_memory": dict(map=str, default=None),
# Container execution settings
"use_container": dict(map=bool, default=True),
"galaxy_user_id": dict(
map=str, valid=lambda s: s == "$uid" or isinstance(s, int) or not s or str(s).isdigit(), default=None
),
"galaxy_group_id": dict(
map=str, valid=lambda s: s == "$gid" or isinstance(s, int) or not s or str(s).isdigit(), default=None
),
# Custom VM image (optional)
"custom_vm_image": dict(map=str, default=None),
# Job cleanup: if true, delete GCP Batch jobs after Galaxy marks them complete
"delete_completed_jobs": dict(map=bool, default=True),
# Prefix for GCP Batch job IDs (helps identify which Galaxy server submitted a job)
"job_id_prefix": dict(map=str, default="galaxy-job"),
# Object store fallback (for future use)
"use_object_store": dict(map=bool, default=False),
"object_store_path": dict(map=str, default=None),
}
class GoogleCloudBatchJobRunner(AsynchronousJobRunner):
"""
Job runner that submits jobs to Google Cloud Batch.
@@ -54,54 +104,7 @@ class GoogleCloudBatchJobRunner(AsynchronousJobRunner):
"""Initialize the Google Cloud Batch job runner."""
log.debug("Starting GoogleCloudBatchJobRunner.__init__")
# Define runner parameter specifications
runner_param_specs = {
"project_id": dict(map=str, default=None),
"region": dict(map=str, default="us-central1"),
"zone": dict(map=str, default=None),
"service_account_file": dict(map=str, default=None),
"service_account_email": dict(map=str, default=None),
"machine_type": dict(map=str, default="n2-standard-4"),
"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=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),
# Extra docker volume mounts (format: "/host/path:/container/path[:ro],...")
"docker_extra_volumes": dict(map=str, default=None),
# Network configuration for NFS access
"network": dict(map=str, default="default"),
"subnet": dict(map=str, default="default"),
# Compute resource configuration (defaults - will be overridden by job requirements)
"vcpu": dict(map=float, default=1.0),
"memory_mib": dict(map=int, default=DEFAULT_MEMORY_MIB),
# Job-specific resource requests (same as Kubernetes runner)
"requests_cpu": dict(map=str, default=None),
"requests_memory": dict(map=str, default=None),
"limits_cpu": dict(map=str, default=None),
"limits_memory": dict(map=str, default=None),
# Container execution settings
"use_container": dict(map=bool, default=True),
"galaxy_user_id": dict(
map=str, valid=lambda s: s == "$uid" or isinstance(s, int) or not s or str(s).isdigit(), default=None
),
"galaxy_group_id": dict(
map=str, valid=lambda s: s == "$gid" or isinstance(s, int) or not s or str(s).isdigit(), default=None
),
# Custom VM image (optional)
"custom_vm_image": dict(map=str, default=None),
# Job cleanup: if true, delete GCP Batch jobs after Galaxy marks them complete
"delete_completed_jobs": dict(map=bool, default=True),
# Prefix for GCP Batch job IDs (helps identify which Galaxy server submitted a job)
"job_id_prefix": dict(map=str, default="galaxy-job"),
# Object store fallback (for future use)
"use_object_store": dict(map=bool, default=False),
"object_store_path": dict(map=str, default=None),
}
kwargs.update({"runner_param_specs": runner_param_specs})
kwargs.update({"runner_param_specs": RUNNER_PARAM_SPECS})
super().__init__(app, nworkers, **kwargs)
# Initialize Google Cloud Batch client
@@ -213,7 +216,7 @@ class GoogleCloudBatchJobRunner(AsynchronousJobRunner):
params = self._get_job_params(job_destination)
# Generate unique job name
prefix = params.get("job_id_prefix", "galaxy-job")
prefix = params.get("job_id_prefix") or "galaxy-job"
job_name = f"{prefix}-{int(time.time())}-{os.urandom(4).hex()}-{job_wrapper.get_id_tag()}"
# Create the batch job specification
@@ -269,7 +272,10 @@ class GoogleCloudBatchJobRunner(AsynchronousJobRunner):
"service_account_email",
"job_id_prefix",
]:
params[key] = job_destination.params.get(key, self.runner_params.get(key))
# Subscript access on runner_params (a defaultdict) so unset keys fall
# back to the spec defaults defined in runner_param_specs; .get() would
# bypass __missing__ and yield None instead of the configured default.
params[key] = job_destination.params.get(key, self.runner_params[key])
log.debug("Finished _get_job_params")
return params
test/unit/app/jobs/test_gcp_batch_runner.py
+71 -1
View File
@@ -9,7 +9,10 @@ from typing import (
import pytest
from galaxy.jobs.runners import RunnerParams
from galaxy.jobs.runners.gcp_batch import GoogleCloudBatchJobRunner
from galaxy.jobs.runners.gcp_batch import (
GoogleCloudBatchJobRunner,
RUNNER_PARAM_SPECS,
)
from galaxy.jobs.runners.util.gcp_batch import (
convert_cpu_to_milli,
convert_duration_to_seconds,
@@ -357,3 +360,70 @@ class TestMonitorSleepTime:
# 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
def _make_runner(runner_params=None):
"""Build a GoogleCloudBatchJobRunner without running __init__ (no GCP client).
_get_job_params only depends on self.runner_params, so we set that directly.
"""
runner = object.__new__(GoogleCloudBatchJobRunner)
runner.runner_params = RunnerParams(specs=RUNNER_PARAM_SPECS, params=runner_params or {})
return runner
# Keys that _get_job_params copies from the destination / runner config. Derived
# dynamically (rather than hard-coded) so the parametrized tests below automatically
# cover any parameter added to _get_job_params in the future.
JOB_PARAM_KEYS = sorted(_make_runner()._get_job_params(SimpleNamespace(params={})).keys())
class TestGetJobParams:
"""Tests for GoogleCloudBatchJobRunner._get_job_params default resolution."""
@pytest.mark.parametrize("key", JOB_PARAM_KEYS)
def test_unset_param_falls_back_to_spec_default(self, key):
"""Every copied parameter resolves to its spec default when nothing overrides it.
This is the regression guard: .get() on the RunnerParams defaultdict would
bypass __missing__ and yield None instead of the configured default.
"""
runner = _make_runner()
destination = SimpleNamespace(params={})
params = runner._get_job_params(destination)
assert params[key] == RUNNER_PARAM_SPECS[key]["default"]
@pytest.mark.parametrize("key", JOB_PARAM_KEYS)
def test_destination_overrides_every_param(self, key):
"""A value on the job destination takes precedence over the spec default for every param.
Destination params are not passed through the RunnerParams spec mapping, so a
plain string sentinel is a valid override for every key.
"""
runner = _make_runner()
sentinel = "destination-sentinel-value"
destination = SimpleNamespace(params={key: sentinel})
params = runner._get_job_params(destination)
assert params[key] == sentinel
def test_runner_config_overrides_default(self):
"""A value set in the runner (plugin) config is used when the destination is silent."""
runner = _make_runner({"job_id_prefix": "from-config"})
destination = SimpleNamespace(params={})
params = runner._get_job_params(destination)
assert params["job_id_prefix"] == "from-config"
def test_destination_overrides_runner_config(self):
"""Destination params win over runner config, which wins over the spec default."""
runner = _make_runner({"job_id_prefix": "from-config"})
destination = SimpleNamespace(params={"job_id_prefix": "from-destination"})
params = runner._get_job_params(destination)
assert params["job_id_prefix"] == "from-destination"