Merge pull request opendatahub-io#10 from kubeflow/master

[pull] master from kubeflow:master

Author: rimolive

sdk/python/kfp_tekton/compiler/_data_passing_rewriter.py

@@ -764,3 +764,25 @@ def _append_original_pr_name_env(task_template):
     for step in task_template['taskSpec']['steps']:
         if step['name'] == 'main':
             _append_original_pr_name_env_to_step(step)
+
+
+def fix_big_data_passing_using_volume(workflow, pipeline_conf):
+    if workflow['spec'].get('workspaces') and pipeline_conf.data_passing_method._volume is not None:
+        volume_dict = pipeline_conf.data_passing_method._volume.to_dict()
+        volume_dict_pvc = volume_dict.get('persistent_volume_claim')
+        if volume_dict_pvc:
+            for workspace in workflow['spec']['workspaces']:
+                if workspace['name'] == workflow['metadata']['name']:
+                    workspace.pop('volumeClaimTemplate')
+                    temp_vc = {}
+                    for key, value in volume_dict_pvc.items():
+                        if value != None:
+                            if key == 'claim_name':
+                                temp_vc['claimName'] = value
+                            if key == 'read_only':
+                                temp_vc['readOnly'] = value
+                    workspace['persistentVolumeClaim'] = temp_vc
+                    if pipeline_conf.data_passing_method._path_prefix != None:
+                        workspace['subPath'] = pipeline_conf.data_passing_method._path_prefix
+                    break
+    return workflow

sdk/python/kfp_tekton/compiler/compiler.py

@@ -39,7 +39,7 @@
 from kfp.dsl._metadata import _extract_pipeline_metadata
 # KFP-Tekton imports
 from kfp_tekton.compiler import __tekton_api_version__ as tekton_api_version
-from kfp_tekton.compiler._data_passing_rewriter import fix_big_data_passing, BIG_DATA_PATH_FORMAT
+from kfp_tekton.compiler._data_passing_rewriter import fix_big_data_passing, fix_big_data_passing_using_volume, BIG_DATA_PATH_FORMAT
 from kfp_tekton.compiler._k8s_helper import convert_k8s_obj_to_json, sanitize_k8s_name, sanitize_k8s_object
 from kfp_tekton.compiler._op_to_template import _op_to_template
 from kfp_tekton.compiler._tekton_handler import _handle_tekton_pipeline_variables, _handle_tekton_custom_task, _process_argo_vars
@@ -1519,6 +1519,9 @@ def _create_workflow(self,
 
     workflow = fix_big_data_passing(workflow, self.loops_pipeline, '-'.join(self._group_names[:-1] + [""]))
 
+    if pipeline_conf and pipeline_conf.data_passing_method is not None:
+      workflow = fix_big_data_passing_using_volume(workflow, pipeline_conf)
+
     workflow.setdefault('metadata', {}).setdefault('annotations', {})['pipelines.kubeflow.org/pipeline_spec'] = \
       json.dumps(pipeline_meta.to_dict(), sort_keys=True)
 

sdk/python/tests/compiler/compiler_tests.py

@@ -639,6 +639,13 @@ def test_big_data_workflow(self):
     from .testdata.big_data_passing import file_passing_pipelines
     self._test_pipeline_workflow(file_passing_pipelines, 'big_data_passing.yaml', skip_noninlined=True)
 
+  def test_big_data_using_volume_workflow(self):
+    """
+    Test compiling a big data passing workflow.
+    """
+    from .testdata.artifact_passing_using_volume import artifact_passing_pipeline
+    self._test_pipeline_workflow(artifact_passing_pipeline, 'artifact_passing_using_volume.yaml', skip_noninlined=True)
+
   def test_create_component_from_func_workflow(self):
     """
     Test compiling a creating component from func workflow.

sdk/python/tests/compiler/testdata/artifact_passing_using_volume.py

@@ -16,35 +16,73 @@
 
 from pathlib import Path
 
-import kfp
-from kfp.components import load_component_from_file
+import kfp as kfp
+from kfp.components import load_component_from_file, create_component_from_func
+from typing import NamedTuple
 
 test_data_dir = Path(__file__).parent / 'test_data'
-producer_op = load_component_from_file(str(test_data_dir / 'produce_2.component.yaml'))
-processor_op = load_component_from_file(str(test_data_dir / 'process_2_2.component.yaml'))
-consumer_op = load_component_from_file(str(test_data_dir / 'consume_2.component.yaml'))
+producer_op = load_component_from_file(
+    str(test_data_dir / 'produce_2.component.yaml'))
+processor_op = load_component_from_file(
+    str(test_data_dir / 'process_2_2.component.yaml'))
+consumer_op = load_component_from_file(
+    str(test_data_dir / 'consume_2.component.yaml'))
+
+
+def metadata_and_metrics() -> NamedTuple(
+    "Outputs",
+    [("mlpipeline_ui_metadata", "UI_metadata"), ("mlpipeline_metrics", "Metrics"
+                                                )],
+):
+    metadata = {
+        "outputs": [{
+            "storage": "inline",
+            "source": "*this should be bold*",
+            "type": "markdown"
+        }]
+    }
+    metrics = {
+        "metrics": [
+            {
+                "name": "train-accuracy",
+                "numberValue": 0.9,
+            },
+            {
+                "name": "test-accuracy",
+                "numberValue": 0.7,
+            },
+        ]
+    }
+    from collections import namedtuple
+    import json
+
+    return namedtuple("output",
+                      ["mlpipeline_ui_metadata", "mlpipeline_metrics"])(
+                          json.dumps(metadata), json.dumps(metrics))
 
 
 @kfp.dsl.pipeline()
 def artifact_passing_pipeline():
     producer_task = producer_op()
-    processor_task = processor_op(producer_task.outputs['output_1'], producer_task.outputs['output_2'])
-    consumer_task = consumer_op(processor_task.outputs['output_1'], processor_task.outputs['output_2'])
+    processor_task = processor_op(producer_task.outputs['output_1'],
+                                  producer_task.outputs['output_2'])
+    consumer_task = consumer_op(processor_task.outputs['output_1'],
+                                processor_task.outputs['output_2'])
 
+    markdown_task = create_component_from_func(func=metadata_and_metrics)()
     # This line is only needed for compiling using dsl-compile to work
-    kfp.dsl.get_pipeline_conf().data_passing_method = volume_based_data_passing_method
+    kfp.dsl.get_pipeline_conf(
+    ).data_passing_method = volume_based_data_passing_method
 
 
 from kubernetes.client.models import V1Volume, V1PersistentVolumeClaimVolumeSource
 from kfp.dsl import data_passing_methods
 
-
 volume_based_data_passing_method = data_passing_methods.KubernetesVolume(
     volume=V1Volume(
         name='data',
         persistent_volume_claim=V1PersistentVolumeClaimVolumeSource(
-            claim_name='data-volume',
-        ),
+            claim_name='data-volume',),
     ),
     path_prefix='artifact_data/',
 )