Merge branch 'dev' into Improvement-17025

Author: SbloodyS

.github/workflows/stale.yml

@@ -32,8 +32,8 @@ jobs:
           # Stale Issues
           days-before-issue-stale: 30
           days-before-issue-close: 7
-          # We do not stale Issues with label Waiting for reply,DSIP,security,roadmap,help wanted,good first issue,community
-          exempt-issue-labels: 'Waiting for reply,DSIP,security,roadmap,help wanted,good first issue,community'
+          # We do not stale Issues with below labels
+          exempt-issue-labels: 'Waiting for reply,DSIP,security,roadmap,help wanted,good first issue,community,priority:low,priority:middle,priority:high'
           stale-issue-message: >
             This issue has been automatically marked as stale because it has not had recent activity
             for 30 days. It will be closed in next 7 days if no further activity occurs.

dolphinscheduler-dao/src/main/resources/sql/upgrade/3.2.0_schema/mysql/dolphinscheduler_dml.sql

@@ -30,10 +30,11 @@ END IF;
 END;
 d//
 
+delimiter ;
+
 -- If the admin account is not associated with a tenant, the admin's tenant will be set to the default tenant.
 UPDATE `t_ds_user` SET `tenant_id` = '-1' WHERE (`user_name` = 'admin') AND (`tenant_id` = '0');
 
-delimiter ;
 CALL dolphin_t_ds_tenant_insert_default();
 DROP PROCEDURE dolphin_t_ds_tenant_insert_default;
 

dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/AbstractCommandExecutor.java

@@ -216,13 +216,15 @@ public void cancelApplication() throws InterruptedException {
             return;
         }
 
-        // soft kill
-        log.info("Begin to kill process process, pid is : {}", taskRequest.getProcessId());
-        process.destroy();
-        if (!process.waitFor(5, TimeUnit.SECONDS)) {
-            process.destroyForcibly();
+        // Try to kill process tree
+        boolean killed = ProcessUtils.kill(taskRequest);
+        if (killed) {
+            log.info("Process tree for task: {} is killed or already finished, pid: {}",
+                    taskRequest.getTaskAppId(), taskRequest.getProcessId());
+        } else {
+            log.error("Failed to kill process tree for task: {}, pid: {}",
+                    taskRequest.getTaskAppId(), taskRequest.getProcessId());
         }
-        log.info("Success kill task: {}, pid: {}", taskRequest.getTaskAppId(), taskRequest.getProcessId());
     }
 
     private void collectPodLogIfNeeded() {

dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/shell/BaseLinuxShellInterceptorBuilder.java

@@ -171,6 +171,7 @@ private List<String> bootstrapCommandInResourceLimitMode() {
         }
 
         bootstrapCommand.add(String.format("--uid=%s", runUser));
+        bootstrapCommand.add(shellAbsolutePath().toString());
         return bootstrapCommand;
     }
 }

dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/utils/ProcessUtils.java

