mlx

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/configuration/kubernetes/KubernetesConfiguration.java

@@ -7,6 +7,7 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.util.ResourceUtils;
 
+import java.io.File;
 import java.io.FileReader;
 import java.io.IOException;
 
@@ -15,7 +16,10 @@ public class KubernetesConfiguration {
 
     @Bean
     public ApiClient apiClient() throws IOException {
-        FileReader fileReader = new FileReader(ResourceUtils.getFile("classpath:kubernetes/config"));
+        File config = ResourceUtils.getFile("classpath:kubernetes/config");  // resource
+//        File config = new File("D:\\idea-project\\simulation-cloud\\simulation-resource-scheduler\\src\\main\\resources\\kubernetes\\config");  //windows
+//        File config = new File("/root/.kube/config");   //linux
+        FileReader fileReader = new FileReader(config);
         return ClientBuilder.kubeconfig(KubeConfig.loadKubeConfig(fileReader)).build();
     }
 

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

@@ -6,9 +6,7 @@ import api.common.pojo.dto.ProjectMessageDTO;
 import api.common.pojo.param.KafkaParameter;
 import api.common.pojo.param.MinioParameter;
 import api.common.pojo.param.RedisParameter;
-import api.common.util.JsonUtil;
-import api.common.util.StringUtil;
-import api.common.util.TimeUtil;
+import api.common.util.*;
 import com.css.simulation.resource.scheduler.feign.CommonService;
 import com.css.simulation.resource.scheduler.mapper.*;
 import com.css.simulation.resource.scheduler.pojo.to.*;
@@ -37,7 +35,7 @@ import java.util.List;
 @Slf4j
 public class ManualProjectConsumer {
 
-    private final String USER_ID = "simulation-resource-scheduler";
+    private static final String USER_ID = "simulation-resource-scheduler";
 
     @Autowired
     ProjectMapper projectMapper;
@@ -54,25 +52,23 @@ public class ManualProjectConsumer {
     @Autowired
     SensorOgtMapper sensorOgtMapper;
     @Autowired
-     CommonService commonService;
+    CommonService commonService;
     @Autowired
-     AlgorithmMapper algorithmMapper;
+    AlgorithmMapper algorithmMapper;
     @Autowired
-     ApiClient apiClient;
-    @Value("${spring.kafka.consumer.topic.manual-project}")
-     String manualProjectTopic;
-    @Value("${scheduler.manual-project.result-path-minio}")
-     String manualProjectResultPath;
-    @Value("${scheduler.manual-project.algorithm-tar-path-linux}")
-     String algorithmTarPathLinux;
+    ApiClient apiClient;
+    @Value("${scheduler.manual-project.topic}")
+    String manualProjectTopic;
+    @Value("${scheduler.linux-temp-path}")
+    String linuxTempPath;
 
+//
+//    @KafkaListener(groupId = "simulation-resource-scheduler", topics = "${scheduler.manual-project.topic}")
+//    public void testConsumer(ConsumerRecord<String, String> projectRecord) {
+//        System.out.println("------- 消费成功：" + projectRecord.value());
+//    }
 
-    @KafkaListener(groupId = "simulation-resource-scheduler", topics = "${spring.kafka.consumer.topic.manual-project}")
-    public void testConsumer(ConsumerRecord<String, String> projectRecord) {
-        System.out.println("------- 消费成功：" + projectRecord.value());
-    }
-
-    @KafkaListener(groupId = "simulation-resource-scheduler", topics = "manualProject")
+    @KafkaListener(groupId = "simulation-resource-scheduler", topics = "${scheduler.manual-project.topic}")
     public void parseProject(ConsumerRecord<String, String> projectRecord) throws IOException, ApiException {
         //1 读取 kafka 的 project 信息
         /*
@@ -133,7 +129,7 @@ public class ManualProjectConsumer {
 
         for (ScenePO scenePO : sceneList) {
             String taskId = StringUtil.getRandomUUID();
-            String resultPath = manualProjectResultPath + projectId + "/" + taskId;
+            String resultPath = linuxTempPath + projectId + "/" + taskId;
             // 保存任务信息
             TaskPO taskPO = TaskPO.builder() // run_start_time 和 run_end_time 不填
                     .id(taskId)
@@ -204,17 +200,26 @@ public class ManualProjectConsumer {
         // -------------------------------- 4 算法（一期按单机版做） --------------------------------
         // 私有仓库导入算法镜像（搭建私有仓库）
         String algorithmId = projectMessageDTO.getAlgorithmId();    // 算法 id
-        //4-1 根据算法 id 获取算法文件地址
-        String minioPath = algorithmMapper.selectMinioPathById(algorithmId);
-        String linuxPath = algorithmTarPathLinux + minioPath.substring(1);
-        // 下载算法文件到本地（ 2 到仓库服务器）
-        Response response = commonService.download(new MinioParameter(minioPath));
-        InputStream inputStream = response.body().asInputStream();
-        //4-2 本地执行 docker load 算法文件成镜像（二期用 docker-java 操作 或 ssh 连接）
-
-
+        //4-1 根据算法 id 获取算法文件地址、是否已导入成镜像。
+        AlgorithmPO algorithmPO = algorithmMapper.selectById(algorithmId);
+        String minioPath = algorithmPO.getMinioPath();
+        String dockerImage;
+        if ("0".equals(algorithmPO.getDockerImport())) {
+            dockerImage = "algorithm_" + algorithmId + ":latest";
+            String algorithmTarLinuxTempPath = linuxTempPath + minioPath;
+            // 下载算法文件到本地（ 2 到仓库服务器）
+            Response response = commonService.download(new MinioParameter(minioPath));
+            InputStream inputStream = response.body().asInputStream();
+            FileUtil.writeInputStreamToLocalFile(inputStream, algorithmTarLinuxTempPath);
+            //4-2 本地执行 docker load 算法文件成镜像（集群版可改成用 docker-java 操作仓库）
+            LinuxUtil.execute("docker import " + algorithmTarLinuxTempPath + " " + dockerImage);
+        } else if ("1".equals(algorithmPO.getDockerImport()) && StringUtil.isNotEmpty(algorithmPO.getDockerImage())) {
+            dockerImage = algorithmPO.getDockerImage();
+        } else {
+            throw new RuntimeException("算法 " + algorithmId + "的 mysql 数据有误！");
+        }
         // -------------------------------- 5 创建 pod 开始执行 --------------------------------
-        int completions = sceneList.size();
+        int completions = sceneList.size();     // 结束标
         int parallelism = projectMessageDTO.getParallelism();    // 并行度
         BatchV1Api batchV1Api = new BatchV1Api(apiClient);
         V1Job yaml = (V1Job) Yaml.load(ResourceUtils.getFile("classpath:kubernetes/template/job-template.yaml"));
@@ -222,34 +227,29 @@ public class ManualProjectConsumer {
         //2 kind
         //3 metadata
         V1ObjectMeta metadata = yaml.getMetadata();
-        metadata.setName("job2");
+        metadata.setName("project_" + projectId);
         yaml.setMetadata(metadata);
-
-        //4 spec
-        V1JobSpec v1JobSpec = yaml.getSpec();
-        //4-1 job
-        v1JobSpec.setCompletions(completions);
-        v1JobSpec.setParallelism(parallelism);
-        //4-2 pod
-        V1PodTemplateSpec v1PodTemplateSpec = v1JobSpec.getTemplate();
-
-        v1PodTemplateSpec.setMetadata(metadata);
-        V1PodSpec v1PodSpec = v1PodTemplateSpec.getSpec();
-        v1PodSpec.setRestartPolicy("Never");
-        //4-3 container
-        List<V1Container> containers = new ArrayList<>();
-        V1Container v1Container = v1PodSpec.getContainers().get(0);
-        v1Container.setImage("perl");
-        v1Container.setImagePullPolicy("IfNotPresent");
-        v1Container.setName("pi");
-        v1Container.setCommand(Arrays.asList("perl", "-Mbignum=bpi", "-wle", "print bpi(2000)"));
-        containers.add(v1Container);
+        //4 job
+        V1JobSpec job = yaml.getSpec();
+        job.setCompletions(completions); // 这个标准是什么？
+        job.setParallelism(parallelism);
+        //5 pod
+        V1PodSpec v1PodSpec = job.getTemplate().getSpec();
+        //6 container
+        List<V1Container> containers = v1PodSpec.getContainers();
+        for (V1Container container : containers) {
+            String name = container.getName();
+            if ("vtd".equals(name)) {
+                container.setName("vtd_" + projectId);
+            }
+            if ("algorithm".equals(name)) {
+                container.setName("algorithm_" + projectId);
+                container.setImage(dockerImage);
+            }
+        }
         //4-4 创建
-        v1PodSpec.setContainers(containers);
-        v1PodTemplateSpec.setSpec(v1PodSpec);
-        v1JobSpec.setTemplate(v1PodTemplateSpec);
-        yaml.setSpec(v1JobSpec);
-        batchV1Api.createNamespacedJob("test-namespace1", yaml, null, null, null);
+        yaml.setSpec(job);
+        batchV1Api.createNamespacedJob("simulation-task", yaml, null, null, null);
 
     }
 

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/controller/TaskController.java

@@ -2,19 +2,28 @@ package com.css.simulation.resource.scheduler.controller;
 
 
 import com.css.simulation.resource.scheduler.service.TaskService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.cloud.context.config.annotation.RefreshScope;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.RestController;
 
-import javax.annotation.Resource;
-
+@RefreshScope
 @RestController
 @RequestMapping("/task")
 public class TaskController {
 
-    @Resource
+    @Autowired
     private TaskService taskService;
+    @Value("${hello}")
+    private String hello;
+
+    @GetMapping("/hello")
+    public String hello(){
+        return hello;
+    }
 
     /**
      * Pod 的心跳接口

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

@@ -1,12 +1,10 @@
 package com.css.simulation.resource.scheduler.mapper;
 
 
-import com.css.simulation.resource.scheduler.pojo.po.ScenePO;
+import com.css.simulation.resource.scheduler.pojo.po.AlgorithmPO;
 import org.apache.ibatis.annotations.*;
 import org.apache.ibatis.type.JdbcType;
 
-import java.util.List;
-
 /**
  * 场景查询
  * scene_natural
@@ -16,10 +14,17 @@ import java.util.List;
 @Mapper
 public interface AlgorithmMapper {
 
-    @Select("select minio_path\n" +
+    @Results(id = "index", value = {
+            @Result(column = "minio_path", property = "minioPath", jdbcType = JdbcType.VARCHAR),
+            @Result(column = "docker_import", property = "dockerImport", jdbcType = JdbcType.VARCHAR),
+            @Result(column = "docker_image", property = "dockerImage", jdbcType = JdbcType.VARCHAR)
+    })
+    @Select("select minio_path,\n" +
+            "   docker_import,\n" +
+            "   docker_image\n" +
             "from algorithm\n" +
             "where is_deleted = '0'\n" +
             "  and id = #{id}")
-    String selectMinioPathById(@Param("id") String id);
+    AlgorithmPO selectById(@Param("id") String id);
 
 }

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/pojo/po/AlgorithmPO.java

@@ -0,0 +1,14 @@
+package com.css.simulation.resource.scheduler.pojo.po;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class AlgorithmPO {
+    private String minioPath;
+    private String dockerImport;
+    private String dockerImage;
+}

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

@@ -15,7 +15,7 @@ import java.util.List;
 @Component
 public class TickScheduler {
 
-    @Value("${spring.kafka.consumer.topic.manual-project}")
+    @Value("${scheduler.manual-project.topic}")
     private String manualProjectTopic;
     @Autowired
     private CommonService commonService;

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

@@ -48,18 +48,18 @@ public class TaskService {
     private TaskIndexMapper taskIndexMapper;
     @Autowired
     private IndexTemplateMapper indexTemplateMapper;
-    @Value("${spring.kafka.consumer.topic.manual-project}")
+    @Value("${scheduler.manual-project.topic}")
     private String manualProjectTopic;
-    @Value("${scheduler.manual-project.score.hostname}")
+    @Value("${scheduler.score.hostname}")
     static String hostname;
-    @Value("${scheduler.manual-project.score.username}")
+    @Value("${scheduler.score.username}")
     static String username;
-    @Value("${scheduler.manual-project.score.password}")
+    @Value("${scheduler.score.password}")
     static String password;
-    @Value("${scheduler.manual-project.score.py-path}")
+    @Value("${scheduler.score.py-path}")
     static String pyPath;
-    @Value("${scheduler.manual-project.result-path-linux}")
-    String manualProjectResultPathLinux;
+    @Value("${scheduler.linux-temp-path}")
+    String linuxTempPath;
 
     public void taskTick(String taskId) {
         // 刷新 redis 心跳时间
@@ -115,7 +115,7 @@ public class TaskService {
                         // 计算每个任务的得分
                         ScoreTO score;
                         String runResultMinio = task2.getRunResult();
-                        String runResultLinux = manualProjectResultPathLinux + runResultMinio.substring(1);
+                        String runResultLinux = linuxTempPath + runResultMinio;
                         String command = "python3 " + pyPath + " " + runResultLinux + " " + task2.getSceneType();  // 默认使用场景名称找打分脚本
 //                        String command = "python3 " + pyPath + " " + runResultLinux + " " + task2.getSceneType() + " " + ruleDetails; // 指定打分脚本
                         try {

simulation-resource-scheduler/src/main/resources/kubernetes/template/job/job-template.yaml

@@ -2,15 +2,61 @@
 apiVersion: batch/v1
 kind: Job
 metadata:
-  name: pi    # job 名称
+  name: job-vtd-perception-1
+  namespace: default
+  labels:
+    user: EY
 spec:
-  completions: 1
-  parallelism: 1
   template:
+    metadata:
+      name: pod-vtd-perception
     spec:
-      restartPolicy: Never
       containers:
-        - image: perl
-          imagePullPolicy: IfNotPresent
-          name: pi
-          command: [ "perl", "-Mbignum=bpi","-wle","print bpi(2000)" ]
+        - name: vtd
+          image: vtd.run.perception:latest
+          imagePullPolicy: Never
+          command: [ "/Controller/VTDController", "/Controller/config/local_docker.ini" ]
+          env:
+            - name: NodeID
+              valueFrom:
+                fieldRef:
+                  fieldPath: metadata.name
+            - name: LM_LICENSE_FILE
+              value: 27500@192.168.10.179
+          ports:
+            - name: ROS-Master-IP
+              containerPort: 11311
+              protocol: TCP
+          volumeMounts:
+            - name: nvidia0
+              mountPath: /dev/nvidia0
+            - name: nvidiactl
+              mountPath: /dev/nvidiactl
+          securityContext:
+            privileged: true
+        - name: algorithm
+          image: sharing_van_algorithm:latest
+          imagePullPolicy: Never
+          command: [ "/simulation_workspace/start.sh" ]
+          env:
+            - name: VtdIP
+              valueFrom:
+                fieldRef:
+                  fieldPath: status.podIP
+            - name: VehicleID
+              value: 1
+            - name: SensorPort
+              value: 50000
+          ports:
+            - name: ROS-Master-IP
+              containerPort: 11311
+              protocol: TCP
+      restartPolicy: Never
+      volumes:
+        - name: nvidia0
+          hostPath:
+            path: /dev/nvidia0
+        - name: nvidiactl
+          hostPath:
+            path: /dev/nvidiactl
+

simulation-resource-server/pom.xml

@@ -8,7 +8,6 @@
         <version>1.0</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
-    <groupId>com.css</groupId>
     <artifactId>simulation-resource-server</artifactId>
 
     <properties>