Merge pull request #47 from zhp8341/bug_fix_optimize

新增客户端日志查看功能

Author: zhp8341

docs/sql/flink_web.sql

@@ -96,6 +96,8 @@ CREATE TABLE `job_run_log` (
   PRIMARY KEY (`id`)
 ) ENGINE=InnoDB  DEFAULT CHARSET=utf8mb4 COMMENT='运行任务日志';
 
+ALTER TABLE job_run_log add `run_ip`  varchar(64) DEFAULT NULL COMMENT '任务运行所在的机器' AFTER local_log ;
+
 -- ----------------------------
 -- Table structure for savepoint_backup
 -- ----------------------------

flink-streaming-web-common/src/main/java/com/flink/streaming/web/common/MessageConstants.java

@@ -28,4 +28,6 @@ public class MessageConstants {
 
     public static final String MESSAGE_009 = "没有获取到savepointPath路径目录  任务:{}";
 
+    public static final String MESSAGE_010 = "无法获取 user.name";
+
 }

flink-streaming-web-common/src/main/java/com/flink/streaming/web/common/util/FileUtils.java

@@ -1,7 +1,5 @@
 package com.flink.streaming.web.common.util;
 
-import cn.hutool.core.date.DateUtil;
-import cn.hutool.core.lang.UUID;
 import cn.hutool.core.util.StrUtil;
 import com.flink.streaming.web.common.SystemConstants;
 import com.flink.streaming.web.enums.SysErrorEnum;
@@ -13,7 +11,6 @@
 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.OutputStreamWriter;
-import java.util.Date;
 
 /**
  * @author zhuhuipei
@@ -104,18 +101,6 @@ public static void writeText(String filePath, String content, boolean isAppend)
         }
     }
 
-    public static String downJar(String downUrl,String sysHome){
-        String extJarPath = sysHome + "tmp/udf_jar/" + DateUtil.formatDate(new Date()) + "/" + UUID.fastUUID();
-        FileUtils.mkdirs(extJarPath);
-        String fileName = System.currentTimeMillis() + "_udf.jar";
-        String[] cmds = {"curl", "-o", extJarPath + "/" + fileName, downUrl};
-
-
-        return "";
-
-    }
-
-
     public static void mkdirs(String path) {
         File dir = new File(path);
         if (!dir.exists()) {

flink-streaming-web-common/src/main/java/com/flink/streaming/web/common/util/IpUtil.java

@@ -1,5 +1,7 @@
 package com.flink.streaming.web.common.util;
 
+import com.flink.streaming.web.exceptions.BizException;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 
 import java.net.InetAddress;
@@ -11,6 +13,7 @@
  * @date 2018/7/19
  * @time 下午6:18
  */
+@Slf4j
 public class IpUtil {
 
     private static String ip;
@@ -54,9 +57,22 @@ private String getIp() {
         return ipAddrStr;
     }
 
+    public static String getHostName() {
+        try {
+            InetAddress addr = InetAddress.getLocalHost();
+            return addr.getHostName(); //获取本机计算机名称
+        } catch (Exception e) {
+            log.error("getHostName is error", e);
+            throw new BizException(e.getMessage());
+        }
+
+
+    }
+
     public static void main(String[] args) {
         System.out.println(IpUtil.getInstance().getLocalIP());
         System.out.println(IpUtil.getInstance().getLocalIP());
-        //System.out.println(IpUtil.getInstance().getLocalIP());
+        System.out.println(getHostName());
+        System.out.println(System.getProperty("user.name"));
     }
 }

flink-streaming-web-common/src/main/java/com/flink/streaming/web/common/util/LinuxInfoUtil.java

@@ -0,0 +1,23 @@
+package com.flink.streaming.web.common.util;
+
+import com.flink.streaming.web.common.MessageConstants;
+import com.flink.streaming.web.exceptions.BizException;
+import org.apache.commons.lang3.StringUtils;
+
+/**
+ * @author zhuhuipei
+ * @Description:
+ * @date 2021/5/5
+ * @time 10:46
+ */
+public class LinuxInfoUtil {
+
+    public static String  loginName(){
+        String userName=System.getProperty("user.name");
+        if (StringUtils.isNotEmpty(userName)){
+            return userName;
+        }
+        throw  new BizException(MessageConstants.MESSAGE_010);
+
+    }
+}

flink-streaming-web-common/src/main/java/com/flink/streaming/web/model/dto/JobRunLogDTO.java

