WIP: feat(sdk)Add PodmanRunner

sdk/CONTRIBUTING.md

@@ -1,23 +1,23 @@
-## Contributing to the `kfp` SDK
+# Contributing to the `kfp` SDK
 
 For developing KFP v2 SDK, use the `master` branch.
 
 For developing KFP v1 SDK, use the [sdk/release-1.8 branch](https://github.com/kubeflow/pipelines/tree/sdk/release-1.8).
 
 For general contribution guidelines including pull request conventions, see [pipelines/CONTRIBUTING.md](https://github.com/kubeflow/pipelines/blob/master/CONTRIBUTING.md).
 
-### Pre-requisites 
+## Pre-requisites
 
-Clone the repo: 
+Clone the repo:
 
 ```bash
 git clone https://github.com/kubeflow/pipelines.git && cd pipelines
 ```
 
-We suggest using a tool like [virtual env](https://docs.python.org/3/library/venv.html) or something similar for isolating 
-your environment and/or packages for you development environment. For this setup we'll stick with virtual env. 
+We suggest using a tool like [virtual env](https://docs.python.org/3/library/venv.html) or something similar for isolating
+your environment and/or packages for you development environment. For this setup we'll stick with virtual env.
 
-For supported python versions, see the sdk [setup.py](https://github.com/kubeflow/pipelines/blob/master/sdk/python/setup.py). 
+For supported python versions, see the sdk [setup.py](https://github.com/kubeflow/pipelines/blob/master/sdk/python/setup.py).
 
 ```bash
 # optional, replace with your tool of choice
@@ -47,89 +47,106 @@ The SDK also relies on a couple other python packages also found within KFP.
 These consists of the [api proto package](https://github.com/kubeflow/pipelines/tree/master/api) and the kfp [kubernetes_platform](https://github.com/kubeflow/pipelines/tree/master/kubernetes_platform) package.
 
 For the proto code, we need protobuf-compiler to generate the python code. Read [here](../kubernetes_platform#dependencies) more about how to install this
-dependency. 
+dependency.
 
 You can install both packages either in development mode if you are planning to do active development on these, or simply do a regular install.
 
 To install the proto package:
+
 ```bash
 pushd api
 make generate python-dev # omit -dev for regular install
 popd
 ```
 
 To install the kubernetes_platform package:
+
 ```bash
 pushd kubernetes_platform
 make python-dev # omit -dev for regular install
 popd
 ```
 
-### Testing
+## Testing
+
 We suggest running unit tests using [`pytest`](https://docs.pytest.org/en/7.1.x/). From the project root, the following runs all KFP SDK unit tests:
-```sh
+
+```bash
 pytest
 ```
 
 To run tests in parallel for faster execution, you can run the tests using the `pytest-xdist` plugin:
 
-```sh
+```bash
 pytest -n auto
 ```
 
-### Code Style
+## Code Style
+
 Dependencies for code style checks/changes can be found in the kfp SDK [requirements-dev.txt](https://github.com/kubeflow/pipelines/blob/master/sdk/python/requirements-dev.txt).
 
+### Style Guide [Required]
 
-#### Style Guide [Required]
 The KFP SDK follows the [Google Python Style Guide](https://google.github.io/styleguide/pyguide.html).
 
-#### Formatting [Required]
+### Formatting [Required]
+
 Please format your code using [yapf](https://github.com/google/yapf) according to the [`.style.yapf`](https://github.com/kubeflow/pipelines/blob/master/.style.yapf) file.
 
 From the project root, run the following code to format your code:
-```sh
+
+```bash
 yapf --in-place --recursive ./sdk/python
 ```
 
-#### Docformatter [Required]
+### Docformatter [Required]
+
 We encourage you to lint your docstrings using [docformatter](https://github.com/PyCQA/docformatter).
 
 From the project root, run the following code to lint your docstrings:
-```sh
+
+```bash
 docformatter --in-place --recursive ./sdk/python
 ```
 
-#### Formatting Imports [Required]
+### Formatting Imports [Required]
+
 Please organize your imports using [isort](https://pycqa.github.io/isort/index.html) according to the [`.isort.cfg`](https://github.com/kubeflow/pipelines/blob/master/.isort.cfg) file.
 
 From the project root, run the following code to format your code:
-```sh
+
+```bash
 isort sdk/python
 ```
 
-#### Pylint [Encouraged]
+### Pylint [Encouraged]
+
 We encourage you to lint your code using [pylint](https://pylint.org/) according to the project [`.pylintrc`](https://github.com/kubeflow/pipelines/blob/master/.pylintrc) file.
 
 From the project root, run the following code to lint your code:
-```sh
+
+```bash
 pylint ./sdk/python/kfp
 ```
 
 Note: `kfp` is not currently fully pylint-compliant. Consider substituting the path argument with the files touched by your development.
 
-#### Static Type Checking [Encouraged]
+### Static Type Checking [Encouraged]
+
 Please use [mypy](https://mypy.readthedocs.io/en/stable/) to check your type annotations.
 
 From the project root, run the following code to lint your docstrings:
-```sh
+
+```bash
 mypy ./sdk/python/kfp/
 ```
+
 Note: `kfp` is not currently fully mypy-compliant. Consider substituting the path argument with the files touched by your development.
 
+### Pre-commit [Recommended]
 
-#### Pre-commit [Recommended]
 Consider using [`pre-commit`](https://github.com/pre-commit/pre-commit) with the provided [`.pre-commit-config.yaml`](https://github.com/kubeflow/pipelines/blob/master/.pre-commit-config.yaml) to implement the above changes:
-```sh
+
+```bash
 pre-commit install
 ```

sdk/python/kfp/local/__init__.py

@@ -16,10 +16,12 @@
 
 from kfp.local.config import DockerRunner
 from kfp.local.config import init
+from kfp.local.config import PodmanRunner
 from kfp.local.config import SubprocessRunner
 
 __all__ = [
     'init',
     'SubprocessRunner',
     'DockerRunner',
+    'PodmanRunner',
 ]

sdk/python/kfp/local/config.py

@@ -59,6 +59,20 @@ def __post_init__(self):
             ) from e
 
 
+@dataclasses.dataclass
+class PodmanRunner:
+    """Runner that indicates that local tasks should be run as a container
+    using Podman."""
+
+    def __post_init__(self):
+        try:
+            import podman  # noqa
+        except ImportError as e:
+            raise ImportError(
+                f"Package 'podman' must be installed to use {PodmanRunner.__name__!r}. Install it using 'pip install podman'."
+            ) from e
+
+
 class LocalExecutionConfig:
     instance = None
 
@@ -78,7 +92,7 @@ def __init__(
         pipeline_root: str,
         raise_on_error: bool,
     ) -> None:
-        permitted_runners = (SubprocessRunner, DockerRunner)
+        permitted_runners = (SubprocessRunner, DockerRunner, PodmanRunner)
         if not isinstance(runner, permitted_runners):
             raise ValueError(
                 f'Got unknown runner {runner} of type {runner.__class__.__name__}. Runner should be one of the following types: {". ".join(prunner.__name__ for prunner in permitted_runners)}.'
@@ -97,7 +111,7 @@ def validate(cls):
 
 def init(
     # annotate with subclasses, not parent class, for more helpful ref docs
-    runner: Union[SubprocessRunner, DockerRunner],
+    runner: Union[SubprocessRunner, DockerRunner, PodmanRunner],
     pipeline_root: str = './local_outputs',
     raise_on_error: bool = True,
 ) -> None:
@@ -106,7 +120,7 @@ def init(
     Once called, components can be invoked locally outside of a pipeline definition.
 
     Args:
-        runner: The runner to use. Supported runners: kfp.local.SubprocessRunner and kfp.local.DockerRunner.
+        runner: The runner to use. Supported runners: kfp.local.SubprocessRunner, kfp.local.DockerRunner, and kfp.local.PodmanRunner.
         pipeline_root: Destination for task outputs.
         raise_on_error: If True, raises an exception when a local task execution fails. If False, fails gracefully and does not terminate the current program.
     """

sdk/python/kfp/local/podman_task_handler.py

@@ -0,0 +1,106 @@
+import os
+from typing import Any, Dict, List
+
+from kfp.local import config
+from kfp.local import status
+from kfp.local import task_handler_interface
+
+
+class PodmanTaskHandler(task_handler_interface.ITaskHandler):
+    """The task handler corresponding to PodmanRunner."""
+
+    def __init__(
+        self,
+        image: str,
+        full_command: List[str],
+        pipeline_root: str,
+        runner: config.PodmanRunner,
+    ) -> None:
+        self.image = image
+        self.full_command = full_command
+        self.pipeline_root = pipeline_root
+        self.runner = runner
+
+    def get_volumes_to_mount(self) -> Dict[str, Any]:
+        """Gets the volume configuration to mount the pipeline root to the
+        container so that outputs can be obtained outside of the container."""
+        if not os.path.isabs(self.pipeline_root):
+            # Defensive check that is enforced by upstream code.
+            # users should not hit this
+            raise ValueError(
+                "'pipeline_root' should be an absolute path to correctly construct the volume mount specification."
+            )
+        return {self.pipeline_root: {'bind': self.pipeline_root, 'mode': 'rw'}}
+
+    def run(self) -> status.Status:
+        """Runs the container and returns the status."""
+        # nest podman import in case not available in user env so that
+        # this module is runnable, even if not using PodmanRunner
+        import podman
+        client = podman.PodmanClient()
+        try:
+            volumes = self.get_volumes_to_mount()
+            return_code = run_container(
+                client=client,
+                image=self.image,
+                command=self.full_command,
+                volumes=volumes,
+            )
+        finally:
+            client.close()
+        return status.Status.SUCCESS if return_code == 0 else status.Status.FAILURE
+
+
+def add_latest_tag_if_not_present(image: str) -> str:
+    """Adds the 'latest' tag if no tag is present in the image name."""
+    if ':' not in image:
+        return f'{image}:latest'
+    return image
+
+
+def run_container(
+    client: 'podman.PodmanClient',
+    image: str,
+    command: List[str],
+    volumes: Dict[str, Dict[str, str]],
+) -> int:
+    """Runs a container using Podman.
+
+    Args:
+        client: The Podman client instance.
+        image: The container image to use.
+        command: The command to run in the container.
+        volumes: Dictionary mapping host paths to volume configuration.
+
+    Returns:
+        The exit code of the container.
+    """
+
+    image = add_latest_tag_if_not_present(image=image)
+
+    image_exists = any(
+        image in existing_image.tags for existing_image in client.images.list())
+    if image_exists:
+        print(f'Found image {image!r}\n')
+    else:
+        print(f'Pulling image {image!r}')
+        repository, tag = image.split(':')
+        client.images.pull(repository=repository, tag=tag)
+        print('Image pull complete\n')
+
+    # Create and run container
+    container = client.containers.create(
+        image=image,
+        command=command,
+        volumes=volumes,
+        detach=True,
+    )
+
+    # Start container and stream logs
+    container.start()
+    for line in container.logs(stream=True):
+        print(line.decode(), end='')
+
+    # Wait for container to finish and get exit code
+    container.wait()
+    return container.exit_code