From 1d4b95b42d3b41198a57a9270d684721512f04fe Mon Sep 17 00:00:00 2001 From: Tommy Li Date: Thu, 4 Jan 2024 10:53:19 -0800 Subject: [PATCH 1/2] update pipelineloop v2 to have failed and cancelled status --- .../pipelinelooprun/pipelinelooprun.go | 29 ++++++++++++++++++- 1 file changed, 28 insertions(+), 1 deletion(-) diff --git a/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun.go b/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun.go index 82afb9bd20..997b7c5fe8 100644 --- a/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun.go +++ b/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun.go @@ -433,18 +433,45 @@ func (c *Reconciler) reconcile(ctx context.Context, customRun *tektonv1beta1.Cus } // CustomRun is cancelled, just cancel all the running instance and return if customRun.IsCancelled() { + var DAGStatus pb.Execution_State if len(failedPrs) > 0 { customRun.Status.MarkCustomRunFailed(pipelineloopv1alpha1.PipelineLoopRunReasonFailed.String(), "CustomRun %s/%s was failed", customRun.Namespace, customRun.Name) + DAGStatus = pb.Execution_FAILED } else { reason := pipelineloopv1alpha1.PipelineLoopRunReasonCancelled.String() if customRun.HasTimedOut(c.clock) { // This check is only possible if we are on tekton 0.27.0 + reason = string(tektonv1beta1.CustomRunReasonTimedOut) } customRun.Status.MarkCustomRunFailed(reason, "CustomRun %s/%s was cancelled", customRun.Namespace, customRun.Name) + DAGStatus = pb.Execution_CANCELED + } + if c.runKFPV2Driver == "true" { + pipelineRunParams := customRun.Spec.Params + options, err := kfptask.ParseParams(customRun) + if err != nil { + logger.Errorf("Run %s/%s is invalid because of %s", customRun.Namespace, customRun.Name, err) + customRun.Status.MarkCustomRunFailed(kfptask.ReasonFailedValidation, + "Run can't be run because it has an invalid param - %v", err) + return err + } + var executionID string + for _, pipelineRunParam := range pipelineRunParams { + if pipelineRunParam.Name == "dag-execution-id" { + executionID = pipelineRunParam.Value.StringVal + break + } + } + kfptask.UpdateOptionsDAGExecutionID(options, executionID) + DAGErr := kfptask.UpdateDAGPublisher(ctx, options, DAGStatus) + if err != nil { + logger.Errorf("kfp publisher failed when reconciling Run %s/%s: %v", customRun.Namespace, customRun.Name, DAGErr) + customRun.Status.MarkCustomRunFailed(kfptask.ReasonDriverError, + "kfp publisher execution failed: %v", DAGErr) + return DAGErr + } } - for _, currentRunningPr := range currentRunningPrs { logger.Infof("CustomRun %s/%s is cancelled. Cancelling PipelineRun %s.", customRun.Namespace, customRun.Name, currentRunningPr.Name) if _, err := c.pipelineClientSet.TektonV1().PipelineRuns(customRun.Namespace).Patch(ctx, currentRunningPr.Name, types.JSONPatchType, cancelPatchBytes, metav1.PatchOptions{}); err != nil { From 687de02e90463008367b85de58eb8d40d8ee4a52 Mon Sep 17 00:00:00 2001 From: Tommy Li Date: Thu, 4 Jan 2024 17:08:23 -0800 Subject: [PATCH 2/2] fix execution id update bug --- .../pkg/reconciler/pipelinelooprun/pipelinelooprun.go | 9 --------- 1 file changed, 9 deletions(-) diff --git a/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun.go b/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun.go index 997b7c5fe8..93f564550e 100644 --- a/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun.go +++ b/tekton-catalog/pipeline-loops/pkg/reconciler/pipelinelooprun/pipelinelooprun.go @@ -448,7 +448,6 @@ func (c *Reconciler) reconcile(ctx context.Context, customRun *tektonv1beta1.Cus DAGStatus = pb.Execution_CANCELED } if c.runKFPV2Driver == "true" { - pipelineRunParams := customRun.Spec.Params options, err := kfptask.ParseParams(customRun) if err != nil { logger.Errorf("Run %s/%s is invalid because of %s", customRun.Namespace, customRun.Name, err) @@ -456,14 +455,6 @@ func (c *Reconciler) reconcile(ctx context.Context, customRun *tektonv1beta1.Cus "Run can't be run because it has an invalid param - %v", err) return err } - var executionID string - for _, pipelineRunParam := range pipelineRunParams { - if pipelineRunParam.Name == "dag-execution-id" { - executionID = pipelineRunParam.Value.StringVal - break - } - } - kfptask.UpdateOptionsDAGExecutionID(options, executionID) DAGErr := kfptask.UpdateDAGPublisher(ctx, options, DAGStatus) if err != nil { logger.Errorf("kfp publisher failed when reconciling Run %s/%s: %v", customRun.Namespace, customRun.Name, DAGErr)