Path refactoring 4

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)
 

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

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:

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

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

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}