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

Davidhua1996 · Davidhua1996 · commit 7ebc490e288b · 2022-10-20T19:12:19.000+08:00
diff --git a/streamis-jobmanager/streamis-job-manager/streamis-job-manager-service/src/main/scala/com/webank/wedatasphere/streamis/jobmanager/manager/service/TaskMonitorService.scala b/streamis-jobmanager/streamis-job-manager/streamis-job-manager-service/src/main/scala/com/webank/wedatasphere/streamis/jobmanager/manager/service/TaskMonitorService.scala
@@ -18,6 +18,7 @@ package com.webank.wedatasphere.streamis.jobmanager.manager.service
 import java.util
 import java.util.Date
 import java.util.concurrent.{Future, TimeUnit}
+
 import com.google.common.collect.Sets
 import com.webank.wedatasphere.streamis.jobmanager.launcher.JobLauncherAutoConfiguration
 import com.webank.wedatasphere.streamis.jobmanager.launcher.conf.JobConfKeyConstants
@@ -30,7 +31,6 @@ import com.webank.wedatasphere.streamis.jobmanager.manager.conf.JobConf
 import com.webank.wedatasphere.streamis.jobmanager.manager.dao.{StreamJobMapper, StreamTaskMapper}
 import com.webank.wedatasphere.streamis.jobmanager.manager.entity.{StreamJob, StreamTask}
 import com.webank.wedatasphere.streamis.jobmanager.manager.utils.StreamTaskUtils
-
 import javax.annotation.{PostConstruct, PreDestroy, Resource}
 import org.apache.commons.lang.exception.ExceptionUtils
 import org.apache.linkis.common.exception.ErrorException
