[incubator-devlake] branch main updated: fix: unable to cancel pipeline with skip-on-fail (#4182)

[li...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

likyh pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-devlake.git


The following commit(s) were added to refs/heads/main by this push:
     new dd391d0cb fix: unable to cancel pipeline with skip-on-fail (#4182)
dd391d0cb is described below

commit dd391d0cbe51cefaa0ad4d5c84bcc28f95c8047b
Author: Klesh Wong <zh...@merico.dev>
AuthorDate: Tue Jan 10 23:58:57 2023 +0800

    fix: unable to cancel pipeline with skip-on-fail (#4182)
---
 services/pipeline_runner.go               |  7 ++++---
 test/services/pipeline_runner_e2e_test.go | 14 +++++++-------
 2 files changed, 11 insertions(+), 10 deletions(-)

diff --git a/services/pipeline_runner.go b/services/pipeline_runner.go
index a84bc816e..60e8d7fa1 100644
--- a/services/pipeline_runner.go
+++ b/services/pipeline_runner.go
@@ -113,6 +113,7 @@ func runPipeline(pipelineId uint64) errors.Error {
 	} else {
 		err = pipelineRun.runPipelineStandalone()
 	}
+	isCancelled := errors.Is(err, context.Canceled)
 	if err != nil {
 		err = errors.Default.Wrap(err, fmt.Sprintf("Error running pipeline %d.", pipelineId))
 	}
@@ -130,7 +131,7 @@ func runPipeline(pipelineId uint64) errors.Error {
 		dbPipeline.Message = err.Error()
 		dbPipeline.ErrorName = err.Messages().Format()
 	}
-	dbPipeline.Status, err = ComputePipelineStatus(dbPipeline)
+	dbPipeline.Status, err = ComputePipelineStatus(dbPipeline, isCancelled)
 	if err != nil {
 		globalPipelineLog.Error(err, "compute pipeline status failed")
 		return err
@@ -148,7 +149,7 @@ func runPipeline(pipelineId uint64) errors.Error {
 // 1. TASK_COMPLETED: all tasks were executed sucessfully
 // 2. TASK_FAILED: SkipOnFail=false with failed task(s)
 // 3. TASK_PARTIAL: SkipOnFail=true with failed task(s)
-func ComputePipelineStatus(pipeline *models.DbPipeline) (string, errors.Error) {
+func ComputePipelineStatus(pipeline *models.DbPipeline, isCancelled bool) (string, errors.Error) {
 	tasks, err := GetLatestTasksOfPipeline(pipeline)
 	if err != nil {
 		return "", err
@@ -168,7 +169,7 @@ func ComputePipelineStatus(pipeline *models.DbPipeline) (string, errors.Error) {
 		}
 	}
 
-	if running > 0 || (pipeline.SkipOnFail && pending > 0) {
+	if running > 0 || (!isCancelled && pipeline.SkipOnFail && pending > 0) {
 		return "", errors.Default.New("unexpected status, did you call computePipelineStatus at a wrong timing?")
 	}
 
diff --git a/test/services/pipeline_runner_e2e_test.go b/test/services/pipeline_runner_e2e_test.go
index eff85467a..46640f792 100644
--- a/test/services/pipeline_runner_e2e_test.go
+++ b/test/services/pipeline_runner_e2e_test.go
@@ -70,7 +70,7 @@ func TestComputePipelineStatus(t *testing.T) {
 	assert.NotZero(t, task_row2_col1.ID)
 
 	// pipeline.status == "failed" if SkipOnFailed=false and any tasks failed
-	status, err := services.ComputePipelineStatus(pipeline)
+	status, err := services.ComputePipelineStatus(pipeline, false)
 	if !assert.Nil(t, err) {
 		println(err.Messages().Format())
 	}
@@ -83,7 +83,7 @@ func TestComputePipelineStatus(t *testing.T) {
 	task_row2_col1.Status = models.TASK_COMPLETED
 	err = db.Update(task_row2_col1)
 	assert.Nil(t, err)
-	status, err = services.ComputePipelineStatus(pipeline)
+	status, err = services.ComputePipelineStatus(pipeline, false)
 	if !assert.Nil(t, err) {
 		println(err.Messages().Format())
 	}
@@ -92,7 +92,7 @@ func TestComputePipelineStatus(t *testing.T) {
 	pipeline.SkipOnFail = true
 	err = db.Update(pipeline)
 	assert.Nil(t, err)
-	status, err = services.ComputePipelineStatus(pipeline)
+	status, err = services.ComputePipelineStatus(pipeline, false)
 	assert.Nil(t, err)
 	assert.Equal(t, models.TASK_COMPLETED, status)
 
@@ -100,7 +100,7 @@ func TestComputePipelineStatus(t *testing.T) {
 	task_row1_col1.Status = models.TASK_FAILED
 	err = db.Update(task_row1_col1)
 	assert.Nil(t, err)
-	status, err = services.ComputePipelineStatus(pipeline)
+	status, err = services.ComputePipelineStatus(pipeline, false)
 	assert.Nil(t, err)
 	assert.Equal(t, models.TASK_PARTIAL, status)
 
@@ -114,7 +114,7 @@ func TestComputePipelineStatus(t *testing.T) {
 	task_row2_col1.Status = models.TASK_FAILED
 	err = db.Update(task_row2_col1)
 	assert.Nil(t, err)
-	status, err = services.ComputePipelineStatus(pipeline)
+	status, err = services.ComputePipelineStatus(pipeline, false)
 	assert.Nil(t, err)
 	assert.Equal(t, models.TASK_FAILED, status)
 
@@ -152,7 +152,7 @@ func TestComputePipelineStatus(t *testing.T) {
 	assert.Nil(t, err)
 	assert.NotZero(t, task_row2_col1.ID)
 
-	status, err = services.ComputePipelineStatus(pipeline)
+	status, err = services.ComputePipelineStatus(pipeline, false)
 	assert.Nil(t, err)
 	assert.Equal(t, models.TASK_COMPLETED, status)
 
@@ -160,7 +160,7 @@ func TestComputePipelineStatus(t *testing.T) {
 	task_row1_col1_rerun.Status = models.TASK_CANCELLED
 	err = db.Update(task_row1_col1_rerun)
 	assert.Nil(t, err)
-	status, err = services.ComputePipelineStatus(pipeline)
+	status, err = services.ComputePipelineStatus(pipeline, false)
 	assert.Nil(t, err)
 	assert.Equal(t, models.TASK_PARTIAL, status)
 }