limit task retry when system error

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    = 144
+	PipelineTaskMaxRetryDuration = 24 * time.Hour
 )
 
 type PipelineTaskDTO struct {
@@ -267,6 +269,17 @@ func (t *PipelineTaskInspect) ConvertErrors() {
 	}
 }
 
+func (t *PipelineTaskInspect) IsErrorsExceed() (bool, *PipelineTaskErrResponse) {
+	now := time.Now()
+	for _, g := range t.Errors {
+		if (!g.Ctx.StartTime.IsZero() && g.Ctx.StartTime.Add(PipelineTaskMaxRetryDuration).Before(now)) ||
+			g.Ctx.Count >= PipelineTaskMaxRetryLimit {
+			return true, g
+		}
+	}
+	return false, nil
+}
+
 func (l *PipelineTaskLoop) Duplicate() *PipelineTaskLoop {
 	if l == nil {
 		return nil

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,21 @@ 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)}})
+	isExceed, _ := timeExceedInspect.IsErrorsExceed()
+	assert.Equal(t, true, isExceed)
+
+	countExceedInspect := &PipelineTaskInspect{}
+	for i := 0; i < 143; i++ {
+		countExceedInspect.Errors = countExceedInspect.AppendError(&PipelineTaskErrResponse{Msg: "xxx"})
+		isExceed, _ = countExceedInspect.IsErrorsExceed()
+		assert.Equal(t, false, isExceed)
+	}
+	countExceedInspect.Errors = countExceedInspect.AppendError(&PipelineTaskErrResponse{Msg: "xxx"})
+	isExceed, _ = countExceedInspect.IsErrorsExceed()
+	assert.Equal(t, true, isExceed)
+}

modules/pipeline/pipengine/reconciler/task.go

@@ -96,6 +96,11 @@ 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 isExceed, errCtx := tr.Task.Inspect.IsErrorsExceed(); isExceed {
+					rlog.TErrorf(tr.P.ID, tr.Task.ID, "failed to handle taskOp: %s, errors exceed limit, stop retry, retry times: %d, start itme: %s, pipelineID: %d, taskID: %d",
+						taskOp.Op(), errCtx.Ctx.Count, errCtx.Ctx.StartTime.Format("2006-01-02 15:04:05"), tr.P.ID, tr.Task.ID)
+					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)

modules/pipeline/pipengine/reconciler/taskrun/framework.go

@@ -129,7 +129,8 @@ 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) {
+		isExceed, _ := tr.Task.Inspect.IsErrorsExceed()
+		if result != nil && !errorsx.IsContainUserError(result) && !isExceed {
 			tr.Task.Status = oldStatus
 		}