mlx

api-common/src/main/java/api/common/util/SshUtil.java

@@ -132,6 +132,36 @@ public class SshUtil {
 
     }
 
+    /**
+     * 获取 cpu 总量
+     */
+    public static long cpuTotal(ClientSession session) throws IOException, InterruptedException {
+
+        Map<?, ?> map1 = cpuInfo(session);
+        Thread.sleep(5 * 1000);
+        Map<?, ?> map2 = cpuInfo(session);
+
+        long user1 = Long.parseLong(map1.get("user").toString());
+        long nice1 = Long.parseLong(map1.get("nice").toString());
+        long system1 = Long.parseLong(map1.get("system").toString());
+        long idle1 = Long.parseLong(map1.get("idle").toString());
+
+        long user2 = Long.parseLong(map2.get("user").toString());
+        long nice2 = Long.parseLong(map2.get("nice").toString());
+        long system2 = Long.parseLong(map2.get("system").toString());
+        long idle2 = Long.parseLong(map2.get("idle").toString());
+
+        long total1 = user1 + system1 + nice1;
+        long total2 = user2 + system2 + nice2;
+        double total = total2 - total1;
+
+        long totalIdle1 = user1 + nice1 + system1 + idle1;
+        long totalIdle2 = user2 + nice2 + system2 + idle2;
+
+        return totalIdle2 - totalIdle1;
+
+    }
+
 
     /**
      * 功能：CPU使用信息
@@ -164,6 +194,20 @@ public class SshUtil {
         return map;
     }
 
+
+    /**
+     * 显存使用率
+     *
+     * @param session 会话
+     * @return 显存总量，单位 MB
+     */
+    public static double gpuUsage(ClientSession session) throws DocumentException, IOException {
+        List<GpuDTO> gpuDTOList = SshUtil.gpuInfo(session);
+        int used = gpuDTOList.stream().mapToInt(gpu -> Integer.parseInt(gpu.getUsedMemory().split(" ")[0])).sum();
+        int total = gpuDTOList.stream().mapToInt(gpu -> Integer.parseInt(gpu.getTotalMemory().split(" ")[0])).sum();
+        return used * 1.0 / total;
+    }
+
     /**
      * 显存总量
      *
@@ -347,7 +391,7 @@ public class SshUtil {
      * @param session 会话
      * @return 剩余可用磁盘(kb)
      */
