[KFP] Merge KFP2 Feature Branch

.github/workflows/ci.yaml

@@ -138,19 +138,25 @@ jobs:
         run: MLRUN_DOCKER_REGISTRY=ghcr.io/ MLRUN_DOCKER_CACHE_FROM_TAG=${{ steps.docker_cache.outputs.tag }} make test-migrations-dockerized
 
   package-tests:
-    name: Run package tests (Python ${{ matrix.python-version }})
+    name: Run package tests (Python ${{ matrix.python-version }}; Pipeline ${{ matrix.pipeline-adapter }})
     runs-on: ubuntu-latest
     strategy:
       matrix:
         # 3.9 is the current >= 1.3.0 python version
         python-version: [3.9]
+        default-pipeline-adapter: ["kfp-v1-8"]
+        pipeline-adapter: ["kfp-v1-8", "kfp-v2"]
     steps:
     - uses: actions/checkout@v3
     - name: Set up python ${{ matrix.python-version }}
       uses: actions/setup-python@v4
       with:
         python-version: ${{ matrix.python-version }}
+        pipeline-adapter: ${{ matrix.pipeline-adapter }}
         cache: 'pip'
+    - name: Change default pipeline adapter for MLRun
+      if: matrix.default-pipeline-adapter != matrix.pipeline-adapter
+      run: sed -i -e 's/${{ matrix.default-pipeline-adapter }}/${{ matrix.pipeline-adapter }}/g' requirements.txt
     - name: Install automation scripts dependencies and add mlrun to dev packages
       run: pip install -r automation/requirements.txt && pip install -e .
     - name: Test package

dockerfiles/mlrun-api/Dockerfile

@@ -74,7 +74,9 @@ ENV UVICORN_TIMEOUT_KEEP_ALIVE=${MLRUN_HTTPDB__HTTP_CONNECTION_TIMEOUT_KEEP_ALIV
 
 COPY . .
 
-RUN python -m pip install .[complete-api]
+RUN python -m pip install .[complete-api] &&\
+    pip install ./pipeline-adapters/mlrun-pipelines-kfp-common &&\
+    pip install ./pipeline-adapters/mlrun-pipelines-kfp-v1-8 \
 
 VOLUME /mlrun/db
 

dockerfiles/test/Dockerfile

@@ -73,6 +73,8 @@ RUN python -m pip install \
 
 COPY . .
 
-RUN pip install -e .[complete]
+RUN pip install -e .[complete] &&\
+    pip install -e ./pipeline-adapters/mlrun-pipelines-kfp-common &&\
+    pip install -e ./pipeline-adapters/mlrun-pipelines-kfp-v1-8
 
 ENV NO_COLOR=1

mlrun/__init__.py

@@ -22,11 +22,16 @@
     "handler",
     "ArtifactType",
     "get_secret_or_env",
+    "mount_v3io",
+    "v3io_cred",
+    "auto_mount",
+    "VolumeMount",
 ]
 
 from os import environ, path
 
 import dotenv
+import mlrun_pipelines
 
 from .config import config as mlconf
 from .datastore import DataItem, store_manager
@@ -35,7 +40,6 @@
 from .execution import MLClientCtx
 from .model import RunObject, RunTemplate, new_task
 from .package import ArtifactType, DefaultPackager, Packager, handler
