support pipeline level imagepullsecret in DSL (#745)

* support pipeline level imagepullsecret in DSL

* use kubernetes native input parameter for imagepullsecrets

* expose a module level function to configure the pipeline settings for the current default pipeline

sdk/python/kfp/compiler/compiler.py

@@ -476,6 +476,11 @@ def _create_pipeline_workflow(self, args, pipeline):
         'serviceAccountName': 'pipeline-runner'
       }
     }
+    if len(pipeline.conf.image_pull_secrets) > 0:
+      image_pull_secrets = []
+      for image_pull_secret in pipeline.conf.image_pull_secrets:
+        image_pull_secrets.append(K8sHelper.convert_k8s_obj_to_json(image_pull_secret))
+      workflow['spec']['imagePullSecrets'] = image_pull_secrets
     if exit_handler:
       workflow['spec']['onExit'] = exit_handler.name
     if volumes:

sdk/python/kfp/dsl/__init__.py

@@ -14,7 +14,7 @@
 
 
 from ._pipeline_param import PipelineParam
-from ._pipeline import Pipeline, pipeline
+from ._pipeline import Pipeline, pipeline, get_pipeline_conf
 from ._container_op import ContainerOp
 from ._ops_group import OpsGroup, ExitHandler, Condition
 from ._python_component import python_component

sdk/python/kfp/dsl/_pipeline.py

@@ -37,6 +37,28 @@ def _pipeline(func):
 
   return _pipeline
 
+class PipelineConf():
+  """PipelineConf contains pipeline level settings
+  """
+  def __init__(self):
+    self.image_pull_secrets = []
+
+  def set_image_pull_secrets(self, image_pull_secrets):
+    """ configure the pipeline level imagepullsecret
+
+    Args:
+      image_pull_secrets: a list of Kubernetes V1LocalObjectReference
+      For detailed description, check Kubernetes V1LocalObjectReference definition
+      https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1LocalObjectReference.md
+    """
+    self.image_pull_secrets = image_pull_secrets
+
+def get_pipeline_conf():
+  """Configure the pipeline level setting to the current pipeline
+    Note: call the function inside the user defined pipeline function.
+  """
+  return Pipeline.get_default_pipeline().conf
+
 class Pipeline():
   """A pipeline contains a list of operators.
 
@@ -86,6 +108,7 @@ def __init__(self, name: str):
     # Add the root group.
     self.groups = [_ops_group.OpsGroup('pipeline', name=name)]
     self.group_id = 0
+    self.conf = PipelineConf()
 
   def __enter__(self):
     if Pipeline._default_pipeline:
@@ -139,3 +162,5 @@ def get_next_group_id(self):
 
     self.group_id += 1
     return self.group_id
+
+

sdk/python/tests/compiler/compiler_tests.py

@@ -232,3 +232,7 @@ def test_py_volume(self):
   def test_py_retry(self):
     """Test retry functionality."""
     self._test_py_compile('retry')
+
+  def test_py_image_pull_secret(self):
+    """Test pipeline imagepullsecret."""
+    self._test_py_compile('imagepullsecret')

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

@@ -0,0 +1,50 @@
+# Copyright 2018 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+import kfp.dsl as dsl
+from kubernetes import client as k8s_client
+
+
+class GetFrequentWordOp(dsl.ContainerOp):
+  """A get frequent word class representing a component in ML Pipelines.
+
+  The class provides a nice interface to users by hiding details such as container,
+  command, arguments.
+  """
+  def __init__(self, name, message):
+    """Args:
+         name: An identifier of the step which needs to be unique within a pipeline.
+         message: a dsl.PipelineParam object representing an input message.
+    """
+    super(GetFrequentWordOp, self).__init__(
+        name=name,
+        image='python:3.5-jessie',
+        command=['sh', '-c'],
+        arguments=['python -c "from collections import Counter; '
+                   'words = Counter(\'%s\'.split()); print(max(words, key=words.get))" '
+                   '| tee /tmp/message.txt' % message],
+        file_outputs={'word': '/tmp/message.txt'})
+
+@dsl.pipeline(
+  name='Save Most Frequent',
+  description='Get Most Frequent Word and Save to GCS'
+)
+def save_most_frequent_word(message: str):
+  """A pipeline function describing the orchestration of the workflow."""
+
+  counter = GetFrequentWordOp(
+          name='get-Frequent',
+          message=message)
+  dsl.get_pipeline_conf().set_image_pull_secrets([k8s_client.V1ObjectReference(name="secretA")])

sdk/python/tests/compiler/testdata/imagepullsecret.yaml

@@ -0,0 +1,69 @@
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: save-most-frequent-
+spec:
+  arguments:
+    parameters:
+    - name: message
+  entrypoint: save-most-frequent
+  imagePullSecrets:
+  - name: secretA
+  serviceAccountName: pipeline-runner
+  templates:
+  - container:
+      args:
+      - python -c "from collections import Counter; words = Counter('{{inputs.parameters.message}}'.split());
+        print(max(words, key=words.get))" | tee /tmp/message.txt
+      command:
+      - sh
+      - -c
+      image: python:3.5-jessie
+    inputs:
+      parameters:
+      - name: message
+    name: get-frequent
+    outputs:
+      artifacts:
+      - name: mlpipeline-ui-metadata
+        path: /mlpipeline-ui-metadata.json
+        s3:
+          accessKeySecret:
+            key: accesskey
+            name: mlpipeline-minio-artifact
+          bucket: mlpipeline
+          endpoint: minio-service.kubeflow:9000
+          insecure: true
+          key: runs/{{workflow.uid}}/{{pod.name}}/mlpipeline-ui-metadata.tgz
+          secretKeySecret:
+            key: secretkey
+            name: mlpipeline-minio-artifact
+      - name: mlpipeline-metrics
+        path: /mlpipeline-metrics.json
+        s3:
+          accessKeySecret:
+            key: accesskey
+            name: mlpipeline-minio-artifact
+          bucket: mlpipeline
+          endpoint: minio-service.kubeflow:9000
+          insecure: true
+          key: runs/{{workflow.uid}}/{{pod.name}}/mlpipeline-metrics.tgz
+          secretKeySecret:
+            key: secretkey
+            name: mlpipeline-minio-artifact
+      parameters:
+      - name: get-frequent-word
+        valueFrom:
+          path: /tmp/message.txt
+  - dag:
+      tasks:
+      - arguments:
+          parameters:
+          - name: message
+            value: '{{inputs.parameters.message}}'
+        name: get-frequent
+        template: get-frequent
+    inputs:
+      parameters:
+      - name: message
+    name: save-most-frequent