-    public static double diskAvailable(ClientSession session) throws IOException {
+    public static long diskAvailable(ClientSession session) throws IOException {
         String execute = execute(session, "df --total");
         String[] split = execute.split("\n");
         String[] split1 = split[split.length - 1].split("\\s+");
@@ -364,7 +408,7 @@ public class SshUtil {
      * @param session 会话
      * @return 总磁盘(kb)
      */
-    public static double diskTotal(ClientSession session) throws IOException {
+    public static long diskTotal(ClientSession session) throws IOException {
         String execute = execute(session, "df --total");
         String[] split = execute.split("\n");
         String[] split1 = split[split.length - 1].split("\\s+");

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/consumer/ManualProjectConsumer.java

@@ -96,7 +96,7 @@ public class ManualProjectConsumer {
         String projectJson = projectRecord.value();
         ProjectMessageDTO projectMessageDTO = JsonUtil.jsonToBean(projectJson, ProjectMessageDTO.class);
         String projectId = projectMessageDTO.getProjectId();    // 项目 id
-        projectMapper.updateProjectState(projectId, DictConstants.PROJECT_RUNNING);   // 修改该 project 的状态为执行中，同时将已完成任务重置为 0 方便测试。
+        projectMapper.resetProjectState(projectId, DictConstants.PROJECT_RUNNING);   // 修改该 project 的状态为执行中，同时将已完成任务重置为 0 方便测试。
         taskMapper.updateStateByProjectId(projectId, DictConstants.TASK_PENDING);    // 将该 project 下所有任务重置为待执行方便测试。
 
 

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

@@ -5,6 +5,8 @@ import com.css.simulation.resource.scheduler.pojo.po.ProjectPO;
 import org.apache.ibatis.annotations.*;
 import org.apache.ibatis.type.JdbcType;
 
+import java.sql.Timestamp;
+
 @Mapper
 public interface ProjectMapper {
 
@@ -23,7 +25,13 @@ public interface ProjectMapper {
             "set now_run_state  = #{state},\n" +
             "    task_completed = '0'\n" +
             "where id = #{id}")
-    void updateProjectState(@Param("id") String id, @Param("state") String state);
+    void resetProjectState(@Param("id") String id, @Param("state") String state);
+
+    @Update("update simulation_manual_project\n" +
+            "set now_run_state  = #{state},\n" +
+            "    finish_time = #{finishTime}\n" +
+            "where id = #{id}")
+    void updateProjectState(@Param("id") String id, @Param("state") String state,@Param("finishTime") Timestamp finishTime);
 
 
     @Update("update simulation_manual_project\n" +

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

@@ -18,6 +18,7 @@ public interface TaskIndexMapper {
             "                                             target,\n" +
             "                                             not_standard_scene_num,\n" +
             "                                             score,\n" +
+            "                                             score_explain,\n" +
             "                                             create_time,\n" +
             "                                             create_user_id,\n" +
             "                                             modify_time,\n" +
@@ -28,6 +29,7 @@ public interface TaskIndexMapper {
             "        #{leaf.target},\n" +
             "        #{leaf.notStandardSceneNum},\n" +
             "        #{leaf.score},\n" +
+            "        #{leaf.scoreExplain},\n" +
             "        #{leaf.createTime},\n" +
             "        #{leaf.createUserId},\n" +
             "        #{leaf.modifyTime},\n" +

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

@@ -5,6 +5,7 @@ import com.css.simulation.resource.scheduler.pojo.po.TaskPO;
 import org.apache.ibatis.annotations.*;
 import org.apache.ibatis.type.JdbcType;
 
+import java.sql.Timestamp;
 import java.util.List;
 
 /**
@@ -37,6 +38,21 @@ public interface TaskMapper {
     List<TaskPO> selectTaskListByProjectId(@Param("projectId") String projectId);
 
 
+    @Update("update simulation_manual_project_task\n" +
+            "set run_state = #{runState},run_start_time = #{runStartTime}\n" +
+            "where id = #{id}")
+    void updateStateWithStartTime(@Param("id") String id, @Param("runState") String runState,@Param("runStartTime") Timestamp runStartTime);
+
+    @Update("update simulation_manual_project_task\n" +
+            "set run_state = #{runState},run_end_time = #{runStopTime},run_result='Success'\n" +
+            "where id = #{id}")
+    void updateSuccessStateWithStopTime(@Param("id") String id, @Param("runState") String runState,@Param("runStopTime") Timestamp runStopTime);
+
+    @Update("update simulation_manual_project_task\n" +
+            "set run_state = #{runState},run_end_time = #{runStopTime},run_result='Failed'\n" +
+            "where id = #{id}")
+    void updateFailStateWithStopTime(@Param("id") String id, @Param("runState") String runState,@Param("runStopTime") Timestamp runStopTime);
+
     @Update("update simulation_manual_project_task\n" +
             "set run_state = #{runState}\n" +
             "where id = #{id}")

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/service/TaskService.java

@@ -106,12 +106,16 @@ public class TaskService {
     @SneakyThrows
     public void taskState(String taskId, String state, String podName) {
         log.info("------- /state 修改任务 " + taskId + "的状态：" + state + "，pod 名称为：" + podName);
-        taskMapper.updateState(taskId, state);
-        if ("Aborted".equals(state)
-                || "PendingAnalysis".equals(state)
-                || "Terminated".equals(state)
-        ) {
+        if ("Running".equals(state)) {
+            taskMapper.updateStateWithStartTime(taskId, state, TimeUtil.getNowForMysql());
+        } else if ("Aborted".equals(state) || "Terminated".equals(state)) {
             LinuxUtil.execute("kubectl delete pod " + podName);
+            taskMapper.updateFailStateWithStopTime(taskId, state, TimeUtil.getNowForMysql());
+        } else if ( "PendingAnalysis".equals(state)) {
+            LinuxUtil.execute("kubectl delete pod " + podName);
+            taskMapper.updateSuccessStateWithStopTime(taskId, state, TimeUtil.getNowForMysql());
+        }else {
+            taskMapper.updateState(taskId, state);
         }
         ProjectPO projectPO = projectMapper.selectById(taskId);
         String projectId = projectPO.getId();
@@ -124,7 +128,7 @@ public class TaskService {
         if (taskNum != endTaskNum) {  // 已结束任务数等于所有任务数量，才会准备打分；否则退出。
             return;
         }
-        projectMapper.updateProjectState(projectId, DictConstants.PROJECT_COMPLETED);   // 修改该 project 的状态为已完成
+        projectMapper.updateProjectState(projectId, DictConstants.PROJECT_COMPLETED, TimeUtil.getNowForMysql());   // 修改该 project 的状态为已完成
         LinuxUtil.execute("kubectl delete job project-" + projectId);
         SshClient clientKafka = SshUtil.getClient();
         ClientSession sessionKafka = SshUtil.getSession(clientKafka, hostnameKafka, usernameKafka, passwordKafka);