处理 pod 超时

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/manager/TaskManager.java

@@ -131,6 +131,7 @@ public class TaskManager {
                 }
             } catch (io.kubernetes.client.openapi.ApiException apiException) {
                 log.error("TaskManager--isCompleted pod " + podName + " 已经被手动删除，该项目可能已经失败或删除。");
+                return false;
             }
             // -------------------------------- 判断项目是否结束 --------------------------------
             ProjectMessageDTO projectMessageDTO = JsonUtil.jsonToBean(stringRedisTemplate.opsForValue().get(redisPrefix.getProjectRunningKey()), ProjectMessageDTO.class);

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/mapper/TaskMapper.java

@@ -45,11 +45,14 @@ public interface TaskMapper {
     TaskPO selectById(@Param("taskId")String taskId);
 
     @ResultMap("task")
-    @Select("select id, p_id, create_user_id\n" +
-            "from simulation_manual_project_task\n" +
-            "where is_deleted = '0'\n" +
-            "  and run_state = 'Running'")
-    List<TaskPO> selectByRunState(@Param("runState") String runState);
+    @Select("select smpt.id, smpt.p_id, smpt.create_user_id\n" +
+            "from simulation_manual_project_task smpt\n" +
+            "         left join simulation_manual_project smp on smpt.p_id = smp.id\n" +
+            "         left join simulation_automatic_subproject sas on smpt.p_id = sas.id\n" +
+            "where smpt.is_deleted = '0'\n" +
+            "  and run_state = 'Running'\n" +
+            "  and (smp.is_deleted = 0 or sas.is_deleted = 0)")
+    List<TaskPO> selectByRunStateAndProjectIsNotDeleted(@Param("runState") String runState);
 
 
 

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/scheduler/ProjectScheduler.java

@@ -15,13 +15,13 @@ import com.css.simulation.resource.scheduler.pojo.po.UserPO;
 import com.css.simulation.resource.scheduler.pojo.to.PrefixTO;
 import com.css.simulation.resource.scheduler.service.TaskService;
 import com.css.simulation.resource.scheduler.util.ProjectUtil;
-import io.kubernetes.client.openapi.ApiClient;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
 
 import javax.annotation.Resource;
 import java.util.ArrayList;
@@ -41,6 +41,8 @@ public class ProjectScheduler {
     String password;
     @Value("${scheduler.linux-path.job-yaml}")
     String jobYaml;
+    @Value("${scheduler.kubernetes.pod-timeout}")
+    Long podTimeOut;    // 超时时间，单位毫秒
     // -------------------------------- Comment --------------------------------
     @Resource
     StringRedisTemplate stringRedisTemplate;
@@ -55,8 +57,6 @@ public class ProjectScheduler {
     @Resource
     AutoSubProjectMapper autoSubProjectMapper;
     @Resource
-    ApiClient apiClient;
-    @Resource
     ProjectConsumer projectConsumer;
     @Resource
     ProjectUtil projectUtil;
@@ -185,28 +185,30 @@ public class ProjectScheduler {
      * 同时也可处理 pod 莫名关闭，因为关闭之后也会超时
      */
     @Scheduled(fixedDelay = 60 * 1000)
+    @Transactional
     public void taskTimeout() {
-        long timeout = 2 * 60 * 1000L;
-        List<TaskPO> executingTaskList = taskMapper.selectByRunState(DictConstants.TASK_RUNNING);
-        if (executingTaskList != null && executingTaskList.size() > 0) {
-            log.info("ProjectScheduler--taskTimeout 正在运行的任务有：" + executingTaskList);
-            for (TaskPO task : executingTaskList) {
-                String userId = task.getCreateUserId();
-                String projectId = task.getPId();
-                String taskId = task.getId();
-                PrefixTO redisPrefix = projectUtil.getRedisPrefixByUserIdAndProjectIdAndTaksId(userId, projectId, taskId);
-                // 获取心跳时间
-                String tickTime = stringRedisTemplate.opsForValue().get(redisPrefix.getTaskTickKey());
-                if (StringUtil.isEmpty(tickTime)) {
-                    log.error(redisPrefix.getTaskTickKey() + "，该 key 的心跳时间为空！");
-                    continue;
-                }
-                long lastTickTime = Long.parseLong(tickTime);
-                // 如果心跳超时则更改任务状态为 Aborted
-                if (TimeUtil.getNow() - lastTickTime > timeout) {
-                    String podName = stringRedisTemplate.opsForValue().get(redisPrefix.getTaskPodKey());
-                    taskService.taskState(taskId, DictConstants.TASK_ABORTED, podName);
-                }
+        //1 查询任务信息（需要过滤已经删除的项目，因为页面上删除项目不会删除任务）
+        List<TaskPO> executingTaskList = taskMapper.selectByRunStateAndProjectIsNotDeleted(DictConstants.TASK_RUNNING);
+        if (CollectionUtil.isEmpty(executingTaskList)) {
+            return;
+        }
+        log.info("ProjectScheduler--taskTimeout 正在运行的任务有：" + executingTaskList);
+        for (TaskPO task : executingTaskList) {
+            String userId = task.getCreateUserId();
+            String projectId = task.getPId();
+            String taskId = task.getId();
+            PrefixTO redisPrefix = projectUtil.getRedisPrefixByUserIdAndProjectIdAndTaksId(userId, projectId, taskId);
+            // 获取心跳时间
+            String tickTime = stringRedisTemplate.opsForValue().get(redisPrefix.getTaskTickKey());
+            if (StringUtil.isEmpty(tickTime)) {
+                log.error(redisPrefix.getTaskTickKey() + "，该 key 的心跳时间为空！");
+                continue;
+            }
+            long lastTickTime = Long.parseLong(tickTime);
+            // 如果心跳超时则更改任务状态为 Aborted
+            if (TimeUtil.getNow() - lastTickTime > podTimeOut) {
+                String podName = stringRedisTemplate.opsForValue().get(redisPrefix.getTaskPodKey());
+                taskService.taskState(taskId, DictConstants.TASK_ABORTED, podName);
             }
         }
     }