Repository: kubeflow/kfp-tekton (language: python)

Issue:
PipelineLoop controller cannot handle nested inline custom task
/kind bug

**What steps did you take and what happened:**
The current PipelineLoop controller can handle inline custom taskSpec on the first layer. However, for the inline taskSpec in the nested layer, it's still expecting the old client behavior.

```
PipelineLoop kubeflow/withitem-multiple-nesting-pipeline-0f753-withitem-multipl-hkmtz can't be Run; it has an invalid spec: missing field(s): spec.pipelineSpec.tasks[2].taskSpec.steps
```

We need to fix this and let all the nested spec to follow the new Tekton v0.25 API.


**What did you expect to happen:**


**Additional information:**
[Miscellaneous information that will assist in solving the issue.]


**Environment:**

* Python Version (use `python --version`):
* SDK Version: 
* Tekton Version (use `tkn version`):
* Kubernetes Version (use `kubectl version`):
* OS (e.g. from `/etc/os-release`):



Discussion / hints:
To reproduce this issue, deploy KFP-Tekton 0.9.0-rc0 with #635 and try the nested loop examples `withitem_multi_nested.yaml` and `withitem_multi_nested.yaml` from #633 
The error is coming from https://github.com/kubeflow/kfp-tekton/blob/master/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun.go#L187-L192

/assign @ScrapCodes 

