apitest action don't set task pointers value

Author: chengjoey

modules/pipeline/pipengine/actionexecutor/plugins/apitest/apitest.go

@@ -18,6 +18,8 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/sirupsen/logrus"
+
 	"github.com/erda-project/erda/apistructs"
 	"github.com/erda-project/erda/modules/pipeline/dbclient"
 	"github.com/erda-project/erda/modules/pipeline/pipengine/actionexecutor/plugins/apitest/logic"
@@ -57,7 +59,41 @@ func (d *define) Create(ctx context.Context, task *spec.PipelineTask) (interface
 }
 
 func (d *define) Start(ctx context.Context, task *spec.PipelineTask) (interface{}, error) {
-	logic.Do(ctx, task)
+
+	go func(ctx context.Context, task *spec.PipelineTask) {
+		executorDoneCh, ok := ctx.Value(spec.MakeTaskExecutorCtxKey(task)).(chan interface{})
+		if !ok {
+			logrus.Warnf("apitest: failed to get executor channel, pipelineID: %d, taskID: %d", task.PipelineID, task.ID)
+		}
+
+		var status = apistructs.PipelineStatusFailed
+		defer func() {
+			if r := recover(); r != nil {
+				logrus.Errorf("api-test logic do panic recover:%s", r)
+			}
+			// if executor chan is nil, task framework can loop query meta get status
+			if executorDoneCh != nil {
+				executorDoneCh <- apistructs.PipelineStatusDesc{Status: status}
+			}
+		}()
+
+		logic.Do(ctx, task)
+
+		latestTask, err := d.dbClient.GetPipelineTask(task.ID)
+		if err != nil {
+			logrus.Errorf("failed to query latest task, err: %v \n", err)
+			return
+		}
+
+		meta := latestTask.Result.Metadata
+		for _, metaField := range meta {
+			if metaField.Name == logic.MetaKeyResult {
+				if metaField.Value == logic.ResultSuccess {
+					status = apistructs.PipelineStatusSuccess
+				}
+			}
+		}
+	}(ctx, task)
 	return nil, nil
 }
 
@@ -70,7 +106,7 @@ func (d *define) Status(ctx context.Context, task *spec.PipelineTask) (apistruct
 	if err != nil {
 		return apistructs.PipelineStatusDesc{}, fmt.Errorf("failed to query latest task, err: %v", err)
 	}
-	*task = latestTask
+	//*task = latestTask
 
 	if task.Status.IsEndStatus() {
 		return apistructs.PipelineStatusDesc{Status: task.Status}, nil
@@ -92,12 +128,14 @@ func (d *define) Status(ctx context.Context, task *spec.PipelineTask) (apistruct
 			if metaField.Value == logic.ResultSuccess {
 				return apistructs.PipelineStatusDesc{Status: apistructs.PipelineStatusSuccess}, nil
 			}
-			return apistructs.PipelineStatusDesc{Status: apistructs.PipelineStatusFailed}, nil
+			if metaField.Value == logic.ResultFailed {
+				return apistructs.PipelineStatusDesc{Status: apistructs.PipelineStatusFailed}, nil
+			}
 		}
 	}
 
 	// return created status to do start step
-	return apistructs.PipelineStatusDesc{Status: apistructs.PipelineStatusCreated}, nil
+	return apistructs.PipelineStatusDesc{Status: apistructs.PipelineStatusRunning}, nil
 }
 
 func (d *define) Inspect(ctx context.Context, task *spec.PipelineTask) (apistructs.TaskInspect, error) {

modules/pipeline/pipengine/actionexecutor/plugins/apitest/logic/meta.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"encoding/json"
 	"strconv"
+	"time"
 
 	"github.com/erda-project/erda/apistructs"
 	"github.com/erda-project/erda/modules/actionagent"
@@ -26,6 +27,7 @@ import (
 	"github.com/erda-project/erda/pkg/apitestsv2"
 	"github.com/erda-project/erda/pkg/apitestsv2/cookiejar"
 	"github.com/erda-project/erda/pkg/encoding/jsonparse"
+	"github.com/erda-project/erda/pkg/loop"
 )
 
 const (
@@ -112,11 +114,15 @@ func writeMetaFile(ctx context.Context, task *spec.PipelineTask, meta *Meta) {
 	cb.PipelineTaskID = task.ID
 
 	cbData, _ := json.Marshal(&cb)
-	err := pipelinefunc.CallbackActionFunc(cbData)
-	if err != nil {
-		log.Errorf("failed to callback, err: %v", err)
-		return
-	}
-
+	// apitest should ensure that callback to pipeline after doing request
+	_ = loop.New(loop.WithDeclineRatio(2), loop.WithDeclineLimit(10*time.Second)).
+		Do(func() (bool, error) {
+			err := pipelinefunc.CallbackActionFunc(cbData)
+			if err != nil {
+				log.Errorf("failed to callback, err: %v", err)
+				return false, err
+			}
+			return true, nil
+		})
 	return
 }

modules/pipeline/pipengine/reconciler/reconcile.go

@@ -104,13 +104,10 @@ func (r *Reconciler) reconcile(ctx context.Context, pipelineID uint64) error {
 			if err != nil {
 				return
 			}
-
-			executorChan := make(chan interface{})
-			ctx = context.WithValue(ctx, spec.MakeTaskExecutorCtxKey(task), executorChan)
 			tr := taskrun.New(ctx, task,
 				ctx.Value(ctxKeyPipelineExitCh).(chan struct{}), ctx.Value(ctxKeyPipelineExitChCancelFunc).(context.CancelFunc),
 				r.TaskThrottler, executor, &p, r.bdl, r.dbClient, r.js,
-				r.actionAgentSvc, r.extMarketSvc, executorChan)
+				r.actionAgentSvc, r.extMarketSvc)
 
 			// tear down task
 			defer func() {

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

@@ -53,7 +53,7 @@ type TaskRun struct {
 	PExitCh        <-chan struct{}
 	PExitChCancel  context.CancelFunc
 	PExit          bool
-	ExecutorDoneCh chan interface{} // executorDoneCh allow action executor return directly
+	ExecutorDoneCh chan interface{}
 
 	// 轮训状态间隔期间可能任务已经是终态，FakeTimeout = true
 	FakeTimeout bool
@@ -70,10 +70,11 @@ func New(ctx context.Context, task *spec.PipelineTask,
 	executor types.ActionExecutor, p *spec.Pipeline, bdl *bundle.Bundle, dbClient *dbclient.Client, js jsonstore.JsonStore,
 	actionAgentSvc *actionagentsvc.ActionAgentSvc,
 	extMarketSvc *extmarketsvc.ExtMarketSvc,
-	executorCh chan interface{},
 ) *TaskRun {
+	// make executor has buffer, don't block task framework
+	executorCH := make(chan interface{}, 1)
 	return &TaskRun{
-		Ctx:       ctx,
+		Ctx:       context.WithValue(ctx, spec.MakeTaskExecutorCtxKey(task), executorCH),
 		Task:      task,
 		Executor:  executor,
 		Throttler: throttler,
@@ -91,7 +92,7 @@ func New(ctx context.Context, task *spec.PipelineTask,
 
 		PExitCh:        pExitCh,
 		PExitChCancel:  pExitChCancel,
-		ExecutorDoneCh: executorCh,
+		ExecutorDoneCh: executorCH,
 
 		ActionAgentSvc: actionAgentSvc,
 		ExtMarketSvc:   extMarketSvc,