Add option to produce pipelinerun taskspec as taskref (#810)

* add option to produce taskspec as taskref

* add rbac for task templates

Author: Tomcli

manifests/kustomize/base/installs/multi-user/api-service/cluster-role.yaml

@@ -54,6 +54,7 @@ rules:
   - taskruns
   - conditions
   - runs
+  - tasks
   verbs:
   - create
   - get

manifests/kustomize/base/installs/multi-user/pipelines-ui/cluster-role.yaml

@@ -46,6 +46,7 @@ rules:
   - pipelineruns
   - taskruns
   - conditions
+  - tasks
   verbs:
   - create
   - get

manifests/kustomize/base/installs/multi-user/scheduled-workflow/cluster-role.yaml

@@ -41,6 +41,7 @@ rules:
   - taskruns
   - conditions
   - runs
+  - tasks
   verbs:
   - create
   - get

manifests/kustomize/base/pipeline/ml-pipeline-apiserver-role.yaml

@@ -33,6 +33,7 @@ rules:
   - taskruns
   - conditions
   - runs
+  - tasks
   verbs:
   - create
   - get

manifests/kustomize/base/pipeline/ml-pipeline-scheduledworkflow-role.yaml

@@ -43,6 +43,7 @@ rules:
   - taskruns
   - conditions
   - runs
+  - tasks
   verbs:
   - create
   - get

manifests/kustomize/base/pipeline/ml-pipeline-ui-role.yaml

@@ -48,6 +48,7 @@ rules:
   - pipelineruns
   - taskruns
   - conditions
+  - tasks
   verbs:
   - create
   - get

manifests/kustomize/base/pipeline/pipeline-runner-role.yaml

@@ -85,6 +85,7 @@ rules:
   - taskruns
   - conditions
   - runs
+  - tasks
   verbs:
   - create
   - get

samples/flip-coin/condition-with-taskref.py

@@ -0,0 +1,69 @@
+# Copyright 2020 kubeflow.org
+#
+# 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.
+
+from kfp import dsl
+from kfp import components
+
+def random_num(low:int, high:int) -> int:
+    """Generate a random number between low and high."""
+    import random
+    result = random.randint(low, high)
+    print(result)
+    return result
+
+def flip_coin() -> str:
+    """Flip a coin and output heads or tails randomly."""
+    import random
+    result = 'heads' if random.randint(0, 1) == 0 else 'tails'
+    print(result)
+    return result
+
+def print_msg(msg: str):
+    """Print a message."""
+    print(msg)
+
+
+flip_coin_op = components.create_component_from_func(
+    flip_coin, base_image='python:alpine3.6')
+print_op = components.create_component_from_func(
+    print_msg, base_image='python:alpine3.6')
+random_num_op = components.create_component_from_func(
+    random_num, base_image='python:alpine3.6')
+
+@dsl.pipeline(
+    name='conditional-execution-pipeline',
+    description='Shows how to use dsl.Condition().'
+)
+def flipcoin_pipeline():
+    flip = flip_coin_op()
+    with dsl.Condition(flip.output == 'heads'):
+        random_num_head = random_num_op(0, 9)
+        with dsl.Condition(random_num_head.output > 5):
+            print_op('heads and %s > 5!' % random_num_head.output)
+        with dsl.Condition(random_num_head.output <= 5):
+            print_op('heads and %s <= 5!' % random_num_head.output)
+
+    with dsl.Condition(flip.output == 'tails'):
+        random_num_tail = random_num_op(10, 19)
+        with dsl.Condition(random_num_tail.output > 15):
+            print_op('tails and %s > 15!' % random_num_tail.output)
+        with dsl.Condition(random_num_tail.output <= 15):
+            print_op('tails and %s <= 15!' % random_num_tail.output)
+
+
+if __name__ == '__main__':
+    from kfp_tekton.compiler import TektonCompiler
+    compiler = TektonCompiler()
+    compiler.produce_taskspec = False
+    compiler.compile(flipcoin_pipeline, __file__.replace('.py', '.yaml'))

sdk/python/kfp_tekton/compiler/compiler.py

@@ -26,6 +26,7 @@
 import collections
 from os import environ as env
 from typing import Callable, List, Text, Dict, Any
+import hashlib
 
 import yaml
 # Kubeflow Pipeline imports
@@ -122,6 +123,7 @@ def __init__(self, **kwargs):
     # Input and output artifacts are hash maps for metadata tracking.
     # artifact_items is the artifact dependency map
     # loops_pipeline recorde the loop tasks information for each loops
+    # produce_taskspec Produces task spec as part of Tekton pipelineRuns
     self.input_artifacts = {}
     self.output_artifacts = {}
     self.artifact_items = {}
@@ -134,6 +136,7 @@ def __init__(self, **kwargs):
     self.pipeline_annotations = {}
     self.tekton_inline_spec = True
     self.resource_in_separate_yaml = False
+    self.produce_taskspec = True
     super().__init__(**kwargs)
 
   def _set_pipeline_conf(self, tekton_pipeline_conf: TektonPipelineConf):
@@ -1431,7 +1434,46 @@ def _create_and_write_workflow(self,
       workflow['metadata']['annotations']['tekton.dev/resource_templates'] = json.dumps(loop_package_annotations,
                                                                                         sort_keys=True)
     # Need to compiles after all the CRs being processed.
+    # Convert taskspec into task templates if specified.
+    if not self.produce_taskspec:
+      component_sha = {}
+      for task in workflow['spec']['pipelineSpec']['tasks']:
+        if task.get('taskSpec'):
+          component_spec_digest = hashlib.sha1(json.dumps(task['taskSpec'], sort_keys=True).encode()).hexdigest()
+          if component_spec_digest not in component_sha.keys():
+            task_template = {}
+            task_template['metadata'] = {}
+            if task['taskSpec'].get('metadata', None):
+              task_template['metadata'] = task['taskSpec'].pop('metadata', None)
+            if task['taskSpec'].get('apiVersion', None) and task['taskSpec'].get('kind', None):
+              task_template['apiVersion'] = task['taskSpec']['apiVersion']
+              task_template['kind'] = task['taskSpec']['kind']
+            else:
+              task_template['apiVersion'] = tekton_api_version
+              task_template['kind'] = 'Task'
+            task_template['spec'] = task['taskSpec']
+            task_template['metadata']['name'] = component_spec_digest
+            component_sha[component_spec_digest] = task_template
+          task.pop("taskSpec", None)
+          task['taskRef'] = {'name': component_spec_digest}
+      # Output task templates into individual files if specified, else append task templates to annotations
+      if self.resource_in_separate_yaml:
+        for key, value in component_sha.items():
+          TektonCompiler._write_workflow(workflow=value,
+                                         package_path=os.path.splitext(package_path)[0] +
+                                         key + '.yaml')
+      else:
+        resource_templates = workflow['metadata']['annotations'].get('tekton.dev/resource_templates', [])
+        if resource_templates:
+          resource_templates = json.loads(resource_templates)
+        for value in component_sha.values():
+          resource_templates.append(value)
+        if resource_templates:
+          workflow['metadata']['annotations']['tekton.dev/resource_templates'] = json.dumps(resource_templates,
+                                                                                            sort_keys=True)
+
     TektonCompiler._write_workflow(workflow=workflow, package_path=package_path)   # Tekton change
+
     # Separate custom task CR from the main workflow
     for i in range(len(self.custom_task_crs)):
       TektonCompiler._write_workflow(workflow=self.custom_task_crs[i],