Fix patch:
diff --git a/tekton-catalog/pipeline-loops/cmd/cli/main.go b/tekton-catalog/pipeline-loops/cmd/cli/main.go
index 68535e6fc4..79f8ba4ab0 100644
--- a/tekton-catalog/pipeline-loops/cmd/cli/main.go
+++ b/tekton-catalog/pipeline-loops/cmd/cli/main.go
@@ -25,7 +25,9 @@ import (
 	"os"
 	"strings"
 
+	"github.com/kubeflow/kfp-tekton/tekton-catalog/pipeline-loops/pkg/apis/pipelineloop"
 	pipelineloopv1alpha1 "github.com/kubeflow/kfp-tekton/tekton-catalog/pipeline-loops/pkg/apis/pipelineloop/v1alpha1"
+	"github.com/kubeflow/kfp-tekton/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun"
 	"github.com/tektoncd/pipeline/pkg/apis/pipeline/v1alpha1"
 	"github.com/tektoncd/pipeline/pkg/apis/pipeline/v1beta1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -72,13 +74,13 @@ func main() {
 	if inputFileType == "yaml" {
 		objs, err := readFile(inputFileName)
 		if err != nil {
-			fmt.Printf("Error while reading input spec: %v\n", err)
+			fmt.Printf("\nError while reading input spec: %v\n", err)
 			os.Exit(2)
 		}
 		for _, o := range objs {
 			marshalledBytes, err := o.MarshalJSON()
 			if err != nil {
-				fmt.Printf("Error while marshalling json: %v\n", err)
+				fmt.Printf("\nError while marshalling json: %v\n", err)
 				os.Exit(3)
 			}
 			err = nil
@@ -89,19 +91,19 @@ func main() {
 				err = validateRun(marshalledBytes)
 			case "Pipeline":
 				err = validatePipeline(marshalledBytes)
-			case "PipelineLoop":
+			case pipelineloop.PipelineLoopControllerName:
 				err = validatePipelineLoop(marshalledBytes)
 			case "PipelineRun":
 				err = validatePipelineRun(marshalledBytes)
 			default:
-				fmt.Printf("Warn: Unsupported kind: %s.\n", kind)
+				fmt.Printf("\nWarn: Unsupported kind: %s.\n", kind)
 			}
 			if err != nil {
 				errs = append(errs, err.Error())
 			}
 		}
 		if len(errs) > 0 {
-			fmt.Printf("Validation errors: %s\n", strings.Join(errs, "\n"))
+			fmt.Printf("\nValidation errors: %s\n", strings.Join(errs, "\n"))
 			os.Exit(100)
 		} else {
 			fmt.Printf("\nCongratulations, all checks passed !!\n")
@@ -125,7 +127,7 @@ func validatePipelineSpec(p *v1beta1.PipelineSpec, name string) error {
 	// Here we only need to validate those embedded spec, whose kind is pipelineLoop.
 	if p.Tasks != nil {
 		for _, task := range p.Tasks {
-			if task.TaskSpec != nil && task.TaskSpec.Kind == "PipelineLoop" {
+			if task.TaskSpec != nil && task.TaskSpec.Kind == pipelineloop.PipelineLoopControllerName {
 				err := validatePipelineLoopEmbedded(task.TaskSpec.Spec.Raw)
 				if err != nil {
 					errs = append(errs, err.Error())
@@ -149,7 +151,7 @@ func validateRun(bytes []byte) error {
 	// We do not need to validate Run because it is validated by tekton admission webhook
 	// And r.Spec.Ref is also validated by tekton.
 	// Here we only need to validate the embedded spec. i.e. r.Spec.Spec
-	if r.Spec.Spec != nil && r.Spec.Spec.Kind == "PipelineLoop" {
+	if r.Spec.Spec != nil && r.Spec.Spec.Kind == pipelineloop.PipelineLoopControllerName {
 		if err := validatePipelineLoopEmbedded(r.Spec.Spec.Spec.Raw); err != nil {
 			return fmt.Errorf("Found validation errors in Run: %s \n %s", r.Name, err.Error())
 		}
@@ -172,8 +174,8 @@ func validatePipelineLoopEmbedded(bytes []byte) error {
 		return err
 	}
 	r1 := map[string]interface{}{
-		"kind":       "PipelineLoop",
-		"apiVersion": "custom.tekton.dev/v1alpha1",
+		"kind":       pipelineloop.PipelineLoopControllerName,
+		"apiVersion": pipelineloopv1alpha1.SchemeGroupVersion.String(),
 		"metadata":   metav1.ObjectMeta{Name: "embedded"},
 		"spec":       embeddedSpec,
 	}
@@ -190,14 +192,30 @@ func validatePipelineLoop(bytes []byte) error {
 	if err := json.Unmarshal(bytes, &pipelineLoop); err != nil {
 		return err
 	}
-	ctx := context.TODO()
+	ctx := context.Background()
+	ctx = pipelinelooprun.EnableCustomTaskFeatureFlag(ctx)
 	pipelineLoop.SetDefaults(ctx)
 	if err := pipelineLoop.Validate(ctx); err != nil {
 		return fmt.Errorf("PipelineLoop name:%s\n %s", pipelineLoop.Name, err.Error())
 	}
+	if err, name := validateNestedPipelineLoop(pipelineLoop); err != nil {
+		return fmt.Errorf("Nested PipelineLoop name:%s\n %s", name, err.Error())
+	}
 	return nil
 }
 
+func validateNestedPipelineLoop(pl pipelineloopv1alpha1.PipelineLoop) (error, string) {
+	for _, task := range pl.Spec.PipelineSpec.Tasks {
+		if task.TaskSpec != nil && task.TaskSpec.Kind == pipelineloop.PipelineLoopControllerName {
+			err := validatePipelineLoopEmbedded(task.TaskSpec.Spec.Raw)
+			if err != nil {
+				return err, task.Name
+			}
+		}
+	}
+	return nil, ""
+}
+
 // readFile parses a single file.
 func readFile(pathname string) ([]unstructured.Unstructured, error) {
 	file, err := os.Open(pathname)
diff --git a/tekton-catalog/pipeline-loops/examples/pipelinespec-with-nested-loop-embedded.yaml b/tekton-catalog/pipeline-loops/examples/pipelinespec-with-nested-loop-embedded.yaml
new file mode 100644
index 0000000000..210ebe9f73
--- /dev/null
+++ b/tekton-catalog/pipeline-loops/examples/pipelinespec-with-nested-loop-embedded.yaml
@@ -0,0 +1,96 @@
+apiVersion: tekton.dev/v1beta1
+kind: PipelineRun
+metadata:
+  name: pr-loop-example
+  labels:
+    mylooplabels: mylooplabels
+spec:
+  pipelineSpec:
+    tasks:
+      - name: first-task
+        taskSpec:
+          steps:
+            - name: echo
+              image: ubuntu
+              imagePullPolicy: IfNotPresent
+              script: |
+                #!/usr/bin/env bash
+                echo "I am the first task before the loop task"
+      - name: loop-task
+        runAfter:
+          - first-task
+        params:
+          - name: message
+            value:
+              - I am the first one
+              - I am the second one
+              - I am the third one
+        taskSpec:
+          apiVersion: custom.tekton.dev/v1alpha1
+          kind: PipelineLoop
+          spec:
+            iterateParam: message
+            pipelineSpec:
+              params:
+                - name: message
+                  type: string
+              tasks:
+                - name: echo-loop-task
+                  params:
+                    - name: message
+                      value: $(params.message)
+                  taskSpec:
+                    params:
+                      - name: message
+                        type: string
+                    steps:
+                      - name: echo
+                        image: ubuntu
+                        imagePullPolicy: IfNotPresent
+                        script: |
+                          #!/usr/bin/env bash
+                          echo "$(params.message)"
+                - name: loop-task-inner
+                  runAfter:
+                    - echo-loop-task
+                  params:
+                    - name: message
+                      value:
+                        - I am the inner first one
+                        - I am the inner second one
+                  taskSpec:
+                    apiVersion: custom.tekton.dev/v1alpha1
+                    kind: PipelineLoop
+                    spec:
+                      iterateParam: message
+                      pipelineSpec:
+                        params:
+                          - name: message
+                            type: string
+                        tasks:
+                          - name: echo-loop-task
+                            params:
+                              - name: message
+                                value: $(params.message)
+                            taskSpec:
+                              params:
+                                - name: message
+                                  type: string
+                              steps:
+                                - name: echo
+                                  image: ubuntu
+                                  imagePullPolicy: IfNotPresent
+                                  script: |
+                                    #!/usr/bin/env bash
+                                    echo "$(params.message)"
+      - name: last-task
+        runAfter:
+          - loop-task
+        taskSpec:
+          steps:
+            - name: echo
+              image: ubuntu
+              imagePullPolicy: IfNotPresent
+              script: |
+                #!/usr/bin/env bash
+                echo "I am the last task after the loop task"
diff --git a/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun.go b/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun.go
index 3b96f17e78..2c6c366ce4 100644
--- a/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun.go
+++ b/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun.go
@@ -30,6 +30,7 @@ import (
 	pipelineloopv1alpha1 "github.com/kubeflow/kfp-tekton/tekton-catalog/pipeline-loops/pkg/apis/pipelineloop/v1alpha1"
 	pipelineloopclientset "github.com/kubeflow/kfp-tekton/tekton-catalog/pipeline-loops/pkg/client/clientset/versioned"
 	listerspipelineloop "github.com/kubeflow/kfp-tekton/tekton-catalog/pipeline-loops/pkg/client/listers/pipelineloop/v1alpha1"
+	"github.com/tektoncd/pipeline/pkg/apis/config"
 	"github.com/tektoncd/pipeline/pkg/apis/pipeline"
 	"github.com/tektoncd/pipeline/pkg/apis/pipeline/v1alpha1"
 	"github.com/tektoncd/pipeline/pkg/apis/pipeline/v1beta1"
@@ -167,7 +168,24 @@ func (c *Reconciler) ReconcileKind(ctx context.Context, run *v1alpha1.Run) pkgre
 	return merr
 }
 
+func EnableCustomTaskFeatureFlag(ctx context.Context) context.Context {
+	defaults, _ := config.NewDefaultsFromMap(map[string]string{})
+	featureFlags, _ := config.NewFeatureFlagsFromMap(map[string]string{
+		"enable-custom-tasks": "true",
+	})
+	artifactBucket, _ := config.NewArtifactBucketFromMap(map[string]string{})
+	artifactPVC, _ := config.NewArtifactPVCFromMap(map[string]string{})
+	c := &config.Config{
+		Defaults:       defaults,
+		FeatureFlags:   featureFlags,
+		ArtifactBucket: artifactBucket,
+		ArtifactPVC:    artifactPVC,
+	}
+	return config.ToContext(ctx, c)
+}
+
 func (c *Reconciler) reconcile(ctx context.Context, run *v1alpha1.Run, status *pipelineloopv1alpha1.PipelineLoopRunStatus) error {
+	ctx = EnableCustomTaskFeatureFlag(ctx)
 	logger := logging.FromContext(ctx)
 
 	// Get the PipelineLoop referenced by the Run


Test patch:
diff --git a/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun_test.go b/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun_test.go
index 5b3d42de83..65171d478b 100644
--- a/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun_test.go
+++ b/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun_test.go
@@ -18,6 +18,7 @@ package pipelinelooprun
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"strings"
 	"testing"
@@ -37,6 +38,7 @@ import (
 	"github.com/tektoncd/pipeline/test/names"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	ktesting "k8s.io/client-go/testing"
 	"k8s.io/client-go/tools/record"
@@ -322,6 +324,64 @@ var paraPipeline = &v1beta1.Pipeline{
 	},
 }
 
+func getInnerLoopByte(pipeline *v1beta1.Pipeline) []byte {
+	innerLoop, err := json.Marshal(pipeline)
+	if err != nil {
+		fmt.Println(fmt.Errorf("error while marshalling pipelineLoop %s", err.Error()).Error())
+	}
+	return innerLoop
+}
+
+var nestedPipeline = &v1beta1.Pipeline{
+	ObjectMeta: metav1.ObjectMeta{Name: "nestedPipeline", Namespace: "foo"},
+	Spec: v1beta1.PipelineSpec{
+		Params: []v1beta1.ParamSpec{{
+			Name: "iteration",
+			Type: v1beta1.ParamTypeString,
+		}, {
+			Name: "additional-parameter",
+			Type: v1beta1.ParamTypeString,
+		}},
+		Tasks: []v1beta1.PipelineTask{{
+			Name: "mytask",
+			TaskSpec: &v1beta1.EmbeddedTask{
+				TypeMeta: runtime.TypeMeta{
+					APIVersion: pipelineloopv1alpha1.SchemeGroupVersion.String(),
+					Kind:       pipelineloop.PipelineLoopControllerName,
+				},
+				Spec: runtime.RawExtension{
+					Raw: getInnerLoopByte(paraPipeline),
+				},
+			},
+		}},
+	},
+}
+
+var nestedPipeline2 = &v1beta1.Pipeline{
+	ObjectMeta: metav1.ObjectMeta{Name: "nestedPipeline2", Namespace: "foo"},
+	Spec: v1beta1.PipelineSpec{
+		Params: []v1beta1.ParamSpec{{
+			Name: "iteration",
+			Type: v1beta1.ParamTypeString,
+		}, {
+			Name: "additional-parameter",
+			Type: v1beta1.ParamTypeString,
+		}},
+		Tasks: []v1beta1.PipelineTask{{
+			Name: "mytask",
+			TaskSpec: &v1beta1.EmbeddedTask{
+				TypeMeta: runtime.TypeMeta{
+					APIVersion: pipelineloopv1alpha1.SchemeGroupVersion.String(),
+					Kind:       pipelineloop.PipelineLoopControllerName,
+				},
+				Spec: runtime.RawExtension{
+					Raw: getInnerLoopByte(nestedPipeline),
+				},
+			},
+		}},
+	},
+}
+
 var paraPipelineLoop = &pipelineloopv1alpha1.PipelineLoop{
 	ObjectMeta: metav1.ObjectMeta{Name: "para-pipelineloop", Namespace: "foo"},
 	Spec: pipelineloopv1alpha1.PipelineLoopSpec{
@@ -339,6 +399,14 @@ var nPipelineLoop = &pipelineloopv1alpha1.PipelineLoop{
 	},
 }
 
+var nestedPipelineLoop = &pipelineloopv1alpha1.PipelineLoop{
+	ObjectMeta: metav1.ObjectMeta{Name: "nested-pipelineloop", Namespace: "foo"},
+	Spec: pipelineloopv1alpha1.PipelineLoopSpec{
+		PipelineSpec:   &nestedPipeline2.Spec,
+		IterateNumeric: "iteration",
+	},
+}
+
 var aPipelineLoopWithInlineTask = &pipelineloopv1alpha1.PipelineLoop{
 	ObjectMeta: metav1.ObjectMeta{Name: "a-pipelineloop-with-inline-task", Namespace: "foo"},
 	Spec: pipelineloopv1alpha1.PipelineLoopSpec{
@@ -397,6 +465,37 @@ var runPipelineLoop = &v1alpha1.Run{
 	},
 }
 
+var nestedRunPipelineLoop = &v1alpha1.Run{
+	ObjectMeta: metav1.ObjectMeta{
+		Name:      "run-pipelineloop",
+		Namespace: "foo",
+		Labels: map[string]string{
+			"myTestLabel": "myTestLabelValue",
+		},
+		Annotations: map[string]string{
+			"myTestAnnotation": "myTestAnnotationValue",
+		},
+	},
+	Spec: v1alpha1.RunSpec{
+		Params: []v1beta1.Param{{
+			Name:  "current-item",
+			Value: v1beta1.ArrayOrString{Type: v1beta1.ParamTypeArray, ArrayVal: []string{"item1", "item2"}},
+		}, {
+			Name:  "additional-parameter",
+			Value: v1beta1.ArrayOrString{Type: v1beta1.ParamTypeString, StringVal: "stuff"},
+		}},
+		Spec: &v1alpha1.EmbeddedRunSpec{
+			TypeMeta: runtime.TypeMeta{
+				APIVersion: pipelineloopv1alpha1.SchemeGroupVersion.String(),
+				Kind:       pipelineloop.PipelineLoopControllerName,
+			},
+			Spec: runtime.RawExtension{
+				Raw: getInnerLoopByte(nestedPipeline2),
+			},
+		},
+	},
+}
+
 var paraRunPipelineLoop = &v1alpha1.Run{
 	ObjectMeta: metav1.ObjectMeta{
 		Name:      "run-pipelineloop",
@@ -936,6 +1035,39 @@ var expectedPipelineRunWithInlineTaskIteration1 = &v1beta1.PipelineRun{
 		Timeout: &metav1.Duration{Duration: 5 * time.Minute},
 	},
 }
+var expectedNestedPipelineRun = &v1beta1.PipelineRun{
+	ObjectMeta: metav1.ObjectMeta{
+		Name:      "run-pipelineloop-00001-9l9zj",
+		Namespace: "foo",
+		OwnerReferences: []metav1.OwnerReference{{
+			APIVersion:         "tekton.dev/v1alpha1",
+			Kind:               "Run",
+			Name:               "run-pipelineloop",
+			Controller:         &trueB,
+			BlockOwnerDeletion: &trueB,
+		}},
+		Labels: map[string]string{
+			"custom.tekton.dev/parentPipelineRun":     "",
+			"custom.tekton.dev/pipelineLoop":          "nested-pipelineloop",
+			"tekton.dev/run":                          "run-pipelineloop",
+			"custom.tekton.dev/pipelineLoopIteration": "1",
+			"myTestLabel":                             "myTestLabelValue",
+		},
+		Annotations: map[string]string{
+			"myTestAnnotation": "myTestAnnotationValue",
+		},
+	},
+	Spec: v1beta1.PipelineRunSpec{
+		PipelineSpec: &nestedPipeline2.Spec,
+		Params: []v1beta1.Param{{
+			Name:  "current-item",
+			Value: v1beta1.ArrayOrString{Type: v1beta1.ParamTypeString, StringVal: "item1"},
+		}, {
+			Name:  "additional-parameter",
+			Value: v1beta1.ArrayOrString{Type: v1beta1.ParamTypeString, StringVal: "stuff"},
+		}},
+	},
+}
 
 func TestReconcilePipelineLoopRun(t *testing.T) {
 
@@ -1078,6 +1210,16 @@ func TestReconcilePipelineLoopRun(t *testing.T) {
 		expectedReason:       pipelineloopv1alpha1.PipelineLoopRunReasonRunning,
 		expectedPipelineruns: []*v1beta1.PipelineRun{expectedParaPipelineRun, expectedParaPipelineRun1},
 		expectedEvents:       []string{"Normal Started", "Normal Running Iterations completed: 0"},
+	}, {
+		name:                 "Reconcile a new run with a nested pipelineloop",
+		pipeline:             nestedPipeline2,
+		pipelineloop:         nestedPipelineLoop,
+		run:                  nestedRunPipelineLoop,
+		pipelineruns:         []*v1beta1.PipelineRun{},
+		expectedStatus:       corev1.ConditionUnknown,
+		expectedReason:       pipelineloopv1alpha1.PipelineLoopRunReasonRunning,
+		expectedPipelineruns: []*v1beta1.PipelineRun{expectedNestedPipelineRun},
+		expectedEvents:       []string{"Normal Started", "Normal Running Iterations completed: 0"},
 	}}
 
 	for _, tc := range testcases {