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

Issue:
Support parallelism for pipeline loop
/kind feature
Introduce the parallism parameter to control the concurrency

**Description:**
[A clear and concise description of what your proposal. What problem does it solve?]
Before we have already implemented loop, but all the pipeline can only run in sequential. We need support parallism parameter so that multiple pipelines can be started at once with one loop arg for each one.


Discussion / hints:
/assign
I will do it after the any-sequencer enhancement.
Spent some time on the any-sequencer and loop issues, will start working on this.

Fix patch:
diff --git a/tekton-catalog/pipeline-loops/README.md b/tekton-catalog/pipeline-loops/README.md
index 5162cd9013..2832fb2958 100644
--- a/tekton-catalog/pipeline-loops/README.md
+++ b/tekton-catalog/pipeline-loops/README.md
@@ -31,6 +31,8 @@
 
 - Install pipelineloop controller `kubectl apply -f config/`
 
+# Parallelism
+`Parallelism` is to define the number of pipelineruns can be created at the same time. It must be eqaul to or bigger than 1. If it's not set then the default value will be 1. If it's bigger than the total iterations in the loop then the number will be total iterations.
 
 # Verification
 - check controller and the webhook. `kubectl get po -n tekton-pipelines`
diff --git a/tekton-catalog/pipeline-loops/examples/loop-example-with-parallelism.yaml b/tekton-catalog/pipeline-loops/examples/loop-example-with-parallelism.yaml
new file mode 100644
index 0000000000..8351fd597b
--- /dev/null
+++ b/tekton-catalog/pipeline-loops/examples/loop-example-with-parallelism.yaml
@@ -0,0 +1,72 @@
+apiVersion: custom.tekton.dev/v1alpha1
+kind: PipelineLoop
+metadata:
+  name: pipelineloop
+spec:
+  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)"
+  iterateParam: message
+  parallelism: 2
+---
+apiVersion: tekton.dev/v1beta1
+kind: PipelineRun
+metadata:
+  annotations:
+    tekton.dev/example-loop-pipeline: '{"spec":{"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\necho \"$(params.message)\"\n"}]}}]},"iterateParam":"message"}}'
+  name: para-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
+        taskRef:
+          apiVersion: custom.tekton.dev/v1alpha1
+          kind: PipelineLoop
+          name: pipelineloop
+      - 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/apis/pipelineloop/v1alpha1/pipelineloop_defaults.go b/tekton-catalog/pipeline-loops/pkg/apis/pipelineloop/v1alpha1/pipelineloop_defaults.go
index b730ff2dd1..9cbe32e539 100644
--- a/tekton-catalog/pipeline-loops/pkg/apis/pipelineloop/v1alpha1/pipelineloop_defaults.go
+++ b/tekton-catalog/pipeline-loops/pkg/apis/pipelineloop/v1alpha1/pipelineloop_defaults.go
@@ -18,6 +18,8 @@ package v1alpha1
 
 import (
 	"context"
+	"os"
+	"strconv"
 
 	"knative.dev/pkg/apis"
 )
