mlx

api-common/src/main/java/api/common/pojo/po/home/SystemServerPO.java

@@ -16,7 +16,7 @@ public class SystemServerPO extends CommonPO {
     String serverAddress;
     String serverType;
     Integer memoryUsage;
-    Double memoryAvailable;
+    Integer memoryAvailable;
     Integer memoryTotal;
     Integer diskUsage;
     Double diskAvailable;

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

@@ -62,8 +62,6 @@ public class SshUtil {
     }
 
 
-
-
     /**
      * @param session 会话
      * @param command 命令
@@ -160,6 +158,17 @@ public class SshUtil {
         return map;
     }
 
+    /**
+     * 显存总量
+     *
+     * @param session 会话
+     * @return 显存总量，单位 MB
+     */
+    public static int gpuTotal(ClientSession session) throws DocumentException, IOException {
+        List<GpuDTO> gpuDTOList = SshUtil.gpuInfo(session);
+        return gpuDTOList.stream().mapToInt(gpu -> Integer.parseInt(gpu.getTotalMemory().split(" ")[0])).sum();
+    }
+
 
     /**
      * 功能：CPU使用信息
@@ -240,12 +249,12 @@ public class SshUtil {
     }
 
     /**
-     * 剩余可用内存
+     * 剩余可用内存(kb)
      *
      * @param session 会话
-     * @return 剩余可用内存
+     * @return 剩余可用内存(kb)
      */
-    public static double memoryAvailable(ClientSession session) throws IOException {
+    public static long memoryAvailable(ClientSession session) throws IOException {
         String execute = SshUtil.execute(session, "cat /proc/meminfo");
         String[] split = execute.split("\n");
         Map<String, Object> map = new HashMap<>();
@@ -274,6 +283,41 @@ public class SshUtil {
     }
 
 
+    /**
+     * 总内存(kb)
+     *
+     * @param session 会话
+     * @return 总内存(kb)
+     */
+    public static long memoryTotal(ClientSession session) throws IOException {
+        String execute = SshUtil.execute(session, "cat /proc/meminfo");
+        String[] split = execute.split("\n");
+        Map<String, Object> map = new HashMap<>();
+        for (String line : split) {
+            if (line == null)
+                break;
+            int beginIndex = 0;
+            int endIndex = line.indexOf(":");
+            if (endIndex != -1) {
+                String key = line.substring(beginIndex, endIndex);
+                beginIndex = endIndex + 1;
+                endIndex = line.length();
+                String memory = line.substring(beginIndex, endIndex);
+                String value = memory.replace("kB", "").trim();
+                map.put(key, value);
+            }
+        }
+
+        long memTotal = Long.parseLong(map.get("MemTotal").toString());
+        long memFree = Long.parseLong(map.get("MemFree").toString());
+        long memused = memTotal - memFree;
+        long buffers = Long.parseLong(map.get("Buffers").toString());
+        long cached = Long.parseLong(map.get("Cached").toString());
+
+        return memTotal;
+    }
+
+
     /**
      * 磁盘使用率
      *

simulation-resource-monitor/src/main/java/com/css/simulation/resource/monitor/scheduler/MyScheduler.java

@@ -45,11 +45,13 @@ public class MyScheduler {
                         .serverId(SshUtil.hostname(session))
                         .serverAddress(ip)
                         .serverType(type)
-                        .cpuUsage((int) SshUtil.cpuUsage(session))
                         .memoryUsage((int) SshUtil.memoryUsage(session))
-                        .memoryAvailable(SshUtil.memoryAvailable(session))
+                        .memoryAvailable((int) SshUtil.memoryAvailable(session))
+                        .memoryTotal((int) SshUtil.memoryTotal(session))
                         .diskUsage((int) SshUtil.diskUsage(session))
                         .diskAvailable(SshUtil.diskAvailable(session))
+//                        .diskTotal(SshUtil.diskTotal)
+                        .cpuUsage((int) SshUtil.cpuUsage(session))
 //                        .taskNumber()
 //                        .weight()
 //                        .gpuUsage("gpu".equals(type)?SshUtil.gpu(session).stream().mapto :0)

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

@@ -5,7 +5,6 @@ import io.kubernetes.client.util.ClientBuilder;
 import io.kubernetes.client.util.KubeConfig;
 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;
@@ -16,9 +15,9 @@ public class KubernetesConfiguration {
 
     @Bean
     public ApiClient apiClient() throws IOException {
-        File config = ResourceUtils.getFile("classpath:kubernetes/config");  // 开发环境可用，生产环境不行，无法从jar 包读取
+//        File config = ResourceUtils.getFile("classpath:kubernetes/config");  // 开发环境可用，生产环境不行，无法从jar 包读取
 //        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
+        File config = new File("/root/.kube/config");   //linux
 //
 //        ClassPathResource classPathResource = new ClassPathResource("kubernetes/config");
 //        InputStream inputStream = classPathResource.getInputStream();

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

@@ -65,11 +65,18 @@ public class TaskService {
     @Value("${scheduler.manual-project.result-path-minio}")
     String resultPathMinio;
     @Value("${scheduler.score.hostname}")
-    String hostname;
+    String hostnameScore;
     @Value("${scheduler.score.username}")
-    String username;
+    String usernameScore;
     @Value("${scheduler.score.password}")
-    String password;
+    String passwordScore;
+
+    @Value("${spring.kafka.hostname}")
+    String hostnameKafka;
+    @Value("${spring.kafka.username}")
+    String usernameKafka;
+    @Value("${spring.kafka.password}")
+    String passwordKafka;
     @Value("${scheduler.score.py-path}")
     String pyPath;
     @Value("${scheduler.linux-temp-path}")
@@ -114,17 +121,20 @@ public class TaskService {
         }
         projectMapper.updateProjectState(projectId, DictConstants.PROJECT_COMPLETED);   // 修改该 project 的状态为已完成
         LinuxUtil.execute("kubectl delete job project-" + projectId);
+        ClientSession sessionKafka = SshUtil.getSession(hostnameKafka, usernameKafka, passwordKafka);
+//        SshUtil.execute(sessionKafka, "/opt/module/kafka_2.13-3.1.0/bin/kafka-topics.sh --bootstrap-server " + hostnameKafka + ":9092 --delete --topic " + projectId);
+        SshUtil.execute(sessionKafka, "/opt/module/kafka_2.13-3.1.0/bin/kafka-topics.sh --bootstrap-server " + hostnameKafka + ":9092 --delete --topic test");
         List<TaskPO> taskList = taskMapper.selectTaskListByProjectId(projectId);  // 所有任务信息
         // -------------------------------- 查询叶子指标 --------------------------------
         List<IndexTemplatePO> leafIndexTemplateList = indexTemplateMapper.selectLeafIndexWithRuleDetailsByPackageId(scenePackageId);
         List<TaskIndexPO> leafTaskIndexList = new ArrayList<>();
         log.info("------- /state 共有 " + leafIndexTemplateList.size() + "个叶子节点！");
 
-        ClientSession session = SshUtil.getSession(hostname, username, password);
+        ClientSession sessionScore = SshUtil.getSession(hostnameScore, usernameScore, passwordScore);
         for (int i = 0; i < leafIndexTemplateList.size(); i++) {
             IndexTemplatePO indexTemplatePO = leafIndexTemplateList.get(i);
             String indexId = indexTemplatePO.getIndexId();
-            log.info("------- /state 开始执行对第 " + (i+1) + " 个叶子节点 " + indexId + " 进行打分！");
+            log.info("------- /state 开始执行对第 " + (i + 1) + " 个叶子节点 " + indexId + " 进行打分！");
             String ruleName = indexTemplatePO.getRuleName();    // 打分脚本名称，例如 AEB_1-1
             String ruleDetails = indexTemplatePO.getRuleDetails();    // 打分脚本内容
             String ruleFilePath = pyPath + "script/" + ruleName.split("_")[0] + "/" + ruleName + ".py";
@@ -171,7 +181,7 @@ public class TaskService {
                             }
                             try {
                                 log.info("------- /state 开始执行打分命令：" + command);
-                                score = JsonUtil.jsonToBean(SshUtil.execute(session, command), ScoreTO.class);
+                                score = JsonUtil.jsonToBean(SshUtil.execute(sessionScore, command), ScoreTO.class);
                                 log.info("------- /state 打分结束，结果为：" + score);
                             } catch (IOException e) {
                                 throw new RuntimeException("------- /state 任务 " + task2Id + " 打分出错，命令为：" + command + " 修改状态为：" + DictConstants.TASK_ABORTED + "\n" + e.getMessage());
@@ -208,7 +218,7 @@ public class TaskService {
             leafTaskIndex.setIsDeleted("0");
             leafTaskIndexList.add(leafTaskIndex);
         }
-        SshUtil.stop(session);
+        SshUtil.stop(sessionScore);
 
         // 根据每个指标的得分和权重算出 project 的总得分。
         double totalScore = compute(leafIndexTemplateList);
@@ -247,9 +257,6 @@ public class TaskService {
     }
 
 
-
-
-
     public Boolean taskConfirm(String taskId) {
         // 查询 task 如果不是 pending 则不执行
         String state = taskMapper.selectStateById(taskId);

simulation-resource-server/src/main/java/com/css/simulation/resource/scene/service/HomePageService.java

@@ -46,14 +46,14 @@ public class HomePageService {
         List<SystemServerPO> systemServerPOList = systemServerMapper.selectAll();
 
 
-        int weightSum = systemServerPOList.stream().mapToInt(SystemServerPO::getWeight).sum();  // 权重之和
+//        int weightSum = systemServerPOList.stream().mapToInt(SystemServerPO::getWeight).sum();  // 权重之和
 
         HardwareVO build = HardwareVO.builder()
                 .serverNumber(systemServerPOList.size())
-                .memoryUsage(systemServerPOList.stream().mapToInt(po -> po.getMemoryUsage() * po.getWeight() / weightSum).sum() + "%")
-                .diskUsage(systemServerPOList.stream().mapToInt(po -> po.getDiskUsage() * po.getWeight() / weightSum).sum() + "%")
-                .cpuUsage(systemServerPOList.stream().mapToInt(po -> po.getCpuUsage() * po.getWeight() / weightSum).sum() + "%")
-                .gpuUsage(systemServerPOList.stream().mapToInt(po -> po.getGpuUsage() * po.getWeight() / weightSum).sum() + "%")
+//                .memoryUsage(systemServerPOList.stream().mapToInt(po -> po.getMemoryUsage() * po.getWeight() / weightSum).sum() + "%")
+//                .diskUsage(systemServerPOList.stream().mapToInt(po -> po.getDiskUsage() * po.getWeight() / weightSum).sum() + "%")
+//                .cpuUsage(systemServerPOList.stream().mapToInt(po -> po.getCpuUsage() * po.getWeight() / weightSum).sum() + "%")
+//                .gpuUsage(systemServerPOList.stream().mapToInt(po -> po.getGpuUsage() * po.getWeight() / weightSum).sum() + "%")
                 .build();
 
         return new ResponseBodyVO<>(ResponseBodyVO.Response.SUCCESS, build);