Repository: getindata/kedro-kubeflow
Pull request #10 (2021-01-06 12:02:19)

MLOPS-73 | kubeflow init

szczeles: Changelog entry added + sample file provided
szczeles: Oh, wait, I will copy the file into README md
em-pe: Ignore failing coverage checks for now. I'll add CLI coverage and then restore them. As an admin you should be able to merge despite failing checks 

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 40925d3..4e5963f 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,6 +3,7 @@
 ## [Unreleased]
 
 - Support for inter-steps volume: setup (one volume per pipeline run), initial load (the content of `data/` directory from the image and mount to all the steps for artifacts passing.
+- `kubeflow init` command added to generate sample configuration file.
 
 ## [0.1.8] - 2021-01-05
 
diff --git a/README.md b/README.md
index c8f3c6f..cdedd74 100644
--- a/README.md
+++ b/README.md
@@ -28,9 +28,29 @@ Usage: kedro kubeflow [OPTIONS] COMMAND [ARGS]...
  
  Commands:
    compile          Translates Kedro pipeline into YAML file with Kubeflow pipeline definition
+   init             Initializes configuration for the plugin
    list-pipelines   List deployed pipeline definitions
    run-once         Deploy pipeline as a single run within given experiment.
    schedule         Schedules recurring execution of latest version of the pipeline
    ui               Open Kubeflow Pipelines UI in new browser tab
    upload-pipeline  Uploads pipeline to Kubeflow server
 ```
+
+## Configuration file
+
+`kedro init` generates configuration file for the plugin, but users may want
+to adjust it to the requirements of the environment:
+
+```
+host: http://10.43.77.224
+
+run_config:
+  image: new-kedro-project
+  experiment_name: New Kedro Project
+  run_name: New Kedro Project
+  wait_for_completion: False
+  volume:
+    storageclass: # default
+    size: 1Gi
+    access_modes: [ReadWriteOnce]
+```
diff --git a/kedro_kubeflow/plugin.py b/kedro_kubeflow/plugin.py
index c77c95b..01d1dac 100644
--- a/kedro_kubeflow/plugin.py
+++ b/kedro_kubeflow/plugin.py
@@ -1,9 +1,11 @@
-import click
+from pathlib import Path
 
+import click
 from semver import VersionInfo
 from kedro import __version__ as kedro_version
 from kedro.config import MissingConfigException
 from .kfpclient import KubeflowClient
+from .utils import strip_margin
 
 KEDRO_VERSION = VersionInfo.parse(kedro_version)
 
@@ -17,7 +19,6 @@ if KEDRO_VERSION.match(">=0.17.0"):
         return metadata.project_name
 else: # <0.17
     from kedro.framework.context import load_context
-    from pathlib import Path
 
     def get_context(metadata, env):
         return load_context(Path.cwd(), env=env)
@@ -41,17 +42,22 @@ def kubeflow_group(ctx, metadata, env):
     """Interact with Kubeflow Pipelines"""
     ctx.ensure_object(dict)
     ctx.obj['kedro_ctx'] = get_context(metadata, env)
+    ctx.obj['project_name'] = get_project_name(metadata, env)
+
+
+def setup_context(ctx):
     ctx.obj['config'] = ctx.obj['kedro_ctx'].config_loader.get(CONFIG_FILE_PATTERN)
     if 'host' not in ctx.obj['config'].keys():
         raise MissingConfigException("No 'host' defined in kubeflow.yml")
 
-    ctx.obj['kfp_client'] = KubeflowClient(ctx.obj['config'], get_project_name(metadata, env), ctx.obj['kedro_ctx'])
+    ctx.obj['kfp_client'] = KubeflowClient(ctx.obj['config'], ctx.obj['project_name'], ctx.obj['kedro_ctx'])
 
 
 @kubeflow_group.command()
 @click.pass_context
 def list_pipelines(ctx):
     """List deployed pipeline definitions"""
+    setup_context(ctx)
     click.echo(ctx.obj['kfp_client'].list_pipelines())
 
 
@@ -65,6 +71,7 @@ def list_pipelines(ctx):
 @click.pass_context
 def run_once(ctx, image: str, pipeline: str, experiment_name: str, run_name: str, wait: bool, image_pull_policy: str):
     """Deploy pipeline as a single run within given experiment. Config can be specified in kubeflow.yml as well."""
+    setup_context(ctx)
     conf = ctx.obj['config']
     run_conf = conf.get("run_config", {})
     image = image if image else run_conf['image']
@@ -80,6 +87,7 @@ def run_once(ctx, image: str, pipeline: str, experiment_name: str, run_name: str
 @click.pass_context
 def ui(ctx) -> None:
     """Open Kubeflow Pipelines UI in new browser tab"""
+    setup_context(ctx)
     import webbrowser
     host = ctx.obj['config']['host']
     webbrowser.open_new_tab(host)
@@ -93,6 +101,7 @@ def ui(ctx) -> None:
 @click.pass_context
 def compile(ctx, image, pipeline, output, image_pull_policy) -> None:
     """Translates Kedro pipeline into YAML file with Kubeflow Pipeline definition"""
+    setup_context(ctx)
     conf = ctx.obj['config']
     run_conf = conf.get("run_config", {})
     image = image if image else run_conf['image']
@@ -106,6 +115,7 @@ def compile(ctx, image, pipeline, output, image_pull_policy) -> None:
 @click.pass_context
 def upload_pipeline(ctx, image, pipeline, image_pull_policy) -> None:
     """Uploads pipeline to Kubeflow server"""
+    setup_context(ctx)
     conf = ctx.obj['config']
     run_conf = conf.get("run_config", {})
     image = image if image else run_conf['image']
@@ -118,9 +128,35 @@ def upload_pipeline(ctx, image, pipeline, image_pull_policy) -> None:
 @click.pass_context
 def schedule(ctx, experiment_name: str, cron_expression: str):
     """Schedules recurring execution of latest version of the pipeline"""
+    setup_context(ctx)
     conf = ctx.obj['config']
     run_conf = conf.get("run_config", {})
     experiment_name = experiment_name if experiment_name else run_conf['experiment_name']
 
     ctx.obj['kfp_client'].schedule(experiment_name, cron_expression)
 
+@kubeflow_group.command()
+@click.argument("kfp_url", type=str)
+@click.option("-x", "--experiment-name", "experiment_name", type=str, help="Name of experiment associated with this run.", default="default")
+@click.pass_context
+def init(ctx, kfp_url: str, experiment_name: str):
+    """Initializes configuration for the plugin"""
+    image = ctx.obj['kedro_ctx'].project_path.name # default from kedro-docker
+    config = f"""
+    |host: {kfp_url}
+    |
+    |run_config:
+    |  image: {image}
+    |  experiment_name: {ctx.obj['project_name']}
+    |  run_name: {ctx.obj['project_name']}
+    |  wait_for_completion: False
+    |  volume:
+    |    storageclass: # default
+    |    #size: 1Gi
+    |    #access_modes: [ReadWriteOnce]
+    """
+    config_path = Path.cwd().joinpath('conf/base/kubeflow.yaml')
+    with open(config_path, 'w') as f:
+        f.write(strip_margin(config))
+
+    click.echo(f"Configuration generated in {config_path}")
diff --git a/kedro_kubeflow/utils.py b/kedro_kubeflow/utils.py
new file mode 100644
index 0000000..57ba283
--- /dev/null
+++ b/kedro_kubeflow/utils.py
@@ -0,0 +1,4 @@
+import re
+
+def strip_margin(text: str) -> str:
+    return re.sub("\n[ \t]*\\|", "\n", text).strip()


Test patch:
diff --git a/tests/test_mod.py b/tests/test_mod.py
index 8a8d7dd..4b93e76 100644
--- a/tests/test_mod.py
+++ b/tests/test_mod.py
@@ -10,6 +10,7 @@ import kfp
 from google.auth.exceptions import DefaultCredentialsError
 
 from kedro_kubeflow.kfpclient import KubeflowClient
+from kedro_kubeflow.utils import strip_margin
 from kedro.pipeline import Pipeline, node
 
 
@@ -18,10 +19,6 @@ def identity(input1: str):
 
 
 class TestKubeflowClient(unittest.TestCase):
-    @staticmethod
-    def _strip_margin(text: str) -> str:
-        return re.sub("\n[ \t]*\\|", "\n", text).strip()
-
     def create_pipeline(self):
         return Pipeline(
             [
@@ -76,7 +73,7 @@ class TestKubeflowClient(unittest.TestCase):
         |Name      ID
         |--------  ------
         |somename  someid"""
-        self.assertEqual(output, TestKubeflowClient._strip_margin(expected_output))
+        self.assertEqual(output, strip_margin(expected_output))
 
     def test_should_run_pipeline_without_waiting(self):
         # given