feat(runner): include processed record count in subtask finish logs

Author: lrf-nitro

backend/core/plugin/plugin_task.go

@@ -53,6 +53,7 @@ type ExecContext interface {
 	GetData() interface{}
 	SetProgress(current int, total int)
 	IncProgress(quantity int)
+	GetProgress() int
 }
 
 // SubTaskContext This interface define all resources that needed for subtask execution

backend/core/runner/run_task.go

@@ -336,7 +336,7 @@ func RunPluginSubTasks(
 			logger.Info("executing subtask %s", subtaskMeta.Name)
 			start := time.Now()
 			err = runSubtask(basicRes, subtaskCtx, task.ID, subtaskNumber, subtaskMeta.EntryPoint)
-			logger.Info("subtask %s finished in %d ms", subtaskMeta.Name, time.Since(start).Milliseconds())
+			logger.Info("subtask %s finished in %d ms, records processed: %d", subtaskMeta.Name, time.Since(start).Milliseconds(), subtaskCtx.GetProgress())
 			if err != nil {
 				err = errors.SubtaskErr.Wrap(err, fmt.Sprintf("subtask %s ended unexpectedly", subtaskMeta.Name), errors.WithData(&subtaskMeta))
 				logger.Error(err, "")

backend/impls/context/default_exec_context.go

@@ -96,6 +96,10 @@ func (c *defaultExecContext) IncProgress(progressType plugin.ProgressType, quant
 	}
 }
 
+func (c *defaultExecContext) GetProgress() int {
+	return int(atomic.LoadInt64(&c.current))
+}
+
 func (c *defaultExecContext) fork(name string) *defaultExecContext {
 	return newDefaultExecContext(
 		c.ctx,

backend/server/services/remote/bridge/context.go

@@ -138,4 +138,11 @@ func (r remoteContextImpl) IncProgress(quantity int) {
 	}
 }
 
+func (r remoteContextImpl) GetProgress() int {
+	if r.parent != nil {
+		return r.parent.GetProgress()
+	}
+	return 0
+}
+
 var _ RemoteContext = (*remoteContextImpl)(nil)