Enable ephemeral containers

Author: GlassOfWhiskey

.github/workflows/ci-tests.yaml

@@ -150,7 +150,7 @@ jobs:
           python -m pip install -r docs/requirements.txt
       - name: "Build documentation and check for consistency"
         env:
-          CHECKSUM: "6642d8d7533c7cf3ec9de7bdbf1871e3a313dd73fa6551fe3bb10e4e94e7ff08"
+          CHECKSUM: "2c3a8784c46b5591461db3df25cf56a431086cb265173e5f648fdde3c9fc23e7"
         run: |
           cd docs
           HASH="$(make checksum | tail -n1)"

docs/source/connector/container.rst

@@ -2,4 +2,4 @@
 ContainerConnector
 ==================
 
-The ``ContainerConnector`` is an abstract connector that serves as a base class to implement software container connectors (e.g., :ref:`Docker <DockerConnector>`, :ref:`Docker Compose <DockerComposeConnector>`, and :ref:`Singularity <SingularityConnector>`). It extends the abstract :ref:`ConnectorWrapper <ConnectorWrapper>` interface, allowing users to spawn software containers on top of local or remote execution environments using the :ref:`stacked locations <Stacked locations>` mechanism. Plus, it prevents :ref:`BatchConnector <BatchConnector>` instances to be wrapped as inner connectors.
+The ``ContainerConnector`` is an abstract connector that serves as a base class to implement software container connectors (e.g., :ref:`Docker <DockerConnector>`, :ref:`Docker Compose <DockerComposeConnector>`, and :ref:`Singularity <SingularityConnector>`). It extends the abstract :ref:`ConnectorWrapper <ConnectorWrapper>` interface, allowing users to spawn software containers on top of local or remote execution environments using the :ref:`stacked locations <Stacked locations>` mechanism. Plus, it prevents :ref:`BatchConnector <BatchConnector>` instances to be wrapped as inner connectors, unless the ``ContainerConnector`` is marked as ``ephemeral``.

docs/source/connector/docker.rst

@@ -2,7 +2,7 @@
 DockerConnector
 ===============
 
-The `Docker <https://www.docker.com/>`_ connector can spawn one or more instances of a Docker container locally on the StreamFlow node. The units of deployment and binding for this connector correspond to the set of homogeneous container instances, while the unit of scheduling is the single instance. It extends the :ref:`ContainerConnector <ContainerConnector>`, which inherits from the :ref:`ConnectorWrapper <ConnectorWrapper>` interface, allowing users to spawn Docker containers on top of local or remote execution environments using the :ref:`stacked locations <Stacked locations>` mechanism.
+The `Docker <https://www.docker.com/>`_ connector can spawn one or more instances of a Docker container locally on the StreamFlow node. It extends the :ref:`ContainerConnector <ContainerConnector>`, which inherits from the :ref:`ConnectorWrapper <ConnectorWrapper>` interface, allowing users to spawn Docker containers on top of local or remote execution environments using the :ref:`stacked locations <Stacked locations>` mechanism. Normally, a single Docker container is reused for multiple workflow commands, reducing cold start overhead. However, when the ``ephemeral`` option is set to ``True``, a fresh container instance is spawned for each command to prevent internal state contamination. In addition, a ``ContainerConnector`` marked as ``ephemeral`` can successfully wrap :ref:`BatchConnector <BatchConnector>` instances.
 
 .. jsonschema:: ../../../streamflow/deployment/connector/schemas/docker.json
     :lift_description: true

docs/source/connector/singularity.rst

@@ -2,7 +2,7 @@
 SingularityConnector
 =====================
 