@@ -85,31 +85,77 @@ private ProcessUtils() {
     private static final Pattern LINUXPATTERN = Pattern.compile("\\((\\d+)\\)");
 
     /**
-     * kill tasks according to different task types.
+     * Terminate the task process, support multi-level signal processing and fallback strategy
+     * @param request Task execution context
+     * @return Whether the process was successfully terminated
      */
-    @Deprecated
     public static boolean kill(@NonNull TaskExecutionContext request) {
         try {
-            log.info("Begin kill task instance, processId: {}", request.getProcessId());
+            log.info("Begin killing task instance, processId: {}", request.getProcessId());
             int processId = request.getProcessId();
             if (processId == 0) {
-                log.error("Task instance kill failed, processId is not exist");
-                return false;
+                log.info("Task instance has already finished, no need to kill");
+                return true;
             }
 
-            String cmd = String.format("kill -9 %s", getPidsStr(processId));
-            cmd = OSUtils.getSudoCmd(request.getTenantCode(), cmd);
-            log.info("process id:{}, cmd:{}", processId, cmd);
+            // Get all child processes
+            String pids = getPidsStr(processId);
+            String[] pidArray = pids.split("\\s+");
+            if (pidArray.length == 0) {
+                log.warn("No valid PIDs found for process: {}", processId);
+                return true;
+            }
+
+            // 1. Try to terminate gracefully (SIGINT)
+            boolean gracefulKillSuccess = sendKillSignal("SIGINT", pids, request.getTenantCode());
+            if (gracefulKillSuccess) {
+                log.info("Successfully killed process tree using SIGINT, processId: {}", processId);
+                return true;
+            }
+
+            // 2. Try to terminate forcefully (SIGTERM)
+            boolean termKillSuccess = sendKillSignal("SIGTERM", pids, request.getTenantCode());
+            if (termKillSuccess) {
+                log.info("Successfully killed process tree using SIGTERM, processId: {}", processId);
+                return true;
+            }
+
+            // 3. As a last resort, use `kill -9`
+            log.warn("SIGINT & SIGTERM failed, using SIGKILL as a last resort for processId: {}", processId);
+            boolean forceKillSuccess = sendKillSignal("SIGKILL", pids, request.getTenantCode());
+            if (forceKillSuccess) {
+                log.info("Successfully sent SIGKILL signal to process tree, processId: {}", processId);
+            } else {
+                log.error("Error sending SIGKILL signal to process tree, processId: {}", processId);
+            }
+            return forceKillSuccess;
 
-            OSUtils.exeCmd(cmd);
-            log.info("Success kill task instance, processId: {}", request.getProcessId());
-            return true;
         } catch (Exception e) {
             log.error("Kill task instance error, processId: {}", request.getProcessId(), e);
             return false;
         }
     }
 
+    /**
+     * Send a kill signal to a process group
+     * @param signal Signal type (SIGINT, SIGTERM, SIGKILL)
+     * @param pids Process ID list
+     * @param tenantCode Tenant code
+     */
+    private static boolean sendKillSignal(String signal, String pids, String tenantCode) {
+        try {
+            String killCmd = String.format("kill -s %s %s", signal, pids);
+            killCmd = OSUtils.getSudoCmd(tenantCode, killCmd);
+            log.info("Sending {} to process group: {}, command: {}", signal, pids, killCmd);
+            OSUtils.exeCmd(killCmd);
+
+            return true;
+        } catch (Exception e) {
+            log.error("Error sending {} to process: {}", signal, pids, e);
+            return false;
+        }
+    }
+
     /**
      * get pids str.
      *

dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/shell/BaseLinuxShellInterceptorBuilderTest.java

@@ -0,0 +1,62 @@
+/*
+ * 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.plugin.task.api.shell;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mockStatic;
+
+import org.apache.dolphinscheduler.common.utils.PropertyUtils;
+import org.apache.dolphinscheduler.plugin.task.api.shell.bash.BashShellInterceptorBuilder;
+import org.apache.dolphinscheduler.plugin.task.api.utils.AbstractCommandExecutorConstants;
+
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+
+class BaseLinuxShellInterceptorBuilderTest {
+
+    @Test
+    void generateBootstrapCommandTest() {
+        BashShellInterceptorBuilder builder = new BashShellInterceptorBuilder()
+                .shellDirectory("/tmp")
+                .shellName("test")
+                .runUser("root")
+                .cpuQuota(10)
+                .memoryQuota(1024)
+                .sudoMode(false);
+        try (MockedStatic<PropertyUtils> mockStatic = mockStatic(PropertyUtils.class)) {
+            // default
+            List<String> defaultCommands = builder.generateBootstrapCommand();
+            assertEquals("bash /tmp/test.sh", String.join(" ", defaultCommands));
+
+            // sudo mode
+            builder.sudoMode(true);
+            List<String> sudoCommands = builder.generateBootstrapCommand();
+            assertEquals("sudo -u root -i /tmp/test.sh", String.join(" ", sudoCommands));
+
+            // resource limit mode
+            mockStatic.when(
+                    () -> PropertyUtils.getBoolean(AbstractCommandExecutorConstants.TASK_RESOURCE_LIMIT_STATE, false))
+                    .thenReturn(true);
+            List<String> limitModeCommands = builder.generateBootstrapCommand();
+            assertEquals("sudo systemd-run -q --scope -p CPUQuota=10% -p MemoryLimit=1024M --uid=root /tmp/test.sh",
+                    String.join(" ", limitModeCommands));
+        }
+    }
+}

dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/utils/ProcessUtilsTest.java

@@ -19,24 +19,39 @@
 
 import org.apache.dolphinscheduler.common.utils.OSUtils;
 import org.apache.dolphinscheduler.plugin.task.api.TaskConstants;
+import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
 
 import org.apache.commons.lang3.SystemUtils;
 
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.MockedStatic;
 import org.mockito.Mockito;
 
 public class ProcessUtilsTest {
 
+    private MockedStatic<OSUtils> mockedOSUtils;
+
+    @BeforeEach
+    void setUp() {
+        mockedOSUtils = Mockito.mockStatic(OSUtils.class);
+    }
+
+    @AfterEach
+    void tearDown() {
+        if (mockedOSUtils != null) {
+            mockedOSUtils.close();
+        }
+    }
     @Test
     public void testGetPidsStr() throws Exception {
         // first
         String pids = "sudo(6279)---558_1497.sh(6282)---sleep(6354)";
         int processId = 6279;
         String exceptPidsStr = "6279 6282 6354";
         String command;
-        MockedStatic<OSUtils> osUtilsMockedStatic = Mockito.mockStatic(OSUtils.class);
         if (SystemUtils.IS_OS_MAC) {
             pids = "-+= 6279 sudo -+- 6282 558_1497.sh --- 6354 sleep";
             command = String.format("%s -sp %d", TaskConstants.PSTREE, processId);
@@ -45,7 +60,7 @@ public void testGetPidsStr() throws Exception {
         } else {
             command = String.format("%s -p %d", TaskConstants.PSTREE, processId);
         }
-        osUtilsMockedStatic.when(() -> OSUtils.exeCmd(command)).thenReturn(pids);
+        mockedOSUtils.when(() -> OSUtils.exeCmd(command)).thenReturn(pids);
         String actualPidsStr = ProcessUtils.getPidsStr(processId);
         Assertions.assertEquals(exceptPidsStr, actualPidsStr);
 
@@ -62,7 +77,7 @@ public void testGetPidsStr() throws Exception {
         } else {
             command2 = String.format("%s -p %d", TaskConstants.PSTREE, processId2);
         }
-        osUtilsMockedStatic.when(() -> OSUtils.exeCmd(command2)).thenReturn(pids2);
+        mockedOSUtils.when(() -> OSUtils.exeCmd(command2)).thenReturn(pids2);
         String actualPidsStr2 = ProcessUtils.getPidsStr(processId2);
         Assertions.assertEquals(exceptPidsStr2, actualPidsStr2);
 
@@ -79,7 +94,7 @@ public void testGetPidsStr() throws Exception {
         } else {
             command3 = String.format("%s -p %d", TaskConstants.PSTREE, processId3);
         }
-        osUtilsMockedStatic.when(() -> OSUtils.exeCmd(command3)).thenReturn(pids3);
+        mockedOSUtils.when(() -> OSUtils.exeCmd(command3)).thenReturn(pids3);
         String actualPidsStr3 = ProcessUtils.getPidsStr(processId3);
         Assertions.assertEquals(exceptPidsStr3, actualPidsStr3);
     }
@@ -95,4 +110,43 @@ public void tetRemoveK8sClientCache() {
         });
     }
 
+    @Test
+    void testKillProcessSuccessWithSigInt() throws Exception {
+        // Arrange
+        TaskExecutionContext taskRequest = Mockito.mock(TaskExecutionContext.class);
+        Mockito.when(taskRequest.getProcessId()).thenReturn(12345);
+        Mockito.when(taskRequest.getTenantCode()).thenReturn("testTenant");
+
+        // Mock getPidsStr
+        mockedOSUtils.when(() -> OSUtils.exeCmd(Mockito.matches(".*pstree.*12345"))).thenReturn("1234 12345");
+
+        // Mock SIGINT command
+        mockedOSUtils.when(() -> OSUtils.getSudoCmd(Mockito.eq("testTenant"), Mockito.matches("kill -s SIGINT.*")))
+                .thenReturn("kill -s SIGINT 12345");
+        mockedOSUtils.when(() -> OSUtils.exeCmd("kill -s SIGINT 12345")).thenReturn("");
+
+        // Mock process check - process dies after SIGINT
+        mockedOSUtils.when(() -> OSUtils.exeCmd("ps -p 12345")).thenReturn(null);
+
+        // Act
+        boolean result = ProcessUtils.kill(taskRequest);
+
+        // Assert
+        Assertions.assertTrue(result);
+        // Verify SIGKILL was never called
+        mockedOSUtils.verify(() -> OSUtils.exeCmd("kill -9 12345"), Mockito.never());
+    }
+
+    @Test
+    void testKillNonExistentProcess() {
+        // Arrange
+        TaskExecutionContext taskRequest = Mockito.mock(TaskExecutionContext.class);
+        Mockito.when(taskRequest.getProcessId()).thenReturn(0);
+
+        // Act
+        boolean result = ProcessUtils.kill(taskRequest);
+
+        // Assert
+        Assertions.assertTrue(result);
+    }
 }