[Improvement][Log] optimize task log path to reduce task running time (#6731)

* [Improvement][Log] optimize task log path to reduce task running time
* Reduce the creation of log directories


Co-authored-by: Kirs <acm_master@163.com>

dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/Constants.java

@@ -184,6 +184,10 @@ public final class Constants {
      */
     public static final String YYYY_MM_DD_HH_MM_SS = "yyyy-MM-dd HH:mm:ss";
 
+    /**
+     * date format of yyyyMMdd
+     */
+    public static final String YYYYMMDD = "yyyyMMdd";
 
     /**
      * date format of yyyyMMddHHmmss

dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/LoggerUtils.java

@@ -24,6 +24,7 @@ import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.InputStreamReader;
 import java.util.ArrayList;
+import java.util.Date;
 import java.util.List;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -71,12 +72,14 @@ public class LoggerUtils {
      * @return task id format
      */
     public static String buildTaskId(String affix,
+                                     Date firstSubmitTime,
                                      Long processDefineCode,
                                      int processDefineVersion,
                                      int processInstId,
                                      int taskId) {
-        // - [taskAppId=TASK-798_1-4084-15210]
-        return String.format(" - %s%s-%s_%s-%s-%s]", TASK_APPID_LOG_FORMAT, affix, processDefineCode, processDefineVersion, processInstId, taskId);
+        // - [taskAppId=TASK-20211107-798_1-4084-15210]
+        String firstSubmitTimeStr = DateUtils.format(firstSubmitTime, Constants.YYYYMMDD);
+        return String.format(" - %s%s-%s-%s_%s-%s-%s]", TASK_APPID_LOG_FORMAT, affix, firstSubmitTimeStr, processDefineCode, processDefineVersion, processInstId, taskId);
     }
 
     /**

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/log/TaskLogDiscriminator.java

@@ -47,7 +47,7 @@ public class TaskLogDiscriminator extends AbstractDiscriminator<ILoggingEvent> {
         String prefix = LoggerUtils.TASK_LOGGER_INFO_PREFIX + "-";
         if (loggerName.startsWith(prefix)) {
             return loggerName.substring(prefix.length(),
-                    loggerName.length() - 1).replace("-","/");
+                    loggerName.length() - 1).replaceFirst("-","/");
         } else {
             return "unknown_task";
         }

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/ConditionTaskProcessor.java

@@ -82,6 +82,7 @@ public class ConditionTaskProcessor extends BaseTaskProcessor {
         );
 
         logger = LoggerFactory.getLogger(LoggerUtils.buildTaskId(LoggerUtils.TASK_LOGGER_INFO_PREFIX,
+                taskInstance.getFirstSubmitTime(),
                 processInstance.getProcessDefinitionCode(),
                 processInstance.getProcessDefinitionVersion(),
                 taskInstance.getProcessInstanceId(),
@@ -144,7 +145,7 @@ public class ConditionTaskProcessor extends BaseTaskProcessor {
     }
 
     private void initTaskParameters() {
-        taskInstance.setLogPath(LogUtils.getTaskLogPath(processInstance.getProcessDefinitionCode(),
+        taskInstance.setLogPath(LogUtils.getTaskLogPath(taskInstance.getFirstSubmitTime(),processInstance.getProcessDefinitionCode(),
                 processInstance.getProcessDefinitionVersion(),
                 taskInstance.getProcessInstanceId(),
                 taskInstance.getId()));

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/DependentTaskProcessor.java

@@ -89,7 +89,8 @@ public class DependentTaskProcessor extends BaseTaskProcessor {
         taskDefinition = processService.findTaskDefinition(
                 taskInstance.getTaskCode(), taskInstance.getTaskDefinitionVersion()
         );
-        taskInstance.setLogPath(LogUtils.getTaskLogPath(processInstance.getProcessDefinitionCode(),
+        taskInstance.setLogPath(LogUtils.getTaskLogPath(taskInstance.getFirstSubmitTime(),
+                processInstance.getProcessDefinitionCode(),
                 processInstance.getProcessDefinitionVersion(),
                 taskInstance.getProcessInstanceId(),
                 taskInstance.getId()));

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/SwitchTaskProcessor.java

@@ -71,7 +71,7 @@ public class SwitchTaskProcessor extends BaseTaskProcessor {
         taskDefinition = processService.findTaskDefinition(
                 taskInstance.getTaskCode(), taskInstance.getTaskDefinitionVersion()
         );
-        taskInstance.setLogPath(LogUtils.getTaskLogPath(processInstance.getProcessDefinitionCode(),
+        taskInstance.setLogPath(LogUtils.getTaskLogPath(taskInstance.getFirstSubmitTime(),processInstance.getProcessDefinitionCode(),
                 processInstance.getProcessDefinitionVersion(),
                 taskInstance.getProcessInstanceId(),
                 taskInstance.getId()));

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/LogUtils.java

@@ -17,11 +17,14 @@
 
 package org.apache.dolphinscheduler.server.utils;
 
+import org.apache.dolphinscheduler.common.Constants;
+import org.apache.dolphinscheduler.common.utils.DateUtils;
 import org.apache.dolphinscheduler.server.log.TaskLogDiscriminator;
 import org.apache.dolphinscheduler.service.queue.entity.TaskExecutionContext;
 
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.util.Date;
 import java.util.Optional;
 
 import org.slf4j.LoggerFactory;
@@ -32,6 +35,8 @@ import ch.qos.logback.core.spi.AppenderAttachable;
 
 public class LogUtils {
 
+    public static final String LOG_TAILFIX = ".log";
+
     private LogUtils() throws IllegalStateException {
         throw new IllegalStateException("Utility class");
     }
@@ -39,7 +44,17 @@ public class LogUtils {
     /**
      * get task log path
      */
-    public static String getTaskLogPath(Long processDefineCode, int processDefineVersion, int processInstanceId, int taskInstanceId) {
+    public static String getTaskLogPath(Date firstSubmitTime, Long processDefineCode, int processDefineVersion, int processInstanceId, int taskInstanceId) {
+        // format /logs/YYYYMMDD/defintion-code_defintion_version-processInstanceId-taskInstanceId.log
+        final String taskLogFileName = new StringBuilder(String.valueOf(processDefineCode))
+            .append(Constants.UNDERLINE)
+            .append(processDefineVersion)
+            .append(Constants.SUBTRACT_CHAR)
+            .append(processInstanceId)
+            .append(Constants.SUBTRACT_CHAR)
+            .append(taskInstanceId)
+            .append(LOG_TAILFIX)
+            .toString();
         // Optional.map will be skipped if null
         return Optional.of(LoggerFactory.getILoggerFactory())
             .map(e -> (AppenderAttachable<ILoggingEvent>) (e.getLogger("ROOT")))
@@ -48,9 +63,8 @@ public class LogUtils {
             .map(TaskLogDiscriminator::getLogBase)
             .map(e -> Paths.get(e)
                 .toAbsolutePath()
-                        .resolve(processDefineCode + "_" + processDefineVersion)
-                        .resolve(String.valueOf(processInstanceId))
-                        .resolve(taskInstanceId + ".log"))
+                .resolve(DateUtils.format(firstSubmitTime,Constants.YYYYMMDD))
+                .resolve(taskLogFileName))
             .map(Path::toString)
             .orElse("");
     }
@@ -59,7 +73,7 @@ public class LogUtils {
      * get task log path by TaskExecutionContext
      */
     public static String getTaskLogPath(TaskExecutionContext taskExecutionContext) {
-        return getTaskLogPath(taskExecutionContext.getProcessDefineCode(),
+        return getTaskLogPath(taskExecutionContext.getFirstSubmitTime(),taskExecutionContext.getProcessDefineCode(),
             taskExecutionContext.getProcessDefineVersion(),
             taskExecutionContext.getProcessInstanceId(),
             taskExecutionContext.getTaskInstanceId());

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/runner/TaskExecuteThread.java

@@ -169,6 +169,7 @@ public class TaskExecuteThread implements Runnable, Delayed {
             }
             TaskRequest taskRequest = JSONUtils.parseObject(JSONUtils.toJsonString(taskExecutionContext), TaskRequest.class);
             String taskLogName = LoggerUtils.buildTaskId(LoggerUtils.TASK_LOGGER_INFO_PREFIX,
+                    taskExecutionContext.getFirstSubmitTime(),
                     taskExecutionContext.getProcessDefineCode(),
                     taskExecutionContext.getProcessDefineVersion(),
                     taskExecutionContext.getProcessInstanceId(),

dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/log/TaskLogDiscriminatorTest.java

@@ -121,7 +121,7 @@ public class TaskLogDiscriminatorTest {
 
             }
         });
-        Assert.assertEquals("1/1/", result);
+        Assert.assertEquals("1/1-", result);
     }
 
     @Test

dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/utils/LogUtilsTest.java

@@ -17,11 +17,14 @@
 
 package org.apache.dolphinscheduler.server.utils;
 
+import org.apache.dolphinscheduler.common.Constants;
+import org.apache.dolphinscheduler.common.utils.DateUtils;
 import org.apache.dolphinscheduler.server.log.TaskLogDiscriminator;
 import org.apache.dolphinscheduler.service.queue.entity.TaskExecutionContext;
 
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.util.Date;
 
 import org.junit.Assert;
 import org.junit.Test;
@@ -38,11 +41,13 @@ public class LogUtilsTest {
 
     @Test
     public void testGetTaskLogPath() {
+        Date firstSubmitTime = new Date();
         TaskExecutionContext taskExecutionContext = new TaskExecutionContext();
         taskExecutionContext.setProcessInstanceId(100);
         taskExecutionContext.setTaskInstanceId(1000);
         taskExecutionContext.setProcessDefineCode(1L);
         taskExecutionContext.setProcessDefineVersion(1);
+        taskExecutionContext.setFirstSubmitTime(firstSubmitTime);
 
         Logger rootLogger = (Logger) LoggerFactory.getILoggerFactory().getLogger("ROOT");
         Assert.assertNotNull(rootLogger);
@@ -60,7 +65,8 @@ public class LogUtilsTest {
 
         Path logPath = Paths.get(".").toAbsolutePath().getParent()
                 .resolve(logBase)
-                .resolve("1_1").resolve("100").resolve("1000.log");
+                .resolve(DateUtils.format(firstSubmitTime, Constants.YYYYMMDD))
+                .resolve("1_1-100-1000.log");
         Assert.assertEquals(logPath.toString(), LogUtils.getTaskLogPath(taskExecutionContext));
     }
 

dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/processor/TaskExecuteProcessorTest.java

@@ -108,6 +108,7 @@ public class TaskExecuteProcessorTest {
                 .thenReturn(workerConfig);
 
         Logger taskLogger = LoggerFactory.getLogger(LoggerUtils.buildTaskId(LoggerUtils.TASK_LOGGER_INFO_PREFIX,
+                taskExecutionContext.getFirstSubmitTime(),
                 taskExecutionContext.getProcessDefineCode(),
                 taskExecutionContext.getProcessDefineVersion(),
                 taskExecutionContext.getProcessInstanceId(),