From be98a95e77d1f9e6fc444663b2ad41deec6d2665 Mon Sep 17 00:00:00 2001 From: mschwoerer <82171591+mschwoer@users.noreply.github.com> Date: Tue, 1 Sep 2026 23:52:20 +0200 Subject: [PATCH 1/2] resolve the docker job paths per view instead of translating between them --- # Conversation that produced these changes --- ## User prompt "now C6" "hm, we need relativize just for the docker host .. could the paths be treated differently there? e.g. just reuse the cluster view paths and do the mounting accordingly?" ## Clarifying round 1 Q: How to handle the docker host path translation? - B: relative path in QuantingEnv (Recommended) <-- chosen - C: keep relativize as it is - A: host mounts mirror cluster paths Co-Authored-By: Claude Opus 5 (1M context) --- airflow_src/dags/impl/processor_impl.py | 13 ++--- airflow_src/plugins/common/quanting_env.py | 3 +- .../plugins/jobs/docker_job_handler.py | 43 +++++++-------- airflow_src/plugins/jobs/job_handler.py | 11 +++- airflow_src/tests/conftest.py | 3 +- .../tests/dags/impl/test_processor_impl.py | 55 ++++++++++--------- airflow_src/tests/helpers.py | 19 ++++++- .../plugins/jobs/test_docker_job_handler.py | 27 ++++++--- .../tests/plugins/jobs/test_job_handler.py | 25 ++++++--- .../PATH_HANDLING_IMPLEMENTATION_PLAN.md | 19 ++++++- shared/yamlsettings.py | 16 ------ 11 files changed, 133 insertions(+), 101 deletions(-) diff --git a/airflow_src/dags/impl/processor_impl.py b/airflow_src/dags/impl/processor_impl.py index d547656d..39244a58 100644 --- a/airflow_src/dags/impl/processor_impl.py +++ b/airflow_src/dags/impl/processor_impl.py @@ -23,7 +23,6 @@ XComKeys, ) from common.paths import ( - get_internal_backup_path, get_internal_output_path, get_internal_output_path_for_raw_file, ) @@ -182,7 +181,6 @@ def _create_quanting_env( ) output_path = CLUSTER_VIEW.resolve(Locations.OUTPUT, relative_output_path) - internal_output_path = get_internal_output_path() / relative_output_path substituted_params = _substitute_config_params( raw_file.id, @@ -225,8 +223,7 @@ def _create_quanting_env( project_id=raw_file.project_id, settings_name=settings.name, settings_version=settings.version, - internal_output_path=str(internal_output_path), - internal_raw_file_path=str(get_internal_backup_path() / relative_raw_file_path), + relative_raw_file_path=str(relative_raw_file_path), config_params=substituted_params, job_engine=settings.job_engine, year_month_folder=get_created_at_year_month(raw_file), @@ -278,8 +275,6 @@ def _check_content(quanting_env: QuantingEnv, settings: Settings) -> list[str]: "raw_file_path", "settings_path", "output_path", - "internal_output_path", - "internal_raw_file_path", "software", ] # these hold resolved paths and are space-separated, so they need the laxer checks @@ -360,7 +355,7 @@ def submit_job( raise AirflowSkipException("Skipping quanting due to instrument settings.") # upfront check 2 - output_path = Path(quanting_env.internal_output_path) + output_path = get_internal_output_path() / quanting_env.relative_output_path if output_path.exists(): msg = f"Output path {output_path} already exists with different content." output_exists_mode = get_airflow_variable( @@ -482,7 +477,7 @@ def check_job_result(*, quanting_env_dict: dict, job_id: str, ti: TaskInstance) JobStates.OUT_OF_MEMORY ): raw_file = get_raw_file_by_id(quanting_env.raw_file_id) - output_path = Path(quanting_env.internal_output_path) + output_path = get_internal_output_path() / quanting_env.relative_output_path if job_status == JobStates.FAILED: if quanting_env.software_type == SoftwareTypes.ALPHADIA: @@ -536,7 +531,7 @@ def compute_metrics( quanting_env = QuantingEnv.from_dict(quanting_env_dict) metrics_type = quanting_env.metrics_type - output_path = Path(quanting_env.internal_output_path) + output_path = get_internal_output_path() / quanting_env.relative_output_path metrics = calc_metrics(output_path, metrics_type=metrics_type) diff --git a/airflow_src/plugins/common/quanting_env.py b/airflow_src/plugins/common/quanting_env.py index c35976c0..f89bd36e 100644 --- a/airflow_src/plugins/common/quanting_env.py +++ b/airflow_src/plugins/common/quanting_env.py @@ -40,8 +40,7 @@ class QuantingEnv(BaseModel): settings_name: str settings_version: int - internal_output_path: str = Field(alias="_INTERNAL_OUTPUT_PATH") - internal_raw_file_path: str = Field(alias="_INTERNAL_RAW_FILE_PATH") + relative_raw_file_path: str = Field(alias="_RELATIVE_RAW_FILE_PATH") config_params: str = Field(alias="_CONFIG_PARAMS") job_engine: str = Field(alias="_JOB_ENGINE") year_month_folder: str = Field(alias="_YEAR_MONTH_FOLDER") diff --git a/airflow_src/plugins/jobs/docker_job_handler.py b/airflow_src/plugins/jobs/docker_job_handler.py index 352a0b72..ae319d96 100644 --- a/airflow_src/plugins/jobs/docker_job_handler.py +++ b/airflow_src/plugins/jobs/docker_job_handler.py @@ -25,7 +25,7 @@ import re import shlex from datetime import datetime -from pathlib import Path +from pathlib import Path, PurePosixPath import docker from airflow.exceptions import AirflowFailException @@ -35,7 +35,7 @@ from docker.models.containers import Container from jobs.job_handler import JobHandler -from shared.keys import InternalPaths +from shared.path_views import AIRFLOW_CONTAINER_VIEW, Locations, View CONTAINER_NAME_PREFIX = "kraken" # docker accepts only [a-zA-Z0-9][a-zA-Z0-9_.-]* as container name, but raw file names may @@ -65,17 +65,17 @@ class DockerJobHandler(JobHandler): """Implementation of JobHandler that runs jobs in Docker containers on the AlphaKraken host.""" - def __init__(self, host_mounts_path: Path): + def __init__(self, docker_host_view: View[PurePosixPath]): """Initialize the docker job handler. Args: - host_mounts_path: Path of the mounts folder as seen by the docker host + docker_host_view: The data directories as seen by the docker host (not by the containers) """ super().__init__() self._client = docker.from_env() - self._host_mounts_path = host_mounts_path + self._docker_host_view = docker_host_view def start_job(self, quanting_env: QuantingEnv) -> str: """Start a job by running a container on the AlphaKraken host. @@ -91,8 +91,12 @@ def start_job(self, quanting_env: QuantingEnv) -> str: # None makes docker use the command defined in the image command = shlex.split(quanting_env.config_params) or None - internal_raw_file_path = Path(quanting_env.internal_raw_file_path) - internal_output_path = Path(quanting_env.internal_output_path) + internal_raw_file_path = AIRFLOW_CONTAINER_VIEW.resolve( + Locations.BACKUP, quanting_env.relative_raw_file_path + ) + internal_output_path = AIRFLOW_CONTAINER_VIEW.resolve( + Locations.OUTPUT, quanting_env.relative_output_path + ) for path in (internal_raw_file_path, internal_output_path): if not path.exists(): raise AirflowFailException(f"Path {path} does not exist in the worker.") @@ -106,11 +110,19 @@ def start_job(self, quanting_env: QuantingEnv) -> str: # bind at the paths the placeholders in the config params resolved to, so that the same # config params work for this engine and for Slurm volumes = { - str(self._to_host_path(internal_raw_file_path)): { + str( + self._docker_host_view.resolve( + Locations.BACKUP, quanting_env.relative_raw_file_path + ) + ): { "bind": quanting_env.raw_file_path, "mode": "ro", }, - str(self._to_host_path(internal_output_path)): { + str( + self._docker_host_view.resolve( + Locations.OUTPUT, quanting_env.relative_output_path + ) + ): { "bind": quanting_env.output_path, "mode": "rw", }, @@ -182,19 +194,6 @@ def _get_image(self, image: str) -> str: return image - def _to_host_path(self, internal_path: Path) -> Path: - """Translate a path within the worker container to the corresponding host path. - - This trick enables to access the files on the container file system with the same paths as on the shared file system. - - E.g. /opt/airflow/mounts/output/P1/out_file.raw/custom - -> /home/kraken-user/alphakraken/production/mounts/output/P1/out_file.raw/custom - for `locations.general.mounts_path: /home/kraken-user/alphakraken/production/mounts`. - """ - return self._host_mounts_path / internal_path.relative_to( - InternalPaths.MOUNTS_PATH - ) - def _get_container(self, job_id: str) -> Container | None: """Get the container with the given id, None if it does not exist (anymore).""" try: diff --git a/airflow_src/plugins/jobs/job_handler.py b/airflow_src/plugins/jobs/job_handler.py index a25db3f8..e7b5f30c 100644 --- a/airflow_src/plugins/jobs/job_handler.py +++ b/airflow_src/plugins/jobs/job_handler.py @@ -11,8 +11,7 @@ from common.quanting_env import QuantingEnv from shared.keys import JobEngines -from shared.path_views import CLUSTER_VIEW, Locations -from shared.yamlsettings import get_host_mounts_path +from shared.path_views import CLUSTER_VIEW, DOCKER_HOST_VIEW, Locations def _get_job_handler(engine: str) -> "JobHandler": @@ -32,8 +31,14 @@ def _get_job_handler(engine: str) -> "JobHandler": f"airflow_src/requirements_docker_job_engine.txt to be installed." ) from e + if not DOCKER_HOST_VIEW.has(Locations.OUTPUT): + raise AirflowFailException( + f"The '{JobEngines.DOCKER}' job engine requires the key " + f"`locations.general.mounts_path` in alphakraken.yaml." + ) + logging.info("Using DockerJobHandler") - return DockerJobHandler(get_host_mounts_path()) + return DockerJobHandler(DOCKER_HOST_VIEW) if engine == JobEngines.FILE_BASED: from jobs._experimental.file_based_job_handler import FileBasedJobHandler diff --git a/airflow_src/tests/conftest.py b/airflow_src/tests/conftest.py index b88ef071..e0c1dac3 100644 --- a/airflow_src/tests/conftest.py +++ b/airflow_src/tests/conftest.py @@ -29,8 +29,7 @@ "project_id": "PID1", "settings_name": "test_settings", "settings_version": 1, - "internal_output_path": "/opt/airflow/mounts/output/PID1/out_test_file.raw/alphadia", - "internal_raw_file_path": "/opt/airflow/mounts/backup/instrument1/1970_01/test_file.raw", + "relative_raw_file_path": "instrument1/1970_01/test_file.raw", "config_params": "", "job_engine": "slurm", "year_month_folder": "1970_01", diff --git a/airflow_src/tests/dags/impl/test_processor_impl.py b/airflow_src/tests/dags/impl/test_processor_impl.py index 37c0b5bd..68d5ee3a 100644 --- a/airflow_src/tests/dags/impl/test_processor_impl.py +++ b/airflow_src/tests/dags/impl/test_processor_impl.py @@ -35,7 +35,7 @@ XComKeys, ) -from airflow_src.tests.helpers import yaml_locations +from airflow_src.tests.helpers import container_locations, yaml_locations from shared.db.models import RawFile, RawFileStatus from shared.keys import JobEngines @@ -105,8 +105,7 @@ def test_create_quanting_env( "SETTINGS_VERSION": 1, "_JOB_ENGINE": "slurm", "_YEAR_MONTH_FOLDER": "1970_01", - "_INTERNAL_OUTPUT_PATH": "/opt/airflow/mounts/output/some_project_id/out_test_file.raw/alphadia", - "_INTERNAL_RAW_FILE_PATH": "/opt/airflow/mounts/backup/instrument1/1970_01/test_file.raw", + "_RELATIVE_RAW_FILE_PATH": "instrument1/1970_01/test_file.raw", "_CONFIG_PARAMS": "", } assert result.to_dict() == expected @@ -189,8 +188,7 @@ def test_create_quanting_env_custom_software( "SETTINGS_VERSION": 1, "_JOB_ENGINE": "slurm", "_YEAR_MONTH_FOLDER": "1970_01", - "_INTERNAL_OUTPUT_PATH": "/opt/airflow/mounts/output/some_project_id/out_test_file.raw/custom", - "_INTERNAL_RAW_FILE_PATH": "/opt/airflow/mounts/backup/instrument1/1970_01/test_file.raw", + "_RELATIVE_RAW_FILE_PATH": "instrument1/1970_01/test_file.raw", "_CONFIG_PARAMS": expected_config_params, } assert result.to_dict() == expected @@ -312,7 +310,7 @@ def test_prepare_job( mock_get_raw_file_by_id.return_value = mock_raw_file mock_settings = MagicMock(config_params=[]) mock_get_settings_by_id.return_value = mock_settings - mock_env = make_quanting_env(internal_output_path="/nonexistent/output/path") + mock_env = make_quanting_env() mock_create_env.return_value = mock_env result = prepare_job(raw_file_id="test_file.raw", settings_id="sid1") @@ -474,8 +472,9 @@ def test_submit_job_executes_ssh_command_and_stores_job_id( ) -> None: """Test that the submit_job function executes the SSH command and stores the job ID.""" # given - output_dir = tmp_path / "PID123" / "out_test_file.raw" / "alphadia" - quanting_env = make_quanting_env(internal_output_path=str(output_dir)) + relative_output_path = "PID123/out_test_file.raw/alphadia" + output_dir = tmp_path / relative_output_path + quanting_env = make_quanting_env(relative_output_path=relative_output_path) mock_raw_file = MagicMock( wraps=RawFile, created_at=datetime.fromtimestamp(0, tz=pytz.UTC), @@ -486,7 +485,8 @@ def test_submit_job_executes_ssh_command_and_stores_job_id( mock_start_job.return_value = "12345" # when - result = submit_job(quanting_env_dict=quanting_env.to_dict()) + with container_locations(output=str(tmp_path)): + result = submit_job(quanting_env_dict=quanting_env.to_dict()) assert result == "12345" assert output_dir.exists() @@ -512,7 +512,7 @@ def test_submit_job_output_folder_exists( # given output_dir = tmp_path / "output" output_dir.mkdir() - quanting_env = make_quanting_env(internal_output_path=str(output_dir)) + quanting_env = make_quanting_env(relative_output_path="output") mock_raw_file = MagicMock( wraps=RawFile, created_at=datetime.fromtimestamp(0, tz=pytz.UTC), @@ -523,7 +523,10 @@ def test_submit_job_output_folder_exists( mock_get_airflow_variable.return_value = "raise" # when - with pytest.raises(AirflowFailException): + with ( + container_locations(output=str(tmp_path)), + pytest.raises(AirflowFailException), + ): submit_job(quanting_env_dict=quanting_env.to_dict()) mock_get_raw_file_by_id.assert_called_once_with("test_file.raw") @@ -544,7 +547,7 @@ def test_submit_job_output_folder_exists_associate( # given output_dir = tmp_path / "output" output_dir.mkdir() - quanting_env = make_quanting_env(internal_output_path=str(output_dir)) + quanting_env = make_quanting_env(relative_output_path="output") mock_raw_file = MagicMock( wraps=RawFile, created_at=datetime.fromtimestamp(0, tz=pytz.UTC), @@ -556,7 +559,8 @@ def test_submit_job_output_folder_exists_associate( mock_get_slurm_job_id_from_log.return_value = "54321" # when - result = submit_job(quanting_env_dict=quanting_env.to_dict()) + with container_locations(output=str(tmp_path)): + result = submit_job(quanting_env_dict=quanting_env.to_dict()) assert result == "54321" @@ -575,7 +579,7 @@ def test_submit_job_output_folder_exists_associate_raise( # given output_dir = tmp_path / "output" output_dir.mkdir() - quanting_env = make_quanting_env(internal_output_path=str(output_dir)) + quanting_env = make_quanting_env(relative_output_path="output") mock_raw_file = MagicMock( wraps=RawFile, created_at=datetime.fromtimestamp(0, tz=pytz.UTC), @@ -587,7 +591,10 @@ def test_submit_job_output_folder_exists_associate_raise( mock_get_slurm_job_id_from_log.return_value = None # when - with pytest.raises(AirflowFailException): + with ( + container_locations(output=str(tmp_path)), + pytest.raises(AirflowFailException), + ): submit_job(quanting_env_dict=quanting_env.to_dict()) @@ -666,17 +673,14 @@ def test_prepare_job_add_mode( # noqa: PLR0913 output="/some_output_path", software="/some_software_base_path", ) -@patch("dags.impl.processor_impl.get_internal_output_path") @patch("dags.impl.processor_impl.get_output_folder_rel_path") def test_create_quanting_env_with_suffix( mock_output_rel_path: MagicMock, - mock_internal_output_path: MagicMock, ) -> None: """Test that _create_quanting_env applies the suffix to all output paths, incl. the config params.""" mock_output_rel_path.return_value = Path( "some_project_id/out_test_file.raw/alphadia" ) - mock_internal_output_path.return_value = Path("/opt/airflow/mounts/output") mock_settings = MagicMock( software_type="custom", @@ -715,10 +719,6 @@ def test_create_quanting_env_with_suffix( result.output_path == "/some_output_path/some_project_id/out_test_file.raw/alphadia.run2" ) - assert ( - result.internal_output_path - == "/opt/airflow/mounts/output/some_project_id/out_test_file.raw/alphadia.run2" - ) assert ( result.config_params == "--out /some_output_path/some_project_id/out_test_file.raw/alphadia.run2" @@ -740,7 +740,7 @@ def test_submit_job_output_folder_exists_add( # noqa: PLR0913 """submit_job raises when output_exists_mode is 'add' but the output path already exists.""" output_dir = tmp_path / "output" output_dir.mkdir() - quanting_env = make_quanting_env(internal_output_path=str(output_dir)) + quanting_env = make_quanting_env(relative_output_path="output") mock_raw_file = MagicMock( wraps=RawFile, created_at=datetime.fromtimestamp(0, tz=pytz.UTC), @@ -749,7 +749,10 @@ def test_submit_job_output_folder_exists_add( # noqa: PLR0913 mock_get_raw_file_by_id.return_value = mock_raw_file mock_get_airflow_variable.return_value = "add" - with pytest.raises(AirflowFailException, match="should have created a unique name"): + with ( + container_locations(output=str(tmp_path)), + pytest.raises(AirflowFailException, match="should have created a unique name"), + ): submit_job(quanting_env_dict=quanting_env.to_dict()) @@ -1068,7 +1071,7 @@ def test_compute_metrics( ) -> None: """Test that compute_metrics makes the expected calls.""" quanting_env = make_quanting_env( - internal_output_path="/opt/airflow/mounts/output/P1/out_test_file.raw/alphadia" + relative_output_path="P1/out_test_file.raw/alphadia" ) mock_calc_metrics.return_value = {"metric1": "value1"} @@ -1092,7 +1095,7 @@ def test_compute_metrics_msqc_software_type( quanting_env = make_quanting_env( software_type="msqc", metrics_type="msqc", - internal_output_path="/opt/airflow/mounts/output/P1/out_test_file.raw/msqc", + relative_output_path="P1/out_test_file.raw/msqc", ) mock_calc_metrics.return_value = {"qc_metric": 42} diff --git a/airflow_src/tests/helpers.py b/airflow_src/tests/helpers.py index 02038ce3..7d8e091f 100644 --- a/airflow_src/tests/helpers.py +++ b/airflow_src/tests/helpers.py @@ -2,10 +2,10 @@ from collections.abc import Iterator from contextlib import contextmanager -from pathlib import PurePosixPath +from pathlib import Path, PurePosixPath from unittest.mock import patch -from shared.path_views import CLUSTER_VIEW +from shared.path_views import AIRFLOW_CONTAINER_VIEW, CLUSTER_VIEW @contextmanager @@ -21,3 +21,18 @@ def yaml_locations(**paths: str) -> Iterator[None]: {key: PurePosixPath(path) for key, path in paths.items()}, ): yield + + +@contextmanager +def container_locations(**paths: str) -> Iterator[None]: + """Override the locations of `AIRFLOW_CONTAINER_VIEW`, e.g. `container_locations(output=str(tmp_path))`. + + Patches the contents of the view object rather than its name, so the patch reaches every + import site and tests stay valid when code moves between modules. + """ + with patch.object( + AIRFLOW_CONTAINER_VIEW, + "_locations", + {key: Path(path) for key, path in paths.items()}, + ): + yield diff --git a/airflow_src/tests/plugins/jobs/test_docker_job_handler.py b/airflow_src/tests/plugins/jobs/test_docker_job_handler.py index 59d4a1aa..75c533ab 100644 --- a/airflow_src/tests/plugins/jobs/test_docker_job_handler.py +++ b/airflow_src/tests/plugins/jobs/test_docker_job_handler.py @@ -1,7 +1,7 @@ """Tests for the docker_job_handler module.""" from collections.abc import Callable -from pathlib import Path +from pathlib import Path, PurePosixPath from unittest.mock import MagicMock, patch import pytest @@ -10,6 +10,7 @@ from common.quanting_env import QuantingEnv from shared.keys import SoftwareTypes +from shared.path_views import Locations, View # `docker` is an optional dependency, cf. requirements_docker_job_engine.txt pytest.importorskip("docker") @@ -18,10 +19,20 @@ MODULE = "jobs.docker_job_handler" -HOST_MOUNTS_PATH = Path("/host/mounts") +DOCKER_HOST_VIEW = View( + "docker host", + { + location: f"/host/mounts/{location}" + for location in [Locations.BACKUP, Locations.OUTPUT] + }, + PurePosixPath, +) -INTERNAL_RAW_FILE_PATH = "/opt/airflow/mounts/backup/test1/2024_07/raw_file_1.raw" -INTERNAL_OUTPUT_PATH = "/opt/airflow/mounts/output/P1/out_raw_file_1.raw/custom" +RELATIVE_RAW_FILE_PATH = "test1/2024_07/raw_file_1.raw" +RELATIVE_OUTPUT_PATH = "P1/out_raw_file_1.raw/custom" + +INTERNAL_RAW_FILE_PATH = f"/opt/airflow/mounts/backup/{RELATIVE_RAW_FILE_PATH}" +INTERNAL_OUTPUT_PATH = f"/opt/airflow/mounts/output/{RELATIVE_OUTPUT_PATH}" # the paths the placeholders in the config params resolved to, cf. `locations.*.absolute_path` RAW_FILE_PATH = "/pool/backup/test1/2024_07/raw_file_1.raw" @@ -41,7 +52,7 @@ def handler() -> MagicMock: with patch(f"{MODULE}.docker.from_env") as mock_from_env: from jobs.docker_job_handler import DockerJobHandler - handler_ = DockerJobHandler(HOST_MOUNTS_PATH) + handler_ = DockerJobHandler(DOCKER_HOST_VIEW) handler_._client = mock_from_env.return_value return handler_ @@ -61,8 +72,8 @@ def sample_quanting_env( config_params=f"{RAW_FILE_PATH} {OUTPUT_PATH} 2", slurm_mem="31G", slurm_cpus_per_task=2, - internal_raw_file_path=INTERNAL_RAW_FILE_PATH, - internal_output_path=INTERNAL_OUTPUT_PATH, + relative_raw_file_path=RELATIVE_RAW_FILE_PATH, + relative_output_path=RELATIVE_OUTPUT_PATH, ) @@ -184,7 +195,7 @@ def test_start_job_should_pass_the_exported_environment( "RAW_FILE_PATH": RAW_FILE_PATH, "SETTINGS_PATH": "/pool/settings/test_settings", "OUTPUT_PATH": OUTPUT_PATH, - "RELATIVE_OUTPUT_PATH": "PID1/out_test_file.raw/alphadia", + "RELATIVE_OUTPUT_PATH": RELATIVE_OUTPUT_PATH, "SPECLIB_FILE_NAME": "some_speclib_file_name", "FASTA_FILE_NAME": "some_fasta_file_name", "CONFIG_FILE_NAME": "some_config_file_name", diff --git a/airflow_src/tests/plugins/jobs/test_job_handler.py b/airflow_src/tests/plugins/jobs/test_job_handler.py index 29d13857..01e66bcd 100644 --- a/airflow_src/tests/plugins/jobs/test_job_handler.py +++ b/airflow_src/tests/plugins/jobs/test_job_handler.py @@ -13,12 +13,12 @@ from airflow_src.tests.helpers import yaml_locations from shared.keys import JobEngines +from shared.path_views import DOCKER_HOST_VIEW # `docker` is an optional dependency, cf. requirements_docker_job_engine.txt HAS_DOCKER = importlib.util.find_spec("docker") is not None SLURM_BASE_DIR = Path("/path/to/slurm_base_path") -HOST_MOUNTS_PATH = Path("/host/mounts") @yaml_locations(slurm=str(SLURM_BASE_DIR)) @@ -37,18 +37,27 @@ def test_get_job_handler_injects_slurm_base_dir() -> None: @pytest.mark.skipif(not HAS_DOCKER, reason="`docker` not installed") -@patch("jobs.job_handler.get_host_mounts_path") @patch("jobs.docker_job_handler.docker.from_env") -def test_get_job_handler_injects_host_mounts_path( +def test_get_job_handler_injects_docker_host_view( mock_from_env: MagicMock, # noqa: ARG001 - mock_get_host_mounts_path: MagicMock, ) -> None: - """Test that the factory reads the mounts path and hands it to the docker handler.""" - mock_get_host_mounts_path.return_value = HOST_MOUNTS_PATH - + """Test that the factory hands the docker host view to the docker handler.""" handler = _get_job_handler(JobEngines.DOCKER) - assert handler._host_mounts_path == HOST_MOUNTS_PATH + assert handler._docker_host_view is DOCKER_HOST_VIEW + + +@pytest.mark.skipif(not HAS_DOCKER, reason="`docker` not installed") +@patch("jobs.docker_job_handler.docker.from_env") +def test_get_job_handler_docker_without_mounts_path( + mock_from_env: MagicMock, # noqa: ARG001 +) -> None: + """Test that a docker host view without locations points to the missing yaml key.""" + with ( + patch.object(DOCKER_HOST_VIEW, "_locations", {}), + pytest.raises(AirflowFailException, match="locations.general.mounts_path"), + ): + _get_job_handler(JobEngines.DOCKER) def test_get_job_handler_docker_without_optional_dependency() -> None: diff --git a/design_docs/PATH_HANDLING_IMPLEMENTATION_PLAN.md b/design_docs/PATH_HANDLING_IMPLEMENTATION_PLAN.md index 3c7b3e3c..0b896d20 100644 --- a/design_docs/PATH_HANDLING_IMPLEMENTATION_PLAN.md +++ b/design_docs/PATH_HANDLING_IMPLEMENTATION_PLAN.md @@ -100,10 +100,23 @@ deliberately not started here; D2 only has to avoid blocking it. ### C6 - Host view cutover - **Goal:** remove the container->host `relative_to` round-trip. -- **Do:** `DockerJobHandler._to_host_path` (`docker_job_handler.py:185-197`) becomes - `DOCKER_HOST_VIEW.resolve(location, rel)`; the handler is constructed with the view rather than with - `get_host_mounts_path()` (`job_handler.py:35`). Delete `get_host_mounts_path`. +- **Do:** the handler is constructed with `DOCKER_HOST_VIEW` rather than with + `get_host_mounts_path()` (`job_handler.py:35`); delete `get_host_mounts_path`. + `DockerJobHandler._to_host_path` disappears entirely: `QuantingEnv` carries + `relative_raw_file_path` and `relative_output_path`, so each side resolves in its own view and + nothing ever crosses between views. + - This replaces `_INTERNAL_OUTPUT_PATH` / `_INTERNAL_RAW_FILE_PATH` with + `_RELATIVE_RAW_FILE_PATH`, which lands §6.5 (`QuantingEnv` carrying both views of the same + object) ahead of D3/D5. Safe: both handlers strip `_`-prefixed keys, so these fields never + leave the process. + - Needs a `container_locations()` test helper next to `yaml_locations()`: tests used to redirect + output to a `tmp_path` by overriding the resolved path in the env, which is no longer a field. + - The factory reports a missing `locations.general.mounts_path` when the `docker` engine is + selected, restoring the actionable message that C3's lenient `DOCKER_HOST_VIEW` gave up. - **Done when:** `shared/yamlsettings.py` no longer exports any path accessor. +- **Note:** the 13 `test_docker_job_handler.py` tests need the optional `docker` package. To run + them without it, put a stub package (`from_env`, `errors.ImageNotFound/NotFound`, + `models.containers.Container`) on `PYTHONPATH`. ### C7 - Consistency test diff --git a/shared/yamlsettings.py b/shared/yamlsettings.py index 41e30fc0..68e9687e 100644 --- a/shared/yamlsettings.py +++ b/shared/yamlsettings.py @@ -104,22 +104,6 @@ def load_alphakraken_yaml(cls) -> dict[str, dict[str, Any]]: ) -def get_host_mounts_path() -> Path: - """Get the path of the mounts folder as seen by the docker host (not by the containers).""" - path = ( - YAMLSETTINGS.get(YamlKeys.LOCATIONS, {}) # type: ignore[possibly-unbound-attribute] - .get(YamlKeys.Locations.GENERAL, {}) - .get(YamlKeys.Locations.MOUNTS_PATH) - ) - - if path is None: - raise KeyError( - f"Key `{YamlKeys.LOCATIONS}.{YamlKeys.Locations.GENERAL}.{YamlKeys.Locations.MOUNTS_PATH}` not found in alphakraken.yaml." - ) - - return Path(path) - - def get_notification_setting(setting_key: str) -> str: """Get a notification setting from the yaml settings.""" setting_value = ( From e0561ecfddc2ba25076838644db8ff6d592138fe Mon Sep 17 00:00:00 2001 From: mschwoerer <82171591+mschwoer@users.noreply.github.com> Date: Wed, 2 Sep 2026 18:18:38 +0200 Subject: [PATCH 2/2] check the deployment configuration against the container view --- # Conversation that produced these changes --- ## User prompt "what is to be decided there?" "yes, go ahead with C7" Co-Authored-By: Claude Opus 5 (1M context) --- .../PATH_HANDLING_IMPLEMENTATION_PLAN.md | 29 ++-- shared/tests/test_deployment_paths.py | 128 ++++++++++++++++++ 2 files changed, 146 insertions(+), 11 deletions(-) create mode 100644 shared/tests/test_deployment_paths.py diff --git a/design_docs/PATH_HANDLING_IMPLEMENTATION_PLAN.md b/design_docs/PATH_HANDLING_IMPLEMENTATION_PLAN.md index 0b896d20..5471716a 100644 --- a/design_docs/PATH_HANDLING_IMPLEMENTATION_PLAN.md +++ b/design_docs/PATH_HANDLING_IMPLEMENTATION_PLAN.md @@ -122,17 +122,24 @@ deliberately not started here; D2 only has to avoid blocking it. - **Goal:** attack §6.6 - `InternalPaths`, `docker-compose.yaml` mount targets and yaml `mount_target` agree by convention only. -- **Do:** a test that parses the volume lists in `docker-compose.yaml:451-538` and each - `envs/alphakraken.*.yaml`, and asserts every `AIRFLOW_CONTAINER_VIEW` location reachable by the worker is - actually mounted there and that `mount_target` matches. - Note the mounts are per service and, for `instruments`/`backup`, per instrument - (`:460,462,520,521,537,538`), so the assertion is "every location has a mount whose target is that - location or a child of it", not a set equality. -- **Done when:** the test fails on the production `backup` mount-depth discrepancy reported in - `BOYSCOUT_20260901_081142.md`, or that discrepancy is resolved first and the test guards it. -- **Note:** decide the production `backup` question (`//samba-pool-1/pool-1` vs - `//samba-pool-1/pool-1/backup`) before writing the assertion - it is a config bug, not a - test-authoring detail. +- **Do:** `shared/tests/test_deployment_paths.py` parses `docker-compose.yaml` and every + `envs/alphakraken.*.yaml` and asserts what is derivable in-repo: + - each bind below the mounts folder puts the data at the same relative path on both sides - + this is what lets the docker host view be derived from `locations.general.mounts_path`; + - nothing is mounted into the containers below a location the container view does not know; + - every location of the container view is backed by at least one bind; + - each `locations..mount_target` equals ``, and each instrument's `mount_target` is + `instruments/`; + - the one deliberate exception - the logs, mounted from `airflow_logs` to `/opt/airflow/logs`, + outside the mounts folder - is pinned so that it stays deliberate. +- **Done when:** each assertion has been shown to fail on a mutation of the config it guards. +- **Note:** `mount_src` is deliberately not checked against `absolute_path`. The former is an SMB + share path, the latter a cluster filesystem path; where a share is rooted on the server is not + derivable from this repo. The production `backup` entry (`//samba-pool-1/pool-1` vs + `/fs/pool-1/backup`) is the only one whose share path does not end in the same component as its + absolute path. That is either a different export convention for `pool-1` (fine) or a real + misconfiguration; `ls /backup` on the worker host settles it - instrument folders + mean fine, a nested `backup` folder means the mount is one level off. ## 2. Dependency order diff --git a/shared/tests/test_deployment_paths.py b/shared/tests/test_deployment_paths.py new file mode 100644 index 00000000..1973813a --- /dev/null +++ b/shared/tests/test_deployment_paths.py @@ -0,0 +1,128 @@ +"""Tests that the deployment configuration agrees with the path views. + +`docker-compose.yaml` and the `envs/alphakraken.*.yaml` files decide where the data actually +shows up in the containers and on the docker host. Nothing but convention (the "DO NOT CHANGE" +comments) keeps them in sync with `path_views`, hence these tests. +""" + +import re +from pathlib import Path + +import pytest +import yaml + +from shared.keys import InternalPaths +from shared.path_views import AIRFLOW_CONTAINER_VIEW, Locations + +_REPO_ROOT = Path(__file__).parents[2] + +# ":[:]", where the host path holds a variable with a colon in it +_BIND = re.compile(r"^(?P.+?):(?P/opt/airflow/[^:]+)(?::[a-z]+)?$") + +# the logs are mounted outside the mounts folder and under a different name, cf. `mount.sh` +_LOGS_MOUNT_TARGET = "airflow_logs" +_LOGS_CONTAINER_PATH = "/opt/airflow/logs" + + +def _mount_binds() -> list[tuple[str, str]]: + """Get all (host path, container path) binds of `docker-compose.yaml` below the mounts folder.""" + compose = yaml.safe_load((_REPO_ROOT / "docker-compose.yaml").read_text()) + + binds = [] + for service in compose["services"].values(): + for volume in service.get("volumes") or []: + match = _BIND.match(str(volume)) + if match and match["container"].startswith(InternalPaths.MOUNTS_PATH): + binds.append((match["host"], match["container"])) + + return binds + + +def _env_yamls() -> list[tuple[str, dict]]: + """Get the name and content of each environment configuration.""" + return [ + (path.name, yaml.safe_load(path.read_text())) + for path in sorted((_REPO_ROOT / "envs").glob("alphakraken.*.yaml")) + ] + + +def test_compose_binds_mirror_the_mounts_folder() -> None: + """Test that each bind puts the data at the same path below the mounts folder on both sides. + + This is what lets the docker host view be derived from `locations.general.mounts_path`. + """ + for host_path, container_path in _mount_binds(): + rel_path = container_path.removeprefix(InternalPaths.MOUNTS_PATH) + + assert host_path.endswith(f"/{rel_path}"), ( + f"bind '{host_path}:{container_path}' does not mirror '{rel_path}'" + ) + + +def test_compose_binds_are_in_a_known_location() -> None: + """Test that nothing is mounted into the containers that the container view does not know.""" + for _, container_path in _mount_binds(): + location = container_path.removeprefix(InternalPaths.MOUNTS_PATH).split("/")[0] + + assert AIRFLOW_CONTAINER_VIEW.has(location), ( + f"'{container_path}' is mounted but '{location}' is no location of the view" + ) + + +def test_every_container_location_is_mounted() -> None: + """Test that each location of the container view is backed by a bind.""" + mounted = { + container_path.removeprefix(InternalPaths.MOUNTS_PATH).split("/")[0] + for _, container_path in _mount_binds() + } + + for location in [Locations.INSTRUMENTS, Locations.BACKUP, Locations.OUTPUT]: + assert location in mounted, f"no bind for location '{location}'" + + +@pytest.mark.parametrize(("file_name", "config"), _env_yamls()) +def test_location_mount_targets_match_the_location_names( + file_name: str, config: dict +) -> None: + """Test that a location is mounted at the folder the container view expects it at.""" + for location, values in config["locations"].items(): + if (mount_target := values.get("mount_target")) is None: + continue + + if location == Locations.LOGS: + assert mount_target == _LOGS_MOUNT_TARGET, file_name + continue + + assert mount_target == location, ( + f"{file_name}: '{location}' is mounted at '{mount_target}'" + ) + + +@pytest.mark.parametrize(("file_name", "config"), _env_yamls()) +def test_instrument_mount_targets_are_below_the_instruments_location( + file_name: str, config: dict +) -> None: + """Test that the instruments are mounted below the folder the container view expects.""" + for instrument_id, values in config["instruments"].items(): + if (mount_target := values.get("mount_target")) is None: + continue + + assert mount_target == f"{Locations.INSTRUMENTS}/{instrument_id}", ( + f"{file_name}: '{instrument_id}' is mounted at '{mount_target}'" + ) + + +def test_the_logs_are_not_mounted_below_the_mounts_folder() -> None: + """Test the one bind that deliberately breaks the mirroring, so that it stays deliberate.""" + compose = yaml.safe_load((_REPO_ROOT / "docker-compose.yaml").read_text()) + + log_binds = { + str(volume) + for service in compose["services"].values() + for volume in service.get("volumes") or [] + if _LOGS_CONTAINER_PATH in str(volume) + } + + assert log_binds == { + f"${{MOUNTS_PATH:?error}}/{_LOGS_MOUNT_TARGET}:{_LOGS_CONTAINER_PATH}:rw" + }