Merge branch 'dev-0.2.3-log-collector' of github.com:WeDataSphere/Streamis into dev-0.2.3-log-collector

Davidhua1996 · Davidhua1996 · commit 7bb566537abb · 2022-10-19T10:39:26.000+08:00
diff --git a/streamis-jobmanager/streamis-job-manager/streamis-job-manager-base/src/main/java/com/webank/wedatasphere/streamis/jobmanager/manager/dao/StreamJobMapper.java b/streamis-jobmanager/streamis-job-manager/streamis-job-manager-base/src/main/java/com/webank/wedatasphere/streamis/jobmanager/manager/dao/StreamJobMapper.java
@@ -30,6 +30,7 @@ List<QueryJobListVo> getJobLists(@Param("projectName") String projectName, @Para
 
     StreamJob getJobById(@Param("jobId") Long jobId);
 
+    List<StreamJob> getJobByName(@Param("jobName") String jobName);
 
     List<StreamJobVersion> getJobVersions(@Param("jobId") Long jobId);
 
diff --git a/streamis-jobmanager/streamis-job-manager/streamis-job-manager-base/src/main/java/com/webank/wedatasphere/streamis/jobmanager/manager/dao/impl/StreamJobMapper.xml b/streamis-jobmanager/streamis-job-manager/streamis-job-manager-base/src/main/java/com/webank/wedatasphere/streamis/jobmanager/manager/dao/impl/StreamJobMapper.xml
@@ -86,6 +86,11 @@
 <!--        linkis_stream_job j,linkis_stream_job_version l WHERE j.id = l.job_id and j.id=#{jobId}-->
     </select>
 
+    <select id="getJobByName" resultType="com.webank.wedatasphere.streamis.jobmanager.manager.entity.StreamJob">
+        SELECT <include refid="Job_Column"/> FROM
+        linkis_stream_job WHERE name = #{jobName}
+    </select>
+
 <!--    <select id="getJobById" resultType="com.webank.wedatasphere.streamis.jobmanager.manager.entity.StreamJob">-->
 <!--        SELECT j.`id`,j.`name`,j.`status`,j.`submit_user`,j.`description`,j.`label`,l.`create_time`,l.`version` as currentVersion,j.`job_type` FROM-->
 <!--        linkis_stream_job j,linkis_stream_job_version l WHERE j.id = l.job_id and j.id=#{jobId} and l.`version` = #{version}-->
diff --git a/streamis-jobmanager/streamis-job-manager/streamis-job-manager-service/src/main/scala/com/webank/wedatasphere/streamis/jobmanager/manager/service/DefaultStreamJobService.scala b/streamis-jobmanager/streamis-job-manager/streamis-job-manager-service/src/main/scala/com/webank/wedatasphere/streamis/jobmanager/manager/service/DefaultStreamJobService.scala
@@ -17,26 +17,26 @@ package com.webank.wedatasphere.streamis.jobmanager.manager.service
 
 import java.util
 import java.util.Date
+
 import com.github.pagehelper.PageInfo
 import com.webank.wedatasphere.streamis.jobmanager.launcher.conf.JobConfKeyConstants
 import com.webank.wedatasphere.streamis.jobmanager.launcher.service.StreamJobConfService
 import com.webank.wedatasphere.streamis.jobmanager.manager.alert.AlertLevel
-import org.apache.linkis.common.exception.ErrorException
-import org.apache.linkis.common.utils.Logging
 import com.webank.wedatasphere.streamis.jobmanager.manager.conf.JobConf
 import com.webank.wedatasphere.streamis.jobmanager.manager.dao.{StreamAlertMapper, StreamJobMapper, StreamTaskMapper}
 import com.webank.wedatasphere.streamis.jobmanager.manager.entity.vo.{QueryJobListVo, TaskCoreNumVo, VersionDetailVo}
-import com.webank.wedatasphere.streamis.jobmanager.manager.entity.{MetaJsonInfo, StreamAlertRecord, StreamJob, StreamJobVersion, StreamJobVersionFiles}
+import com.webank.wedatasphere.streamis.jobmanager.manager.entity._
 import com.webank.wedatasphere.streamis.jobmanager.manager.exception.{JobCreateErrorException, JobFetchErrorException}
 import com.webank.wedatasphere.streamis.jobmanager.manager.transform.JobContentParser
 import com.webank.wedatasphere.streamis.jobmanager.manager.transform.entity.StreamisTransformJobContent
 import com.webank.wedatasphere.streamis.jobmanager.manager.util.{ReaderUtils, ZipHelper}
 import org.apache.commons.lang.StringUtils
+import org.apache.linkis.common.exception.ErrorException
+import org.apache.linkis.common.utils.Logging
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Service
 import org.springframework.transaction.annotation.Transactional
 
-import javax.annotation.Resource
 import scala.collection.JavaConverters._
 
 
@@ -60,6 +60,8 @@ class DefaultStreamJobService extends StreamJobService with Logging {
     this.streamJobMapper.getJobById(jobId)
   }
 
+  override def getJobByName(jobName: String): util.List[StreamJob] = streamJobMapper.getJobByName(jobName)
+
   override def getByProList(projectName: String, userName: String, jobName: String, jobStatus: Integer, jobCreator: String): PageInfo[QueryJobListVo] = {
     val streamJobList = streamJobMapper.getJobLists(projectName, userName, jobName, jobStatus, jobCreator)
     if (streamJobList != null && !streamJobList.isEmpty) {
diff --git a/streamis-jobmanager/streamis-job-manager/streamis-job-manager-service/src/main/scala/com/webank/wedatasphere/streamis/jobmanager/manager/service/DefaultStreamTaskService.scala b/streamis-jobmanager/streamis-job-manager/streamis-job-manager-service/src/main/scala/com/webank/wedatasphere/streamis/jobmanager/manager/service/DefaultStreamTaskService.scala
@@ -483,7 +483,7 @@ class DefaultStreamTaskService extends StreamTaskService with Logging{
       }).asJava
   }
 
-  def getTask(jobId:Long, version: String): FlinkJobInfo ={
+  def getTaskJobInfo(jobId:Long, version: String): FlinkJobInfo ={
     val str = streamTaskMapper.getTask(jobId, version)
     if (StringUtils.isBlank(str)) {
       return new FlinkJobInfo
@@ -524,6 +524,9 @@ class DefaultStreamTaskService extends StreamTaskService with Logging{
     }
   }
 
+
+  override def getLatestTaskByJobId(jobId: Long): StreamTask = streamTaskMapper.getLatestByJobId(jobId)
+
   /**
    * Create new task use the latest job version
    *
@@ -558,6 +561,8 @@ class DefaultStreamTaskService extends StreamTaskService with Logging{
      }
   }
 
+  override def updateTask(streamTask: StreamTask): Unit = streamTaskMapper.updateTask(streamTask)
+
   /**
    * Just launch task by task id
    *
diff --git a/streamis-jobmanager/streamis-job-manager/streamis-job-manager-service/src/main/scala/com/webank/wedatasphere/streamis/jobmanager/manager/service/StreamJobService.scala b/streamis-jobmanager/streamis-job-manager/streamis-job-manager-service/src/main/scala/com/webank/wedatasphere/streamis/jobmanager/manager/service/StreamJobService.scala
@@ -15,6 +15,9 @@ trait StreamJobService {
 
 
   def getJobById(jobId: Long): StreamJob
+
+  def getJobByName(jobName: String): util.List[StreamJob]
+
   /**
    * Page list query
    * @param projectName project name
diff --git a/streamis-jobmanager/streamis-job-manager/streamis-job-manager-service/src/main/scala/com/webank/wedatasphere/streamis/jobmanager/manager/service/StreamTaskService.scala b/streamis-jobmanager/streamis-job-manager/streamis-job-manager-service/src/main/scala/com/webank/wedatasphere/streamis/jobmanager/manager/service/StreamTaskService.scala
@@ -20,7 +20,6 @@ import com.webank.wedatasphere.streamis.jobmanager.launcher.linkis.entity.LogReq
 import com.webank.wedatasphere.streamis.jobmanager.launcher.linkis.job.FlinkJobInfo
 import com.webank.wedatasphere.streamis.jobmanager.manager.entity.StreamTask
 import com.webank.wedatasphere.streamis.jobmanager.manager.entity.vo.{ExecResultVo, JobProgressVo, JobStatusVo, PauseResultVo, StreamTaskListVo}
-
 import java.util
 import java.util.concurrent.Future
 /**
@@ -87,6 +86,8 @@ trait StreamTaskService {
    */
    def launch(taskId: Long, execUser: String): Unit
 
+   def getLatestTaskByJobId(jobId: Long): StreamTask
+
   /**
    * Create new task use the latest job version
    * @param jobId job id
@@ -95,6 +96,8 @@ trait StreamTaskService {
    */
    def createTask(jobId: Long, status: Int, creator: String): StreamTask
 
+   def updateTask(streamTask: StreamTask): Unit
+
   /**
    * Update the task status
    * @param jobId job id
@@ -146,7 +149,7 @@ trait StreamTaskService {
    * @param version version
    * @return
    */
-   def getTask(jobId: Long, version: String): FlinkJobInfo
+   def getTaskJobInfo(jobId: Long, version: String): FlinkJobInfo
 
 
    def getStateInfo(taskId: Long): JobState
diff --git a/streamis-jobmanager/streamis-jobmanager-server/src/main/java/com/webank/wedatasphere/streamis/jobmanager/restful/api/JobRestfulApi.java b/streamis-jobmanager/streamis-jobmanager-server/src/main/java/com/webank/wedatasphere/streamis/jobmanager/restful/api/JobRestfulApi.java
@@ -15,23 +15,30 @@
 
 package com.webank.wedatasphere.streamis.jobmanager.restful.api;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.github.pagehelper.PageHelper;
 import com.github.pagehelper.PageInfo;
 import com.webank.wedatasphere.streamis.jobmanager.exception.JobException;
 import com.webank.wedatasphere.streamis.jobmanager.exception.JobExceptionManager;
 import com.webank.wedatasphere.streamis.jobmanager.launcher.job.JobInfo;
 import com.webank.wedatasphere.streamis.jobmanager.launcher.job.manager.JobLaunchManager;
 import com.webank.wedatasphere.streamis.jobmanager.launcher.linkis.entity.LogRequestPayload;
+import com.webank.wedatasphere.streamis.jobmanager.launcher.linkis.job.FlinkJobInfo;
+import com.webank.wedatasphere.streamis.jobmanager.manager.conf.JobConf;
 import com.webank.wedatasphere.streamis.jobmanager.manager.entity.MetaJsonInfo;
 import com.webank.wedatasphere.streamis.jobmanager.manager.entity.StreamJob;
 import com.webank.wedatasphere.streamis.jobmanager.manager.entity.StreamJobVersion;
+import com.webank.wedatasphere.streamis.jobmanager.manager.entity.StreamTask;
 import com.webank.wedatasphere.streamis.jobmanager.manager.entity.vo.*;
 import com.webank.wedatasphere.streamis.jobmanager.manager.project.service.ProjectPrivilegeService;
 import com.webank.wedatasphere.streamis.jobmanager.manager.service.StreamJobService;
 import com.webank.wedatasphere.streamis.jobmanager.manager.service.StreamTaskService;
 import com.webank.wedatasphere.streamis.jobmanager.manager.transform.entity.StreamisTransformJobContent;
+import com.webank.wedatasphere.streamis.jobmanager.manager.utils.StreamTaskUtils;
+import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang.exception.ExceptionUtils;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.linkis.httpclient.dws.DWSHttpClient;
 import org.apache.linkis.server.Message;
 import org.apache.linkis.server.security.SecurityFilter;
 import org.slf4j.Logger;
@@ -43,10 +50,7 @@
 import javax.annotation.Resource;
 import javax.servlet.http.HttpServletRequest;
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
+import java.util.*;
 
 @RequestMapping(path = "/streamis/streamJobManager/job")
 @RestController
@@ -210,7 +214,7 @@ public Message detailsJob(HttpServletRequest req, @RequestParam(value = "jobId",
         realTimeTrafficDTOS.add(realTimeTrafficDTO);
 
 
-        jobDetailsVO.setLinkisJobInfo(streamTaskService.getTask(jobId,version));
+        jobDetailsVO.setLinkisJobInfo(streamTaskService.getTaskJobInfo(jobId,version));
         jobDetailsVO.setDataNumber(dataNumberDTOS);
         jobDetailsVO.setLoadCondition(loadConditionDTOs);
         jobDetailsVO.setRealTimeTraffic(realTimeTrafficDTOS);
@@ -238,6 +242,102 @@ public Message executeHistoryJob(HttpServletRequest req,
         return Message.ok().data("details", details);
     }
 
+    @RequestMapping(path = "/addTask", method = RequestMethod.GET)
+    public Message addTask(HttpServletRequest req,
+                           @RequestParam(value = "jobName") String jobName,
+                           @RequestParam(value = "appId") String appId,
+                           @RequestParam(value = "appUrl") String appUrl) {
+        String username = SecurityFilter.getLoginUsername(req);
+        LOG.info("User {} try to add a new task for Streamis job {} with appId: {}, appUrl: {}.", username, jobName, appId, appUrl);
+        List<StreamJob> streamJobs = streamJobService.getJobByName(jobName);
+        if(CollectionUtils.isEmpty(streamJobs)) {
+            return Message.error("Not exits Streamis job " + jobName);
+        } else if(streamJobs.size() > 1) {
+            return Message.error("Too many Streamis Job named " + jobName + ", we cannot distinguish between them.");
+        } else if(!"spark.jar".equals(streamJobs.get(0).getJobType())) {
+            return Message.error("Only spark.jar Job support to add new tasks.");
+        }
+        if (!streamJobService.hasPermission(streamJobs.get(0), username) &&
+                !this.privilegeService.hasEditPrivilege(req, streamJobs.get(0).getProjectName())) {
+            return Message.error("Have no permission to add new task for StreamJob [" + jobName + "].");
+        }
+        // 如果存在正在运行的，先将其停止掉
+        StreamTask streamTask = streamTaskService.getLatestTaskByJobId(streamJobs.get(0).getId());
+        if(streamTask != null && JobConf.isRunning(streamTask.getStatus())) {
+            LOG.warn("Streamis Job {} exists running task, update its status to stopped at first.", jobName);
+            streamTask.setStatus((Integer) JobConf.FLINK_JOB_STATUS_STOPPED().getValue());
+            streamTaskService.updateTask(streamTask);
+        } else {
+            // 这里取个巧，从该工程该用户有权限的Job中找到一个Flink的历史作业，作为这个Spark Streaming作业的jobId和jobInfo
+            // 替换掉JobInfo中的 yarn 信息，这样我们前端就可以在不修改任何逻辑的情况下正常展示Spark Streaming作业了
+            PageInfo<QueryJobListVo> jobList = streamJobService.getByProList(streamJobs.get(0).getProjectName(), username, null, 0, null);
+            Optional<QueryJobListVo> copyJob = jobList.getList().stream().min((job1, job2) -> {
+                if (job1.getStatus() > 0) {
+                    return 0;
+                } else {
+                    return 1;
+                }
+            });
+            if(!copyJob.isPresent()) {
+                return Message.error("If no Flink Job has submitted, the register to Streamis cannot be succeeded.");
+            }
+            StreamTask copyTask = streamTaskService.getLatestTaskByJobId(copyJob.get().getId());
+            LOG.warn("Streamis Job {} will bind the linkisJobInfo from history Flink Job {} with linkisJobId: {}, linkisJobInfo: {}.",
+                    jobName, copyJob.get().getName(), copyTask.getLinkisJobId(), copyTask.getLinkisJobInfo());
+            streamTask = streamTaskService.createTask(streamJobs.get(0).getId(), (Integer) JobConf.FLINK_JOB_STATUS_RUNNING().getValue(), username);
+            streamTask.setLinkisJobId(copyTask.getLinkisJobId());
+            streamTask.setLinkisJobInfo(copyTask.getLinkisJobInfo());
+        }
+        streamTask.setStartTime(new Date());
+        streamTask.setLastUpdateTime(new Date());
+        FlinkJobInfo flinkJobInfo;
+        try {
+            flinkJobInfo = DWSHttpClient.jacksonJson().readValue(streamTask.getLinkisJobInfo(), FlinkJobInfo.class);
+        } catch (JsonProcessingException e) {
+            LOG.error("Job {} deserialize the jobInfo from history Job failed!", jobName, e);
+            return Message.error("Deserialize the jobInfo from history Job failed!");
+        }
+        flinkJobInfo.setApplicationId(appId);
+        flinkJobInfo.setApplicationUrl(appUrl);
+        flinkJobInfo.setName(jobName);
+        flinkJobInfo.setStatus(JobConf.getStatusString((Integer) JobConf.FLINK_JOB_STATUS_RUNNING().getValue()));
+        StreamTaskUtils.refreshInfo(streamTask, flinkJobInfo);
+        streamTaskService.updateTask(streamTask);
+        LOG.info("Streamis Job {} has added a new task successfully.", jobName);
+        return Message.ok();
+    }
+
+    @RequestMapping(path = "/stopTask", method = RequestMethod.GET)
+    public Message stopTask(HttpServletRequest req,
+                           @RequestParam(value = "jobName") String jobName,
+                            @RequestParam(value = "appId") String appId,
+                            @RequestParam(value = "appUrl") String appUrl) {
+        String username = SecurityFilter.getLoginUsername(req);
+        LOG.info("User {} try to stop task for Streamis job {} with appId: {}, appUrl: {}.", username, jobName, appId, appUrl);
+        List<StreamJob> streamJobs = streamJobService.getJobByName(jobName);
+        if(CollectionUtils.isEmpty(streamJobs)) {
+            return Message.error("Not exits Streamis job " + jobName);
+        } else if(streamJobs.size() > 1) {
+            return Message.error("Too many Streamis Job named " + jobName + ", we cannot distinguish between them.");
+        } else if(!"spark.jar".equals(streamJobs.get(0).getJobType())) {
+            return Message.error("Only spark.jar Job support to add new tasks.");
+        }
+        if (!streamJobService.hasPermission(streamJobs.get(0), username) &&
+                !this.privilegeService.hasEditPrivilege(req, streamJobs.get(0).getProjectName())) {
+            return Message.error("Have no permission to add new task for StreamJob [" + jobName + "].");
+        }
+        // 如果存在正在运行的，将其停止掉
+        StreamTask streamTask = streamTaskService.getLatestTaskByJobId(streamJobs.get(0).getId());
+        if(streamTask != null && JobConf.isRunning(streamTask.getStatus())) {
+            LOG.warn("Streamis Job {} is exists running task, update its status to stopped.", jobName);
+            streamTask.setStatus((Integer) JobConf.FLINK_JOB_STATUS_STOPPED().getValue());
+            streamTaskService.updateTask(streamTask);
+        } else {
+            LOG.warn("Streamis Job {} is not exists running task, ignore to stop it.", jobName);
+        }
+        return Message.ok();
+    }
+
     @RequestMapping(path = "/progress", method = RequestMethod.GET)
     public Message progressJob(HttpServletRequest req, @RequestParam(value = "jobId", required = false) Long jobId,
                                @RequestParam(value = "version", required = false) String version) throws IOException, JobException {

Original file line number	Diff line number	Diff line change
`@@ -30,6 +30,7 @@ List<QueryJobListVo> getJobLists(@Param("projectName") String projectName, @Para`
`30`	`30`
`31`	`31`	`StreamJob getJobById(@Param("jobId") Long jobId);`
`32`	`32`
	`33`	`+ List<StreamJob> getJobByName(@Param("jobName") String jobName);`
`33`	`34`
`34`	`35`	`List<StreamJobVersion> getJobVersions(@Param("jobId") Long jobId);`
`35`	`36`
Original file line number	Diff line number	Diff line change
`@@ -483,7 +483,7 @@ class DefaultStreamTaskService extends StreamTaskService with Logging{`
`483`	`483`	`}).asJava`
`484`	`484`	`}`
`485`	`485`
`486`		`- def getTask(jobId:Long, version: String): FlinkJobInfo ={`
	`486`	`+ def getTaskJobInfo(jobId:Long, version: String): FlinkJobInfo ={`
`487`	`487`	`val str = streamTaskMapper.getTask(jobId, version)`
`488`	`488`	`if (StringUtils.isBlank(str)) {`
`489`	`489`	`return new FlinkJobInfo`
`@@ -524,6 +524,9 @@ class DefaultStreamTaskService extends StreamTaskService with Logging{`
`524`	`524`	`}`
`525`	`525`	`}`
`526`	`526`
	`527`	`+`
	`528`	`+ override def getLatestTaskByJobId(jobId: Long): StreamTask = streamTaskMapper.getLatestByJobId(jobId)`
	`529`	`+`
`527`	`530`	`/**`
`528`	`531`	`* Create new task use the latest job version`
`529`	`532`	`*`
`@@ -558,6 +561,8 @@ class DefaultStreamTaskService extends StreamTaskService with Logging{`
`558`	`561`	`}`
`559`	`562`	`}`
`560`	`563`
	`564`	`+ override def updateTask(streamTask: StreamTask): Unit = streamTaskMapper.updateTask(streamTask)`
	`565`	`+`
`561`	`566`	`/**`
`562`	`567`	`* Just launch task by task id`
`563`	`568`	`*`