[Feature-17787] Support config max task log size for per task

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/logging/LogClientDelegate.java

@@ -78,7 +78,15 @@ public byte[] getWholeLogBytes(TaskInstance taskInstance) {
         if (checkNodeExists(taskInstance)) {
             TaskInstanceLogFileDownloadResponse response = localLogClient.getWholeLog(taskInstance);
             if (response.getCode() == LogResponseStatus.SUCCESS) {
-                return response.getLogBytes();
+                // For local logs, also get rolling log files if they exist
+                String logPath = taskInstance.getLogPath();
+                java.io.File logFile = new java.io.File(logPath);
+                if (logFile.exists()) {
+                    return org.apache.dolphinscheduler.common.utils.LogUtils
+                            .getFileContentBytesWithRollingLogs(logPath);
+                } else {
+                    return response.getLogBytes();
+                }
             } else {
                 log.warn("get whole log bytes is not success for task instance {}; reason :{}", taskInstance.getId(),
                         response.getMessage());

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/logging/RemoteLogClient.java

@@ -33,7 +33,7 @@ public class RemoteLogClient {
      * @return Returns the log content in byte array format.
      */
     public byte[] getWholeLog(TaskInstance taskInstance) {
-        return LogUtils.getFileContentBytesFromRemote(taskInstance.getLogPath());
+        return LogUtils.getFileContentBytesWithRollingLogs(taskInstance.getLogPath());
     }
 
     /**

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

@@ -29,7 +29,10 @@
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
+import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -79,11 +82,21 @@ public static List<String> readPartFileContentFromLocal(String filePath,
                                                             int limit) {
         File file = new File(filePath);
         if (file.exists() && file.isFile()) {
-            try (Stream<String> stream = Files.lines(Paths.get(filePath))) {
-                return stream.skip(skipLine).limit(limit).collect(Collectors.toList());
-            } catch (IOException e) {
-                log.error("read file error", e);
-                throw new RuntimeException(String.format("Read file: %s error", filePath), e);
+            log.info("readPartFileContentFromLocal Reading log file");
+            // Check if there are rolling log files
+            List<File> logFiles = getRollingLogFiles(filePath);
+
+            if (logFiles.size() > 1) {
+                // Handle rolling log files
+                return readFromRollingLogFiles(logFiles, skipLine, limit);
+            } else {
+                // Handle single log file
+                try (Stream<String> stream = Files.lines(Paths.get(filePath))) {
+                    return stream.skip(skipLine).limit(limit).collect(Collectors.toList());
+                } catch (IOException e) {
+                    log.error("read file error", e);
+                    throw new RuntimeException(String.format("Read file: %s error", filePath), e);
+                }
             }
         } else {
             throw new RuntimeException("The file path: " + filePath + " not exists");
@@ -169,4 +182,154 @@ public static String getLocalLogBaseDir() {
         return loggerContext.getProperty("log.base.ctx");
     }
 
+    /**
+     * Get all rolling log files for a given base file path.
+     * Returns a sorted list containing the base file and its rolled versions (e.g., .1, .2, etc.)
+     * ordered from newest to oldest (base file first, then .1, .2, etc.)
+     */
+    private static List<File> getRollingLogFiles(String basePath) {
+        List<File> allFiles = new ArrayList<>();
+
+        File baseFile = new File(basePath);
+        File parentDir = baseFile.getParentFile();
+        String fileName = baseFile.getName();
+
+        // Add the base file if it exists
+        if (baseFile.exists()) {
+            allFiles.add(baseFile);
+        }
+
+        // Look for rolling files with pattern: basePath.N
+        if (parentDir != null) {
+            File[] files = parentDir.listFiles((dir, name) -> name.startsWith(fileName + ".") &&
+                    Pattern.matches(Pattern.quote(fileName) + "\\.\\d+", name));
+
+            if (files != null) {
+                allFiles.addAll(Arrays.asList(files));
+            }
+        }
+
+        // Sort all files in reverse order based on rolling number
+        // Base file (without number) is treated as having number 0, so it comes last
+        // descending order (larger numbers first)
+        allFiles.sort((file1, file2) -> {
+            int num1 = isRollingFile(file1) ? extractRollingNumber(file1) : 0;
+            int num2 = isRollingFile(file2) ? extractRollingNumber(file2) : 0;
+            return Integer.compare(num2, num1);
+        });
+
+        return allFiles;
+    }
+
+    /**
+     * Extract the rolling number from a file name (e.g., from "xxx.log.3" extract 3)
+     */
+    private static int extractRollingNumber(File file) {
+        String fileName = file.getName();
+        int lastDotIndex = fileName.lastIndexOf('.');
+        if (lastDotIndex != -1 && lastDotIndex < fileName.length() - 1) {
+            try {
+                return Integer.parseInt(fileName.substring(lastDotIndex + 1));
+            } catch (NumberFormatException e) {
+                return Integer.MAX_VALUE; // Put invalid files at the end
+            }
+        }
+        return Integer.MAX_VALUE;
+    }
+
+    /**
+     * Check if the file is a rolling file (has a number suffix like .1, .2, etc.)
+     */
+    private static boolean isRollingFile(File file) {
+        String fileName = file.getName();
+        // Check if the filename matches the pattern of a rolling file (basename.number)
+        int lastDotIndex = fileName.lastIndexOf('.');
+        if (lastDotIndex != -1 && lastDotIndex < fileName.length() - 1) {
+            String suffix = fileName.substring(lastDotIndex + 1);
+            return suffix.matches("\\d+");
+        }
+        return false;
+    }
+
+    /**
+     * Read lines from multiple rolling log files in order
+     */
+    private static List<String> readFromRollingLogFiles(List<File> logFiles, int skipLine, int limit) {
+        List<String> allLines = new ArrayList<>();
+
+        // Read all lines from all log files in order
+        for (File file : logFiles) {
+            log.info("Reading log file: {}", file.getAbsolutePath());
+            try (Stream<String> stream = Files.lines(file.toPath())) {
+                List<String> fileLines = stream.collect(Collectors.toList());
+                allLines.addAll(fileLines);
+            } catch (IOException e) {
+                log.error("Error reading file: " + file.getAbsolutePath(), e);
+                throw new RuntimeException(String.format("Read file: %s error", file.getAbsolutePath()), e);
+            }
+        }
+
+        // Apply skip and limit with overflow protection
+        int startIndex = Math.min(skipLine, allLines.size());
+        // Prevent integer overflow when calculating end index
+        int endIndex;
+        if (limit > allLines.size() - startIndex) {
+            endIndex = allLines.size();
+        } else {
+            endIndex = startIndex + limit;
+        }
+
+        return allLines.subList(startIndex, endIndex);
+    }
+
+    /**
+     * Get content of multiple log files (including rolling log files) as byte array
+     * Reads files in reverse order (xxx.log.n, xxx.log.n-1, ..., xxx.log)
+     *
+     * @param filePath base file path
+     * @return byte array of all log files content
+     */
+    public static byte[] getFileContentBytesWithRollingLogs(String filePath) {
+        File file = new File(filePath);
+        if (file.exists() && file.isFile()) {
+            // Check if there are rolling log files
+            List<File> logFiles = getRollingLogFiles(filePath);
+
+            if (logFiles.size() > 1) {
+                // Handle multiple log files (base file + rolling files)
+                return getBytesFromMultipleLogFiles(logFiles);
+            } else {
+                // Handle single log file
+                return getFileContentBytesFromLocal(filePath);
+            }
+        } else {
+            throw new RuntimeException("The file path: " + filePath + " not exists");
+        }
+    }
+
+    /**
+     * Read bytes from multiple log files in order
+     */
+    private static byte[] getBytesFromMultipleLogFiles(List<File> logFiles) {
+        List<byte[]> allBytes = new ArrayList<>();
+
+        // Read all bytes from all log files in order
+        for (File file : logFiles) {
+            log.info("Reading log file for download: {}", file.getAbsolutePath());
+            byte[] fileBytes = getFileContentBytesFromLocal(file.getAbsolutePath());
+            allBytes.add(fileBytes);
+        }
+
+        // Combine all bytes
+        int totalLength = allBytes.stream().mapToInt(bytes -> bytes.length).sum();
+        byte[] result = new byte[totalLength];
+        int position = 0;
+
+        for (byte[] bytes : allBytes) {
+            System.arraycopy(bytes, 0, result, position, bytes.length);
+            position += bytes.length;
+        }
+
+        return result;
+    }
 }

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

@@ -0,0 +1,167 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dolphinscheduler.common.utils;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.File;
+import java.net.URL;
+import java.util.List;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class LogUtilsTest {
+
+    private String predefinedLogPath;
+
+    @BeforeEach
+    public void setUp() {
+        URL resourceUrl = getClass().getClassLoader().getResource("log/730.log");
+        if (resourceUrl != null) {
+            predefinedLogPath = resourceUrl.getPath();
+        }
+    }
+
+    @AfterEach
+    public void tearDown() {
+    }
+
+    @Test
+    public void testReadPartFileContentFromLocal_Success() {
+        if (predefinedLogPath != null) {
+            String path = predefinedLogPath.replace("%20", " ");
+            List<String> result = LogUtils.readPartFileContentFromLocal(path, 1, 3);
+
+            assertNotNull(result);
+            assertTrue(result.size() >= 0);
+        }
+    }
+
+    @Test
+    public void testReadPartFileContentFromLocal_SkipNoneLimitAll() {
+        if (predefinedLogPath != null) {
+            String path = predefinedLogPath.replace("%20", " ");
+            List<String> result = LogUtils.readPartFileContentFromLocal(path, 0, 100);
+
+            assertNotNull(result);
+            assertTrue(result.size() >= 0);
+        }
+    }
+
+    @Test
+    public void testReadPartFileContentFromPredefinedRollingFiles() {
+        if (predefinedLogPath != null) {
+            String mainLogPath = predefinedLogPath.replace("%20", " ");
+
+            File mainLogFile = new File(mainLogPath);
+            assertTrue(mainLogFile.exists(), "Main log file should exist");
+
+            File rollingLogFile = new File(mainLogPath + ".1");
+            assertTrue(rollingLogFile.exists(), "Rolling log file should exist");
+
+            List<String> result = LogUtils.readPartFileContentFromLocal(mainLogPath, 0, 50);
+
+            assertNotNull(result);
+            assertTrue(result.size() > 0, "Should have some content from the log files");
+
+            System.out.println("Number of lines read: " + result.size());
+            for (int i = 0; i < Math.min(5, result.size()); i++) {
+                System.out.println("Line " + i + ": " + result.get(i));
+            }
+        }
+    }
+
+    @Test
+    public void testReadPartFileContentFromLocal_SkipAll() {
+        if (predefinedLogPath != null) {
+            String path = predefinedLogPath.replace("%20", " ");
+            List<String> result = LogUtils.readPartFileContentFromLocal(path, 1000, 5);
+
+            assertNotNull(result);
+            assertTrue(result.isEmpty());
+        }
+    }
+
+    @Test
+    public void testReadPartFileContentFromLocal_LimitZero() {
+        if (predefinedLogPath != null) {
+            String path = predefinedLogPath.replace("%20", " ");
+            List<String> result = LogUtils.readPartFileContentFromLocal(path, 0, 0);
+
+            assertNotNull(result);
+            assertTrue(result.isEmpty());
+        }
+    }
+
+    @Test
+    public void testReadPartFileContentFromLocal_FileDoesNotExist() {
+        assertThrows(RuntimeException.class, () -> {
+            LogUtils.readPartFileContentFromLocal("/non/existent/file.log", 0, 5);
+        });
+    }
+
+    @Test
+    public void testReadPartFileContentFromTwoSpecificRollingFiles() {
+        String resourcePath = "log/730.log";
+        URL resourceUrl = getClass().getClassLoader().getResource(resourcePath);
+
+        if (resourceUrl != null) {
+            String mainLogPath = resourceUrl.getPath().replace("%20", " ");
+
+            File mainLogFile = new File(mainLogPath);
+            File rollingLogFile = new File(mainLogPath + ".1");
+
+            assertTrue(mainLogFile.exists(), "Main log file (730.log) should exist");
+            assertTrue(rollingLogFile.exists(), "Rolling log file (730.log.1) should exist");
+
+            List<String> result = LogUtils.readPartFileContentFromLocal(mainLogPath, 0, 100);
+
+            assertNotNull(result);
+            assertTrue(result.size() > 0, "Should read content from both log files");
+
+            System.out.println("Total lines from both files (730.log and 730.log.1): " + result.size());
+        }
+    }
+
+    @Test
+    public void testGetFileContentBytesWithRollingLogs() {
+        String resourcePath = "log/730.log";
+        URL resourceUrl = getClass().getClassLoader().getResource(resourcePath);
+
+        if (resourceUrl != null) {
+            String mainLogPath = resourceUrl.getPath().replace("%20", " ");
+
+            File mainLogFile = new File(mainLogPath);
+            File rollingLogFile = new File(mainLogPath + ".1");
+
+            assertTrue(mainLogFile.exists(), "Main log file (730.log) should exist");
+            assertTrue(rollingLogFile.exists(), "Rolling log file (730.log.1) should exist");
+
+            byte[] result = LogUtils.getFileContentBytesWithRollingLogs(mainLogPath);
+
+            assertNotNull(result);
+            assertTrue(result.length > 0, "Should read bytes from both log files");
+
+            System.out.println("Total bytes from both files (730.log and 730.log.1): " + result.length);
+        }
+    }
+}