Introduce file staging delegation via JSON staging manifest

dev-requirements.txt

@@ -21,7 +21,7 @@ flake8
 
 mypy<=1.0.1  # https://github.com/pydantic/pydantic/issues/5192
 types-paramiko
-types-pkg-resources
+types-setuptools
 types-PyYAML
 types-pycurl
 types-requests

docker/coexecutor/Dockerfile

@@ -1,41 +1,79 @@
-FROM conda/miniconda3
+# use the root of the repository as context, i.e. `docker build . -f ./docker/coexecutor/Dockerfile`
+
+FROM python:3.12-bookworm as build_wheel
+
+ENV PIP_ROOT_USER_ACTION=ignore
+
+WORKDIR /build
+
+# install requirements
+COPY requirements.txt .
+COPY dev-requirements.txt .
+RUN pip install --no-cache-dir --upgrade pip \
+    && pip install --no-cache-dir setuptools -r requirements.txt -r dev-requirements.txt
+
+# build Pulsar wheel
+COPY . .
+RUN python setup.py sdist bdist_wheel
+
+
+FROM python:3.12-bookworm
 
 ENV PYTHONUNBUFFERED 1
+ENV PIP_ROOT_USER_ACTION=ignore
 ENV DEBIAN_FRONTEND noninteractive
 ENV PULSAR_CONFIG_CONDA_PREFIX /usr/local
 
