Application initialization is encapsulated into an interface, including configuration and task upload

Author: v_xiangbiaowu

docs/zh_CN/0.2.0/使用文档/Streamis用户手册.md

@@ -60,6 +60,7 @@
 
 	xxx.zip
 	    ├── meta.json
+	    ├── conf.json  #可选文件
 	    ├── test.sql
 	    ├── test.jar
 	    ├── file3
@@ -109,6 +110,34 @@
 }
 ```
 
+其中conf.json是job的配置文件，是一个可选的文件，格式为：
+```
+{
+    "wds.linkis.flink.resource": {      //资源配置
+        "wds.linkis.flink.app.parallelism": "4",              #parallelism并行度，默认4
+        "wds.linkis.flink.jobmanager.memory": "1024",         #JobManager Memory，默认1024
+        "wds.linkis.flink.taskmanager.cpus": "2",             #TaskManager CPUs，默认2
+        "wds.linkis.flink.taskmanager.memory": "4096",        #TaskManager Memory，默认4096
+        "wds.linkis.flink.taskmanager.numberOfTaskSlots": "2",#TaskManager Slot数量，默认2
+        "wds.linkis.rm.yarnqueue": ""                         #使用Yarn队列
+    },
+    "wds.linkis.flink.custom": {        #Flink参数，K/V可自定义
+        "aa": "aaa", 
+        "bb": "bbb",
+        ...
+    },
+    "wds.linkis.flink.produce": {       #生产配置
+        "wds.linkis.flink.checkpoint.switch": "OFF",           #Checkpoint开关
+        "wds.linkis.flink.savepoint.path": "",                  #快照文件位置
+        "wds.linkis.flink.app.fail-restart.switch": "OFF",     #作业失败自动拉起开关
+        "wds.linkis.flink.app.start-auto-restore.switch": "ON" #作业启动状态自恢复
+    },
+    "wds.linkis.flink.authority": {      #权限设置
+        "wds.linkis.flink.authority.visible": ""        #可见人员
+    }
+}
+```
+
 ### 4.2.2 示例
 
 ​	streamisjobtest为flinksql文件，meta.json是该任务的元数据信息。

streamis-jobmanager/streamis-job-manager/streamis-job-manager-base/src/main/java/com/webank/wedatasphere/streamis/jobmanager/manager/util/ReaderUtils.java

@@ -32,6 +32,7 @@
 public class ReaderUtils {
     private static final String metaFileName = "meta.txt";
     private static final String metaFileJsonName = "meta.json";
+    private static final String confFileJsonName = "conf.json";
     private static final String type = "type";
     private static final String fileName = "filename";
     private static final String projectName = "projectname";
@@ -86,7 +87,7 @@ public List<String> listFiles(String path) throws FileException {
         File[] files = file.listFiles();
         if (files != null) {
             for (File filePath : files) {
-                if (!metaFileJsonName.equals(filePath.getName()) && !filePath.getName().endsWith(".zip")) {
+                if (!metaFileJsonName.equals(filePath.getName()) &&!confFileJsonName.equals(filePath.getName()) && !filePath.getName().endsWith(".zip")) {
                     if (!checkName(filePath.getName())) {
                         throw FileExceptionManager.createException(30601,filePath.getName());
                     }
@@ -151,6 +152,38 @@ private InputStream generateInputStream(String basePath) throws IOException, Fil
         return IoUtils.generateInputInputStream(basePath + File.separator + metaFileJsonName);
     }
 
+    public Map<String, Object> parseConfJson(String dirPath) throws IOException, FileException {
+        return parseJson(dirPath, confFileJsonName);
+    }
+
+    public Map<String, Object> parseJson(String dirPath,String fimeName) throws IOException, FileException {
+        getBasePath(dirPath);
+        try (InputStream inputStream = generateInputStream(basePath,fimeName);
+             InputStreamReader streamReader = new InputStreamReader(inputStream);
+             BufferedReader reader = new BufferedReader(streamReader);) {
+            return readJson(reader,HashMap.class);
+        }
+    }
+
+    private InputStream generateInputStream(String basePath,String fimeName) throws IOException, FileException {
+        File metaFile = new File(basePath + File.separator + fimeName);
+        if (!metaFile.exists()) {
+            throw new FileException(30603, fimeName);
+        }
+        return IoUtils.generateInputInputStream(basePath + File.separator + fimeName);
+    }
+
+    private <T> T readJson(BufferedReader reader,Class<T> valueType) throws IOException {
+        String line = null;
+        StringBuilder sb = new StringBuilder();
+        while ((line = reader.readLine()) != null) {
+            sb.append(line);
+        }
+        ObjectMapper objectMapper = new ObjectMapper();
+        return objectMapper.readValue(sb.toString(), valueType);
+    }
+
+
     private PublishRequestVo read(InputStream inputStream) throws IOException, FileException {
         try (InputStreamReader streamReader = new InputStreamReader(inputStream);
              BufferedReader reader = new BufferedReader(streamReader);) {

streamis-jobmanager/streamis-job-manager/streamis-job-manager-service/src/main/scala/com/webank/wedatasphere/streamis/jobmanager/manager/service/DefaultStreamJobService.scala

@@ -16,7 +16,8 @@
 package com.webank.wedatasphere.streamis.jobmanager.manager.service
 
 import java.util
-import java.util.Date
+import java.util.{Date, Map}
+
 import com.github.pagehelper.PageInfo
 import com.webank.wedatasphere.streamis.jobmanager.launcher.conf.JobConfKeyConstants
 import com.webank.wedatasphere.streamis.jobmanager.launcher.service.StreamJobConfService
@@ -35,8 +36,9 @@ import org.apache.commons.lang.StringUtils
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Service
 import org.springframework.transaction.annotation.Transactional
-
 import javax.annotation.Resource
+import org.apache.commons.collections.MapUtils
+
 import scala.collection.JavaConverters._
 
 
@@ -178,8 +180,15 @@ class DefaultStreamJobService extends StreamJobService with Logging {
     }
     validateUpload(metaJsonInfo.getProjectName, metaJsonInfo.getJobName, userName)
     //  生成StreamJob，根据StreamJob生成StreamJobVersion
-    val version = createStreamJob(metaJsonInfo, userName)
-    //  上传所有非meta.json的文件
+    val version = createStreamJob(metaJsonInfo, userName);
+    val configContent = readerUtils.parseConfJson(inputPath).asInstanceOf[util.Map[String, Any]]
+    if(MapUtils.isNotEmpty(configContent)){
+      info(s"mapContent ${configContent} parsed according to the conf.json file")
+      streamJobConfService.saveJobConfig(version.getJobId, configContent)
+    }else{
+      logger.warn("the conf.json file does not exist in the ZIP format material package")
+    }
+    //  上传所有非meta.json、conf.json、.zip的文件
     uploadFiles(metaJsonInfo, version, inputZipPath)
     version
   }

streamis-jobmanager/streamis-jobmanager-server/src/main/java/com/webank/wedatasphere/streamis/jobmanager/restful/api/UploadRestfulApi.java

@@ -24,8 +24,10 @@
 import com.webank.wedatasphere.streamis.jobmanager.manager.util.IoUtils;
 import com.webank.wedatasphere.streamis.jobmanager.manager.util.ZipHelper;
 import org.apache.commons.io.FileUtils;
+import org.apache.commons.io.FilenameUtils;
 import org.apache.commons.io.IOUtils;
 import org.apache.commons.lang.exception.ExceptionUtils;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.linkis.server.Message;
 import org.apache.linkis.server.security.SecurityFilter;
 import org.slf4j.Logger;
@@ -99,4 +101,44 @@ public Message uploadJar(HttpServletRequest request,
             IOUtils.closeQuietly(is);
         }
     }
+
+    @RequestMapping(path = "/initJob", method = RequestMethod.POST)
+    public Message initJob(HttpServletRequest request,
+                           @RequestParam(name = "projectName", required = false) String projectName,
+                           @RequestParam(name = "jobFilePath", required = false) String jobFilePath) throws IOException, JobException {
+        String userName = SecurityFilter.getLoginUsername(request);
+        if (StringUtils.isBlank(projectName)){
+            return Message.error("Project name cannot be empty(项目名不能为空，请指定)");
+        }
+        if (StringUtils.isBlank(jobFilePath)){
+            return Message.error("jobFilePath can not be blank");
+        }
+        if(!ZipHelper.isZip(jobFilePath)){
+            throw JobExceptionManager.createException(30302);
+        }
+        if (!projectPrivilegeService.hasEditPrivilege(request, projectName)) return Message.error("the current user has no operation permission");
+
+        InputStream is = null;
+        OutputStream os = null;
+        try{
+            LOG.info("user {} start initJob to project {} according to jobFilePath {}.",userName, projectName,jobFilePath);
+            String fileName = FilenameUtils.getName(jobFilePath);
+            String inputPath = IoUtils.generateIOPath(userName, "streamis", fileName);
+            File file = new File(inputPath);
+            if(file.getParentFile().exists()){
+                FileUtils.deleteDirectory(file.getParentFile());
+            }
+            is = IoUtils.generateInputInputStream(jobFilePath);
+            os = IoUtils.generateExportOutputStream(inputPath);
+            IOUtils.copy(is, os);
+            StreamJobVersion job = streamJobService.uploadJob(projectName, userName, inputPath);
+            return Message.ok().data("jobId",job.getJobId());
+        } catch (Exception e){
+            LOG.error("Failed to init {} to project {} for user {}.", jobFilePath, projectName, userName, e);
+            return Message.error(ExceptionUtils.getRootCauseMessage(e));
+        } finally{
+            IOUtils.closeQuietly(os);
+            IOUtils.closeQuietly(is);
+        }
+    }
 }