-The `Singularity <https://sylabs.io/singularity>`_ connector can spawn one or more instances of a Singularity container locally on the StreamFlow node. The units of deployment and binding for this connector correspond to the set of homogeneous container instances, while the unit of scheduling is the single instance. It extends the :ref:`ContainerConnector <ContainerConnector>`, which inherits from the :ref:`ConnectorWrapper <ConnectorWrapper>` interface, allowing users to spawn Singularity containers on top of local or remote execution environments using the :ref:`stacked locations <Stacked locations>` mechanism.
+The `Singularity <https://sylabs.io/singularity>`_ connector can spawn one or more instances of a Singularity container locally on the StreamFlow node. It extends the :ref:`ContainerConnector <ContainerConnector>`, which inherits from the :ref:`ConnectorWrapper <ConnectorWrapper>` interface, allowing users to spawn Singularity containers on top of local or remote execution environments using the :ref:`stacked locations <Stacked locations>` mechanism. Normally, a single Singularity instance is reused for multiple workflow commands, reducing cold start overhead. However, when the ``ephemeral`` option is set to ``True``, a fresh container instance is spawned for each command to prevent internal state contamination. In addition, a ``ContainerConnector`` marked as ``ephemeral`` can successfully wrap :ref:`BatchConnector <BatchConnector>` instances.
 
 .. jsonschema:: ../../../streamflow/deployment/connector/schemas/singularity.json
     :lift_description: true

docs/source/cwl/docker-requirement.rst

@@ -43,6 +43,7 @@ By default, StreamFlow automatically maps a step with the ``DockerRequirement``
         type: docker
         config:
           image: node:slim
+          ephemeral: true
 
 StreamFlow also supports the possibility to map a CWL ``DockerRequirement`` onto different types of connectors through the :ref:`CWLDockerTranslator <CWLDockerTranslator>` extension point. In particular, the ``docker`` section of a workflow configuration can bind each step or subworkflow to a specific translator type, making it possible to convert a pure CWL workflow with ``DockerRequirement`` features into a hybrid workflow. The available translator types are: ``docker``, ``kubernetes``, ``none`` and ``singularity``.
 

docs/source/cwl/docker/docker.rst

@@ -2,7 +2,7 @@
 DockerCWLDockerTranslator
 =========================
 
-The Docker :ref:`CWLDockerTranslator <CWLDockerTranslator>` instantiates a :ref:`DockerConnector <DockerConnector>` instance with the given configuration for every CWL :ref:`DockerRequirement <CWL Docker Requirement>` specification in the selected subworkflow.
+The Docker :ref:`CWLDockerTranslator <CWLDockerTranslator>` instantiates a :ref:`DockerConnector <DockerConnector>` instance with the given configuration for every CWL :ref:`DockerRequirement <CWL Docker Requirement>` specification in the selected subworkflow. Note that the resulting ``DockerConnector`` instance spawns ``ephemeral`` containers, making it able to wrap also :ref:`BatchConnector <BatchConnector>` instances for HPC deployments.
 
 .. jsonschema:: ../../../../streamflow/cwl/requirement/docker/schemas/docker.json
     :lift_description: true

docs/source/cwl/docker/kubernetes.rst

@@ -2,7 +2,7 @@
 KubernetesCWLDockerTranslator
 =============================
 
-The Kubernetes :ref:`CWLDockerTranslator <CWLDockerTranslator>` instantiates a :ref:`KubernetesConnector <KubernetesConnector>` instance with the given configuration for every CWL :ref:`DockerRequirement <CWL Docker Requirement>` specification in the selected subworkflow.
+The Kubernetes :ref:`CWLDockerTranslator <CWLDockerTranslator>` instantiates a :ref:`KubernetesConnector <KubernetesConnector>` instance with the given configuration for every CWL :ref:`DockerRequirement <CWL Docker Requirement>` specification in the selected subworkflow. Note that, unlike other ``CWLDockerTranslator`` classes, the ``KubernetesConnector`` does not support ``ephemeral`` containers.
 
 .. jsonschema:: ../../../../streamflow/cwl/requirement/docker/schemas/kubernetes.json
     :lift_description: true

docs/source/cwl/docker/singularity.rst

@@ -2,7 +2,7 @@
 SingularityCWLDockerTranslator
 ==============================
 
