mlx

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/common/util/KubernetesUtil.java

@@ -13,6 +13,7 @@ import org.apache.sshd.client.session.ClientSession;
 
 import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 @Slf4j
@@ -208,8 +209,7 @@ public class KubernetesUtil {
      * @param namespaceName namespace 名称
      * @param podName       pod 名称
      */
-    @SneakyThrows
-    public static void deletePod(ApiClient apiClient, String namespaceName, String podName) {
+    public static void deletePod(ApiClient apiClient, String namespaceName, String podName) throws InterruptedException, ApiException {
         deletePodSync(apiClient, namespaceName, podName);
     }
 
@@ -221,8 +221,22 @@ public class KubernetesUtil {
      * @param namespaceName namespace 名称
      * @param podName       pod 名称
      */
-    @SneakyThrows
-    public static void deletePodSync(ApiClient apiClient, String namespaceName, String podName) {
+    public static void deletePod2(ApiClient apiClient, String namespaceName, String podName) throws InterruptedException, ApiException {
+        deletePodSync(apiClient, namespaceName, podName);
+        log.info("deletePod2() 等待 pod " + podName + " 的资源释放完成。");
+        TimeUnit.SECONDS.sleep(7);
+        log.info("deletePod2()  pod " + podName + " 资源释放完成。");
+    }
+
+    /**
+     * 删除 pod
+     * 异步删除会等待 pod 完全删除完毕，即资源释放完全
+     *
+     * @param apiClient     api 客户端
+     * @param namespaceName namespace 名称
+     * @param podName       pod 名称
+     */
+    public static void deletePodSync(ApiClient apiClient, String namespaceName, String podName) throws ApiException, InterruptedException {
         log.info("KubernetesUtil--deletePod 删除 " + namespaceName + ":" + podName);
         CoreV1Api coreV1Api = new CoreV1Api(apiClient);
         coreV1Api.deleteNamespacedPod(podName, namespaceName, null, null, null, null, null, null);

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/common/util/ProjectUtil.java

@@ -14,6 +14,7 @@ import com.css.simulation.resource.scheduler.service.domain.KubernetesNodeTO;
 import com.css.simulation.resource.scheduler.service.domain.NodeTO;
 import com.css.simulation.resource.scheduler.service.domain.PrefixTO;
 import io.kubernetes.client.openapi.ApiClient;
+import io.kubernetes.client.openapi.ApiException;
 import lombok.SneakyThrows;
 import lombok.Synchronized;
 import lombok.extern.slf4j.Slf4j;
@@ -58,9 +59,10 @@ public class ProjectUtil {
     @Resource
     private KubernetesConfiguration kubernetesConfiguration;
     @Resource
-    private StringRedisTemplate stringRedisTemplate;
-    @Resource
     private ApiClient apiClient;
+    @Resource
+    private StringRedisTemplate stringRedisTemplate;
+
 
 
     @SneakyThrows
@@ -114,13 +116,16 @@ public class ProjectUtil {
             log.info("deletePod() 等待 pod " + podName + " 的资源释放完成。");
             TimeUnit.SECONDS.sleep(10);
             log.info("deletePod() 删除 pod 并删除 redis 键值对：" + key);
+        } catch (ApiException apiException) {
+            log.info("deletePod() pod " + podName + " 已删除。");
         } catch (Exception e) {
             e.printStackTrace();
-            log.error("deletePod() pod " + podName + " 不存在。", e);
+            log.error("deletePod() 删除 pod " + podName + " 报错。", e);
         }
     }
 
 
+
     public String getNodeNameOfPod(String podName) {
         return stringRedisTemplate.opsForValue().get("pod:" + podName + ":node");
     }

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

@@ -3,6 +3,8 @@ package com.css.simulation.resource.scheduler.dao.manager;
 import api.common.pojo.common.ResponseBodyVO;
 import api.common.pojo.constants.DictConstants;
 import api.common.util.*;
+import com.css.simulation.resource.scheduler.common.configuration.kubernetes.KubernetesConfiguration;
+import com.css.simulation.resource.scheduler.common.util.KubernetesUtil;
 import com.css.simulation.resource.scheduler.dao.mapper.*;
 import com.css.simulation.resource.scheduler.service.feign.VideoService;
 import com.css.simulation.resource.scheduler.dao.entity.IndexTemplatePO;
@@ -17,6 +19,7 @@ import com.css.simulation.resource.scheduler.common.util.ProjectUtil;
 import com.css.simulation.resource.scheduler.common.util.RedisUtil;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import io.kubernetes.client.openapi.ApiClient;
 import io.minio.MinioClient;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
@@ -92,6 +95,10 @@ public class TaskManager {
     private SqlSessionFactory sqlSessionFactory;
     @Resource
     private TaskLock taskLock;
+    @Resource
+    private KubernetesConfiguration kubernetesConfiguration;
+    @Resource
+    private ApiClient apiClient;
 
     public void batchInsertTask(List<TaskPO> taskPOList) {
         try (SqlSession sqlSession = sqlSessionFactory.openSession(ExecutorType.BATCH, false)) {
@@ -161,6 +168,10 @@ public class TaskManager {
                 if (!result) {
                     log.info("isCompleted() 项目 " + projectId + " 还未运行完成。");
                     projectUtil.createNextPod3(projectId, nodeName, podName);
+                } else {
+                    //如果项目已完成先把 pod 删除，并归还并行度
+                    KubernetesUtil.deletePod(apiClient, kubernetesConfiguration.getNamespace(), podName);
+                    projectUtil.addOneParallelismToNode(nodeName);
                 }
             } catch (Exception exception) {
                 log.info("isCompleted() 报错。", exception);
@@ -503,16 +514,16 @@ public class TaskManager {
 //        SshUtil.stop(clientKafka, sessionKafka);
 
 
-        // 归还并行度
-        Set<String> nodeOfPodKeySet = RedisUtil.getKeySetByPrefix(stringRedisTemplate, "pod:project-" + projectId);
-        for (String nodeOfPodKey : nodeOfPodKeySet) {
-            String podName = nodeOfPodKey.split(":")[1];
-            String nodeName = projectUtil.getNodeNameOfPod(podName);
-            // 删除 pod
-            projectUtil.deletePod(podName);
-            // 节点并行度加一
-            projectUtil.addOneParallelismToNode(nodeName);
-        }
+//        // 归还并行度
+//        Set<String> nodeOfPodKeySet = RedisUtil.getKeySetByPrefix(stringRedisTemplate, "pod:project-" + projectId);
+//        for (String nodeOfPodKey : nodeOfPodKeySet) {
+//            String podName = nodeOfPodKey.split(":")[1];
+//            String nodeName = projectUtil.getNodeNameOfPod(podName);
+//            // 删除 pod
+//            projectUtil.deletePod(podName);
+//            // 节点并行度加一
+//            projectUtil.addOneParallelismToNode(nodeName);
+//        }
 
         // 删除 redis 中的 项目运行信息 键值对
         RedisUtil.deleteByPrefix(stringRedisTemplate, redisPrefix.getProjectRunningKey());

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/common/scheduler/AlgorithmScheduler.java → simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/web/scheduler/AlgorithmScheduler.java

@@ -1,4 +1,4 @@
-package com.css.simulation.resource.scheduler.common.scheduler;
+package com.css.simulation.resource.scheduler.web.scheduler;
 
 import api.common.util.SshUtil;
 import com.css.simulation.resource.scheduler.common.configuration.docker.DockerConfiguration;

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/common/scheduler/ClusterScheduler.java → simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/web/scheduler/ClusterScheduler.java

@@ -1,4 +1,4 @@
-package com.css.simulation.resource.scheduler.common.scheduler;
+package com.css.simulation.resource.scheduler.web.scheduler;
 
 import api.common.pojo.constants.DictConstants;
 import api.common.util.CollectionUtil;