diff --git a/ruoyi-modules/management-platform/src/main/java/com/ruoyi/platform/scheduling/AppScheduleTask.java b/ruoyi-modules/management-platform/src/main/java/com/ruoyi/platform/scheduling/AppScheduleTask.java index fa003b4b..d1c36fa8 100644 --- a/ruoyi-modules/management-platform/src/main/java/com/ruoyi/platform/scheduling/AppScheduleTask.java +++ b/ruoyi-modules/management-platform/src/main/java/com/ruoyi/platform/scheduling/AppScheduleTask.java @@ -144,16 +144,10 @@ public class AppScheduleTask { */ private void startCSAppTask(TrainingTask trainingTask) throws Exception { //异步调用超算任务执行接口 - JSONObject taskInfo = JSONUtil.parseObj(trainingTask.getTaskInfo()); - String clusterId = (String) taskInfo.get("type"); - Integer ntasks = (Integer) taskInfo.get("ntasks"); - Integer nodes = (Integer) taskInfo.get("nodes"); - Integer objectId = (Integer) taskInfo.getByPath("input_file.file_id"); - String name = taskInfo.get("name").toString(); // 异步提交任务 CompletableFuture future = CompletableFuture.supplyAsync(() -> { try { - return csCollectService.submit(clusterId, name, "fileName", objectId, ntasks, nodes); + return csCollectService.submit(trainingTask); } catch (Exception e) { throw new RuntimeException("异步提交任务失败", e); } diff --git a/ruoyi-modules/management-platform/src/main/java/com/ruoyi/platform/service/CSCollectService.java b/ruoyi-modules/management-platform/src/main/java/com/ruoyi/platform/service/CSCollectService.java index baf31a46..aef41d8b 100644 --- a/ruoyi-modules/management-platform/src/main/java/com/ruoyi/platform/service/CSCollectService.java +++ b/ruoyi-modules/management-platform/src/main/java/com/ruoyi/platform/service/CSCollectService.java @@ -2,6 +2,7 @@ package com.ruoyi.platform.service; import cn.hutool.json.JSONArray; import cn.hutool.json.JSONObject; +import com.ruoyi.platform.domain.TrainingTask; import org.springframework.web.multipart.MultipartFile; import java.util.HashMap; @@ -12,7 +13,7 @@ public interface CSCollectService { Integer uploadFile(String appCode, MultipartFile file) throws Exception; - String submit(String clusterId, String jobName, String fileName, Integer objectId, Integer ntasks, Integer nodes) throws Exception; + String submit(TrainingTask trainingTask) throws Exception; String getJobLogs(Long taskId) throws Exception; diff --git a/ruoyi-modules/management-platform/src/main/java/com/ruoyi/platform/service/impl/CSCollectServiceImpl.java b/ruoyi-modules/management-platform/src/main/java/com/ruoyi/platform/service/impl/CSCollectServiceImpl.java index 4c13e521..0f1673d0 100644 --- a/ruoyi-modules/management-platform/src/main/java/com/ruoyi/platform/service/impl/CSCollectServiceImpl.java +++ b/ruoyi-modules/management-platform/src/main/java/com/ruoyi/platform/service/impl/CSCollectServiceImpl.java @@ -9,11 +9,11 @@ import cn.hutool.json.JSONArray; import cn.hutool.json.JSONObject; import cn.hutool.json.JSONUtil; import com.ruoyi.common.redis.service.RedisService; +import com.ruoyi.platform.domain.TrainingTask; import com.ruoyi.platform.domain.service.CSAuthInfo; import com.ruoyi.platform.service.CSCollectService; import com.ruoyi.platform.utils.HttpUtils; import com.ruoyi.platform.utils.MinioUtil; -import org.apache.http.client.methods.CloseableHttpResponse; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import lombok.extern.slf4j.Slf4j; @@ -23,7 +23,9 @@ import org.springframework.web.multipart.MultipartFile; import javax.annotation.Resource; import java.io.*; +import java.util.ArrayList; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; @@ -106,30 +108,23 @@ public class CSCollectServiceImpl implements CSCollectService { /** * 任务提交执行 * - * @param clusterId - * @param jobName - * @param fileName - * @param objectId - * @return taskId * @throws Exception + * example */ @Override - public String submit(String clusterId, String jobName, String fileName, Integer objectId, Integer ntasks, Integer nodes) throws Exception { - String url = csUrl + "/jsm/v2/jobs/submit"; - String submitParam = "{\"userID\":137,\"jobSetInfo\":{\"jobs\":[{\"localJobID\":\"1\",\"type\":\"HPC\",\"name\":\"test1166\",\"clusterId\":\"1865927992266463180\"," + - "\"backend\":\"slurm\",\"app\":\"lammps\",\"operateType\":\"\",\"parameters\":{\"inputFile\":\"in.lj\",\"hpcBindingFiles\":[{\"paramName\":\"inputFile\"," + - "\"resource\":{\"type\":\"object\",\"objectID\":70797}}],\"ntasks\":\"1\",\"nodes\":\"2\"}},{\"localJobID\":\"4\",\"type\":\"DataReturn\"," + - "\"targetJob\":[{\"targetJobID\":\"1\",\"inputParams\":{\"PackageName\":\"Name\",\"ClusterID\":\"ClusterID\",\"Output\":\"Output\"}}]}]}}"; + public String submit(TrainingTask trainingTask) throws Exception { + JSONObject taskInfo = JSONUtil.parseObj(trainingTask.getTaskInfo()); + JSONObject taskParam = JSONUtil.parseObj(trainingTask.getTaskParam()); + String clusterId = (String) taskInfo.get("type"); + String ntasks = (String) taskInfo.get("ntasks"); + String nodes = (String) taskInfo.get("nodes"); + String jobName = taskInfo.get("name").toString(); - JSONObject params = JSONUtil.parseObj(submitParam); CSAuthInfo csAuthInfo = getZSAuthInfoFromRedis(); - params.set("userID", csAuthInfo.getUserId()); - params.putByPath("jobSetInfo.jobs[0].name", jobName); - params.putByPath("jobSetInfo.jobs[0].clusterId", clusterId); - params.putByPath("jobSetInfo.jobs[0].parameters.inputFile", fileName); - params.putByPath("jobSetInfo.jobs[0].parameters.ntasks", ntasks.toString()); - params.putByPath("jobSetInfo.jobs[0].parameters.nodes", nodes.toString()); - params.putByPath("jobSetInfo.jobs[0].parameters.hpcBindingFiles[0].resource.objectID", objectId); + List> hpcBindingFiles = buildHpcBindingFiles(taskParam, taskInfo); + JSONObject params = convertToJobRequest(clusterId, jobName, hpcBindingFiles, csAuthInfo.getUserId(), ntasks, nodes); + + String url = csUrl + "/jsm/v2/jobs/submit"; String res = HttpUtils.sendBodyPostWithToken(url, params, csAuthInfo.getToken()); JSONObject resData = returnResDataWithOK(res); JSONObject taskIdObj = JSONUtil.parseObj(getByPath(resData, "message")); @@ -177,6 +172,119 @@ public class CSCollectServiceImpl implements CSCollectService { return resData.get("outLogs").toString(); } + /** + * 根据 taskParam（表单定义）和 taskInfo（表单提交值）提取文件字段，构建 hpcBindingFiles 列表 + * @return hpcBindingFiles 列表，每个元素包含 paramName 和 objectID + */ + public static List> buildHpcBindingFiles(JSONObject taskParam, JSONObject taskInfo) { + List> bindingFiles = new ArrayList<>(); + // 遍历 schema 的所有字段 + for (Map.Entry entry : taskParam.entrySet()) { + String fieldName = entry.getKey(); + JSONObject fieldDef = (JSONObject) entry.getValue(); + + // 判断是否为文件字段（file_field 为 true） + Boolean isFileField = fieldDef.getBool("file_field"); + if (isFileField != null && isFileField) { + // 从 formData 中获取该字段的值 + Object fileValue = taskInfo.get(fieldName); + if (fileValue instanceof JSONObject) { + JSONObject fileObj = (JSONObject) fileValue; + // 提取 file_id 和 filename + Integer objectId = fileObj.getInt("file_id"); + + // 构建 binding 元素 + Map binding = new HashMap<>(); + binding.put("paramName", fieldName); + binding.put("objectID", objectId); + bindingFiles.add(binding); + } + } + } + + return bindingFiles; + } + + /** + * 将任务提交参数转换为后端所需格式 + * @param clusterId 集群ID + * @param name 任务名称 + * @param hpcBindingFiles 文件绑定列表，每个元素包含 paramName 和 objectID + * @param userId 用户ID（原入参中的 objectId 视为用户ID） + * @param ntasks 并行任务数 + * @param nodes 节点数 + * @return 符合JSON串3结构的JSONObject + */ + public JSONObject convertToJobRequest(String clusterId, String name, + List> hpcBindingFiles, + Integer userId, String ntasks, String nodes) { + // 根对象 + JSONObject result = new JSONObject(); + result.set("userID", userId); + + // jobSetInfo 对象 + JSONObject jobSetInfo = new JSONObject(); + JSONArray jobs = new JSONArray(); + + // ----- HPC 任务 (localJobID = "1") ----- + JSONObject hpcJob = new JSONObject(); + hpcJob.set("localJobID", "1"); + hpcJob.set("type", "HPC"); + hpcJob.set("name", name); + hpcJob.set("clusterId", clusterId); + hpcJob.set("backend", "slurm"); + hpcJob.set("app", "specfem3d_globe"); + hpcJob.set("operateType", ""); + + // parameters 对象 + JSONObject parameters = new JSONObject(); + parameters.set("ntasks", ntasks); + parameters.set("nodes", nodes); + + // hpcBindingFiles 数组 + JSONArray bindingArray = new JSONArray(); + if (hpcBindingFiles != null) { + for (Map fileInfo : hpcBindingFiles) { + JSONObject binding = new JSONObject(); + binding.set("paramName", fileInfo.get("paramName")); + + JSONObject resource = new JSONObject(); + resource.set("type", "object"); + resource.set("objectID", fileInfo.get("objectID")); + + binding.set("resource", resource); + bindingArray.set(binding); + } + } + parameters.set("hpcBindingFiles", bindingArray); + hpcJob.set("parameters", parameters); + jobs.set(hpcJob); + + // ----- DataReturn 任务 (localJobID = "4") ----- + JSONObject dataReturnJob = new JSONObject(); + dataReturnJob.set("localJobID", "4"); + dataReturnJob.set("type", "DataReturn"); + + JSONArray targetJobArray = new JSONArray(); + JSONObject targetJob = new JSONObject(); + targetJob.set("targetJobID", "1"); + + JSONObject inputParams = new JSONObject(); + inputParams.set("PackageName", "Name"); + inputParams.set("ClusterID", "ClusterID"); + inputParams.set("Output", "Output"); + targetJob.set("inputParams", inputParams); + + targetJobArray.set(targetJob); + dataReturnJob.set("targetJob", targetJobArray); + jobs.set(dataReturnJob); + + jobSetInfo.set("jobs", jobs); + result.set("jobSetInfo", jobSetInfo); + + return result; + } + /** * 创建包 *