-The Singularity :ref:`CWLDockerTranslator <CWLDockerTranslator>` instantiates a :ref:`SingularityConnector <SingularityConnector>` instance with the given configuration for every CWL :ref:`DockerRequirement <CWL Docker Requirement>` specification in the selected subworkflow.
+The Singularity :ref:`CWLDockerTranslator <CWLDockerTranslator>` instantiates a :ref:`SingularityConnector <SingularityConnector>` instance with the given configuration for every CWL :ref:`DockerRequirement <CWL Docker Requirement>` specification in the selected subworkflow. Note that the resulting ``SingularityConnector`` instance spawns ``ephemeral`` containers, making it able to wrap also :ref:`BatchConnector <BatchConnector>` instances for HPC deployments.
 
 .. jsonschema:: ../../../../streamflow/cwl/requirement/docker/schemas/docker.json
     :lift_description: true

docs/source/ext/cwl-docker-translator.rst

@@ -28,7 +28,8 @@ Implementations
 ===================================================     ================================================================
 Type                                                    Class
 ===================================================     ================================================================
-:ref:`docker <DockerCWLDockerTranslator>`               streamflow.cwl.requirement.docker.DockerCWLDockerTranslator
-:ref:`kubernetes <KubernetesCWLDockerTranslator>`       streamflow.cwl.requirement.docker.KubernetesCWLDockerTranslator
-:ref:`singularity <SingularityCWLDockerTranslator>`     streamflow.cwl.requirement.docker.SingularityCWLDockerTranslator
+:ref:`docker <DockerCWLDockerTranslator>`               streamflow.cwl.requirement.docker.docker.DockerCWLDockerTranslator
+:ref:`kubernetes <KubernetesCWLDockerTranslator>`       streamflow.cwl.requirement.docker.kubernetes.KubernetesCWLDockerTranslator
+:ref:`none <NoContainerCWLDockerTranslator>`            streamflow.cwl.requirement.docker.nocontainer.NoContainerCWLDockerTranslator
+:ref:`singularity <SingularityCWLDockerTranslator>`     streamflow.cwl.requirement.docker.singularity.SingularityCWLDockerTranslator
 ===================================================     ================================================================

streamflow/cwl/requirement/docker/docker.py

