This is an automated email from the ASF dual-hosted git repository.
lidongdai pushed a commit to branch 1.3.6-prepare
in repository https://gitbox.apache.org/repos/asf/incubator-dolphinscheduler.git
The following commit(s) were added to refs/heads/1.3.6-prepare by this push:
new 92b406a [1.3.6-prepare][FIX-#4617][LoggerServer] task log can not
refresh in … (#5074)
92b406a is described below
commit 92b406a624159b3bdb0c4059dc700c3d4972ff74
Author: Kirs <[email protected]>
AuthorDate: Wed Mar 31 00:28:00 2021 +0800
[1.3.6-prepare][FIX-#4617][LoggerServer] task log can not refresh in …
(#5074)
* [1.3.6-prepare][FIX-#4617][LoggerServer] task log can not refresh in time
#4618
issue #4617
pr #4618
* code style
---
.../worker/task/AbstractCommandExecutor.java | 49 ++++--
.../worker/shell/ShellCommandExecutorTest.java | 184 +++++++++++++++++++--
2 files changed, 197 insertions(+), 36 deletions(-)
diff --git
a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/AbstractCommandExecutor.java
b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/AbstractCommandExecutor.java
index 6f8f3f9..c951349 100644
---
a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/AbstractCommandExecutor.java
+++
b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/AbstractCommandExecutor.java
@@ -84,6 +84,9 @@ public abstract class AbstractCommandExecutor {
*/
protected TaskExecutionContext taskExecutionContext;
+
+ protected boolean logOutputIsSuccess = false;
+
/**
* taskExecutionContextCacheManager
*/
@@ -349,28 +352,38 @@ public abstract class AbstractCommandExecutor {
*/
private void parseProcessOutput(Process process) {
String threadLoggerInfoName =
String.format(LoggerUtils.TASK_LOGGER_THREAD_NAME + "-%s",
taskExecutionContext.getTaskAppId());
+ ExecutorService getOutputLogService =
ThreadUtils.newDaemonSingleThreadExecutor(threadLoggerInfoName + "-" +
"getOutputLogService");
+ getOutputLogService.submit(() -> {
+ BufferedReader inReader = null;
+ try {
+ inReader = new BufferedReader(new
InputStreamReader(process.getInputStream()));
+ String line;
+ while ((line = inReader.readLine()) != null) {
+ logBuffer.add(line);
+ }
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ } finally {
+ logOutputIsSuccess = true;
+ close(inReader);
+ }
+ });
+ getOutputLogService.shutdown();
ExecutorService parseProcessOutputExecutorService =
ThreadUtils.newDaemonSingleThreadExecutor(threadLoggerInfoName);
- parseProcessOutputExecutorService.submit(new Runnable(){
- @Override
- public void run() {
- BufferedReader inReader = null;
-
- try {
- inReader = new BufferedReader(new
InputStreamReader(process.getInputStream()));
- String line;
-
- long lastFlushTime = System.currentTimeMillis();
-
- while ((line = inReader.readLine()) != null) {
- logBuffer.add(line);
+ parseProcessOutputExecutorService.submit(() -> {
+ try {
+ long lastFlushTime = System.currentTimeMillis();
+ while (logBuffer.size() > 0 || !logOutputIsSuccess) {
+ if (logBuffer.size() > 0) {
lastFlushTime = flush(lastFlushTime);
+ } else {
+ Thread.sleep(Constants.DEFAULT_LOG_FLUSH_INTERVAL);
}
- } catch (Exception e) {
- logger.error(e.getMessage(),e);
- } finally {
- clear();
- close(inReader);
}
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ } finally {
+ clear();
}
});
parseProcessOutputExecutorService.shutdown();
diff --git
a/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/shell/ShellCommandExecutorTest.java
b/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/shell/ShellCommandExecutorTest.java
index acc7a22..ff4eb48 100644
---
a/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/shell/ShellCommandExecutorTest.java
+++
b/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/shell/ShellCommandExecutorTest.java
@@ -16,40 +16,62 @@
*/
package org.apache.dolphinscheduler.server.worker.shell;
-import com.alibaba.fastjson.JSON;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.lang.reflect.Method;
+
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
-import org.apache.dolphinscheduler.common.model.TaskNode;
+import org.apache.dolphinscheduler.common.utils.OSUtils;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
-import org.apache.dolphinscheduler.common.utils.LoggerUtils;
+import org.apache.dolphinscheduler.server.entity.TaskExecutionContext;
+import org.apache.dolphinscheduler.server.worker.task.AbstractCommandExecutor;
import org.apache.dolphinscheduler.server.worker.task.AbstractTask;
-import org.apache.dolphinscheduler.server.worker.task.TaskManager;
import org.apache.dolphinscheduler.server.worker.task.TaskProps;
import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
import org.apache.dolphinscheduler.service.process.ProcessService;
+
+import org.junit.Assert;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.springframework.context.ApplicationContext;
import java.util.Date;
+import java.util.List;
/**
- * python shell command executor test
+ * python shell command executor test
*/
-@Ignore
+@RunWith(PowerMockRunner.class)
+@PrepareForTest(OSUtils.class)
+@PowerMockIgnore({"javax.management.*"})
public class ShellCommandExecutorTest {
private static final Logger logger =
LoggerFactory.getLogger(ShellCommandExecutorTest.class);
private ProcessService processService = null;
+ private ApplicationContext applicationContext;
@Before
- public void before(){
- processService =
SpringApplicationContext.getBean(ProcessService.class);
+ public void before() {
+ applicationContext = PowerMockito.mock(ApplicationContext.class);
+ processService = PowerMockito.mock(ProcessService.class);
+ SpringApplicationContext springApplicationContext = new
SpringApplicationContext();
+ springApplicationContext.setApplicationContext(applicationContext);
+
PowerMockito.when(applicationContext.getBean(ProcessService.class)).thenReturn(processService);
}
+ @Ignore
@Test
public void test() throws Exception {
@@ -64,19 +86,18 @@ public class ShellCommandExecutorTest {
taskProps.setTaskInstanceId(7657);
-
TaskInstance taskInstance = processService.findTaskInstanceById(7657);
String taskJson = taskInstance.getTaskJson();
- TaskNode taskNode = JSON.parseObject(taskJson, TaskNode.class);
- taskProps.setTaskParams(taskNode.getParams());
+// TaskNode taskNode = JSON.parseObject(taskJson, TaskNode.class);
+// taskProps.setTaskParams(taskNode.getParams());
// custom logger
- Logger taskLogger =
LoggerFactory.getLogger(LoggerUtils.buildTaskId(LoggerUtils.TASK_LOGGER_INFO_PREFIX,
- taskInstance.getProcessDefinitionId(),
- taskInstance.getProcessInstanceId(),
- taskInstance.getId()));
+// Logger taskLogger =
LoggerFactory.getLogger(LoggerUtils.buildTaskId(LoggerUtils.TASK_LOGGER_INFO_PREFIX,
+// taskInstance.getProcessDefinitionId(),
+// taskInstance.getProcessInstanceId(),
+// taskInstance.getId()));
// AbstractTask task = TaskManager.newTask(taskInstance.getTaskType(),
taskProps, taskLogger);
@@ -92,14 +113,141 @@ public class ShellCommandExecutorTest {
task.handle();
ExecutionStatus status = ExecutionStatus.SUCCESS;
- if (task.getExitStatusCode() == Constants.EXIT_CODE_SUCCESS){
+ if (task.getExitStatusCode() == Constants.EXIT_CODE_SUCCESS) {
status = ExecutionStatus.SUCCESS;
- }else if (task.getExitStatusCode() == Constants.EXIT_CODE_KILL){
+ } else if (task.getExitStatusCode() == Constants.EXIT_CODE_KILL) {
status = ExecutionStatus.KILL;
- }else {
+ } else {
status = ExecutionStatus.FAILURE;
}
logger.info(status.toString());
}
+
+ @Test
+ public void testParseProcessOutput() {
+ Class<AbstractCommandExecutor> shellCommandExecutorClass =
AbstractCommandExecutor.class;
+ try {
+
+ Method method =
shellCommandExecutorClass.getDeclaredMethod("parseProcessOutput",
Process.class);
+ method.setAccessible(true);
+ Object[] arg1s = {new Process() {
+ @Override
+ public OutputStream getOutputStream() {
+ return new OutputStream() {
+ @Override
+ public void write(int b) throws IOException {
+ logger.info("unit test");
+ }
+ };
+ }
+
+ @Override
+ public InputStream getInputStream() {
+ return new InputStream() {
+ @Override
+ public int read() throws IOException {
+ return 0;
+ }
+ };
+ }
+
+ @Override
+ public InputStream getErrorStream() {
+ return null;
+ }
+
+ @Override
+ public int waitFor() throws InterruptedException {
+ return 0;
+ }
+
+ @Override
+ public int exitValue() {
+ return 0;
+ }
+
+ @Override
+ public void destroy() {
+ logger.info("unit test");
+ }
+ } };
+ method.invoke(new AbstractCommandExecutor(null, new
TaskExecutionContext(), logger) {
+ @Override
+ protected String buildCommandFilePath() {
+ return null;
+ }
+
+ @Override
+ protected String commandInterpreter() {
+ return null;
+ }
+
+ @Override
+ protected void createCommandFileIfNotExists(String
execCommand, String commandFile) throws IOException {
+ logger.info("unit test");
+ }
+ }, arg1s);
+ } catch (Exception e) {
+ logger.error(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testFindAppId() {
+ Class<AbstractCommandExecutor> shellCommandExecutorClass =
AbstractCommandExecutor.class;
+ try {
+
+ Method method =
shellCommandExecutorClass.getDeclaredMethod("findAppId", new
Class[]{String.class});
+ method.setAccessible(true);
+ Object[] arg1s = {"11111"};
+ String result = (String) method.invoke(new
AbstractCommandExecutor(null, null, null) {
+ @Override
+ protected String buildCommandFilePath() {
+ return null;
+ }
+
+ @Override
+ protected String commandInterpreter() {
+ return null;
+ }
+
+ @Override
+ protected void createCommandFileIfNotExists(String
execCommand, String commandFile) throws IOException {
+ logger.info("unit test");
+ }
+ }, arg1s);
+ } catch (Exception e) {
+ logger.error(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testConvertFile2List() {
+ Class<AbstractCommandExecutor> shellCommandExecutorClass =
AbstractCommandExecutor.class;
+ try {
+ Method method =
shellCommandExecutorClass.getDeclaredMethod("convertFile2List", String.class);
+ method.setAccessible(true);
+ Object[] arg1s = {"/opt/1.txt"};
+ List<String> result = (List<String>) method.invoke(new
AbstractCommandExecutor(null, null, null) {
+ @Override
+ protected String buildCommandFilePath() {
+ return null;
+ }
+
+ @Override
+ protected String commandInterpreter() {
+ return null;
+ }
+
+ @Override
+ protected void createCommandFileIfNotExists(String
execCommand, String commandFile) throws IOException {
+ logger.info("unit test");
+ }
+ }, arg1s);
+ Assert.assertTrue(true);
+ } catch (Exception e) {
+ logger.error(e.getMessage());
+ }
+ }
}
\ No newline at end of file