@@ -84,58 +84,64 @@ class TaskMonitorService extends Logging {
       return
     }
     streamTasks.filter(shouldMonitor).foreach { streamTask =>
-      streamTask.setLastUpdateTime(new Date)
-      streamTaskMapper.updateTask(streamTask)
       val job = streamJobMapper.getJobById(streamTask.getJobId)
-      info(s"Try to update status of StreamJob-${job.getName}.")
-      val retryHandler = new RetryHandler {}
-      retryHandler.setRetryNum(3)
-      retryHandler.setRetryMaxPeriod(2000)
-      retryHandler.addRetryException(classOf[ErrorException])
-      var jobInfo:JobInfo = null
-      Utils.tryCatch {
-        jobInfo = retryHandler.retry(refresh(streamTask, jobLaunchManager), s"Task-Monitor-${job.getName}")
-      } { ex => {
-        error(s"Fetch StreamJob-${job.getName} failed, maybe the Linkis cluster is wrong, please be noticed!", ex)
-        val errorMsg = ExceptionUtils.getRootCauseMessage(ex)
-        if (errorMsg != null && errorMsg.contains("Not exists EngineConn")) {
-          streamTask.setStatus(JobConf.FLINK_JOB_STATUS_FAILED.getValue)
-          streamTask.setErrDesc("Not exists EngineConn.")
-        } else {
-          // 连续三次还是出现异常，说明Linkis的Manager已经不能正常提供服务，告警并不再尝试获取状态，等待下次尝试
-          val users = getAlertUsers(job)
-          users.add(job.getCreateBy)
-          alert(jobService.getAlertLevel(job), s"请求LinkisManager失败，Linkis集群出现异常，请关注！影响任务[${job.getName}]", users, streamTask)
-        }
-      }
-      }
-      streamTaskMapper.updateTask(streamTask)
-      if(streamTask.getStatus == JobConf.FLINK_JOB_STATUS_FAILED.getValue) {
-        warn(s"StreamJob-${job.getName} is failed, please be noticed.")
-        var extraMessage = ""
-        Option(jobInfo) match {
-          case Some(flinkJobInfo: FlinkJobInfo) =>
-            extraMessage = s",${flinkJobInfo.getApplicationId}"
-          case _ =>
-        }
-        // Need to add restart feature if user sets the restart parameters.
-        var alertMsg = s"Streamis 流式应用[${job.getName}${extraMessage}]已经失败, 请登陆Streamis查看应用日志."
-        this.streamJobConfMapper.getRawConfValue(job.getId, JobConfKeyConstants.FAIL_RESTART_SWITCH.getValue) match {
-          case "ON" =>
-            alertMsg = s"${alertMsg} 现将自动拉起该应用"
-            Utils.tryCatch{
-              info(s"Start to reLaunch the StreamisJob [${job.getName}], now to submit and schedule it...")
-              // Use submit user to start job
-              val future: Future[String] = streamTaskService.asyncExecute(job.getId, 0L, job.getSubmitUser, true)
-            }{
-              case e:Exception =>
-                warn(s"Fail to reLaunch the StreamisJob [${job.getName}]", e)
-            }
-          case _ =>
-        }
+      if(!JobConf.SUPPORTED_MANAGEMENT_JOB_TYPES.getValue.contains(job.getJobType)) {
         val userList = Sets.newHashSet(job.getSubmitUser, job.getCreateBy)
         userList.addAll(getAlertUsers(job))
+        val alertMsg = s"Spark Streaming应用[${job.getName}]已经超过 ${Utils.msDurationToString(System.currentTimeMillis - streamTask.getLastUpdateTime.getTime)} 没有更新状态, 请及时确认应用是否正常！"
         alert(jobService.getAlertLevel(job), alertMsg, new util.ArrayList[String](userList), streamTask)
+      } else {
+        streamTask.setLastUpdateTime(new Date)
+        streamTaskMapper.updateTask(streamTask)
+        info(s"Try to update status of StreamJob-${job.getName}.")
+        val retryHandler = new RetryHandler {}
+        retryHandler.setRetryNum(3)
+        retryHandler.setRetryMaxPeriod(2000)
+        retryHandler.addRetryException(classOf[ErrorException])
+        var jobInfo:JobInfo = null
+        Utils.tryCatch {
+          jobInfo = retryHandler.retry(refresh(streamTask, jobLaunchManager), s"Task-Monitor-${job.getName}")
+        } { ex =>
+          error(s"Fetch StreamJob-${job.getName} failed, maybe the Linkis cluster is wrong, please be noticed!", ex)
+          val errorMsg = ExceptionUtils.getRootCauseMessage(ex)
+          if (errorMsg != null && errorMsg.contains("Not exists EngineConn")) {
+            streamTask.setStatus(JobConf.FLINK_JOB_STATUS_FAILED.getValue)
+            streamTask.setErrDesc("Not exists EngineConn.")
+          } else {
+            // 连续三次还是出现异常，说明Linkis的Manager已经不能正常提供服务，告警并不再尝试获取状态，等待下次尝试
+            val users = getAlertUsers(job)
+            users.add(job.getCreateBy)
+            alert(jobService.getAlertLevel(job), s"请求LinkisManager失败，Linkis集群出现异常，请关注！影响任务[${job.getName}]", users, streamTask)
+          }
+        }
+        streamTaskMapper.updateTask(streamTask)
+        if(streamTask.getStatus == JobConf.FLINK_JOB_STATUS_FAILED.getValue) {
+          warn(s"StreamJob-${job.getName} is failed, please be noticed.")
+          var extraMessage = ""
+          Option(jobInfo) match {
+            case Some(flinkJobInfo: FlinkJobInfo) =>
+              extraMessage = s",${flinkJobInfo.getApplicationId}"
+            case _ =>
+          }
+          // Need to add restart feature if user sets the restart parameters.
+          var alertMsg = s"Streamis 流式应用[${job.getName}${extraMessage}]已经失败, 请登陆Streamis查看应用日志."
+          this.streamJobConfMapper.getRawConfValue(job.getId, JobConfKeyConstants.FAIL_RESTART_SWITCH.getValue) match {
+            case "ON" =>
+              alertMsg = s"${alertMsg} 现将自动拉起该应用"
+              Utils.tryCatch{
+                info(s"Start to reLaunch the StreamisJob [${job.getName}], now to submit and schedule it...")
+                // Use submit user to start job
+                val future: Future[String] = streamTaskService.asyncExecute(job.getId, 0L, job.getSubmitUser, true)
+              }{
+                case e:Exception =>
+                  warn(s"Fail to reLaunch the StreamisJob [${job.getName}]", e)
+              }
+            case _ =>
+          }
+          val userList = Sets.newHashSet(job.getSubmitUser, job.getCreateBy)
+          userList.addAll(getAlertUsers(job))
+          alert(jobService.getAlertLevel(job), alertMsg, new util.ArrayList[String](userList), streamTask)
+        }
       }
     }
     info("All StreamTasks status have updated.")
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
@@ -22,6 +22,7 @@
 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.job.state.JobStateInfo;
 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;
@@ -51,6 +52,7 @@
 import javax.servlet.http.HttpServletRequest;
 import java.io.IOException;
 import java.util.*;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 @RequestMapping(path = "/streamis/streamJobManager/job")
@@ -277,6 +279,7 @@ public Message addTask(HttpServletRequest req,
         if(streamTask != null && JobConf.isRunning(streamTask.getStatus())) {
             LOG.warn("Streamis Job {} exists running task, update its status from Running to stopped at first.", jobName);
             streamTask.setStatus((Integer) JobConf.FLINK_JOB_STATUS_STOPPED().getValue());
+            streamTask.setErrDesc("stopped by App's new task.");
             streamTaskService.updateTask(streamTask);
         } else if(streamTask == null) {
             // 这里取个巧，从该工程该用户有权限的Job中找到一个Flink的历史作业，作为这个Spark Streaming作业的jobId和jobInfo
@@ -306,21 +309,65 @@ public Message addTask(HttpServletRequest req,
         }
         streamTask.setStartTime(new Date());
         streamTask.setLastUpdateTime(new Date());
+        StreamTask finalStreamTask = streamTask;
+        return withFlinkJobInfo(jobName, streamTask.getLinkisJobInfo(), flinkJobInfo -> {
+            flinkJobInfo.setApplicationId(appId);
+            flinkJobInfo.setApplicationUrl(appUrl);
+            flinkJobInfo.setName(jobName);
+            flinkJobInfo.setStatus(JobConf.getStatusString((Integer) JobConf.FLINK_JOB_STATUS_RUNNING().getValue()));
+            StreamTaskUtils.refreshInfo(finalStreamTask, flinkJobInfo);
+            streamTaskService.updateTask(finalStreamTask);
+            LOG.info("Streamis Job {} has added a new task successfully.", jobName);
+            return Message.ok();
+        });
+    }
+
+    private Message withFlinkJobInfo(String jobName, String flinkJobInfoStr, Function<FlinkJobInfo, Message> flinkJobInfoFunction) {
         FlinkJobInfo flinkJobInfo;
         try {
-            flinkJobInfo = DWSHttpClient.jacksonJson().readValue(streamTask.getLinkisJobInfo(), FlinkJobInfo.class);
+            flinkJobInfo = DWSHttpClient.jacksonJson().readValue(flinkJobInfoStr, 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();
+            LOG.error("Job {} deserialize the flinkJobInfo string to object failed!", jobName, e);
+            return Message.error("Deserialize the flinkJobInfo string to object failed!");
+        }
+        return flinkJobInfoFunction.apply(flinkJobInfo);
+    }
+
+    @RequestMapping(path = "/updateTask", method = RequestMethod.GET)
+    public Message updateTask(HttpServletRequest req,
+                              @RequestParam(value = "jobName") String jobName,
+                              @RequestParam(value = "appId") String appId,
+                              @RequestParam(value = "metrics") String metrics) {
+        String username = SecurityFilter.getLoginUsername(req);
+        LOG.info("User {} try to update task for Streamis job {} with appId: {}.", username, jobName, appId);
+        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 update task.");
+        }
+        StreamTask streamTask = streamTaskService.getLatestTaskByJobId(streamJobs.get(0).getId());
+        if(streamTask == null) {
+            LOG.warn("Job {} is not exists running task, ignore to update its metrics.", jobName);
+            return Message.ok("not exists running task, ignore it.");
+        }
+        return withFlinkJobInfo(jobName, streamTask.getLinkisJobInfo(), flinkJobInfo -> {
+           if(!flinkJobInfo.getApplicationId().equals(appId)) {
+               LOG.warn("Job {} with running task <appId: {}> is not equals to the request appId: {}, ignore to update its metrics.",
+                       jobName, flinkJobInfo.getApplicationId(), appId);
+               return Message.ok("the request appId is not equals to the running task appId " + flinkJobInfo.getApplicationId());
+           }
+            JobStateInfo jobStateInfo = new JobStateInfo();
+           jobStateInfo.setTimestamp(System.currentTimeMillis());
+           jobStateInfo.setLocation(metrics);
+           flinkJobInfo.setJobStates(new JobStateInfo[]{jobStateInfo});
+            StreamTaskUtils.refreshInfo(streamTask, flinkJobInfo);
+            streamTaskService.updateTask(streamTask);
+            LOG.info("Streamis Job {} has updated the task metrics successfully.", jobName);
+            return Message.ok();
+        });
     }
 
     @RequestMapping(path = "/stopTask", method = RequestMethod.GET)
@@ -345,13 +392,23 @@ public Message stopTask(HttpServletRequest req,
         // 如果存在正在运行的，将其停止掉
         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);
+            return withFlinkJobInfo(jobName, streamTask.getLinkisJobInfo(), flinkJobInfo -> {
+                if(flinkJobInfo.getApplicationId().equals(appId)) {
+                    LOG.warn("Streamis Job {} is exists running task, update its status to stopped.", jobName);
+                    streamTask.setStatus((Integer) JobConf.FLINK_JOB_STATUS_STOPPED().getValue());
+                    streamTask.setErrDesc("stopped by App itself.");
+                    streamTaskService.updateTask(streamTask);
+                    return Message.ok();
+                } else {
+                    LOG.warn("Job {} with running task <appId: {}> is not equals to the request appId: {}, ignore to stop it.",
+                            jobName, flinkJobInfo.getApplicationId(), appId);
+                    return Message.ok("the request appId is not equals to the running task appId " + flinkJobInfo.getApplicationId());
+                }
+            });
         } else {
             LOG.warn("Streamis Job {} is not exists running task, ignore to stop it.", jobName);
+            return Message.ok();
         }
-        return Message.ok();
     }
 
     @RequestMapping(path = "/progress", method = RequestMethod.GET)
@@ -411,6 +468,12 @@ public Message getLog(HttpServletRequest req,
         logType = StringUtils.isBlank(logType) ? "client" : logType;
         String username = SecurityFilter.getLoginUsername(req);
         StreamJob streamJob = this.streamJobService.getJobById(jobId);
+        if(streamJob == null) {
+            return Message.error("not exists job " + jobId);
+        } else if(!JobConf.SUPPORTED_MANAGEMENT_JOB_TYPES().getValue().contains(streamJob.getJobType()) &&
+            "client".equals(logType)) {
+            return Message.error("Job " + streamJob.getName() + " is not supported to get client logs.");
+        }
         if (!streamJobService.hasPermission(streamJob, username) &&
                 !this.privilegeService.hasAccessPrivilege(req, streamJob.getProjectName())) {
             return Message.error("Have no permission to fetch logs from StreamJob [" + jobId + "]");
