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..5471716a 100644 --- a/design_docs/PATH_HANDLING_IMPLEMENTATION_PLAN.md +++ b/design_docs/PATH_HANDLING_IMPLEMENTATION_PLAN.md @@ -100,26 +100,46 @@ 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 - **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" + } 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 = (