@@ -31,4 +33,19 @@ func (tl *PipelineLoop) SetDefaults(ctx context.Context) {
 
 // SetDefaults set any defaults for the PipelineLoop spec
 func (tls *PipelineLoopSpec) SetDefaults(ctx context.Context) {
+	if tls.Parallelism == 0 {
+		parallelism := os.Getenv("LOOP_PARALLELISM")
+		if parallelism == "" {
+			tls.Parallelism = 1
+			return
+		}
+		i, err := strconv.Atoi(parallelism)
+		if err != nil {
+			// fall back to default 1
+			tls.Parallelism = 1
+			return
+		} else {
+			tls.Parallelism = i
+		}
+	}
 }
diff --git a/tekton-catalog/pipeline-loops/pkg/apis/pipelineloop/v1alpha1/pipelineloop_types.go b/tekton-catalog/pipeline-loops/pkg/apis/pipelineloop/v1alpha1/pipelineloop_types.go
index 4d81e2692e..fd955a1ea1 100644
--- a/tekton-catalog/pipeline-loops/pkg/apis/pipelineloop/v1alpha1/pipelineloop_types.go
+++ b/tekton-catalog/pipeline-loops/pkg/apis/pipelineloop/v1alpha1/pipelineloop_types.go
@@ -57,6 +57,10 @@ type PipelineLoopSpec struct {
 	// +optional
 	Timeout *metav1.Duration `json:"timeout,omitempty"`
 
+	// Parallelism represents how many pipelines can be triggered simutaniously by the loop.
+	// +optional
+	Parallelism int `json:"parallelism,omitempty"`
+
 	// Retries represents how many times a task should be retried in case of task failure.
 	// +optional
 	Retries int `json:"retries,omitempty"`
@@ -113,12 +117,16 @@ func (t PipelineLoopRunReason) String() string {
 type PipelineLoopRunStatus struct {
 	// PipelineLoopSpec contains the exact spec used to instantiate the Run
 	PipelineLoopSpec *PipelineLoopSpec `json:"pipelineLoopSpec,omitempty"`
+	// current running pipelinerun number
+	// +optional
+	CurrentRunning int `json:"currentRunning,omitempty"`
 	// map of PipelineLoopPipelineRunStatus with the PipelineRun name as the key
 	// +optional
 	PipelineRuns map[string]*PipelineLoopPipelineRunStatus `json:"pipelineRuns,omitempty"`
 }
 
-// PipelineLoopPipelineRunStatus contains the iteration number for a PipelineRun and the PipelineRun's Status
+// PipelineLoopPipelineRunStatus contains the iteration number for a PipelineRun,
+// current running pipeline number, and the PipelineRun's Status
 type PipelineLoopPipelineRunStatus struct {
 	// iteration number
 	Iteration int `json:"iteration,omitempty"`
diff --git a/tekton-catalog/pipeline-loops/pkg/apis/pipelineloop/v1alpha1/pipelineloop_validation.go b/tekton-catalog/pipeline-loops/pkg/apis/pipelineloop/v1alpha1/pipelineloop_validation.go
index ccdc0b6746..4db3132a37 100644
--- a/tekton-catalog/pipeline-loops/pkg/apis/pipelineloop/v1alpha1/pipelineloop_validation.go
+++ b/tekton-catalog/pipeline-loops/pkg/apis/pipelineloop/v1alpha1/pipelineloop_validation.go
@@ -37,6 +37,9 @@ func (tl *PipelineLoop) Validate(ctx context.Context) *apis.FieldError {
 
 // Validate PipelineLoopSpec
 func (tls *PipelineLoopSpec) Validate(ctx context.Context) *apis.FieldError {
+	if tls.Parallelism < 1 {
+		return apis.ErrInvalidValue(tls.Parallelism, "spec.Parallelism")
+	}
 	// Validate Task reference or inline task spec.
 	if err := validateTask(ctx, tls); err != nil {
 		return err
diff --git a/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun.go b/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun.go
index 9b2daca720..2c363a75ef 100644
--- a/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun.go
+++ b/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun.go
@@ -184,83 +184,64 @@ func (c *Reconciler) reconcile(ctx context.Context, run *v1alpha1.Run, status *p
 	}
 
 	// Update status of PipelineRuns.  Return the PipelineRun representing the highest loop iteration.
-	highestIteration, highestIterationPr, err := c.updatePipelineRunStatus(logger, run, status)
+	highestIteration, currentRunningPrs, failedPrs, err := c.updatePipelineRunStatus(logger, run, status)
 	if err != nil {
 		return fmt.Errorf("error updating PipelineRun status for Run %s/%s: %w", run.Namespace, run.Name, err)
 	}
 
-	// Check the status of the PipelineRun for the highest iteration.
-	if highestIterationPr != nil {
-		// If it's not done, wait for it to finish or cancel it if the run is cancelled.
-		if !highestIterationPr.IsDone() {
-			if run.IsCancelled() {
-				logger.Infof("Run %s/%s is cancelled.  Cancelling PipelineRun %s.", run.Namespace, run.Name, highestIterationPr.Name)
-				b, err := getCancelPatch()
-				if err != nil {
-					return fmt.Errorf("Failed to make patch to cancel PipelineRun %s: %v", highestIterationPr.Name, err)
-				}
-				if _, err := c.pipelineClientSet.TektonV1beta1().PipelineRuns(run.Namespace).Patch(ctx, highestIterationPr.Name, types.JSONPatchType, b, metav1.PatchOptions{}); err != nil {
-					run.Status.MarkRunRunning(pipelineloopv1alpha1.PipelineLoopRunReasonCouldntCancel.String(),
-						"Failed to patch PipelineRun `%s` with cancellation: %v", highestIterationPr.Name, err)
-					return nil
-				}
-				// Update status. It is still running until the PipelineRun is actually cancelled.
-				run.Status.MarkRunRunning(pipelineloopv1alpha1.PipelineLoopRunReasonRunning.String(),
-					"Cancelling PipelineRun %s", highestIterationPr.Name)
+	// Run is cancelled, just cancel all the running instance and return
+	if run.IsCancelled() {
+		if len(failedPrs) > 0 {
+			run.Status.MarkRunFailed(pipelineloopv1alpha1.PipelineLoopRunReasonCancelled.String(),
+				"Run %s/%s was cancelled",
+				run.Namespace, run.Name)
+		}
+		for _, currentRunningPr := range currentRunningPrs {
+			logger.Infof("Run %s/%s is cancelled.  Cancelling PipelineRun %s.", run.Namespace, run.Name, currentRunningPr.Name)
+			b, err := getCancelPatch()
+			if err != nil {
+				return fmt.Errorf("Failed to make patch to cancel PipelineRun %s: %v", currentRunningPr.Name, err)
+			}
+			if _, err := c.pipelineClientSet.TektonV1beta1().PipelineRuns(run.Namespace).Patch(ctx, currentRunningPr.Name, types.JSONPatchType, b, metav1.PatchOptions{}); err != nil {
+				run.Status.MarkRunRunning(pipelineloopv1alpha1.PipelineLoopRunReasonCouldntCancel.String(),
+					"Failed to patch PipelineRun `%s` with cancellation: %v", currentRunningPr.Name, err)
 				return nil
 			}
-			run.Status.MarkRunRunning(pipelineloopv1alpha1.PipelineLoopRunReasonRunning.String(),
-				"Iterations completed: %d", highestIteration-1)
-			return nil
 		}
+		return nil
+	}
 
-		if !highestIterationPr.Status.GetCondition(apis.ConditionSucceeded).IsTrue() {
-			if run.IsCancelled() {
-				run.Status.MarkRunFailed(pipelineloopv1alpha1.PipelineLoopRunReasonCancelled.String(),
-					"Run %s/%s was cancelled",
-					run.Namespace, run.Name)
-			} else {
-				run.Status.MarkRunFailed(pipelineloopv1alpha1.PipelineLoopRunReasonFailed.String(),
-					"PipelineRun %s has failed", highestIterationPr.Name)
-			}
-			return nil
-		}
+	// Run may be marked succeeded already by updatePipelineRunStatus
+	if run.IsSuccessful() {
+		return nil
+	}
 
-		// Mark run successful if the condition are met.
-		// if the last loop task is skipped, but the highestIterationPr successed. Mark run success.
-		// lastLoopTask := highestIterationPr.ObjectMeta.Annotations["last-loop-task"]
-		lastLoopTask := ""
-		for key, val := range run.ObjectMeta.Labels {
-			if key == "last-loop-task" {
-				lastLoopTask = val
-			}
-		}
-		if lastLoopTask != "" {
-			skippedTaskList := highestIterationPr.Status.SkippedTasks
-			for _, task := range skippedTaskList {
-				if task.Name == lastLoopTask {
-					// Mark run successful and stop the loop pipelinerun
-					run.Status.MarkRunSucceeded(pipelineloopv1alpha1.PipelineLoopRunReasonSucceeded.String(),
-						"PipelineRuns completed successfully with the conditions are met")
-					run.Status.Results = []runv1alpha1.RunResult{{
-						Name:  "condition",
-						Value: "pass",
-					}}
-					return nil
-				}
-			}
-		}
+	// Check the status of the PipelineRun for the highest iteration.
+	for _, failedPr := range failedPrs {
+		run.Status.MarkRunFailed(pipelineloopv1alpha1.PipelineLoopRunReasonFailed.String(),
+			"PipelineRun %s has failed", failedPr.Name)
+		return nil
 	}
 
+	// Mark run status Running
+	run.Status.MarkRunRunning(pipelineloopv1alpha1.PipelineLoopRunReasonRunning.String(),
+		"Iterations completed: %d", highestIteration-len(currentRunningPrs))
+
 	// Move on to the next iteration (or the first iteration if there was no PipelineRun).
 	// Check if the Run is done.
 	nextIteration := highestIteration + 1
 	if nextIteration > totalIterations {
+		// Still running which we already marked, just waiting
+		if len(currentRunningPrs) > 0 {
+			logger.Infof("Already started all pipelineruns for the loop, totally %d pipelineruns, waiting for complete.", totalIterations)
+			return nil
+		}
+		// All task finished
 		run.Status.MarkRunSucceeded(pipelineloopv1alpha1.PipelineLoopRunReasonSucceeded.String(),
 			"All PipelineRuns completed successfully")
 		run.Status.Results = []runv1alpha1.RunResult{{
 			Name:  "condition",
-			Value: "fail",
+			Value: "succeeded",
 		}}
 		return nil
 	}
@@ -271,20 +252,35 @@ func (c *Reconciler) reconcile(ctx context.Context, run *v1alpha1.Run, status *p
 			run.Namespace, run.Name)
 		return nil
 	}
-
-	// Create a PipelineRun to run this iteration.
-	pr, err := c.createPipelineRun(ctx, logger, pipelineLoopSpec, run, nextIteration)
-	if err != nil {
-		return fmt.Errorf("error creating PipelineRun from Run %s: %w", run.Name, err)
+	actualParallelism := 1
+	// if Parallelism is bigger then totalIterations means there's no limit
+	if status.PipelineLoopSpec.Parallelism > totalIterations {
+		actualParallelism = totalIterations
+	} else if status.PipelineLoopSpec.Parallelism > 0 {
+		actualParallelism = status.PipelineLoopSpec.Parallelism
 	}
-
-	status.PipelineRuns[pr.Name] = &pipelineloopv1alpha1.PipelineLoopPipelineRunStatus{
-		Iteration: nextIteration,
-		Status:    &pr.Status,
+	if len(currentRunningPrs) >= actualParallelism {
+		logger.Infof("Currently %d pipelinerun started, meet parallelism %d, waiting...", len(currentRunningPrs), actualParallelism)
+		return nil
 	}
 
-	run.Status.MarkRunRunning(pipelineloopv1alpha1.PipelineLoopRunReasonRunning.String(),
-		"Iterations completed: %d", highestIteration)
+	// Create PipelineRun to run this iteration based on parallelism
+	for i := 0; i < actualParallelism-len(currentRunningPrs); i++ {
+		pr, err := c.createPipelineRun(ctx, logger, pipelineLoopSpec, run, nextIteration)
+		if err != nil {
+			return fmt.Errorf("error creating PipelineRun from Run %s: %w", run.Name, err)
+		}
+
+		status.PipelineRuns[pr.Name] = &pipelineloopv1alpha1.PipelineLoopPipelineRunStatus{
+			Iteration: nextIteration,
+			Status:    &pr.Status,
+		}
+		nextIteration++
+		if nextIteration > totalIterations {
+			logger.Infof("Started all pipelineruns for the loop, totally %d pipelineruns.", totalIterations)
+			return nil
+		}
+	}
 
 	return nil
 }
@@ -387,20 +383,22 @@ func (c *Reconciler) updateLabelsAndAnnotations(ctx context.Context, run *v1alph
 	return nil
 }
 
-func (c *Reconciler) updatePipelineRunStatus(logger *zap.SugaredLogger, run *v1alpha1.Run, status *pipelineloopv1alpha1.PipelineLoopRunStatus) (int, *v1beta1.PipelineRun, error) {
+func (c *Reconciler) updatePipelineRunStatus(logger *zap.SugaredLogger, run *v1alpha1.Run, status *pipelineloopv1alpha1.PipelineLoopRunStatus) (int, []*v1beta1.PipelineRun, []*v1beta1.PipelineRun, error) {
 	highestIteration := 0
-	var highestIterationPr *v1beta1.PipelineRun = nil
+	var currentRunningPrs []*v1beta1.PipelineRun
+	var failedPrs []*v1beta1.PipelineRun
 	if status.PipelineRuns == nil {
 		status.PipelineRuns = make(map[string]*pipelineloopv1alpha1.PipelineLoopPipelineRunStatus)
 	}
 	pipelineRunLabels := getPipelineRunLabels(run, "")
 	pipelineRuns, err := c.pipelineRunLister.PipelineRuns(run.Namespace).List(labels.SelectorFromSet(pipelineRunLabels))
 	if err != nil {
-		return 0, nil, fmt.Errorf("could not list PipelineRuns %#v", err)
+		return 0, nil, nil, fmt.Errorf("could not list PipelineRuns %#v", err)
 	}
 	if pipelineRuns == nil || len(pipelineRuns) == 0 {
-		return 0, nil, nil
+		return 0, nil, nil, nil
 	}
+	status.CurrentRunning = 0
 	for _, pr := range pipelineRuns {
 		lbls := pr.GetLabels()
 		iterationStr := lbls[pipelineloop.GroupName+pipelineLoopIterationLabelKey]
@@ -409,18 +407,50 @@ func (c *Reconciler) updatePipelineRunStatus(logger *zap.SugaredLogger, run *v1a
 			run.Status.MarkRunFailed(pipelineloopv1alpha1.PipelineLoopRunReasonFailedValidation.String(),
 				"Error converting iteration number in PipelineRun %s:  %#v", pr.Name, err)
 			logger.Errorf("Error converting iteration number in PipelineRun %s:  %#v", pr.Name, err)
-			return 0, nil, nil
+			return 0, nil, nil, nil
+		}
+		// when we just create pr in a forloop, the started time may be empty
+		if !pr.IsDone() {
+			status.CurrentRunning++
+			currentRunningPrs = append(currentRunningPrs, pr)
 		}
+		if pr.IsDone() && !pr.Status.GetCondition(apis.ConditionSucceeded).IsTrue() {
+			failedPrs = append(failedPrs, pr)
+		}
+
+		// Mark run successful if the condition are met.
+		// if the last loop task is skipped, but the highestIterationPr successed. Mark run success.
+		// lastLoopTask := highestIterationPr.ObjectMeta.Annotations["last-loop-task"]
+		lastLoopTask := ""
+		for key, val := range run.ObjectMeta.Labels {
+			if key == "last-loop-task" {
+				lastLoopTask = val
+			}
+		}
+		if lastLoopTask != "" {
+			skippedTaskList := pr.Status.SkippedTasks
+			for _, task := range skippedTaskList {
+				if task.Name == lastLoopTask {
+					// Mark run successful and stop the loop pipelinerun
+					run.Status.MarkRunSucceeded(pipelineloopv1alpha1.PipelineLoopRunReasonSucceeded.String(),
+						"PipelineRuns completed successfully with the conditions are met")
+					run.Status.Results = []runv1alpha1.RunResult{{
+						Name:  "condition",
+						Value: "pass",
+					}}
+				}
+			}
+		}
+
 		status.PipelineRuns[pr.Name] = &pipelineloopv1alpha1.PipelineLoopPipelineRunStatus{
 			Iteration: iteration,
 			Status:    &pr.Status,
 		}
 		if iteration > highestIteration {
 			highestIteration = iteration
-			highestIterationPr = pr
 		}
 	}
-	return highestIteration, highestIterationPr, nil
+	return highestIteration, currentRunningPrs, failedPrs, nil
 }
 
 func getCancelPatch() ([]byte, error) {


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 bdc5c7c1cf..18f128dd73 100644
--- a/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun_test.go
+++ b/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun_test.go
@@ -151,17 +151,18 @@ func getPipelineLoopController(t *testing.T, d test.Data, pipelineloops []*pipel
 	}, cancel
 }
 
-func getCreatedPipelinerun(t *testing.T, clients test.Clients) *v1beta1.PipelineRun {
+func getCreatedPipelinerun(t *testing.T, clients test.Clients) []*v1beta1.PipelineRun {
 	t.Log("actions", clients.Pipeline.Actions())
+	var createdPr []*v1beta1.PipelineRun
 	for _, a := range clients.Pipeline.Actions() {
 		if a.GetVerb() == "create" {
 			obj := a.(ktesting.CreateAction).GetObject()
 			if pr, ok := obj.(*v1beta1.PipelineRun); ok {
-				return pr
+				createdPr = append(createdPr, pr)
 			}
 		}
 	}
-	return nil
+	return createdPr
 }
 
 func checkEvents(fr *record.FakeRecorder, testName string, wantEvents []string) error {
@@ -298,6 +299,38 @@ var nPipeline = &v1beta1.Pipeline{
 	},
 }
 
+var paraPipeline = &v1beta1.Pipeline{
+	ObjectMeta: metav1.ObjectMeta{Name: "para-pipeline", Namespace: "foo"},
+	Spec: v1beta1.PipelineSpec{
+		Params: []v1beta1.ParamSpec{{
+			Name: "current-item",
+			Type: v1beta1.ParamTypeString,
+		}, {
+			Name: "additional-parameter",
+			Type: v1beta1.ParamTypeString,
+		}},
+		Tasks: []v1beta1.PipelineTask{{
+			Name: "mytask",
+			TaskSpec: &v1beta1.EmbeddedTask{
+				TaskSpec: v1beta1.TaskSpec{
+					Steps: []v1beta1.Step{{
+						Container: corev1.Container{Name: "foo", Image: "bar"},
+					}},
+				},
+			},
+		}},
+	},
+}
+
+var paraPipelineLoop = &pipelineloopv1alpha1.PipelineLoop{
+	ObjectMeta: metav1.ObjectMeta{Name: "para-pipelineloop", Namespace: "foo"},
+	Spec: pipelineloopv1alpha1.PipelineLoopSpec{
+		PipelineRef:  &v1beta1.PipelineRef{Name: "para-pipeline"},
+		IterateParam: "current-item",
+		Parallelism:  2,
+	},
+}
+
 var nPipelineLoop = &pipelineloopv1alpha1.PipelineLoop{
 	ObjectMeta: metav1.ObjectMeta{Name: "n-pipelineloop", Namespace: "foo"},
 	Spec: pipelineloopv1alpha1.PipelineLoopSpec{
@@ -364,6 +397,33 @@ var runPipelineLoop = &v1alpha1.Run{
 	},
 }
 
+var paraRunPipelineLoop = &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"},
+		}},
+		Ref: &v1alpha1.TaskRef{
+			APIVersion: pipelineloopv1alpha1.SchemeGroupVersion.String(),
+			Kind:       pipelineloop.PipelineLoopControllerName,
+			Name:       "para-pipelineloop",
+		},
+	},
+}
+
 var conditionRunPipelineLoop = &v1alpha1.Run{
 	ObjectMeta: metav1.ObjectMeta{
 		Name:      "run-pipelineloop",
@@ -623,6 +683,74 @@ var expectedPipelineRunIterationDict = &v1beta1.PipelineRun{
 	},
 }
 
+var expectedParaPipelineRun = &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":          "para-pipelineloop",
+			"tekton.dev/run":                          "run-pipelineloop",
+			"custom.tekton.dev/pipelineLoopIteration": "1",
+			"myTestLabel":                             "myTestLabelValue",
+		},
+		Annotations: map[string]string{
+			"myTestAnnotation": "myTestAnnotationValue",
+		},
+	},
+	Spec: v1beta1.PipelineRunSpec{
+		PipelineRef: &v1beta1.PipelineRef{Name: "para-pipeline"},
+		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"},
+		}},
+	},
+}
+
+var expectedParaPipelineRun1 = &v1beta1.PipelineRun{
+	ObjectMeta: metav1.ObjectMeta{
+		Name:      "run-pipelineloop-00002-mz4c7",
+		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":          "para-pipelineloop",
+			"tekton.dev/run":                          "run-pipelineloop",
+			"custom.tekton.dev/pipelineLoopIteration": "2",
+			"myTestLabel":                             "myTestLabelValue",
+		},
+		Annotations: map[string]string{
+			"myTestAnnotation": "myTestAnnotationValue",
+		},
+	},
+	Spec: v1beta1.PipelineRunSpec{
+		PipelineRef: &v1beta1.PipelineRef{Name: "para-pipeline"},
+		Params: []v1beta1.Param{{
+			Name:  "current-item",
+			Value: v1beta1.ArrayOrString{Type: v1beta1.ParamTypeString, StringVal: "item2"},
+		}, {
+			Name:  "additional-parameter",
+			Value: v1beta1.ArrayOrString{Type: v1beta1.ParamTypeString, StringVal: "stuff"},
+		}},
+	},
+}
+
 var expectedPipelineRunIteration1 = &v1beta1.PipelineRun{
 	ObjectMeta: metav1.ObjectMeta{
 		Name:      "run-pipelineloop-00001-9l9zj",
@@ -940,6 +1068,16 @@ func TestReconcilePipelineLoopRun(t *testing.T) {
 		expectedReason:       pipelineloopv1alpha1.PipelineLoopRunReasonCancelled,
 		expectedPipelineruns: []*v1beta1.PipelineRun{successful(expectedPipelineRunIteration1)},
 		expectedEvents:       []string{"Warning Failed Run " + runPipelineLoop.Namespace + "/" + runPipelineLoop.Name + " was cancelled"},
+	}, {
+		name:                 "Reconcile a new run with a pipelineloop with Parallelism specified",
+		pipeline:             paraPipeline,
+		pipelineloop:         paraPipelineLoop,
+		run:                  paraRunPipelineLoop,
+		pipelineruns:         []*v1beta1.PipelineRun{},
+		expectedStatus:       corev1.ConditionUnknown,
+		expectedReason:       pipelineloopv1alpha1.PipelineLoopRunReasonRunning,
+		expectedPipelineruns: []*v1beta1.PipelineRun{expectedParaPipelineRun, expectedParaPipelineRun1},
+		expectedEvents:       []string{"Normal Started", "Normal Running Iterations completed: 0"},
 	}}
 
 	for _, tc := range testcases {
@@ -979,17 +1117,17 @@ func TestReconcilePipelineLoopRun(t *testing.T) {
 			// If the number of expected PipelineRuns is greater than the original number of PipelineRuns
 			// then the test expects a new PipelineRun to be created.  The new PipelineRun must be the
 			// last one in the list of expected PipelineRuns.
-			createdPipelinerun := getCreatedPipelinerun(t, clients)
+			createdPipelineruns := getCreatedPipelinerun(t, clients)
 			if len(tc.expectedPipelineruns) > len(tc.pipelineruns) {
-				if createdPipelinerun == nil {
+				if len(createdPipelineruns) == 0 {
 					t.Errorf("A PipelineRun should have been created but was not")
 				} else {
-					if d := cmp.Diff(tc.expectedPipelineruns[len(tc.expectedPipelineruns)-1], createdPipelinerun); d != "" {
+					if d := cmp.Diff(tc.expectedPipelineruns, createdPipelineruns); d != "" {
 						t.Errorf("Expected PipelineRun was not created. Diff %s", diff.PrintWantGot(d))
 					}
 				}
 			} else {
-				if createdPipelinerun != nil {
+				if len(createdPipelineruns) > 0 {
 					t.Errorf("A PipelineRun was created which was not expected")
 				}
 			}