feat(pipelineloop): Support last_idx and last_elem (kubeflow#1044)

ScrapCodes · Tomcli · web-flow · commit 30eb17811a7e · 2022-09-30T16:15:55.000Z
* WIP: Support last_idx and last_elem * WIP: SDK changes to support last-idx variable. * fix * fix sdk param * fix controller to correctly update results * fix tests * fix unit tests and lint * fix test cases to be executable (opendatahub-io#5) * fix unit tests and lint * fix test cases to be executatble * removed unused code and added test * fix test Co-authored-by: Tommy Li <Tommy.chaoping.li@ibm.com>
diff --git a/sdk/python/kfp_tekton/compiler/compiler.py b/sdk/python/kfp_tekton/compiler/compiler.py
@@ -144,6 +144,7 @@ def __init__(self, **kwargs):
     self.produce_taskspec = True
     self.security_context = None
     self.automount_service_account_token = None
+    self.group_names = {}
     super().__init__(**kwargs)
 
   def _set_pipeline_conf(self, tekton_pipeline_conf: TektonPipelineConf):
@@ -205,11 +206,13 @@ def _group_to_dag_template(self, group, inputs, outputs, dependencies, pipeline_
     pipeline_name_copy = sanitize_k8s_name(pipeline_name, max_length=LOOP_PIPELINE_NAME_LENGTH)
     sub_group_name_copy = sanitize_k8s_name(sub_group.name, max_length=LOOP_GROUP_NAME_LENGTH, rev_truncate=True)
     self._group_names = [pipeline_name_copy, sub_group_name_copy]
+    raw_group_name = '-'.join(self._group_names)
     if self.uuid:
       self._group_names.insert(1, self.uuid)
     # pipeline name (max 40) + loop id (max 5) + group name (max 16) + two connecting dashes (2) = 63 (Max size for CRD names)
     group_name = '-'.join(self._group_names) if group_type == "loop" or \
         group_type == "graph" or group_type == 'addon' else sub_group.name
+    self.group_names[raw_group_name] = group_name
     template = {
       'metadata': {
         'name': group_name,
@@ -667,6 +670,8 @@ def _get_dependencies(self, pipeline, root_group, op_groups,
                   upstream_op = pipeline.ops[upstream_op_name]
               elif upstream_op_name in opsgroups:
                   upstream_op = opsgroups[upstream_op_name]
+              elif "for-loop" in upstream_op_name:
+                continue
               else:
                   raise ValueError('compiler cannot find the ' +
                                     upstream_op_name)
@@ -740,7 +745,9 @@ def _get_inputs_outputs(
         if param.value:
           continue
         if param.op_name:
-          upstream_op = pipeline.ops[param.op_name]
+          upstream_op = pipeline.ops.get(param.op_name, None)
+          if not upstream_op:
+            continue
           upstream_groups, downstream_groups = \
             self._get_uncommon_ancestors(op_groups, opsgroup_groups, upstream_op, op)
           for i, group_name in enumerate(downstream_groups):
@@ -1823,6 +1830,14 @@ def _create_and_write_workflow(self,
           workflow['metadata']['annotations']['tekton.dev/resource_templates'] = json.dumps(resource_templates,
                                                                                             sort_keys=True)
 
+    # Inject uuid to loop parameter task name if exist
+    if self.uuid:
+      for k, v in self.group_names.items():
+        if workflow['spec'].get('pipelineSpec'):
+          for task in workflow['spec']['pipelineSpec']['tasks']:
+            for param in task.get('params', []):
+              if isinstance(param['value'], str):
+                param['value'] = param['value'].replace(k, v)
     TektonCompiler._write_workflow(workflow=workflow, package_path=package_path)   # Tekton change
 
     # Separate custom task CR from the main workflow
diff --git a/sdk/python/kfp_tekton/tekton.py b/sdk/python/kfp_tekton/tekton.py
@@ -280,6 +280,9 @@ def add_pod_label(self, name: str, value: str):
   def _next_id(self):
     return str(_pipeline.Pipeline.get_default_pipeline().get_next_group_id())
 
+  def _seek_id(self):
+    return str(_pipeline.Pipeline.get_default_pipeline().group_id)
+
   def __init__(self,
                loop_args: Union[str,
                                 _for_loop.ItemList,
@@ -324,6 +327,12 @@ def __init__(self,
                 value=separator
             )
 
+    # param_name is unique on the pipeline level but not cluster level. Therefore, we still need to replace it to a cluster
+    # unique uuid during our compiler step.
+    param_name = "-".join([_pipeline.Pipeline.get_default_pipeline().name, self.type, str(int(self._seek_id()) + 1)])
+    self.last_idx = dsl.PipelineParam(name="last-idx", op_name=param_name)
+    self.last_elem = dsl.PipelineParam(name="last-elem", op_name=param_name)
+
   def __enter__(self) -> Union[Tuple[TektonLoopIterationNumber, LoopArguments], _for_loop.LoopArguments]:
     rev = super().__enter__()
     if self.call_enumerate:
diff --git a/sdk/python/tests/compiler/compiler_tests.py b/sdk/python/tests/compiler/compiler_tests.py
@@ -64,6 +64,13 @@
 
 class TestTektonCompiler(unittest.TestCase):
 
+  def test_last_idx(self):
+    """
+    Test compiling a initial container workflow.
+    """
+    from .testdata.last_idx import pipeline
+    self._test_pipeline_workflow(pipeline, 'last_idx.yaml', skip_noninlined=False)
+
   def test_init_container_workflow(self):
     """
     Test compiling a initial container workflow.
diff --git a/sdk/python/tests/compiler/testdata/last_idx.py b/sdk/python/tests/compiler/testdata/last_idx.py
@@ -0,0 +1,50 @@
+# 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_tekton.compiler import TektonCompiler
+from kfp import dsl
+from kfp.components import load_component_from_text
+from kfp_tekton import tekton
+
+PrintOp = load_component_from_text("""
+  name: print
+  inputs:
+  - name: msg
+  implementation:
+    container:
+      image: alpine:3.6
+      command:
+      - sh
+      - -c
+      - |
+        set -e
+        echo $0
+      - {inputValue: msg }
+""")
+
+
+@dsl.pipeline(name='pipeline')
+def pipeline(param: int = 10):
+    loop_args = "1,2"
+    loop = tekton.Loop.from_string(loop_args, separator=',')
+    with loop.enumerate() as (idx, el):
+        # loop enter-local variables, not available outside of the loop
+        PrintOp(f"it no {idx}: {el}")
+    # using loop's fields, not enter-local variables
+    PrintOp(f"last it: no {loop.last_idx}: {loop.last_elem}")
+
+
+if __name__ == "__main__":
+    TektonCompiler().compile(pipeline, __file__.replace('.py', '.yaml'))
diff --git a/sdk/python/tests/compiler/testdata/last_idx.yaml b/sdk/python/tests/compiler/testdata/last_idx.yaml
@@ -0,0 +1,121 @@
+# Copyright 2021 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.
+
+apiVersion: tekton.dev/v1beta1
+kind: PipelineRun
+metadata:
+  name: pipeline
+  annotations:
+    tekton.dev/output_artifacts: '{}'
+    tekton.dev/input_artifacts: '{"print-2": [{"name": "pipeline-for-loop-4-last-elem",
+      "parent_task": "pipeline-for-loop-4"}, {"name": "pipeline-for-loop-4-last-idx",
+      "parent_task": "pipeline-for-loop-4"}]}'
+    tekton.dev/artifact_bucket: mlpipeline
+    tekton.dev/artifact_endpoint: minio-service.kubeflow:9000
+    tekton.dev/artifact_endpoint_scheme: http://
+    tekton.dev/artifact_items: '{"print": [], "print-2": []}'
+    sidecar.istio.io/inject: "false"
+    tekton.dev/template: ''
+    pipelines.kubeflow.org/big_data_passing_format: $(workspaces.$TASK_NAME.path)/artifacts/$ORIG_PR_NAME/$TASKRUN_NAME/$TASK_PARAM_NAME
+    pipelines.kubeflow.org/pipeline_spec: '{"inputs": [{"default": "10", "name": "param",
+      "optional": true, "type": "Integer"}], "name": "pipeline"}'
+  labels:
+    pipelines.kubeflow.org/pipelinename: ''
+    pipelines.kubeflow.org/generation: ''
+spec:
+  params:
+  - name: param
+    value: '10'
+  pipelineSpec:
+    params:
+    - name: param
+      default: '10'
+    tasks:
+    - name: print-2
+      params:
+      - name: pipeline-for-loop-4-last-elem
+        value: $(tasks.pipeline-for-loop-4.results.last-elem)
+      - name: pipeline-for-loop-4-last-idx
+        value: $(tasks.pipeline-for-loop-4.results.last-idx)
+      taskSpec:
+        steps:
+        - name: main
+          command:
+          - sh
+          - -c
+          - |
+            set -e
+            echo $0
+          - 'last it: no $(inputs.params.pipeline-for-loop-4-last-idx): $(inputs.params.pipeline-for-loop-4-last-elem)'
+          image: alpine:3.6
+        params:
+        - name: pipeline-for-loop-4-last-elem
+        - name: pipeline-for-loop-4-last-idx
+        metadata:
+          labels:
+            pipelines.kubeflow.org/cache_enabled: "true"
+          annotations:
+            pipelines.kubeflow.org/component_spec_digest: '{"name": "print", "outputs":
+              [], "version": "print@sha256=fab857e67527b293d70a6637e295a10c70b9b8b114e2b64cd5a17a33e3cab2a1"}'
+    - name: pipeline-for-loop-4
+      params:
+      - name: loop-item-param-2
+        value: 1,2
+      - name: loop-item-param-3
+        value: ','
+      taskSpec:
+        apiVersion: custom.tekton.dev/v1alpha1
+        kind: PipelineLoop
+        spec:
+          pipelineSpec:
+            params:
+            - name: iteration-number-5
+              type: string
+            - name: loop-item-param-2
+              type: string
+            tasks:
+            - name: print
+              params:
+              - name: iteration-number-5
+                value: $(params.iteration-number-5)
+              - name: loop-item-param-2
+                value: $(params.loop-item-param-2)
+              taskSpec:
+                steps:
+                - name: main
+                  command:
+                  - sh
+                  - -c
+                  - |
+                    set -e
+                    echo $0
+                  - 'it no $(inputs.params.iteration-number-5): $(inputs.params.loop-item-param-2)'
+                  image: alpine:3.6
+                params:
+                - name: iteration-number-5
+                  type: string
+                - name: loop-item-param-2
+                  type: string
+                metadata:
+                  labels:
+                    pipelines.kubeflow.org/cache_enabled: "true"
+                  annotations:
+                    pipelines.kubeflow.org/component_spec_digest: '{"name": "print",
+                      "outputs": [], "version": "print@sha256=fab857e67527b293d70a6637e295a10c70b9b8b114e2b64cd5a17a33e3cab2a1"}'
+          iterationNumberParam: iteration-number-5
+          iterateParam: loop-item-param-2
+          iterateParamStringSeparator: loop-item-param-3
+        metadata:
+          labels:
+            pipelines.kubeflow.org/cache_enabled: "true"
diff --git a/sdk/python/tests/compiler/testdata/last_idx_noninlined.yaml b/sdk/python/tests/compiler/testdata/last_idx_noninlined.yaml
@@ -0,0 +1,93 @@
+# Copyright 2021 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.
+
+apiVersion: tekton.dev/v1beta1
+kind: PipelineRun
+metadata:
+  name: pipeline
+  annotations:
+    tekton.dev/output_artifacts: '{}'
+    tekton.dev/input_artifacts: '{"print-2": [{"name": "pipeline-for-loop-4-last-elem",
+      "parent_task": "pipeline-for-loop-4"}, {"name": "pipeline-for-loop-4-last-idx",
+      "parent_task": "pipeline-for-loop-4"}]}'
+    tekton.dev/artifact_bucket: mlpipeline
+    tekton.dev/artifact_endpoint: minio-service.kubeflow:9000
+    tekton.dev/artifact_endpoint_scheme: http://
+    tekton.dev/artifact_items: '{"print": [], "print-2": []}'
+    sidecar.istio.io/inject: "false"
+    tekton.dev/template: ''
+    pipelines.kubeflow.org/big_data_passing_format: $(workspaces.$TASK_NAME.path)/artifacts/$ORIG_PR_NAME/$TASKRUN_NAME/$TASK_PARAM_NAME
+    pipelines.kubeflow.org/pipeline_spec: '{"inputs": [{"default": "10", "name": "param",
+      "optional": true, "type": "Integer"}], "name": "pipeline"}'
+    tekton.dev/resource_templates: '[{"apiVersion": "custom.tekton.dev/v1alpha1",
+      "kind": "PipelineLoop", "metadata": {"name": "pipeline-for-loop-4"}, "spec":
+      {"iterateParam": "loop-item-param-2", "iterateParamStringSeparator": "loop-item-param-3",
+      "iterationNumberParam": "iteration-number-5", "pipelineSpec": {"params": [{"name":
+      "iteration-number-5", "type": "string"}, {"name": "loop-item-param-2", "type":
+      "string"}], "tasks": [{"name": "print", "params": [{"name": "iteration-number-5",
+      "value": "$(params.iteration-number-5)"}, {"name": "loop-item-param-2", "value":
+      "$(params.loop-item-param-2)"}], "taskSpec": {"metadata": {"annotations": {"pipelines.kubeflow.org/component_spec_digest":
+      "{\"name\": \"print\", \"outputs\": [], \"version\": \"print@sha256=fab857e67527b293d70a6637e295a10c70b9b8b114e2b64cd5a17a33e3cab2a1\"}"},
+      "labels": {"pipelines.kubeflow.org/cache_enabled": "true"}}, "params": [{"name":
+      "iteration-number-5", "type": "string"}, {"name": "loop-item-param-2", "type":
+      "string"}], "steps": [{"command": ["sh", "-c", "set -e\necho $0\n", "it no $(inputs.params.iteration-number-5):
+      $(inputs.params.loop-item-param-2)"], "image": "alpine:3.6", "name": "main"}]}}]}}}]'
+  labels:
+    pipelines.kubeflow.org/pipelinename: ''
+    pipelines.kubeflow.org/generation: ''
+spec:
+  params:
+  - name: param
+    value: '10'
+  pipelineSpec:
+    params:
+    - name: param
+      default: '10'
+    tasks:
+    - name: print-2
+      params:
+      - name: pipeline-for-loop-4-last-elem
+        value: $(tasks.pipeline-for-loop-4.results.last-elem)
+      - name: pipeline-for-loop-4-last-idx
+        value: $(tasks.pipeline-for-loop-4.results.last-idx)
+      taskSpec:
+        steps:
+        - name: main
+          command:
+          - sh
+          - -c
+          - |
+            set -e
+            echo $0
+          - 'last it: no $(inputs.params.pipeline-for-loop-4-last-idx): $(inputs.params.pipeline-for-loop-4-last-elem)'
+          image: alpine:3.6
+        params:
+        - name: pipeline-for-loop-4-last-elem
+        - name: pipeline-for-loop-4-last-idx
+        metadata:
+          labels:
+            pipelines.kubeflow.org/cache_enabled: "true"
+          annotations:
+            pipelines.kubeflow.org/component_spec_digest: '{"name": "print", "outputs":
+              [], "version": "print@sha256=fab857e67527b293d70a6637e295a10c70b9b8b114e2b64cd5a17a33e3cab2a1"}'
+    - name: pipeline-for-loop-4
+      taskRef:
+        apiVersion: custom.tekton.dev/v1alpha1
+        kind: PipelineLoop
+        name: pipeline-for-loop-4
+      params:
+      - name: loop-item-param-2
+        value: 1,2
+      - name: loop-item-param-3
+        value: ','
diff --git a/sdk/python/tests/compiler/testdata/long_recursive_group_name.py b/sdk/python/tests/compiler/testdata/long_recursive_group_name.py
@@ -38,7 +38,7 @@ def PrintOp(name: str, msg: str):
     inputs:
     - {name: input_text, type: String, description: 'Represents an input parameter.'}
     outputs:
-    - {name: output_value, type: String, description: 'Represents an output paramter.'}
+    - {name: output_value, type: String, description: 'Represents an output parameter.'}
     implementation:
         container:
             image: alpine:3.6
diff --git a/sdk/python/tests/compiler/testdata/long_recursive_group_name.yaml b/sdk/python/tests/compiler/testdata/long_recursive_group_name.yaml
@@ -40,8 +40,8 @@ metadata:
       "condition-cel-outcome", "value": "$(tasks.condition-cel.results.outcome)"}],
       "taskSpec": {"metadata": {"annotations": {"pipelines.kubeflow.org/component_spec_digest":
       "{\"name\": \"print-iter\", \"outputs\": [{\"description\": \"Represents an
-      output paramter.\", \"name\": \"output_value\", \"type\": \"String\"}], \"version\":
-      \"print-iter@sha256=1390d83fb5d8aa6c6a41af858140c4c362850b13718dedacfc3fb0a633596313\"}"},
+      output parameter.\", \"name\": \"output_value\", \"type\": \"String\"}], \"version\":
+      \"print-iter@sha256=cefce1330b3d50a4e85b6ac51edd51f250a5cf18193353f27a2e13a9d7535555\"}"},
       "labels": {"pipelines.kubeflow.org/cache_enabled": "true"}}, "params": [{"name":
       "condition-cel-outcome", "type": "string"}], "results": [{"description": "/tmp/outputs/output_value/data",
       "name": "output-value", "type": "string"}], "steps": [{"command": ["sh", "-c",
diff --git a/sdk/python/tests/compiler/testdata/long_recursive_group_name_noninlined.yaml b/sdk/python/tests/compiler/testdata/long_recursive_group_name_noninlined.yaml
@@ -40,8 +40,8 @@ metadata:
       "condition-cel-outcome", "value": "$(tasks.condition-cel.results.outcome)"}],
       "taskSpec": {"metadata": {"annotations": {"pipelines.kubeflow.org/component_spec_digest":
       "{\"name\": \"print-iter\", \"outputs\": [{\"description\": \"Represents an
-      output paramter.\", \"name\": \"output_value\", \"type\": \"String\"}], \"version\":
-      \"print-iter@sha256=1390d83fb5d8aa6c6a41af858140c4c362850b13718dedacfc3fb0a633596313\"}"},
+      output parameter.\", \"name\": \"output_value\", \"type\": \"String\"}], \"version\":
+      \"print-iter@sha256=cefce1330b3d50a4e85b6ac51edd51f250a5cf18193353f27a2e13a9d7535555\"}"},
       "labels": {"pipelines.kubeflow.org/cache_enabled": "true"}}, "params": [{"name":
       "condition-cel-outcome", "type": "string"}], "results": [{"description": "/tmp/outputs/output_value/data",
       "name": "output-value", "type": "string"}], "steps": [{"command": ["sh", "-c",
diff --git a/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun.go b/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun.go
diff --git a/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun_test.go b/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun_test.go