-from .platforms import VolumeMount, auto_mount, mount_v3io, v3io_cred
 from .projects import (
     ProjectMetadata,
     build_function,
@@ -65,6 +69,11 @@
 
 __version__ = Version().get()["version"]
 
+VolumeMount = mlrun_pipelines.common.mounts.VolumeMount
+mount_v3io = mlrun_pipelines.mounts.mount_v3io
+v3io_cred = mlrun_pipelines.mounts.v3io_cred
+auto_mount = mlrun_pipelines.mounts.auto_mount
+
 
 def get_version():
     """get current mlrun version"""

mlrun/__main__.py

@@ -27,6 +27,7 @@
 import dotenv
 import pandas as pd
 import yaml
+from mlrun_pipelines.mounts import auto_mount as auto_mount_modifier
 from tabulate import tabulate
 
 import mlrun
@@ -37,7 +38,6 @@
 from .db import get_run_db
 from .errors import err_to_str
 from .model import RunTemplate
-from .platforms import auto_mount as auto_mount_modifier
 from .projects import load_project
 from .run import (
     get_object,
@@ -466,6 +466,17 @@ def run(
     is_flag=True,
     help="ensure the project exists, if not, create project",
 )
+@click.option(
+    "--state-file-path", default="/tmp/state", help="path to file with state data"
+)
+@click.option(
+    "--image-file-path", default="/tmp/image", help="path to file with image data"
+)
+@click.option(
+    "--full-image-file-path",
+    default="/tmp/fullimage",
+    help="path to file with full image data",
+)
 def build(
     func_url,
     name,
@@ -485,6 +496,9 @@ def build(
     skip,
     env_file,
     ensure_project,
+    state_file_path,
+    image_file_path,
+    full_image_file_path,
 ):
     """Build a container image from code and requirements."""
 
@@ -574,12 +588,12 @@ def build(
         state = func.status.state
         image = func.spec.image
         if kfp:
-            with open("/tmp/state", "w") as fp:
+            with open(state_file_path, "w") as fp:
                 fp.write(state or "none")
             full_image = func.full_image_path(image) or ""
-            with open("/tmp/image", "w") as fp:
+            with open(image_file_path, "w") as fp:
                 fp.write(image)
-            with open("/tmp/fullimage", "w") as fp:
+            with open(full_image_file_path, "w") as fp:
                 fp.write(full_image)
             print("Full image path = ", full_image)
 

mlrun/artifacts/model.py

@@ -93,8 +93,8 @@ def __init__(
         self.model_file = model_file
         self.metrics = metrics or {}
         self.parameters = paraemeters or {}
-        self.inputs = inputs or []
-        self.outputs = outputs or []
+        self.inputs: list[Feature] = inputs or []
+        self.outputs: list[Feature] = outputs or []
         self.framework = framework
         self.algorithm = algorithm
         self.feature_vector = feature_vector

mlrun/db/httpdb.py

@@ -15,7 +15,6 @@
 import enum
 import http
 import re
-import tempfile
 import time
 import traceback
 import typing
@@ -26,9 +25,9 @@
 from typing import Optional, Union
 from urllib.parse import urlparse
 
-import kfp
 import requests
 import semver
+from mlrun_pipelines.utils import compile_pipeline
 
 import mlrun
 import mlrun.common.schemas
@@ -52,7 +51,6 @@
     datetime_to_iso,
     dict_to_json,
     logger,
-    new_pipe_metadata,
     normalize_name,
     version,
 )
@@ -591,7 +589,7 @@ def get_log(self, uid, project="", offset=0, size=None):
         if offset < 0:
             raise MLRunInvalidArgumentError("Offset cannot be negative")
         if size is None:
-            size = int(config.httpdb.logs.pull_logs_default_size_limit)
+            size = int(mlrun.mlconf.httpdb.logs.pull_logs_default_size_limit)
         elif size == -1:
             logger.warning(
                 "Retrieving all logs. This may be inefficient and can result in a large log."
@@ -637,23 +635,25 @@ def watch_log(self, uid, project="", watch=True, offset=0):
 
         state, text = self.get_log(uid, project, offset=offset)
         if text:
-            print(text.decode(errors=config.httpdb.logs.decode.errors))
+            print(text.decode(errors=mlrun.mlconf.httpdb.logs.decode.errors))
         nil_resp = 0
         while True:
             offset += len(text)
             # if we get 3 nil responses in a row, increase the sleep time to 10 seconds
             # TODO: refactor this to use a conditional backoff mechanism
             if nil_resp < 3:
-                time.sleep(int(config.httpdb.logs.pull_logs_default_interval))
+                time.sleep(int(mlrun.mlconf.httpdb.logs.pull_logs_default_interval))
             else:
                 time.sleep(
-                    int(config.httpdb.logs.pull_logs_backoff_no_logs_default_interval)
+                    int(
+                        mlrun.mlconf.httpdb.logs.pull_logs_backoff_no_logs_default_interval
+                    )
                 )
             state, text = self.get_log(uid, project, offset=offset)
             if text:
                 nil_resp = 0
                 print(
-                    text.decode(errors=config.httpdb.logs.decode.errors),
+                    text.decode(errors=mlrun.mlconf.httpdb.logs.decode.errors),
                     end="",
                 )
             else:
@@ -1850,14 +1850,11 @@ def submit_pipeline(
         if isinstance(pipeline, str):
             pipe_file = pipeline
         else:
-            pipe_file = tempfile.NamedTemporaryFile(suffix=".yaml", delete=False).name
-            conf = new_pipe_metadata(
+            pipe_file = compile_pipeline(
                 artifact_path=artifact_path,
                 cleanup_ttl=cleanup_ttl,
-                op_transformers=ops,
-            )
-            kfp.compiler.Compiler().compile(
-                pipeline, pipe_file, type_check=False, pipeline_conf=conf
+                ops=ops,
+                pipeline=pipeline,
             )
 
         if pipe_file.endswith(".yaml"):

mlrun/launcher/base.py

@@ -18,10 +18,11 @@
 import uuid
 from typing import Any, Callable, Optional, Union
 
+import mlrun_pipelines.common.ops
+
 import mlrun.common.schemas
 import mlrun.config
 import mlrun.errors
-import mlrun.kfpops
 import mlrun.lists
 import mlrun.model
 import mlrun.runtimes
@@ -390,7 +391,7 @@ def _wrap_run_result(
             return
 
         if result and runtime.kfp and err is None:
-            mlrun.kfpops.write_kfpmeta(result)
+            mlrun_pipelines.common.ops.write_kfpmeta(result)
 
         self._log_track_results(runtime.is_child, result, run)
 

mlrun/platforms/__init__.py

@@ -17,22 +17,23 @@
 from pprint import pprint
 from time import sleep
 
-from .iguazio import (
-    V3ioStreamClient,
-    VolumeMount,
-    add_or_refresh_credentials,
-    is_iguazio_session_cookie,
-    mount_v3io,
-    v3io_cred,
-)
-from .other import (
+from mlrun_pipelines.common.mounts import VolumeMount
+from mlrun_pipelines.mounts import (
     auto_mount,
     mount_configmap,
     mount_hostpath,
     mount_pvc,
     mount_s3,
     mount_secret,
+    mount_v3io,
     set_env_variables,
+    v3io_cred,
+)
+
+from .iguazio import (
+    V3ioStreamClient,
+    add_or_refresh_credentials,
+    is_iguazio_session_cookie,
 )