debug run one task when submit

weibocom · Oct 14, 2024 · 3f9d73f · 3f9d73f
1 parent 0353b75
commit 3f9d73f
Show file tree

Hide file tree

Showing 10 changed files with 35 additions and 20 deletions.
diff --git a/...picene-core/src/main/java/com/weibo/rill/flow/olympicene/core/runtime/DAGInteraction.java b/...picene-core/src/main/java/com/weibo/rill/flow/olympicene/core/runtime/DAGInteraction.java
@@ -32,7 +32,7 @@ public interface DAGInteraction {
     /**
      * 提交需执行的DAG任务
      */
-    void submit(String executionId, DAG dag, Map<String, Object> data, DAGSettings settings, NotifyInfo notifyInfo);
+    void submit(String executionId, String taskName, DAG dag, Map<String, Object> data, DAGSettings settings, NotifyInfo notifyInfo);
 
     /**
      * 完成task后调用接口

diff --git a/...icene-traversal/src/main/java/com/weibo/rill/flow/olympicene/traversal/DAGOperations.java b/...icene-traversal/src/main/java/com/weibo/rill/flow/olympicene/traversal/DAGOperations.java
@@ -231,9 +231,9 @@ public void redoTask(String executionId, List<String> taskNames, Map<String, Obj
         dagTraversal.submitTraversal(executionId, null);
     }
 
-    public void submitDAG(String executionId, DAG dag, DAGSettings settings, Map<String, Object> data, NotifyInfo notifyInfo) {
+    public void submitDAG(String executionId, String taskName, DAG dag, DAGSettings settings, Map<String, Object> data, NotifyInfo notifyInfo) {
         log.info("submitDAG task begin to execute executionId:{} notifyInfo:{}", executionId, notifyInfo);
-        ExecutionResult executionResult = dagRunner.submitDAG(executionId, dag, settings, data, notifyInfo);
+        ExecutionResult executionResult = dagRunner.submitDAG(executionId, taskName, dag, settings, data, notifyInfo);
         Optional.ofNullable(getTimeoutSeconds(new HashMap<>(), executionResult.getContext(), dag.getTimeline()))
                 .ifPresent(timeoutSeconds -> timeCheckRunner.addDAGToTimeoutCheck(executionId, timeoutSeconds));
         dagTraversal.submitTraversal(executionId, null);

diff --git a/...ympicene-traversal/src/main/java/com/weibo/rill/flow/olympicene/traversal/Olympicene.java b/...ympicene-traversal/src/main/java/com/weibo/rill/flow/olympicene/traversal/Olympicene.java
@@ -60,16 +60,16 @@ public Olympicene(DAGInfoStorage dagInfoStorage, DAGOperations dagOperations, Ex
      * 提交需执行的DAG任务
      */
     public void submit(String executionId, DAG dag, Map<String, Object> data) {
-        submit(executionId, dag, data, DAGSettings.DEFAULT, null);
+        submit(executionId, null, dag, data, DAGSettings.DEFAULT, null);
     }
 
     /**
      * 提交需执行的DAG任务
      */
     @Override
-    public void submit(String executionId, DAG dag, Map<String, Object> data, DAGSettings settings, NotifyInfo notifyInfo) {
+    public void submit(String executionId, String taskName, DAG dag, Map<String, Object> data, DAGSettings settings, NotifyInfo notifyInfo) {
         runNotify(executionId, NotifyType.SUBMIT, notifyInfo,
-                () -> dagOperations.submitDAG(executionId, dag, settings, data, notifyInfo));
+                () -> dagOperations.submitDAG(executionId, taskName, dag, settings, data, notifyInfo));
     }
 
     public void runNotify(String executionId, NotifyType notifyType, NotifyInfo notifyInfo, Runnable actions) {
@@ -186,7 +186,7 @@ public DAGResult run(String executionId, DAG dag, Map<String, Object> data, DAGS
 
         dagResultHandler.initEnv(executionId);
         doRunNotify(executionId, NotifyType.RUN, notifyInfo,
-                () -> dagOperations.submitDAG(executionId, dag, settings, data, notifyInfo));
+                () -> dagOperations.submitDAG(executionId, null, dag, settings, data, notifyInfo));
         return dagResultHandler.getDAGResult(executionId, timeoutInMillisecond);
     }
 }
diff --git a/...e-traversal/src/main/java/com/weibo/rill/flow/olympicene/traversal/runners/DAGRunner.java b/...e-traversal/src/main/java/com/weibo/rill/flow/olympicene/traversal/runners/DAGRunner.java
@@ -67,8 +67,8 @@ public DAGRunner(DAGContextStorage dagContextStorage, DAGInfoStorage dagInfoStor
         this.dagStorageProcedure = dagStorageProcedure;
     }
 
