limit task retry when system error

erda-project · Dec 13, 2021 · 26f9358 · 26f9358
1 parent dd78a2f
commit 26f9358
Show file tree

Hide file tree

Showing 4 changed files with 40 additions and 2 deletions.
diff --git a/apistructs/pipeline_task.go b/apistructs/pipeline_task.go
@@ -23,7 +23,9 @@ import (
 
 const (
 	// TerminusDefineTag add this tag env to container for collecting logs
-	TerminusDefineTag = "TERMINUS_DEFINE_TAG"
+	TerminusDefineTag            = "TERMINUS_DEFINE_TAG"
+	PipelineTaskMaxRetryLimit    = 100
+	PipelineTaskMaxRetryDuration = 24 * time.Hour
 )
 
 type PipelineTaskDTO struct {
@@ -267,6 +269,20 @@ func (t *PipelineTaskInspect) ConvertErrors() {
 	}
 }
 
+func (t *PipelineTaskInspect) IsErrorsExceed() bool {
+	var exceed bool
+	now := time.Now()
+	for _, g := range t.Errors {
+		if !g.Ctx.StartTime.IsZero() && g.Ctx.StartTime.Add(PipelineTaskMaxRetryDuration).Before(now) {
+			exceed = true
+		}
+		if g.Ctx.Count >= PipelineTaskMaxRetryLimit {
+			exceed = true
+		}
+	}
+	return exceed
+}
+
 func (l *PipelineTaskLoop) Duplicate() *PipelineTaskLoop {
 	if l == nil {
 		return nil

diff --git a/apistructs/pipeline_task_test.go b/apistructs/pipeline_task_test.go
@@ -17,6 +17,9 @@ package apistructs
 import (
 	"fmt"
 	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func TestPipelineTaskLoop_Duplicate(t *testing.T) {
@@ -95,3 +98,18 @@ func TestPipelineTaskLoop_IsEmpty(t *testing.T) {
 		})
 	}
 }
+
+func TestIsErrorsExceed(t *testing.T) {
+	now := time.Now()
+	timeExceedInspect := &PipelineTaskInspect{}
+	timeExceedInspect.Errors = timeExceedInspect.AppendError(&PipelineTaskErrResponse{Msg: "xxx", Ctx: PipelineTaskErrCtx{StartTime: now.Add(-25 * time.Hour)}})
+	assert.Equal(t, true, timeExceedInspect.IsErrorsExceed())
+
+	countExceedInspect := &PipelineTaskInspect{}
+	for i := 0; i < 99; i++ {
+		countExceedInspect.Errors = countExceedInspect.AppendError(&PipelineTaskErrResponse{Msg: "xxx"})
+		assert.Equal(t, false, countExceedInspect.IsErrorsExceed())
+	}
+	countExceedInspect.Errors = countExceedInspect.AppendError(&PipelineTaskErrResponse{Msg: "xxx"})
+	assert.Equal(t, true, countExceedInspect.IsErrorsExceed())
+}
diff --git a/modules/pipeline/pipengine/reconciler/task.go b/modules/pipeline/pipengine/reconciler/task.go
@@ -96,6 +96,10 @@ func reconcileTask(tr *taskrun.TaskRun) error {
 					rlog.TErrorf(tr.P.ID, tr.Task.ID, "failed to handle taskOp: %s, user abnormalErr: %v, don't need retry", taskOp.Op(), abnormalErr)
 					return abnormalErr
 				}
+				if tr.Task.Inspect.IsErrorsExceed() {
+					rlog.TErrorf(tr.P.ID, tr.Task.ID, "failed to handle taskOp: %s, errors exceed limit, stop retry", taskOp.Op())
+					return abnormalErr
+				}
 				// don't contain user error mean err is platform error, should retry always
 				rlog.TErrorf(tr.P.ID, tr.Task.ID, "failed to handle taskOp: %s, abnormalErr: %v, continue retry, retry times: %d", taskOp.Op(), abnormalErr, platformErrRetryTimes)
 				resetTaskForAbnormalRetry(tr, platformErrRetryTimes)

diff --git a/modules/pipeline/pipengine/reconciler/taskrun/framework.go b/modules/pipeline/pipengine/reconciler/taskrun/framework.go
@@ -129,7 +129,7 @@ func (tr *TaskRun) waitOp(itr TaskOp, o *Elem) (result error) {
 		}
 
 		// if result only contain platform error, task will retry, so don't set status changed
-		if result != nil && !errorsx.IsContainUserError(result) {
+		if result != nil && !errorsx.IsContainUserError(result) && !tr.Task.Inspect.IsErrorsExceed() {
 			tr.Task.Status = oldStatus
 		}