mlx

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

@@ -88,7 +88,7 @@ public class ManualProjectConsumer {
     @KafkaListener(groupId = "simulation-resource-scheduler", topics = "${scheduler.manual-project.topic}")
     @SneakyThrows
     public void cacheManualProject(ConsumerRecord<String, String> projectRecord) {
-        log.info("------- ManualProjectConsumer--cacheManualProject 接收到项目开始消息为：" + projectRecord);
+        log.info("ManualProjectConsumer--cacheManualProject 接收到项目开始消息为：" + projectRecord);
         String projectJson = projectRecord.value();
         //1 读取 kafka 的 project 信息
         ProjectMessageDTO projectMessageDTO = JsonUtil.jsonToBean(projectJson, ProjectMessageDTO.class);
@@ -128,9 +128,11 @@ public class ManualProjectConsumer {
         Set<String> runningProjectSet = redisTemplate.keys(manualProjectTopic + ":cluster:" + clusterId + ":running" + "*");
         int runningProjectNumber = CollectionUtil.isEmpty(runningProjectSet) ? 0 : runningProjectSet.size();
         if (runningProjectNumber < simulationLicenseNumber) {
+            log.info("ManualProjectConsumer--cacheManualProject 集群 " + clusterId + " 将项目 " + projectId + " 执行！");
             redisTemplate.opsForValue().set(manualProjectTopic + ":cluster:" + clusterId + ":running" + projectId, projectJson);
             parseManualProject(projectJson);
         } else {
+            log.info("ManualProjectConsumer--cacheManualProject 集群 " + clusterId + " 将项目 " + projectId + " 放入等待队列！");
             redisTemplate.opsForValue().set(manualProjectTopic + ":cluster:" + clusterId + ":waiting" + projectId, projectJson);
         }
     }
@@ -142,7 +144,7 @@ public class ManualProjectConsumer {
      * @param projectJson 项目启动消息
      */
     @SneakyThrows
-    public void parseManualProject( String projectJson) {
+    public void parseManualProject(String projectJson) {
 
         // -------------------------------- 0 准备 --------------------------------
         log.info("------- ManualProjectConsumer 接收到项目开始消息为：" + projectJson);

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

@@ -72,7 +72,6 @@ public class TaskManager {
     RequestConfig requestConfig;
 
 
-
     @SneakyThrows
     public boolean isProjectCompleted(String userId, String projectId, String taskId, String state, String podName, ClientSession session) {
 
@@ -176,11 +175,10 @@ public class TaskManager {
         ClusterPO clusterPO = clusterMapper.selectByUserId(userId);
         String clusterId = clusterPO.getId();
         stringRedisTemplate.delete(manualProjectTopic + ":cluster:" + clusterId + ":running" + projectId);
-
     }
 
     @SneakyThrows
-    public void score(String userId, String projectId, ClientSession session){
+    public void score(String userId, String projectId, ClientSession session) {
         // -------------------------------- 打分 --------------------------------
         ProjectPO projectPO = manualProjectMapper.selectById(projectId);
         String packageId = projectPO.getScenePackageId();  // 场景测试包 id，指标根 id
@@ -217,19 +215,19 @@ public class TaskManager {
                     .filter(task -> indexId.equals(task.getLastTargetId()))
                     .collect(Collectors.toList());
             log.info("TaskService--state 叶子节点 " + indexId + " 包括 " + taskListOfLeafIndex.size() + " 个成功运行结束任务！");
-            log.info("TaskService--state 计算叶子节点 " + indexId + " 的得分：" + taskListOfLeafIndex);
+            // 计算叶子指标的得分
             taskListOfLeafIndex.forEach(taskOfLeaf -> {
-                String runState = taskOfLeaf.getRunState();
-                if (DictConstants.TASK_ANALYSIS.equals(runState)) {
-                    String task2Id = taskOfLeaf.getId();
-                    taskMapper.updateSuccessStateWithStopTime(task2Id, DictConstants.TASK_ANALYSING, TimeUtil.getNowForMysql());
-                    // 计算每个任务的得分
-                    String runResultMinio = taskOfLeaf.getRunResultFilePath() + "/Ego.csv";
-                    String runResultLinux = linuxTempPath + runResultMinio;
-                    String scoreCommand = "python3 " + pyPath + "main.py " + runResultLinux + " " + taskOfLeaf.getSceneType() + " " + ruleName; // 指定打分脚本
-                    String scoreResult;
-                    ScoreTO score;
-                    try {
+                String task2Id = taskOfLeaf.getId();
+                try {
+                    String runState = taskOfLeaf.getRunState();
+                    if (DictConstants.TASK_ANALYSIS.equals(runState)) {
+                        taskMapper.updateSuccessStateWithStopTime(task2Id, DictConstants.TASK_ANALYSING, TimeUtil.getNowForMysql());
+                        // 计算每个任务的得分
+                        String runResultMinio = taskOfLeaf.getRunResultFilePath() + "/Ego.csv";
+                        String runResultLinux = linuxTempPath + runResultMinio;
+                        String scoreCommand = "python3 " + pyPath + "main.py " + runResultLinux + " " + taskOfLeaf.getSceneType() + " " + ruleName; // 指定打分脚本
+                        String scoreResult;
+                        ScoreTO score;
                         try {
                             log.info("TaskService--state 下载 minio 上的结果文件 " + runResultMinio + " 到本地：" + runResultLinux);
                             MinioUtil.downloadToFile(minioClient, bucketName, runResultMinio, runResultLinux);  // 也可改成下载到指定ip的服务器上，需要保证和打分脚本在一台机器上。
@@ -245,27 +243,26 @@ public class TaskManager {
                         } catch (IOException e) {
                             throw new RuntimeException("------- TaskService--state 项目" + projectId + "的任务" + task2Id + " 打分出错，命令为：" + scoreCommand + " 修改状态为：" + DictConstants.TASK_ABORTED + "\n" + e.getMessage());
                         }
-                    } catch (Exception e) {
-                        taskOfLeaf.setRunState(DictConstants.TASK_ABORTED);
-                        taskMapper.updateFailStateWithStopTime(task2Id, DictConstants.TASK_ABORTED, TimeUtil.getNowForMysql(), DictConstants.TASK_ERROR_REASON_5);
-                        log.error(e.getMessage());
-                        return; // 如果打分失败则开始下一个打分
+                        taskOfLeaf.setReturnSceneId(score.getUnit_scene_ID());
+                        taskOfLeaf.setScore(score.getUnit_scene_score());
+                        taskOfLeaf.setTargetEvaluate(score.getEvaluate_item());
+                        taskOfLeaf.setScoreExplain(score.getScore_description());
+                        taskOfLeaf.setModifyUserId(userId);
+                        taskOfLeaf.setModifyTime(TimeUtil.getNowForMysql());
+                        scoreExplain.set(score.getScore_description());
+                        taskOfLeaf.setRunState(DictConstants.TASK_COMPLETED);
+                        taskMapper.updateSuccessStateAndScoreResultWithStopTime(
+                                taskOfLeaf,
+                                DictConstants.TASK_COMPLETED,
+                                TimeUtil.getNowForMysql()
+                        );
                     }
-                    assert score != null;
-                    taskOfLeaf.setReturnSceneId(score.getUnit_scene_ID());
-                    taskOfLeaf.setScore(score.getUnit_scene_score());
-                    taskOfLeaf.setTargetEvaluate(score.getEvaluate_item());
-                    taskOfLeaf.setScoreExplain(score.getScore_description());
-                    taskOfLeaf.setModifyUserId(userId);
-                    taskOfLeaf.setModifyTime(TimeUtil.getNowForMysql());
-                    scoreExplain.set(score.getScore_description());
-
-                    taskOfLeaf.setRunState(DictConstants.TASK_COMPLETED);
-                    taskMapper.updateSuccessStateAndScoreResultWithStopTime(
-                            taskOfLeaf,
-                            DictConstants.TASK_COMPLETED,
-                            TimeUtil.getNowForMysql()
-                    );
+                } catch (Exception e) {
+                    taskOfLeaf.setRunState(DictConstants.TASK_ABORTED);
+                    taskOfLeaf.setScore(0.0);
+                    taskMapper.updateFailStateWithStopTime(task2Id, DictConstants.TASK_ABORTED, TimeUtil.getNowForMysql(), DictConstants.TASK_ERROR_REASON_5);
+                    log.error(e.getMessage());
+                    // 如果打分失败则开始下一个打分
                 }
             });
 
@@ -398,9 +395,6 @@ public class TaskManager {
     }
 
 
-
-
-
     public Boolean taskConfirm(String taskId) {
         // 查询 task 如果不是 pending 则不执行
         String state = taskMapper.selectStateById(taskId);

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

@@ -3,6 +3,7 @@ package com.css.simulation.resource.scheduler.mapper;
 
 import org.apache.ibatis.annotations.Mapper;
 import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Select;
 import org.apache.ibatis.annotations.Update;
 
 @Mapper
@@ -14,4 +15,8 @@ public interface AutoSubProjectMapper {
     void updateProjectStateById(@Param("subProjectId") String subProjectId, @Param("nowRunState") String nowRunState);
 
 
+    @Select("select create_user_id\n" +
+            "from simulation_automatic_subproject\n" +
+            "where id = #{subProjectId}")
+    String selectCreateUserById(@Param("subProjectId") String subProjectId);
 }