-    public ExecutionResult submitDAG(String executionId, DAG dag, DAGSettings settings, Map<String, Object> data, NotifyInfo notifyInfo) {
-        ExecutionResult ret = ExecutionResult.builder().build();
+    public ExecutionResult submitDAG(String executionId, String taskName, DAG dag, DAGSettings settings, Map<String, Object> data, NotifyInfo notifyInfo) {
+        ExecutionResult ret = ExecutionResult.builder().needRetry(false).retryIntervalInSeconds(0).build();
 
         dagStorageProcedure.lockAndRun(LockerKey.buildDagInfoLockName(executionId), () -> {
             DAGInfo currentExecutionIdDagInfo = dagInfoStorage.getBasicDAGInfo(executionId);
@@ -104,6 +104,15 @@ public ExecutionResult submitDAG(String executionId, DAG dag, DAGSettings settin
                     .dagInvokeMsg(dagInvokeMsg)
                     .dagStatus(DAGStatus.RUNNING)
                     .make();
+
+            if (MapUtils.isNotEmpty(dagInfoToUpdate.getTasks()) && taskName != null) {
+                for (Map.Entry<String, TaskInfo> taskInfoEntry : dagInfoToUpdate.getTasks().entrySet()) {
+                    if (!taskInfoEntry.getKey().equals(taskName)) {
+                        taskInfoEntry.getValue().setTaskStatus(TaskStatus.SKIPPED);
+                    }
+                }
+            }
+
             ret.setDagInfo(dagInfoToUpdate);
             Optional.ofNullable(dagInvokeMsg)
                     .map(DAGInvokeMsg::getExecutionRoutes)
@@ -259,7 +268,7 @@ public ExecutionResult finishDAG(String executionId, DAGInfo dagInfo, DAGStatus
         }
 
         log.info("finishDAG finish, executionId:{}", executionId);
-        return ExecutionResult.builder().dagInfo(wholeDagInfo).context(context).build();
+        return ExecutionResult.builder().dagInfo(wholeDagInfo).context(context).needRetry(false).retryIntervalInSeconds(0).build();
     }
 
     private void updateDAGInvokeStartTime(DAGInfo dagInfo) {

diff --git a/...e-traversal/src/test/groovy/com/weibo/rill/flow/olympicene/traversal/InvokeMsgTest.groovy b/...e-traversal/src/test/groovy/com/weibo/rill/flow/olympicene/traversal/InvokeMsgTest.groovy
@@ -83,7 +83,7 @@ class InvokeMsgTest extends Specification {
                 .parentDAGTaskInfoName("A")
                 .parentDAGTaskExecutionType(FunctionPattern.FLOW_SYNC)
                 .build()
-        olympicene.submit("smallFlow", smallFlow, [:], DAGSettings.DEFAULT, smallFlowSubmit)
+        olympicene.submit("smallFlow", null, smallFlow, [:], DAGSettings.DEFAULT, smallFlowSubmit)
 
         NotifyInfo smallFlowNotify = NotifyInfo.builder()
                 .taskInfoName("B_0-C")

diff --git a/...ne-traversal/src/test/groovy/com/weibo/rill/flow/olympicene/traversal/MultiDAGTest.groovy b/...ne-traversal/src/test/groovy/com/weibo/rill/flow/olympicene/traversal/MultiDAGTest.groovy
@@ -66,7 +66,7 @@ class MultiDAGTest extends Specification {
         NotifyInfo notifyInfo = NotifyInfo.builder()
                 .parentDAGExecutionId("level1")
                 .parentDAGTaskInfoName("A").build()
-        olympicene.submit("level2", dag, [:], DAGSettings.DEFAULT, notifyInfo)
+        olympicene.submit("level2", null, dag, [:], DAGSettings.DEFAULT, notifyInfo)
         TaskInfo level1TaskA = dagStorage.getDAGInfo("level1").getTask("A")
         DAGInfo level2 = dagStorage.getDAGInfo("level2")
 
@@ -87,12 +87,12 @@ class MultiDAGTest extends Specification {
         NotifyInfo notifyInfoLevel2 = NotifyInfo.builder()
                 .parentDAGExecutionId("level1")
                 .parentDAGTaskInfoName("A").build()
-        olympicene.submit("level2", dag, [:], dagSettings, notifyInfoLevel2)
+        olympicene.submit("level2", null, dag, [:], dagSettings, notifyInfoLevel2)
 
         NotifyInfo notifyInfoLevel3 = NotifyInfo.builder()
                 .parentDAGExecutionId("level2")
                 .parentDAGTaskInfoName("A").build()
-        olympicene.submit("level3", dag, [:], dagSettings, notifyInfoLevel3)
+        olympicene.submit("level3", null, dag, [:], dagSettings, notifyInfoLevel3)
 
         then:
         def e = thrown(DAGTraversalException)

diff --git a/...-service/src/main/java/com/weibo/rill/flow/service/dispatcher/FlowProtocolDispatcher.java b/...-service/src/main/java/com/weibo/rill/flow/service/dispatcher/FlowProtocolDispatcher.java
@@ -77,7 +77,7 @@ public String handle(Resource resource, DispatchInfo dispatchInfo) {
         DAGSettings dagSettings = DAGSettings.builder()
                 .ignoreExist(false)
                 .dagMaxDepth(bizDConfs.getFlowDAGMaxDepth()).build();
-        olympicene.submit(executionId, dag, data, dagSettings, notifyInfo);
+        olympicene.submit(executionId, null, dag, data, dagSettings, notifyInfo);
         dagResourceStatistic.updateFlowTypeResourceStatus(parentDAGExecutionId, parentTaskName, resource.getResourceName(), dag);
         ProfileActions.recordTinyDAGSubmit(executionId);
         // 记录prometheus

diff --git a/rill-flow-service/src/main/java/com/weibo/rill/flow/service/facade/OlympiceneFacade.java b/rill-flow-service/src/main/java/com/weibo/rill/flow/service/facade/OlympiceneFacade.java
@@ -107,18 +107,23 @@ public Map<String, Object> submit(Long uid, String descriptorId, String callback
             String businessId = DescriptorIdUtil.changeDescriptorIdToBusinessId(descriptorId);
             Map<String, Object> context = dagContextInitializer.newSubmitContextBuilder(businessId).withData(data).withIdentity(descriptorId).build();
 
-            return submit(uid, descriptorId, context, callback, resourceCheckConfig);
+            return submit(uid, descriptorId, null, context, callback, resourceCheckConfig);
         };
 
         return profileRecordService.runNotifyAndRecordProfile(url, descriptorId, submitActions);
     }
 
 
     public Map<String, Object> submit(User flowUser, String descriptorId, Map<String, Object> context, String callback, ResourceCheckConfig resourceCheckConfig) {
-        return submit(Optional.ofNullable(flowUser).map(User::getUid).orElse(0L), descriptorId, context, callback, resourceCheckConfig);
+        return submit(Optional.ofNullable(flowUser).map(User::getUid).orElse(0L), descriptorId, null, context, callback, resourceCheckConfig);
     }
 
-    public Map<String, Object> submit(Long uid, String descriptorId, Map<String, Object> context, String callback, ResourceCheckConfig resourceCheckConfig) {
+
+    public Map<String, Object> submit(User flowUser, String descriptorId, Map<String, Object> context, String callback, ResourceCheckConfig resourceCheckConfig, String taskName) {
+        return submit(Optional.ofNullable(flowUser).map(User::getUid).orElse(0L), descriptorId, taskName, context, callback, resourceCheckConfig);
+    }
+
+    public Map<String, Object> submit(Long uid, String descriptorId, String taskName, Map<String, Object> context, String callback, ResourceCheckConfig resourceCheckConfig) {
         String dagDescriptor = descriptorManager.getDagDescriptor(uid, context, descriptorId);
         DAG dag = dagStringParser.parse(dagDescriptor);
         String executionId = ExecutionIdUtil.generateExecutionId(dag);
@@ -132,7 +137,7 @@ public Map<String, Object> submit(Long uid, String descriptorId, Map<String, Obj
                     .build();
         }
         context.put("flow_execution_id", executionId);
-        olympicene.submit(executionId, dag, context, DAGSettings.DEFAULT, notifyInfo);
+        olympicene.submit(executionId, taskName, dag, context, DAGSettings.DEFAULT, notifyInfo);
         Map<String, Object> ret = Maps.newHashMap();
         ret.put("execution_id", executionId);
         return ret;

diff --git a/...ow-service/src/test/groovy/com/weibo/rill/flow/service/facade/OlympiceneFacadeTest.groovy b/...ow-service/src/test/groovy/com/weibo/rill/flow/service/facade/OlympiceneFacadeTest.groovy
@@ -74,7 +74,7 @@ class OlympiceneFacadeTest extends Specification {
         user.getUid() >> 1L
         expect:
         facade.submit(1L, "testBusiness:testFeatureName", new JSONObject(["resourceName": "testCallbackUrl"]).toJSONString(), null, new JSONObject(["a": 1]), null)
-        facade.submit(1L, "testBusiness:testFeatureName", ["resourceName": "testCallbackUrl"], null, null)
+        facade.submit(1L, "testBusiness:testFeatureName", null, ["resourceName": "testCallbackUrl"], null, null)
         facade.submit(user, "testBusiness:testFeatureName", ["resourceName": "testCallbackUrl"], null, null)
     }
 

diff --git a/rill-flow-web/src/main/java/com/weibo/rill/flow/controller/FlowController.java b/rill-flow-web/src/main/java/com/weibo/rill/flow/controller/FlowController.java
@@ -95,6 +95,7 @@ public class FlowController {
     @RequestMapping(value = "submit.json", method = RequestMethod.POST)
     public Map<String, Object> submit(User flowUser,
                                       @ApiParam(value = "工作流ID") @RequestParam(value = "descriptor_id") String descriptorId,
+                                      @ApiParam(value = "单步执行任务名称") @RequestParam(value = "task_name", required = false) String taskName,
                                       @ApiParam(value = "执行完成后的回调地址") @RequestParam(value = "callback", required = false) String callback,
                                       @ApiParam(value = "用于检测资源是否可用的检测规则") @RequestParam(value = "resource_check", required = false) String resourceCheck,
                                       @ApiParam(value = "工作流执行的context信息") @RequestBody(required = false) JSONObject data) {