You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dolphinscheduler.apache.org by "Radeity (via GitHub)" <gi...@apache.org> on 2023/09/25 05:06:18 UTC
[GitHub] [dolphinscheduler] Radeity commented on a diff in pull request #14569: [Feature][lmprovement] Master/Worker restore fault tolerant optimization
Radeity commented on code in PR #14569:
URL: https://github.com/apache/dolphinscheduler/pull/14569#discussion_r1335370167
##########
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnable.java:
##########
@@ -2248,4 +2269,118 @@ private void mergeTaskInstanceVarPool(TaskInstance taskInstance) {
workflowInstance.setVarPool(JSONUtils.toJsonString(processVarPool));
}
+
+ /**
+ * determines whether the process is running
+ *
+ * @param pid process_id
+ * @return boolean
+ */
+ public boolean isProcessRunning(String pid) throws Exception {
Review Comment:
May I ask why you pass `processPath` here? Also, for many types of tasks, `executePath` is generated on `Worker`, you can not use this path in Master.
##########
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnable.java:
##########
@@ -2248,4 +2269,118 @@ private void mergeTaskInstanceVarPool(TaskInstance taskInstance) {
workflowInstance.setVarPool(JSONUtils.toJsonString(processVarPool));
}
+
+ /**
+ * determines whether the process is running
+ *
+ * @param pid process_id
+ * @return boolean
+ */
+ public boolean isProcessRunning(String pid) throws Exception {
+
+ String processPath = pid;
+
+ // build shell commands, use ps-ef to list all processes, and use GREP filters to match the process id
+ String command = "/bin/sh -c \"ps -ef | grep " + processPath + " | grep -v grep\"";
+
+ // use the ProcessBuilder class to create a new process and set its command
+ ProcessBuilder processBuilder = new ProcessBuilder("/bin/sh", "-c", command);
+
+ // start the process
+ Process process = processBuilder.start();
+
+ // reads the output of the command and gets the standard input stream for the process
+ BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()));
+
+ // reads each line in the process input stream
+ String line;
+ while ((line = reader.readLine()) != null) {
+ if (line.contains(processPath)) {
+ return true;
+ }
+ }
+ try {
+ // gets the exit code for the command.
+ int exitCode = process.waitFor();
+ // if the exit code is 0, and the output contains process id, the process exists.
+ return exitCode == 0;
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ return false;
+ }
+
+ /**
+ * determine if the yarn task is running
+ *
+ * @param applicationId applicationId
+ * @return boolean
+ */
+ public boolean isApplicationRunning(String applicationId ) throws Exception {
+
+ // build shell commands, use yarn application -status applicationId
+ String[] command = {"/bin/bash", "-c", "yarn application -status " + applicationId};
+
+ // use the ProcessBuilder class to create a new process and set its command
+ ProcessBuilder processBuilder = new ProcessBuilder(command);
+
+ // start the process
+ Process process = processBuilder.start();
+
+ // reads the output of the command and gets the standard input stream for the process
+ BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()));
+
+ String line;
+ String state = null;
+ // reads each line in the process input stream
+ while ((line = reader.readLine()) != null) {
+ // get the yarn task running status
+ if (line.contains("State")) {
+ state = line.split(":")[1].trim();
+ break;
+ }
+ }
+ try {
+ // gets the exit code for the command.
+ int exitCode = process.waitFor();
+ // if the exit code is 0, and the yarn task is running
+ if (exitCode == 0 && state.equals("RUNNING")) {
+ // process is running
+ return true;
+ } else {
+ return false;
+ }
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ return false;
+ }
+ public boolean determineWhetherTaskIsRunning(TaskInstance task) throws Exception {
+ String pidPath = null;
+ if(task.getExecutePath() != null){
+ pidPath = task.getExecutePath();
+
+ }
+ String pid = getDesiredPath(pidPath);
+
+ // Yarn task is determined by parsing whether the task log contains the content of the application
+ String applicationId = task.getAppLink();
Review Comment:
Currently, DS does not implement methods to split submit and track remote application, you can check `AbstractRemoteTask.handle()`, so we can not get appId by `task.getAppLink()`. Maybe you have to figure out how to take-over remote application effectively.
##########
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnable.java:
##########
@@ -825,6 +825,27 @@ private void initTaskQueue() throws StateEventHandleException, CronParseExceptio
List<TaskInstance> validTaskInstanceList =
taskInstanceDao.queryValidTaskListByWorkflowInstanceId(workflowInstance.getId(),
workflowInstance.getTestFlag());
+ // the task instance needs fault tolerance
+ if(workflowInstance.getRecovery() == Flag.YES){
+ //determine whether the task is running
+ for (int i = 0; i < validTaskInstanceList.size(); i++) {
+ TaskInstance task = validTaskInstanceList.get(i);
+ boolean isRunningTaskFaultTolerance = false;
+ try {
+ isRunningTaskFaultTolerance = determineWhetherTaskIsRunning(task);
Review Comment:
Agree with @ruanwenjun, it's better `TAKE-OVER` the running application, need some detailed design, such as modify `workflowInstanceHost` on Worker.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@dolphinscheduler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org