git

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/app/entity/MultiProjectWaitQueueEntity.java

@@ -23,6 +23,11 @@ public class MultiProjectWaitQueueEntity {
 
     private String projectId;
 
+    private String controllerPath;
+    private String connectorPath;
+
+    private String softwarePath;
+
     private List<MultiTaskMessageEntity> multiTaskMessageEntityList;
 
     private List<MultiSimulationSceneKafkaParam> kafkaParamList;

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

@@ -7,6 +7,7 @@ import api.common.pojo.param.project.MultiCreateYamlRet;
 import api.common.pojo.param.project.MultiSimulationProjectKafkaParam;
 import api.common.pojo.param.project.MultiSimulationProjectParam;
 import api.common.pojo.param.project.MultiSimulationSceneKafkaParam;
+import api.common.pojo.po.group.SimulationMageGroupPO;
 import api.common.pojo.po.project.MultiSimulationProjectTaskRecordPO;
 import api.common.pojo.vo.map.SimulationMapVO;
 import api.common.pojo.vo.project.MultiSimulationProjectVO;
@@ -39,6 +40,7 @@ import io.minio.MinioClient;
 import lombok.SneakyThrows;
 import lombok.Synchronized;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.kafka.clients.admin.Admin;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.springframework.beans.factory.annotation.Value;
@@ -125,6 +127,8 @@ public class ProjectApplicationService {
     private SimulationMapMapper mapMapper;
     @Resource
     private MultiSimulationProjectMapper multiSimulationProjectMapper;
+    @Resource
+    private SimulationMageGroupMapper mageGroupMapper;
 
     @Resource
     private MultiSimulationProjectTaskRecordMapper taskRecordMapper;
@@ -584,6 +588,7 @@ public class ProjectApplicationService {
             parallel = multiTaskMessageEntityList.size();
         }
         Integer runState = multiProjectWaitQueue.getRunState();
+
         // 使用完塞入
         Map<String, Integer> multiNodeMapToUse = projectDomainService.getMultiNodeMapToUse(isChoiceGpu, parallel);
         List<MultiCreateYamlRet> yamlList = new ArrayList<>();
@@ -615,7 +620,9 @@ public class ProjectApplicationService {
                 throw new RuntimeException("未选取到可用的节点");
             }
             MultiSimulationSceneKafkaParam multiSimulationSceneKafkaParam = multiProjectWaitQueue.getKafkaParamList().get(i);
-            MultiCreateYamlRet multiTempYaml = projectDomainService.createMultiTempYaml(projectId, multiSimulationSceneKafkaParam, messageEntity, modelName, partition, offset, isChoiceGpu);
+            MultiCreateYamlRet multiTempYaml = projectDomainService.createMultiTempYaml(projectId, multiSimulationSceneKafkaParam, messageEntity
+                ,multiProjectWaitQueue.getConnectorPath(),multiProjectWaitQueue.getControllerPath()
+                , modelName, partition, offset, isChoiceGpu);
             multiTempYaml.setTaskId(messageEntity.getInfo().getTask_id());
             multiTempYaml.setNodeName(modelName);
             yamlList.add(multiTempYaml);
@@ -1038,6 +1045,7 @@ public class ProjectApplicationService {
     }
 
     @Async("pool1")
+    @SneakyThrows
     public void runMultiProject(MultiSimulationProjectKafkaParam projectStartMessageEntity) {
         MultiProjectWaitQueueEntity multiTaskAndFixData = createMultiTaskAndFixData(projectStartMessageEntity);
         checkIfCanRunMulti(multiTaskAndFixData);
@@ -1123,6 +1131,16 @@ public class ProjectApplicationService {
             KafkaUtil.deleteTopic(kafkaAdminClient, projectId);
 
             List<MultiSimulationSceneKafkaParam> kafkaParamList = projectStartMessageEntity.getKafkaParamList();
+
+            MultiSimulationProjectVO projectVO = multiSimulationProjectMapper.selectMultiSimulationProjectById(projectId);
+            if (Objects.isNull(projectVO)){
+                throw new RuntimeException("未找到有效的仿真任务");
+            }
+            String simulationMageGroupId = projectVO.getSimulationMageGroupId();
+            SimulationMageGroupPO groupPO = mageGroupMapper.selectSimulationMageGroupById(simulationMageGroupId);
+            if (Objects.isNull(groupPO) || StringUtils.isBlank(groupPO.getControllerPath()) || StringUtils.isBlank(groupPO.getConnectorPath())){
+                throw new RuntimeException("仿真镜像组无效");
+            }
             List<MultiTaskMessageEntity> entityList = new ArrayList<>();
             for (MultiSimulationSceneKafkaParam kafkaParam: kafkaParamList) {
                 String taskId = StringUtil.getRandomUUID();
@@ -1226,6 +1244,9 @@ public class ProjectApplicationService {
             MultiProjectWaitQueueEntity build = MultiProjectWaitQueueEntity.builder()
                 .multiTaskMessageEntityList(entityList)
                 .projectId(projectId)
+                .connectorPath(groupPO.getConnectorPath())
+                .controllerPath(groupPO.getControllerPath())
+                .softwarePath(groupPO.getSoftwarePath())
                 .waitingParallelism(entityList.size())
                 .kafkaParamList(kafkaParamList)
                 .runState(-1)
@@ -1344,6 +1365,9 @@ public class ProjectApplicationService {
             waitMulti(MultiProjectWaitQueueEntity.builder().waitingParallelism(parallelism - remainderParallelism)
                 .runState(runSt)
                 .projectId(projectWaitQueueEntity.getProjectId())
+                .softwarePath(projectWaitQueueEntity.getSoftwarePath())
+                .connectorPath(projectWaitQueueEntity.getConnectorPath())
+                .controllerPath(projectWaitQueueEntity.getControllerPath())
                 .multiTaskMessageEntityList(projectWaitQueueEntity.getMultiTaskMessageEntityList())
                 .kafkaParamList(projectWaitQueueEntity.getKafkaParamList())
                 .build());
@@ -1361,6 +1385,9 @@ public class ProjectApplicationService {
             waitMulti(MultiProjectWaitQueueEntity.builder().waitingParallelism(0)
                     .runState(projectWaitQueueEntity.getMultiTaskMessageEntityList().size() -1)
                     .projectId(projectWaitQueueEntity.getProjectId())
+                    .softwarePath(projectWaitQueueEntity.getSoftwarePath())
+                    .connectorPath(projectWaitQueueEntity.getConnectorPath())
+                    .controllerPath(projectWaitQueueEntity.getControllerPath())
                     .kafkaParamList(projectWaitQueueEntity.getKafkaParamList())
                     .multiTaskMessageEntityList(projectWaitQueueEntity.getMultiTaskMessageEntityList()).build()
             );

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

@@ -218,7 +218,9 @@ public class ProjectDomainService {
     }
 
     @SneakyThrows
-    public MultiCreateYamlRet createMultiTempYaml(String projectId, MultiSimulationSceneKafkaParam multiSimulationSceneKafkaParam, MultiTaskMessageEntity messageEntity, String nodeName, int kafkaPartition, long kafkaOffset, String isChoiceGpu) {
+    public MultiCreateYamlRet createMultiTempYaml(String projectId, MultiSimulationSceneKafkaParam multiSimulationSceneKafkaParam, MultiTaskMessageEntity messageEntity,
+                                                  String connectorPath, String controllerPath,
+                                                  String nodeName, int kafkaPartition, long kafkaOffset, String isChoiceGpu) {
         String podName = getMultiRandomPodName(projectId);   // 生成 podName
         String podYaml = getMultiPodYamlName(nodeName, podName);     // 模板文件名称
         String yamlPath = multiPodYamlDirectory + podYaml;
@@ -242,6 +244,10 @@ public class ProjectDomainService {
         podString = podString.replace("cpu-order", "\""+1+"\"");
         podString = podString.replace("vtd-command", kubernetesConfiguration.getMultiCommandVtdGpu());
         podString = podString.replace("kafka-topic", projectId);
+        podString = podString.replace("controller-container", "controller-multi-" + messageEntity.getInfo().getScene_id());
+        podString = podString.replace("controller-image", controllerPath);
+        podString = podString.replace("connector-container", "connector-multi-" + messageEntity.getInfo().getScene_id());
+        podString = podString.replace("connector-image", connectorPath);
 
         List<MultiSimulationSceneCarVO> simulationSceneCarVOList = multiSimulationSceneKafkaParam.getSimulationSceneCarVOList();
 

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/infra/db/mysql/mapper/MultiSimulationProjectMapper.java

@@ -27,7 +27,7 @@ public interface MultiSimulationProjectMapper {
         "where id = #{projectId}")
     int updateMultiSimulationProjectStatus(MultiSimulationProjectParam param);
 
-    @Select("select id,project_key as projectKey,project_name as projectName,project_status as projectStatus,create_time as createTime,project_description as projectDescription, " +
+    @Select("select id,project_key as projectKey,project_name as projectName,project_status as projectStatus,simulation_mage_group_id as simulationMageGroupId,create_time as createTime,project_description as projectDescription, " +
         "project_user_id as projectUserId from multi_simulation_project where deleted = 0 " +
         "and id = #{id} limit 1")
     MultiSimulationProjectVO selectMultiSimulationProjectById(@Param("id") String projectId);

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/infra/db/mysql/mapper/SimulationMageGroupMapper.java

@@ -0,0 +1,12 @@
+package com.css.simulation.resource.scheduler.infra.db.mysql.mapper;
+
+import api.common.pojo.po.group.SimulationMageGroupPO;
+import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Select;
+
+@Mapper
+public interface SimulationMageGroupMapper {
+    @Select("select id,group_name as groupName,software_path as softwarePath,controller_path as controllerPath,connector_path as connectorPath,group_description as groupDescription from simulation_mage_group where id = #{id} and deleted = 0")
+    SimulationMageGroupPO selectSimulationMageGroupById(@Param("id") String id);
+}