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

Issue:
Update SDK and pipeline loop controller to support retries.
/kind feature

**Description:**
https://github.com/tektoncd/pipeline/pull/4327 PR add support for retries in tektoncd for custom tasks. We should support in both 

1. PipelineLoop Controller 
2. SDK

**Additional information:**



Discussion / hints:

Fix patch:
diff --git a/tekton-catalog/pipeline-loops/cmd/webhook/main.go b/tekton-catalog/pipeline-loops/cmd/webhook/main.go
index bdf8dd9a3f..dc4eab292e 100644
--- a/tekton-catalog/pipeline-loops/cmd/webhook/main.go
+++ b/tekton-catalog/pipeline-loops/cmd/webhook/main.go
@@ -22,7 +22,6 @@ import (
 
 	pipelineloopv1alpha1 "github.com/kubeflow/kfp-tekton/tekton-catalog/pipeline-loops/pkg/apis/pipelineloop/v1alpha1"
 	defaultconfig "github.com/tektoncd/pipeline/pkg/apis/config"
-	"github.com/tektoncd/pipeline/pkg/contexts"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
@@ -66,7 +65,7 @@ func newDefaultingAdmissionController(ctx context.Context, cmw configmap.Watcher
 
 		// A function that infuses the context passed to Validate/SetDefaults with custom metadata.
 		func(ctx context.Context) context.Context {
-			return contexts.WithUpgradeViaDefaulting(store.ToContext(ctx))
+			return store.ToContext(ctx)
 		},
 
 		// Whether to disallow unknown fields.
@@ -91,7 +90,7 @@ func newValidationAdmissionController(ctx context.Context, cmw configmap.Watcher
 
 		// A function that infuses the context passed to Validate/SetDefaults with custom metadata.
 		func(ctx context.Context) context.Context {
-			return contexts.WithUpgradeViaDefaulting(store.ToContext(ctx))
+			return store.ToContext(ctx)
 		},
 
 		// Whether to disallow unknown fields.
diff --git a/tekton-catalog/pipeline-loops/examples/loop-example-basic.yaml b/tekton-catalog/pipeline-loops/examples/loop-example-basic.yaml
index a56f221e14..13613a68cf 100644
--- a/tekton-catalog/pipeline-loops/examples/loop-example-basic.yaml
+++ b/tekton-catalog/pipeline-loops/examples/loop-example-basic.yaml
@@ -22,7 +22,7 @@ spec:
             imagePullPolicy: IfNotPresent
             script: |
               #!/usr/bin/env bash
-              echo "$(params.message)"
+              echo "$(params.message)" && exit 1;
   iterateParam: message
 ---
 apiVersion: tekton.dev/v1beta1
@@ -46,6 +46,7 @@ spec:
                 #!/usr/bin/env bash
                 echo "I am the first task before the loop task"
       - name: loop-task
+        retries: 3
         runAfter:
           - first-task
         params:
diff --git a/tekton-catalog/pipeline-loops/go.mod b/tekton-catalog/pipeline-loops/go.mod
index bf0f033628..142653a094 100644
--- a/tekton-catalog/pipeline-loops/go.mod
+++ b/tekton-catalog/pipeline-loops/go.mod
@@ -4,16 +4,17 @@ go 1.13
 
 require (
 	contrib.go.opencensus.io/exporter/stackdriver v0.13.5 // indirect
+	github.com/go-kit/kit v0.10.0 // indirect
 	github.com/go-openapi/validate v0.19.5 // indirect
 	github.com/google/go-cmp v0.5.6
-	github.com/hashicorp/go-multierror v1.1.0
-	github.com/tektoncd/pipeline v0.27.1
-	go.uber.org/zap v1.18.1
+	github.com/hashicorp/go-multierror v1.1.1
+	github.com/tektoncd/pipeline v0.29.1-0.20211111015141-52590be06673
+	go.uber.org/zap v1.19.1
 	gomodules.xyz/jsonpatch/v2 v2.2.0
 	gopkg.in/evanphx/json-patch.v4 v4.9.0 // indirect
 	honnef.co/go/tools v0.0.1-2020.1.5 // indirect
-	k8s.io/api v0.20.7
-	k8s.io/apimachinery v0.20.7
-	k8s.io/client-go v0.20.7
-	knative.dev/pkg v0.0.0-20210730172132-bb4aaf09c430
+	k8s.io/api v0.21.4
+	k8s.io/apimachinery v0.21.4
+	k8s.io/client-go v0.21.4
+	knative.dev/pkg v0.0.0-20211101212339-96c0204a70dc
 )
diff --git a/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun.go b/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun.go
index 6a40dd194b..2cf94cc97f 100644
--- a/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun.go
+++ b/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun.go
@@ -20,6 +20,8 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"k8s.io/apimachinery/pkg/runtime"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"log"
 	"reflect"
 	"strconv"
@@ -278,13 +280,63 @@ func (c *Reconciler) reconcile(ctx context.Context, run *v1alpha1.Run, status *p
 		return nil
 	}
 
+	retriesDone := len(run.Status.RetriesStatus)
+	retries := run.Spec.Retries
+	if retriesDone < retries && failedPrs != nil && len(failedPrs) > 0 {
+		logger.Infof("RetriesDone: %d, Total Retries: %d", retriesDone, retries)
+		run.Status.RetriesStatus = append(run.Status.RetriesStatus, v1alpha1.RunStatus{
+			Status: duckv1.Status{
+				ObservedGeneration: 0,
+				Conditions:         run.Status.Conditions.DeepCopy(),
+				Annotations:        nil,
+			},
+			RunStatusFields: runv1alpha1.RunStatusFields{
+				StartTime:      run.Status.StartTime.DeepCopy(),
+				CompletionTime: run.Status.CompletionTime.DeepCopy(),
+				Results:        nil,
+				RetriesStatus:  nil,
+				ExtraFields:    runtime.RawExtension{},
+			},
+		})
+		// Without immediately updating here, wrong number of retries are performed.
+		_, err := c.pipelineClientSet.TektonV1alpha1().Runs(run.Namespace).UpdateStatus(ctx, run, metav1.UpdateOptions{})
+		if err != nil {
+			return err
+		}
+		for _, failedPr := range failedPrs {
+			// PipelineRun do not support a retry, we dispose off old PR and create a fresh one.
+			// instead of deleting we just label it deleted=True.
+			deletedLabel := map[string]string{"deleted": "True"}
+			mergePatch := map[string]interface{}{
+				"metadata": map[string]interface{}{
+					"labels": deletedLabel,
+				},
+			}
+			patch, err := json.Marshal(mergePatch)
+			if err != nil {
+				return err
+			}
+			_, err = c.pipelineClientSet.TektonV1alpha1().PipelineRuns(failedPr.Namespace).
+				Patch(ctx, failedPr.Name, types.MergePatchType, patch, metav1.PatchOptions{})
+			pr, err := c.createPipelineRun(ctx, logger, pipelineLoopSpec, run, highestIteration, iterationElements)
+			if err != nil {
+				return fmt.Errorf("error creating PipelineRun from Run %s while retrying: %w", run.Name, err)
+			}
+			status.PipelineRuns[pr.Name] = &pipelineloopv1alpha1.PipelineLoopPipelineRunStatus{
+				Iteration: highestIteration,
+				Status:    &pr.Status,
+			}
+			logger.Infof("Retried failed pipelineRun: %s with new pipelineRun: %s", failedPr.Name, pr.Name)
+		}
+		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))
@@ -430,20 +482,6 @@ func (c *Reconciler) createPipelineRun(ctx context.Context, logger *zap.SugaredL
 
 }
 
-// func (c *Reconciler) retryPipelineRun(ctx context.Context, tr *v1beta1.PipelineRun) (*v1beta1.PipelineRun, error) {
-// 	newStatus := *tr.Status.DeepCopy()
-// 	newStatus.RetriesStatus = nil
-// 	tr.Status.RetriesStatus = append(tr.Status.RetriesStatus, newStatus)
-// 	tr.Status.StartTime = nil
-// 	tr.Status.CompletionTime = nil
-// 	tr.Status.PodName = ""
-// 	tr.Status.SetCondition(&apis.Condition{
-// 		Type:   apis.ConditionSucceeded,
-// 		Status: corev1.ConditionUnknown,
-// 	})
-// 	return c.pipelineClientSet.TektonV1beta1().PipelineRuns(tr.Namespace).UpdateStatus(ctx, tr, metav1.UpdateOptions{})
-// }
-
 func (c *Reconciler) updateLabelsAndAnnotations(ctx context.Context, run *v1alpha1.Run) error {
 	newRun, err := c.runLister.Runs(run.Namespace).Get(run.Name)
 	if err != nil {
@@ -484,6 +522,10 @@ func (c *Reconciler) updatePipelineRunStatus(logger *zap.SugaredLogger, run *v1a
 	status.CurrentRunning = 0
 	for _, pr := range pipelineRuns {
 		lbls := pr.GetLabels()
+		if lbls["deleted"] == "True" {
+			// PipelineRun is already retried, skipping...
+			continue
+		}
 		iterationStr := lbls[pipelineloop.GroupName+pipelineLoopIterationLabelKey]
 		iteration, err := strconv.Atoi(iterationStr)
 		if err != nil {


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 f8ab027a38..2d9ce70793 100644
--- a/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun_test.go
+++ b/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun_test.go
@@ -104,6 +104,21 @@ func failed(pr *v1beta1.PipelineRun) *v1beta1.PipelineRun {
 	return prWithStatus
 }
 
+func setRetries(run *v1alpha1.Run, retries int) *v1alpha1.Run {
+	run.Spec.Retries = retries
+	return run
+}
+
+func setDeleted(pr *v1beta1.PipelineRun) *v1beta1.PipelineRun {
+	pr.Labels["deleted"] = "True"
+	return pr
+}
+
+func setPrName(pr *v1beta1.PipelineRun, name string) *v1beta1.PipelineRun {
+	pr.Name = name
+	return pr
+}
+
 // getPipelineLoopController returns an instance of the PipelineLoop controller/reconciler that has been seeded with
 // d, where d represents the state of the system (existing resources) needed for the test.
 func getPipelineLoopController(t *testing.T, d test.Data, pipelineloops []*pipelineloopv1alpha1.PipelineLoop) (test.Assets, func()) {
@@ -837,6 +852,78 @@ var expectedPipelineRunIteration1 = &v1beta1.PipelineRun{
 	},
 }
 
+var expectedPipelineRunFailed = &v1beta1.PipelineRun{
+	ObjectMeta: metav1.ObjectMeta{
+		Name:      "run-pipelineloop-00001-failed",
+		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/originalPipelineRun":   "pr-loop-example",
+			"custom.tekton.dev/parentPipelineRun":     "pr-loop-example",
+			"custom.tekton.dev/pipelineLoop":          "a-pipelineloop",
+			"tekton.dev/run":                          "run-pipelineloop",
+			"custom.tekton.dev/pipelineLoopIteration": "1",
+			"myTestLabel":                             "myTestLabelValue",
+		},
+		Annotations: map[string]string{
+			"custom.tekton.dev/pipelineLoopCurrentIterationItem": `"item1"`,
+			"myTestAnnotation": "myTestAnnotationValue",
+		},
+	},
+	Spec: v1beta1.PipelineRunSpec{
+		PipelineRef: &v1beta1.PipelineRef{Name: "a-pipeline"},
+		Params: []v1beta1.Param{{
+			Name:  "additional-parameter",
+			Value: v1beta1.ArrayOrString{Type: v1beta1.ParamTypeString, StringVal: "stuff"},
+		}, {
+			Name:  "current-item",
+			Value: v1beta1.ArrayOrString{Type: v1beta1.ParamTypeString, StringVal: "item1"},
+		}},
+	},
+}
+
+var expectedPipelineRunRetry = &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/originalPipelineRun":   "pr-loop-example",
+			"custom.tekton.dev/parentPipelineRun":     "pr-loop-example",
+			"custom.tekton.dev/pipelineLoop":          "a-pipelineloop",
+			"tekton.dev/run":                          "run-pipelineloop",
+			"custom.tekton.dev/pipelineLoopIteration": "1",
+			"myTestLabel":                             "myTestLabelValue",
+		},
+		Annotations: map[string]string{
+			"custom.tekton.dev/pipelineLoopCurrentIterationItem": `"item1"`,
+			"myTestAnnotation": "myTestAnnotationValue",
+		},
+	},
+	Spec: v1beta1.PipelineRunSpec{
+		PipelineRef: &v1beta1.PipelineRef{Name: "a-pipeline"},
+		Params: []v1beta1.Param{{
+			Name:  "additional-parameter",
+			Value: v1beta1.ArrayOrString{Type: v1beta1.ParamTypeString, StringVal: "stuff"},
+		}, {
+			Name:  "current-item",
+			Value: v1beta1.ArrayOrString{Type: v1beta1.ParamTypeString, StringVal: "item1"},
+		}},
+	},
+}
+
 var expectedPipelineRunIterateNumeric1 = &v1beta1.PipelineRun{
 	ObjectMeta: metav1.ObjectMeta{
 		Name:      "run-pipelineloop-00001-9l9zj",
@@ -1146,6 +1233,15 @@ func TestReconcilePipelineLoopRun(t *testing.T) {
 		expectedReason:       pipelineloopv1alpha1.PipelineLoopRunReasonFailed,
 		expectedPipelineruns: []*v1beta1.PipelineRun{failed(expectedPipelineRunIteration1)},
 		expectedEvents:       []string{"Warning Failed PipelineRun " + expectedPipelineRunIteration1.Name + " has failed"},
+	}, {
+		name:                 "Reconcile a run with retries after the first PipelineRun has failed",
+		pipeline:             aPipeline,
+		pipelineloop:         aPipelineLoop,
+		run:                  loopRunning(setRetries(runPipelineLoop, 1)),
+		pipelineruns:         []*v1beta1.PipelineRun{failed(expectedPipelineRunFailed)},
+		expectedStatus:       corev1.ConditionUnknown,
+		expectedReason:       pipelineloopv1alpha1.PipelineLoopRunReasonRunning,
+		expectedPipelineruns: []*v1beta1.PipelineRun{setDeleted(failed(expectedPipelineRunFailed)), expectedPipelineRunRetry},
 	}, {
 		name:                 "Reconcile a new run with a pipelineloop with Parallelism specified",
 		pipeline:             paraPipeline,
@@ -1248,7 +1344,16 @@ func TestReconcilePipelineLoopRun(t *testing.T) {
 				if len(createdPipelineruns) == 0 {
 					t.Errorf("A PipelineRun should have been created but was not")
 				} else {
-					if d := cmp.Diff(tc.expectedPipelineruns, createdPipelineruns); d != "" {
+					pipelineRunsExpectedToBeCreated := make([]*v1beta1.PipelineRun, len(createdPipelineruns))
+					i := 0
+					for _, pr := range tc.expectedPipelineruns {
+						if pr.Labels["deleted"] != "True" {
+							pipelineRunsExpectedToBeCreated[i] = pr
+							i = i + 1 // skip the pr that were retried.
+						}
+					}
+
+					if d := cmp.Diff(pipelineRunsExpectedToBeCreated, createdPipelineruns); d != "" {
 						t.Errorf("Expected PipelineRun was not created. Diff %s", diff.PrintWantGot(d))
 					}
 				}
@@ -1260,8 +1365,12 @@ func TestReconcilePipelineLoopRun(t *testing.T) {
 
 			// Verify Run status contains status for all PipelineRuns.
 			expectedPipelineRuns := map[string]pipelineloopv1alpha1.PipelineLoopPipelineRunStatus{}
-			for i, pr := range tc.expectedPipelineruns {
-				expectedPipelineRuns[pr.Name] = pipelineloopv1alpha1.PipelineLoopPipelineRunStatus{Iteration: i + 1, Status: &pr.Status}
+			i := 1
+			for _, pr := range tc.expectedPipelineruns {
+				expectedPipelineRuns[pr.Name] = pipelineloopv1alpha1.PipelineLoopPipelineRunStatus{Iteration: i, Status: &pr.Status}
+				if pr.Labels["deleted"] != "True" {
+					i = i + 1 // iteration remain same, incase previous pr was a retry.
+				}
 			}
 			checkRunStatus(t, reconciledRun, expectedPipelineRuns)