@@ -48,6 +48,7 @@ def __init__(
         entrypoint: str | None = None,
         env: MutableSequence[str] | None = None,
         envFile: MutableSequence[str] | None = None,
+        ephemeral: bool = True,
         expose: MutableSequence[str] | None = None,
         gpus: MutableSequence[str] | None = None,
         groupAdd: MutableSequence[str] | None = None,
@@ -140,6 +141,7 @@ def __init__(
         self.entrypoint: str | None = entrypoint
         self.env: MutableSequence[str] | None = env
         self.envFile: MutableSequence[str] | None = envFile
+        self.ephemeral: bool = ephemeral
         self.expose: MutableSequence[str] | None = expose
         self.gpus: MutableSequence[str] | None = gpus
         self.groupAdd: MutableSequence[str] | None = groupAdd
@@ -218,14 +220,10 @@ def get_target(
         volume = list(self.volume) if self.volume else []
         volume.append(f"{target.workdir}:/tmp/streamflow")
         if output_directory is not None:
-            if target.deployment.type == "local":
-                volume.append(
-                    f"{os.path.join(target.workdir, utils.random_name())}:{output_directory}"
-                )
-            else:
-                volume.append(
-                    f"{posixpath.join(target.workdir, utils.random_name())}:{output_directory}"
-                )
+            path_processor = os.path if target.deployment.type == "local" else posixpath
+            volume.append(
+                f"{path_processor.join(target.workdir, utils.random_name())}:{output_directory}"
+            )
         return Target(
             deployment=DeploymentConfig(
                 name=utils.random_name(),
@@ -264,6 +262,7 @@ def get_target(
                     "entrypoint": self.entrypoint,
                     "env": self.env,
                     "envFile": self.envFile,
+                    "ephemeral": self.ephemeral,
                     "expose": self.expose,
                     "gpus": self.gpus,
                     "groupAdd": self.groupAdd,
@@ -321,6 +320,7 @@ def get_target(
                     "volume": volume,
                     "volumeDriver": self.volumeDriver,
                     "volumesFrom": self.volumesFrom,
+                    "workdir": self.workdir,
                 },
                 workdir="/tmp/streamflow",  # nosec
                 wraps=(

streamflow/cwl/requirement/docker/schemas/docker.json

@@ -60,7 +60,7 @@
       "items": {
         "type": "string"
       },
-      "description": "Command to run when deploying the container"
+      "description": "Command to run when deploying the container. Not supported for ephemeral containers"
     },
     "cpuPeriod": {
       "type": "integer",
@@ -199,6 +199,11 @@
       "uniqueItems": true,
       "description": "Read in a file of environment variables"
     },
+    "ephemeral": {
+      "type": "boolean",
+      "default": true,
+      "description": "Invoke a fresh container instance for every command in the workflow"
+    },
     "expose": {
       "type": "array",
       "items": {

streamflow/cwl/requirement/docker/schemas/singularity.json

@@ -15,6 +15,10 @@
       "type": "string",
       "description": "Apply cgroups from file for container processes (root only)"
     },
+    "arch": {
+      "type": "string",
+      "description": "architecture to pull from library"
+    },
     "bind": {
       "type": "array",
       "items": {
@@ -36,7 +40,7 @@
     },
     "boot": {
       "type": "boolean",
-      "description": "Execute /sbin/init to boot container (root only)"
+      "description": "Execute /sbin/init to boot container (root only). Not supported for ephemeral containers"
     },
     "cleanenv": {
       "type": "boolean",
@@ -48,7 +52,7 @@
       "items": {
         "type": "string"
       },
-      "description": "Command to run when deploying the container"
+      "description": "Command to run when deploying the container. Not supported for ephemeral containers"
     },
     "compat": {
       "type": "boolean",
@@ -65,7 +69,8 @@
     },
     "cpuShares": {
       "type": "integer",
-      "description": "CPU shares for container (default -1)"
+      "description": "CPU shares for container",
+      "default": -1
     },
     "cpus": {
       "type": "string",
@@ -85,7 +90,7 @@
     },
     "dns": {
       "type": "string",
-      "description": "List of DNS server separated by commas to add in resolv.conf"
+      "description": "List of DNS server separated by commas to add in resolv.conf. Not supported for ephemeral containers"
     },
     "dockerHost": {
       "type": "string",
@@ -107,6 +112,11 @@
       "type": "string",
       "description": "Pass environment variables from file to contained process"
     },
+    "ephemeral": {
+      "type": "boolean",
+      "default": true,
+      "description": "Invoke a fresh container instance for every command in the workflow"
+    },
     "fakeroot": {
       "type": "boolean",
       "description": "Run container in new user namespace as uid 0"
@@ -230,12 +240,17 @@
     },
     "pidFile": {
       "type": "string",
-      "description": "Write instance PID to the file with the given name"
+      "description": "Write instance PID to the file with the given name. Not supported for ephemeral containers"
     },
     "pidsLimit": {
       "type": "integer",
       "description": "Limit number of container PIDs, use -1 for unlimited"
     },
+    "pullDir": {
+      "type": "string",
+      "description": "Download images to the specific directory",
+      "default": "The target workdir"
+    },
     "rocm": {
       "type": "boolean",
       "description": "Enable experimental ROCM support"

streamflow/cwl/requirement/docker/singularity.py

@@ -20,6 +20,7 @@ def __init__(
         addCaps: str | None = None,
         allowSetuid: bool = False,
         applyCgroups: str | None = None,
+        arch: str | None = None,
         bind: MutableSequence[str] | None = None,
         blkioWeight: int | None = None,
         blkioWeightDevice: MutableSequence[str] | None = None,
@@ -39,6 +40,7 @@ def __init__(
         dropCaps: str | None = None,
         env: MutableSequence[str] | None = None,
         envFile: str | None = None,
+        ephemeral: bool = True,
         fakeroot: bool = False,
         fusemount: MutableSequence[str] | None = None,
         home: str | None = None,
@@ -66,6 +68,7 @@ def __init__(
         pemPath: str | None = None,
         pidFile: str | None = None,
         pidsLimit: int | None = None,
+        pullDir: str | None = None,
         rocm: bool = False,
         scratch: MutableSequence[str] | None = None,
         security: MutableSequence[str] | None = None,
@@ -79,6 +82,7 @@ def __init__(
         self.addCaps: str | None = addCaps
         self.allowSetuid: bool = allowSetuid
         self.applyCgroups: str | None = applyCgroups
+        self.arch: str | None = arch
         self.bind: MutableSequence[str] | None = bind
         self.blkioWeight: int | None = blkioWeight
         self.blkioWeightDevice: MutableSequence[str] | None = blkioWeightDevice
@@ -98,6 +102,7 @@ def __init__(
         self.dockerHost: str | None = dockerHost
         self.env: MutableSequence[str] | None = env
         self.envFile: str | None = envFile
+        self.ephemeral: bool = ephemeral
         self.fakeroot: bool = fakeroot
         self.fusemount: MutableSequence[str] | None = fusemount
         self.home: str | None = home
@@ -125,6 +130,7 @@ def __init__(
         self.pemPath: str | None = pemPath
         self.pidFile: str | None = pidFile
         self.pidsLimit: int | None = pidsLimit
+        self.pullDir: str | None = pullDir
         self.rocm: bool = rocm
         self.scratch: MutableSequence[str] | None = scratch
         self.security: MutableSequence[str] | None = security
@@ -154,14 +160,10 @@ def get_target(
         bind = list(self.bind) if self.bind else []
         bind.append(f"{target.workdir}:/tmp/streamflow")
         if output_directory is not None:
-            if target.deployment.type == "local":
-                bind.append(
-                    f"{os.path.join(target.workdir, utils.random_name())}:{output_directory}"
-                )
-            else:
-                bind.append(
-                    f"{posixpath.join(target.workdir, utils.random_name())}:{output_directory}"
-                )
+            path_processor = os.path if target.deployment.type == "local" else posixpath
+            bind.append(
+                f"{path_processor.join(target.workdir, utils.random_name())}:{output_directory}"
+            )
         return Target(
             deployment=DeploymentConfig(
                 name=utils.random_name(),
@@ -171,6 +173,7 @@ def get_target(
                     "addCaps": self.addCaps,
                     "allowSetuid": self.allowSetuid,
                     "applyCgroups": self.applyCgroups,
+                    "arch": self.arch,
                     "bind": bind,
                     "blkioWeight": self.blkioWeight,
                     "blkioWeightDevice": self.blkioWeightDevice,
@@ -190,6 +193,7 @@ def get_target(
                     "dockerHost": self.dockerHost,
                     "env": self.env,
                     "envFile": self.envFile,
+                    "ephemeral": self.ephemeral,
                     "fakeroot": self.fakeroot,
                     "fusemount": self.fusemount,
                     "home": self.home,
@@ -217,12 +221,16 @@ def get_target(
                     "pemPath": self.pemPath,
                     "pidFile": self.pidFile,
                     "pidsLimit": self.pidsLimit,
+                    "pullDir": (
+                        self.pullDir if self.pullDir is not None else target.workdir
+                    ),
                     "rocm": self.rocm,
                     "scratch": self.scratch,
                     "security": self.security,
                     "transferBufferSize": self.transferBufferSize,
                     "userns": self.userns,
                     "uts": self.uts,
+                    "workdir": self.workdir,
                     "writable": self.writable,
                     "writableTmpfs": self.writableTmpfs,
                 },