+# set up Galaxy Depot repository (provides SLURM DRMAA packages for Debian Buster and newer releases)
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends ca-certificates curl gnupg \
+    && apt-get clean && rm -rf /var/lib/apt/lists/* \
+    && curl -fsSL "http://keyserver.ubuntu.com/pks/lookup?op=get&search=0x18381AC8832160AF" | gpg --dearmor -o /etc/apt/trusted.gpg.d/galaxy-depot.gpg \
+    && echo "deb https://depot.galaxyproject.org/apt/ $(bash -c '. /etc/os-release; echo ${VERSION_CODENAME:-bookworm}') main" | tee /etc/apt/sources.list.d/galaxy-depot.list
+
+# set up Debian Bullseye repository (and use it only for libslurm36, needed by slurm-drmaa1, and slurm)
+RUN echo "deb http://deb.debian.org/debian/ bullseye main" > /etc/apt/sources.list.d/bullseye.list && \
+cat <<EOF > /etc/apt/preferences.d/bullseye.pref
+Package: *
+Pin: release n=bullseye
+Pin-Priority: -1
+
+Package: libslurm36, slurm
+Pin: release n=bullseye
+Pin-Priority: 100
+EOF
+
+# set up CVMFS repository
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends lsb-release wget \
+    && wget https://ecsft.cern.ch/dist/cvmfs/cvmfs-release/cvmfs-release-latest_all.deb \
+    && dpkg -i cvmfs-release-latest_all.deb && rm -f cvmfs-release-latest_all.deb
+
 # wget, gcc, pip - to build and install Pulsar.
 # bzip2 for Miniconda.
 # TODO: pycurl stuff...
-RUN apt-get update \
-    && apt-get install -y --no-install-recommends apt-transport-https \
+RUN apt-get update && apt-get install -y --no-install-recommends \
     # Install CVMFS client
-    && apt-get install -y --no-install-recommends lsb-release wget \
-    && wget https://ecsft.cern.ch/dist/cvmfs/cvmfs-release/cvmfs-release-latest_all.deb \
-    && dpkg -i cvmfs-release-latest_all.deb \
-    && rm -f cvmfs-release-latest_all.deb \
+    cvmfs cvmfs-config-default \
     # Install packages
-    && apt-get update \
-    && apt-get install -y --no-install-recommends gcc \
-        libcurl4-openssl-dev \
-        cvmfs cvmfs-config-default \
-        slurm-llnl slurm-drmaa-dev \
-        bzip2 \
+    gcc libcurl4-openssl-dev \
+    munge libmunge-dev slurm slurm-drmaa-dev \
+    bzip2 \
     # Install Pulsar Python requirements
     && pip install --no-cache-dir -U pip \
     && pip install --no-cache-dir drmaa wheel kombu pykube pycurl \
             webob psutil PasteDeploy pyyaml paramiko \
     # Remove build deps and cleanup
     && apt-get -y remove gcc wget lsb-release \
     && apt-get -y autoremove \
-    && apt-get autoclean \
-    && rm -rf /var/lib/apt/lists/* /var/log/dpkg.log \
-    && /usr/sbin/create-munge-key
+    && apt-get clean && rm -rf /var/lib/apt/lists/* /var/log/dpkg.log
+
+COPY --from=build_wheel /build/dist/pulsar_app-*-py2.py3-none-any.whl /
 
-ADD pulsar_app-*-py2.py3-none-any.whl /pulsar_app-*-py2.py3-none-any.whl
+SHELL ["/bin/bash", "-c"]
 
-RUN pip install --upgrade setuptools && pip install pyOpenSSL --upgrade && pip install cryptography --upgrade
-RUN pip install --no-cache-dir /pulsar_app-*-py2.py3-none-any.whl[galaxy_extended_metadata] && rm /pulsar_app-*-py2.py3-none-any.whl
+RUN pip install --no-cache-dir --upgrade setuptools pyOpenSSL cryptography
+RUN pip install --no-cache-dir "$(echo /pulsar_app-*-py2.py3-none-any.whl)"[galaxy_extended_metadata] && rm /pulsar_app-*-py2.py3-none-any.whl
 RUN pip install --upgrade 'importlib-metadata<5.0'
 RUN _pulsar-configure-galaxy-cvmfs
 RUN _pulsar-conda-init --conda_prefix=/pulsar_dependencies/conda

pulsar/client/action_mapper.py

@@ -513,6 +513,51 @@ def write_from_path(self, pulsar_path):
         tus_upload_file(self.url, pulsar_path)
 
 
+class JsonTransferAction(BaseAction):
+    """
+    This action indicates that the pulsar server should create a JSON manifest that can be used to stage files by an
+    external system that can stage files in and out of the compute environment.
+    """
+    inject_url = True
+    whole_directory_transfer_supported = False
+    action_type = "json_transfer"
+    staging = STAGING_ACTION_REMOTE
+
+    def __init__(self, source, file_lister=None, url=None, from_path=None, to_path=None):
+        super().__init__(source, file_lister)
+        self.url = url
+
+        self._from_path = from_path
+        self._to_path = to_path
+        # `from_path` and `to_path` are mutually exclusive, only one of them should be set
+
+    @classmethod
+    def from_dict(cls, action_dict):
+        return JsonTransferAction(
+            source=action_dict["source"],
+            url=action_dict["url"],
+            from_path=action_dict.get("from_path"),
+            to_path=action_dict.get("to_path")
+        )
+
+    def to_dict(self):
+        return self._extend_base_dict(**self.to_staging_manifest_entry())
+
+    def write_to_path(self, path):
+        self._from_path, self._to_path = None, path
+
+    def write_from_path(self, pulsar_path: str):
+        self._from_path, self._to_path = pulsar_path, None
+
+    def to_staging_manifest_entry(self):
+        staging_manifest_entry = dict(url=self.url)
+        if self._from_path:
+            staging_manifest_entry["from_path"] = self._from_path
+        if self._to_path:
+            staging_manifest_entry["to_path"] = self._to_path
+        return staging_manifest_entry
+
+
 class RemoteObjectStoreCopyAction(BaseAction):
     """
     """
@@ -664,6 +709,7 @@ def write_to_path(self, path):
 
 
 DICTIFIABLE_ACTION_CLASSES = [
+    JsonTransferAction,
     RemoteCopyAction,
     RemoteTransferAction,
     RemoteTransferTusAction,
@@ -844,6 +890,7 @@ def unstructured_map(self, path):
 
 ACTION_CLASSES: List[Type[BaseAction]] = [
     NoneAction,
+    JsonTransferAction,
     RewriteAction,
     TransferAction,
     CopyAction,
@@ -859,6 +906,7 @@ def unstructured_map(self, path):
 
 __all__ = (
     'FileActionMapper',
+    'JsonTransferAction',
     'path_type',
     'from_dict',
     'MessageAction',

pulsar/client/client.py

@@ -168,7 +168,7 @@ def __init__(self, destination_params, job_id, job_manager_interface):
         self.job_manager_interface = job_manager_interface
 
     def launch(self, command_line, dependencies_description=None, env=None, remote_staging=None, job_config=None,
-               dynamic_file_sources=None, token_endpoint=None):
+               dynamic_file_sources=None, token_endpoint=None, staging_manifest=None):
         """
         Queue up the execution of the supplied `command_line` on the remote
         server. Called launch for historical reasons, should be renamed to
@@ -374,6 +374,90 @@ def _build_setup_message(self, command_line, dependencies_description, env, remo
             launch_params["setup_params"] = setup_params
         return launch_params
 
+    def get_pulsar_app_config(
+        self,
+        pulsar_app_config,
+        container,
+        wait_after_submission,
+        manager_name,
+        manager_type,
+        dependencies_description,
+    ):
+
+        pulsar_app_config = pulsar_app_config or {}
+        manager_config = self._ensure_manager_config(
+            pulsar_app_config,
+            manager_name,
+            manager_type,
+        )
+
+        if (
+            "staging_directory" not in manager_config and "staging_directory" not in pulsar_app_config
+        ):
+            pulsar_app_config["staging_directory"] = CONTAINER_STAGING_DIRECTORY
+
+        if self.amqp_key_prefix:
+            pulsar_app_config["amqp_key_prefix"] = self.amqp_key_prefix
+
+        if "monitor" not in manager_config:
+            manager_config["monitor"] = (
+                MonitorStyle.BACKGROUND.value
+                if wait_after_submission
+                else MonitorStyle.NONE.value
+            )
+        if "persistence_directory" not in pulsar_app_config:
+            pulsar_app_config["persistence_directory"] = os.path.join(
+                CONTAINER_STAGING_DIRECTORY, "persisted_data"
+            )
+        elif "manager" in pulsar_app_config and manager_name != "_default_":
+            log.warning(
+                "'manager' set in app config but client has non-default manager '%s', this will cause communication"
+                " failures, remove `manager` from app or client config to fix",
+                manager_name,
+            )
+
+        using_dependencies = container is None and dependencies_description is not None
+        if using_dependencies and "dependency_resolution" not in pulsar_app_config:
+            # Setup default dependency resolution for container above...
+            dependency_resolution = {
+                "cache": False,
+                "use": True,
+                "default_base_path": "/pulsar_dependencies",
+                "cache_dir": "/pulsar_dependencies/_cache",
+                "resolvers": [
+                    {  # TODO: add CVMFS resolution...
+                        "type": "conda",
+                        "auto_init": True,
+                        "auto_install": True,
+                        "prefix": "/pulsar_dependencies/conda",
+                    },
+                    {
+                        "type": "conda",
+                        "auto_init": True,
+                        "auto_install": True,
+                        "prefix": "/pulsar_dependencies/conda",
+                        "versionless": True,
+                    },
+                ],
+            }
+            pulsar_app_config["dependency_resolution"] = dependency_resolution
+        return pulsar_app_config
+
+    def _ensure_manager_config(self, pulsar_app_config, manager_name, manager_type):
+        if "manager" in pulsar_app_config:
+            manager_config = pulsar_app_config["manager"]
+        elif "managers" in pulsar_app_config:
+            managers_config = pulsar_app_config["managers"]
+            if manager_name not in managers_config:
+                managers_config[manager_name] = {}
+            manager_config = managers_config[manager_name]
+        else:
+            manager_config = {}
+            pulsar_app_config["manager"] = manager_config
+        if "type" not in manager_config:
+            manager_config["type"] = manager_type
+        return manager_config
+
 
 class MessagingClientManagerProtocol(ClientManagerProtocol):
     status_cache: Dict[str, Dict[str, Any]]
@@ -405,7 +489,7 @@ def _build_status_request_message(self):
 class MessageJobClient(BaseMessageJobClient):
 
     def launch(self, command_line, dependencies_description=None, env=None, remote_staging=None, job_config=None,
-               dynamic_file_sources=None, token_endpoint=None):
+               dynamic_file_sources=None, token_endpoint=None, staging_manifest=None):
         """
         """
         launch_params = self._build_setup_message(
@@ -439,7 +523,7 @@ def __init__(self, destination_params, job_id, client_manager, shell):
         self.shell = shell
 
     def launch(self, command_line, dependencies_description=None, env=None, remote_staging=None, job_config=None,
-               dynamic_file_sources=None, token_endpoint=None):
+               dynamic_file_sources=None, token_endpoint=None, staging_manifest=None):
         """
         """
         launch_params = self._build_setup_message(
@@ -491,7 +575,8 @@ def launch(
         dynamic_file_sources=None,
         container_info=None,
         token_endpoint=None,
-        pulsar_app_config=None
+        pulsar_app_config=None,
+        staging_manifest=None,
     ) -> Optional[ExternalId]:
         """
         """
@@ -513,48 +598,15 @@ def launch(
 
         manager_name = self.client_manager.manager_name
         manager_type = "coexecution" if container is not None else "unqueued"
-        pulsar_app_config = pulsar_app_config or {}
-        manager_config = self._ensure_manager_config(
-            pulsar_app_config, manager_name, manager_type,
+        pulsar_app_config = self.get_pulsar_app_config(
+            pulsar_app_config=pulsar_app_config,
+            container=container,
+            wait_after_submission=wait_after_submission,
+            manager_name=manager_name,
+            manager_type=manager_type,
+            dependencies_description=dependencies_description,
         )
 
-        if "staging_directory" not in manager_config and "staging_directory" not in pulsar_app_config:
-            pulsar_app_config["staging_directory"] = CONTAINER_STAGING_DIRECTORY
-
-        if self.amqp_key_prefix:
-            pulsar_app_config["amqp_key_prefix"] = self.amqp_key_prefix
-
-        if "monitor" not in manager_config:
-            manager_config["monitor"] = MonitorStyle.BACKGROUND.value if wait_after_submission else MonitorStyle.NONE.value
-        if "persistence_directory" not in pulsar_app_config:
-            pulsar_app_config["persistence_directory"] = os.path.join(CONTAINER_STAGING_DIRECTORY, "persisted_data")
-        elif "manager" in pulsar_app_config and manager_name != '_default_':
-            log.warning(
-                "'manager' set in app config but client has non-default manager '%s', this will cause communication"
-                " failures, remove `manager` from app or client config to fix", manager_name)
-
-        using_dependencies = container is None and dependencies_description is not None
-        if using_dependencies and "dependency_resolution" not in pulsar_app_config:
-            # Setup default dependency resolution for container above...
-            dependency_resolution = {
-                "cache": False,
-                "use": True,
-                "default_base_path": "/pulsar_dependencies",
-                "cache_dir": "/pulsar_dependencies/_cache",
-                "resolvers": [{  # TODO: add CVMFS resolution...
-                    "type": "conda",
-                    "auto_init": True,
-                    "auto_install": True,
-                    "prefix": '/pulsar_dependencies/conda',
-                }, {
-                    "type": "conda",
-                    "auto_init": True,
-                    "auto_install": True,
-                    "prefix": '/pulsar_dependencies/conda',
-                    "versionless": True,
-                }]
-            }
-            pulsar_app_config["dependency_resolution"] = dependency_resolution
         base64_message = to_base64_json(launch_params)
         base64_app_conf = to_base64_json(pulsar_app_config)
         pulsar_container_image = self.pulsar_container_image
@@ -606,21 +658,6 @@ def _pulsar_script_args(self, manager_name, base64_job, base64_app_conf, wait_ar
         manager_args.extend(["--base64", base64_job, "--app_conf_base64", base64_app_conf])
         return manager_args
 
-    def _ensure_manager_config(self, pulsar_app_config, manager_name, manager_type):
-        if "manager" in pulsar_app_config:
-            manager_config = pulsar_app_config["manager"]
-        elif "managers" in pulsar_app_config:
-            managers_config = pulsar_app_config["managers"]
-            if manager_name not in managers_config:
-                managers_config[manager_name] = {}
-            manager_config = managers_config[manager_name]
-        else:
-            manager_config = {}
-            pulsar_app_config["manager"] = manager_config
-        if "type" not in manager_config:
-            manager_config["type"] = manager_type
-        return manager_config
-
     def _launch_containers(
         self,
         pulsar_submit_container: CoexecutionContainerCommand,