@@ -80,6 +80,8 @@ public class JobRunLogDTO implements Serializable {
      */
     private String localLog;
 
+    private String runIp;
+
 
     public static JobRunLog toEntity(JobRunLogDTO jobRunLogDTO) {
         if (jobRunLogDTO == null) {
@@ -100,6 +102,7 @@ public static JobRunLog toEntity(JobRunLogDTO jobRunLogDTO) {
         jobRunLog.setCreateTime(jobRunLogDTO.getCreateTime());
         jobRunLog.setEditTime(jobRunLogDTO.getEditTime());
         jobRunLog.setLocalLog(jobRunLogDTO.getLocalLog());
+        jobRunLog.setRunIp(jobRunLogDTO.getRunIp());
         return jobRunLog;
     }
 
@@ -123,6 +126,7 @@ public static JobRunLogDTO toDTO(JobRunLog jobRunLog) {
         jobRunLogDTO.setCreator(jobRunLog.getCreator());
         jobRunLogDTO.setEditor(jobRunLog.getEditor());
         jobRunLogDTO.setLocalLog(jobRunLog.getLocalLog());
+        jobRunLogDTO.setRunIp(jobRunLog.getRunIp());
         return jobRunLogDTO;
     }
 

flink-streaming-web-common/src/main/java/com/flink/streaming/web/model/entity/JobRunLog.java

@@ -75,5 +75,7 @@ public class JobRunLog implements Serializable {
      */
     private String localLog;
 
+    private String runIp;
+
 
 }

flink-streaming-web-common/src/main/java/com/flink/streaming/web/model/param/UpsertJobConfigParam.java

@@ -122,6 +122,8 @@ public static JobConfigDTO toDTO(UpsertJobConfigParam upsertJobConfigParam) {
         jobConfigDTO.setStatus(JobConfigStatus.getJobConfigStatus(upsertJobConfigParam.getStauts()));
         if (StringUtils.isNotEmpty(upsertJobConfigParam.getExtJarPath())) {
             jobConfigDTO.setExtJarPath(upsertJobConfigParam.getExtJarPath().trim());
+        }else{
+            jobConfigDTO.setExtJarPath(SystemConstant.SPACE);
         }
         if (StringUtils.isNotEmpty(upsertJobConfigParam.getAlarmTypes())) {
             List<AlarmTypeEnum> list = new ArrayList<>();

flink-streaming-web-common/src/main/java/com/flink/streaming/web/model/vo/JobRunLogVO.java

@@ -4,6 +4,7 @@
 import com.flink.streaming.web.enums.JobStatusEnum;
 import com.flink.streaming.web.model.dto.JobRunLogDTO;
 import lombok.Data;
+import org.apache.commons.lang3.StringUtils;
 import org.springframework.util.CollectionUtils;
 
 import java.io.Serializable;
@@ -77,8 +78,13 @@ public class JobRunLogVO implements Serializable {
      */
     private String localLog;
 
+    /**
+     * 本地客户端日志
+     */
+    private String clinetJobUrl;
 
-    public static JobRunLogVO toVO(JobRunLogDTO jobRunLogDTO, boolean isLocalLog) {
+
+    public static JobRunLogVO toVO(JobRunLogDTO jobRunLogDTO, boolean isLocalLog,Integer port) {
         if (jobRunLogDTO == null) {
             return null;
         }
@@ -97,6 +103,10 @@ public static JobRunLogVO toVO(JobRunLogDTO jobRunLogDTO, boolean isLocalLog) {
         if (isLocalLog) {
             jobRunLogVO.setLocalLog(jobRunLogDTO.getLocalLog());
         }
+        if (port!=null && StringUtils.isNotEmpty(jobRunLogDTO.getRunIp())){
+            jobRunLogVO.setClinetJobUrl(String.format("http://%s:%s/log/getFlinkLocalJobLog",
+                    jobRunLogDTO.getRunIp(),port));
+        }
 
         return jobRunLogVO;
     }
@@ -108,7 +118,7 @@ public static List<JobRunLogVO> toListVO(List<JobRunLogDTO> jobRunLogList, boole
         List<JobRunLogVO> list = new ArrayList<>();
 
         for (JobRunLogDTO jobRunLog : jobRunLogList) {
-            list.add(JobRunLogVO.toVO(jobRunLog, isLocalLog));
+            list.add(JobRunLogVO.toVO(jobRunLog, isLocalLog,null));
         }
         return list;
 

flink-streaming-web-config/src/main/java/com/flink/streaming/web/config/CustomConfig.java

@@ -16,4 +16,7 @@ public class CustomConfig {
 
     @Value("${custom.core.jar.name}")
     private String coreJarName;
+
+    @Value("${server.port}")
+    private  Integer webPort;
 }