Merge branch 'DataLinkDC:dev' into dev

gaoyan1998 · Nov 29, 2023 · ee7414d · ee7414d
2 parents 74a9fce + e32555e
commit ee7414d
Show file tree

Hide file tree

Showing 41 changed files with 1,046 additions and 160 deletions.
diff --git a/dinky-admin/src/main/java/org/dinky/controller/ClusterInstanceController.java b/dinky-admin/src/main/java/org/dinky/controller/ClusterInstanceController.java
@@ -154,8 +154,10 @@ public Result<Void> deleteClusterInstanceById(@RequestParam Integer id) {
             paramType = "query",
             required = true,
             dataTypeClass = JsonNode.class)
-    public Result<List<ClusterInstance>> listClusterInstance(@RequestParam("keyword") String searchKeyWord) {
-        return Result.succeed(clusterInstanceService.selectListByKeyWord(searchKeyWord));
+    public Result<List<ClusterInstance>> listClusterInstance(
+            @RequestParam(defaultValue = "") String searchKeyWord,
+            @RequestParam(defaultValue = "false") boolean isAutoCreate) {
+        return Result.succeed(clusterInstanceService.selectListByKeyWord(searchKeyWord, isAutoCreate));
     }
 
     /**

diff --git a/dinky-admin/src/main/java/org/dinky/controller/HistoryController.java b/dinky-admin/src/main/java/org/dinky/controller/HistoryController.java
@@ -59,7 +59,7 @@ public class HistoryController {
      * @param para
      * @return
      */
-    @PostMapping
+    @PostMapping("/list")
     @ApiOperation("Query History List")
     @ApiImplicitParam(name = "para", value = "Query Parameters", dataType = "JsonNode", paramType = "body")
     public ProTableResult<History> listHistory(@RequestBody JsonNode para) {

diff --git a/dinky-admin/src/main/java/org/dinky/data/model/ext/JobAlertData.java b/dinky-admin/src/main/java/org/dinky/data/model/ext/JobAlertData.java
@@ -29,7 +29,6 @@
 import org.dinky.data.model.job.JobInstance;
 import org.dinky.data.options.JobAlertRuleOptions;
 import org.dinky.job.JobConfig;
-import org.dinky.utils.JsonUtils;
 import org.dinky.utils.TimeUtil;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
@@ -150,7 +149,7 @@ public static JobAlertData buildData(JobInfoDetail jobInfoDetail) {
         builder.alertTime(TimeUtil.nowStr());
 
         JobDataDto jobDataDto = jobInfoDetail.getJobDataDto();
-        JobConfig job = JsonUtils.parseObject(jobInfoDetail.getHistory().getConfigJson(), JobConfig.class);
+        JobConfig job = jobInfoDetail.getHistory().getConfigJson();
         ClusterInstance clusterInstance = jobInfoDetail.getClusterInstance();
         CheckPointOverView checkpoints = jobDataDto.getCheckpoints();
         FlinkJobExceptionsDetail exceptions = jobDataDto.getExceptions();

diff --git a/dinky-admin/src/main/java/org/dinky/data/model/job/History.java b/dinky-admin/src/main/java/org/dinky/data/model/job/History.java
@@ -19,13 +19,20 @@
 
 package org.dinky.data.model.job;
 
+import org.dinky.data.typehandler.JSONObjectHandler;
+import org.dinky.job.JobConfig;
+
 import java.io.Serializable;
 import java.time.LocalDateTime;
 
 import com.baomidou.mybatisplus.annotation.FieldStrategy;
 import com.baomidou.mybatisplus.annotation.TableField;
 import com.baomidou.mybatisplus.annotation.TableName;
-import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.fasterxml.jackson.annotation.JsonFormat;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import com.fasterxml.jackson.datatype.jsr310.deser.LocalDateTimeDeserializer;
+import com.fasterxml.jackson.datatype.jsr310.ser.LocalDateTimeSerializer;
 
 import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
@@ -84,17 +91,20 @@ public class History implements Serializable {
     @ApiModelProperty(value = "Result", dataType = "String")
     private String result;
 
-    @TableField(exist = false)
-    @ApiModelProperty(hidden = true)
-    private ObjectNode config;
-
     @ApiModelProperty(value = "JSON Configuration", dataType = "String")
-    private String configJson;
+    @TableField(typeHandler = JSONObjectHandler.class)
+    private JobConfig configJson;
 
     @ApiModelProperty(value = "Start Time", dataType = "LocalDateTime")
+    @JsonDeserialize(using = LocalDateTimeDeserializer.class)
+    @JsonSerialize(using = LocalDateTimeSerializer.class)
+    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
     private LocalDateTime startTime;
 
     @ApiModelProperty(value = "End Time", dataType = "LocalDateTime")
+    @JsonDeserialize(using = LocalDateTimeDeserializer.class)
+    @JsonSerialize(using = LocalDateTimeSerializer.class)
+    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
     private LocalDateTime endTime;
 
     @ApiModelProperty(value = "Task ID", dataType = "Integer")

diff --git a/dinky-admin/src/main/java/org/dinky/data/typehandler/JSONObjectHandler.java b/dinky-admin/src/main/java/org/dinky/data/typehandler/JSONObjectHandler.java
@@ -28,6 +28,7 @@
 import org.dinky.data.model.mapping.ClusterConfigurationMapping;
 import org.dinky.data.model.mapping.ClusterInstanceMapping;
 import org.dinky.gateway.model.FlinkClusterConfig;
+import org.dinky.job.JobConfig;
 
 import org.apache.ibatis.type.JdbcType;
 import org.apache.ibatis.type.MappedJdbcTypes;
@@ -49,7 +50,8 @@
     ClusterInstanceMapping.class,
     ClusterConfigurationMapping.class,
     FlinkClusterConfig.class,
-    TaskExtConfig.class
+    TaskExtConfig.class,
+    JobConfig.class
 })
 public class JSONObjectHandler<T> extends AbstractJsonTypeHandler<T> {
 

diff --git a/dinky-admin/src/main/java/org/dinky/job/handler/ClearJobHistoryHandler.java b/dinky-admin/src/main/java/org/dinky/job/handler/ClearJobHistoryHandler.java
@@ -21,6 +21,7 @@
 
 import org.dinky.data.model.job.History;
 import org.dinky.data.model.job.JobInstance;
+import org.dinky.service.ClusterInstanceService;
 import org.dinky.service.HistoryService;
 import org.dinky.service.JobHistoryService;
 import org.dinky.service.JobInstanceService;
@@ -38,6 +39,7 @@ public class ClearJobHistoryHandler {
     private JobInstanceService jobInstanceService;
     private JobHistoryService jobHistoryService;
     private HistoryService historyService;
+    private ClusterInstanceService clusterService;
 
     /**
      * Clears job history records based on the specified criteria.
@@ -66,6 +68,9 @@ public void clearJobHistory(Integer maxRetainDays, Integer maxRetainCount) {
                 List<JobInstance> deleteList = jobInstanceService.list(deleteWrapper);
                 List<Integer> historyDeleteIds =
                         deleteList.stream().map(JobInstance::getHistoryId).collect(Collectors.toList());
+                List<Integer> clusterDeleteIds =
+                        deleteList.stream().map(JobInstance::getClusterId).collect(Collectors.toList());
+                clusterService.removeBatchByIds(clusterDeleteIds);
                 jobHistoryService.removeBatchByIds(historyDeleteIds);
                 jobInstanceService.remove(deleteWrapper);
             }

diff --git a/dinky-admin/src/main/java/org/dinky/job/handler/Job2MysqlHandler.java b/dinky-admin/src/main/java/org/dinky/job/handler/Job2MysqlHandler.java
@@ -44,7 +44,6 @@
 import org.dinky.service.JobHistoryService;
 import org.dinky.service.JobInstanceService;
 import org.dinky.service.TaskService;
-import org.dinky.utils.JsonUtils;
 
 import java.time.LocalDateTime;
 
@@ -87,12 +86,11 @@ public boolean init() {
         }
         history.setJobManagerAddress(job.getJobManagerAddress());
         history.setJobName(job.getJobConfig().getJobName());
-        history.setSession(job.getJobConfig().getSession());
         history.setStatus(job.getStatus().ordinal());
         history.setStatement(job.getStatement());
         history.setStartTime(job.getStartTime());
         history.setTaskId(job.getJobConfig().getTaskId());
-        history.setConfigJson(JsonUtils.toJsonString(job.getJobConfig()));
+        history.setConfigJson(job.getJobConfig());
         historyService.save(history);
 
         job.setId(history.getId());

diff --git a/dinky-admin/src/main/java/org/dinky/service/ClusterInstanceService.java b/dinky-admin/src/main/java/org/dinky/service/ClusterInstanceService.java
@@ -145,5 +145,5 @@ public interface ClusterInstanceService extends ISuperService<ClusterInstance> {
      */
     ClusterInstance deploySessionCluster(Integer id);
 
-    List<ClusterInstance> selectListByKeyWord(String searchKeyWord);
+    List<ClusterInstance> selectListByKeyWord(String searchKeyWord, boolean isAutoCreate);
 }
diff --git a/dinky-admin/src/main/java/org/dinky/service/HistoryService.java b/dinky-admin/src/main/java/org/dinky/service/HistoryService.java
@@ -29,15 +29,6 @@
  */
 public interface HistoryService extends ISuperService<History> {
 
-    /**
-     * Remove the history of a Git project based on its ID.
-     *
-     * @param id The ID of the Git project to remove the history for.
-     * @return A boolean value indicating whether the removal was successful.
-     */
-    @Deprecated
-    boolean removeHistoryById(Integer id);
-
     /**
      * Get latest history info by task id.
      *

diff --git a/dinky-admin/src/main/java/org/dinky/service/impl/ClusterInstanceServiceImpl.java b/dinky-admin/src/main/java/org/dinky/service/impl/ClusterInstanceServiceImpl.java
@@ -213,14 +213,15 @@ public ClusterInstance deploySessionCluster(Integer id) {
      * @return
      */
     @Override
-    public List<ClusterInstance> selectListByKeyWord(String searchKeyWord) {
+    public List<ClusterInstance> selectListByKeyWord(String searchKeyWord, boolean isAutoCreate) {
         return getBaseMapper()
                 .selectList(new LambdaQueryWrapper<ClusterInstance>()
-                        .like(ClusterInstance::getName, searchKeyWord)
-                        .or()
-                        .like(ClusterInstance::getAlias, searchKeyWord)
-                        .or()
-                        .like(ClusterInstance::getNote, searchKeyWord));
+                        .and(true, i -> i.eq(ClusterInstance::getAutoRegisters, isAutoCreate))
+                        .and(true, i -> i.like(ClusterInstance::getName, searchKeyWord)
+                                .or()
+                                .like(ClusterInstance::getAlias, searchKeyWord)
+                                .or()
+                                .like(ClusterInstance::getNote, searchKeyWord)));
     }
 
     private boolean checkHealth(ClusterInstance clusterInstance) {

diff --git a/dinky-admin/src/main/java/org/dinky/service/impl/HistoryServiceImpl.java b/dinky-admin/src/main/java/org/dinky/service/impl/HistoryServiceImpl.java
@@ -20,37 +20,29 @@
 package org.dinky.service.impl;
 
 import org.dinky.data.model.job.History;
-import org.dinky.data.result.ResultPool;
 import org.dinky.mapper.HistoryMapper;
 import org.dinky.mybatis.service.impl.SuperServiceImpl;
 import org.dinky.service.HistoryService;
 
 import org.springframework.stereotype.Service;
 
-import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+
+import lombok.RequiredArgsConstructor;
 
 /**
  * HistoryServiceImpl
  *
  * @since 2021/6/26 23:08
  */
 @Service
+@RequiredArgsConstructor
 public class HistoryServiceImpl extends SuperServiceImpl<HistoryMapper, History> implements HistoryService {
-
-    @Override
-    public boolean removeHistoryById(Integer id) {
-        History history = getById(id);
-        if (history != null) {
-            ResultPool.remove(history.getJobId());
-        }
-        return removeById(id);
-    }
-
     @Override
     public History getLatestHistoryById(Integer id) {
-        return baseMapper.selectOne(new QueryWrapper<History>()
-                .eq("task_id", id)
-                .orderByDesc("start_time")
+        return baseMapper.selectOne(new LambdaQueryWrapper<>(History.class)
+                .eq(History::getTaskId, id)
+                .orderByDesc(History::getStartTime)
                 .last("limit 1"));
     }
 }
diff --git a/dinky-admin/src/main/java/org/dinky/service/impl/JobInstanceServiceImpl.java b/dinky-admin/src/main/java/org/dinky/service/impl/JobInstanceServiceImpl.java
@@ -50,7 +50,6 @@
 import org.dinky.service.JobHistoryService;
 import org.dinky.service.JobInstanceService;
 import org.dinky.service.MonitorService;
-import org.dinky.utils.JsonUtils;
 
 import java.util.List;
 import java.util.Map;
@@ -164,7 +163,7 @@ public JobInfoDetail getJobInfoDetailInfo(JobInstance jobInstance) {
         jobInfoDetail.setClusterInstance(clusterInstance);
 
         History history = historyService.getById(jobInstance.getHistoryId());
-        history.setConfig(JsonUtils.parseObject(history.getConfigJson()));
+        history.setConfigJson(history.getConfigJson());
         jobInfoDetail.setHistory(history);
         if (Asserts.isNotNull(history.getClusterConfigurationId())) {
             ClusterConfiguration clusterConfig =

diff --git a/dinky-admin/src/main/resources/mapper/HistoryMapper.xml b/dinky-admin/src/main/resources/mapper/HistoryMapper.xml
@@ -10,7 +10,7 @@
         from
         dinky_history a
         <where>
-            1=1
+            1=1 and a.task_id = #{param.taskId}
             <if test='param.jobId!=null and param.jobId!=""'>
                 and a.job_id = #{param.jobId}
             </if>
@@ -20,9 +20,6 @@
             <if test='param.clusterId!=null and param.clusterId!=""'>
                 and a.cluster_id = #{param.clusterId}
             </if>
-            <if test='param.session!=null and param.session!=""'>
-                and a.session = #{param.session}
-            </if>
             <if test='param.status!=null and param.status!=""'>
                 and a.status = #{param.status}
             </if>

diff --git a/dinky-core/src/main/java/org/dinky/job/JobConfig.java b/dinky-core/src/main/java/org/dinky/job/JobConfig.java
@@ -111,13 +111,6 @@ public class JobConfig {
             notes = "Flag indicating whether to use auto-cancel")
     private boolean useAutoCancel;
 
-    @ApiModelProperty(
-            value = "Session information",
-            dataType = "String",
-            example = "session-123",
-            notes = "Session information")
-    private String session;
-
     @ApiModelProperty(
             value = "Flag indicating whether to use remote execution",
             dataType = "boolean",

diff --git a/dinky-core/src/test/java/org/dinky/core/JobManagerTest.java b/dinky-core/src/test/java/org/dinky/core/JobManagerTest.java
@@ -49,7 +49,6 @@ public void cancelJobSelect() throws Exception {
                 .useResult(true)
                 .useChangeLog(true)
                 .useAutoCancel(true)
-                .session("s1")
                 .clusterId(2)
                 .jobName("Test")
                 .fragment(false)

diff --git a/dinky-web/src/components/CustomEditor/CodeShow/index.tsx b/dinky-web/src/components/CustomEditor/CodeShow/index.tsx
@@ -29,7 +29,7 @@ import { EditorLanguage } from 'monaco-editor/esm/metadata';
 
 import FullscreenBtn from '@/components/CustomEditor/FullscreenBtn';
 import { Editor, Monaco } from '@monaco-editor/react';
-import { CSSProperties, memo, useRef, useState } from 'react';
+import { CSSProperties, useRef, useState } from 'react';
 
 export type CodeShowFormProps = {
   height?: string | number;
@@ -266,4 +266,4 @@ const CodeShow = (props: CodeShowFormProps) => {
   );
 };
 
-export default memo(CodeShow);
+export default CodeShow;
diff --git a/dinky-web/src/components/CustomEditor/languages/constants.ts b/dinky-web/src/components/CustomEditor/languages/constants.ts
@@ -51,3 +51,5 @@ export enum CustomEditorLanguage {
   JavaLog = 'javalog',
   FlinkSQL = 'flinksql'
 }
+
+// EditorLanguage
diff --git a/dinky-web/src/components/CustomEditor/languages/flinksql/index.tsx b/dinky-web/src/components/CustomEditor/languages/flinksql/index.tsx
@@ -30,6 +30,7 @@ export function FlinkSQLLanguage(monaco?: Monaco | undefined, registerCompletion
   monaco?.languages.register({
     id: CustomEditorLanguage.FlinkSQL,
     extensions: ['.sql'],
+    mimetypes: ['text/x-flinksql', 'text/x-flinksql', 'text/x-flinksql', 'text/flinksql'],
     aliases: ['flinksql', 'fsql', 'flinkSQL', 'FlinkSQL']
   });
   buildMonarchTokensProvider(monaco);
@@ -39,4 +40,15 @@ export function FlinkSQLLanguage(monaco?: Monaco | undefined, registerCompletion
     registerFlinkSQLCompilation(monaco);
   }
   buildFlinkSQLConfiguration(monaco);
+
+  monaco?.languages.onLanguageEncountered(CustomEditorLanguage.FlinkSQL, () => {
+    monaco?.editor?.getModels().forEach((model) => {
+      model.onDidChangeLanguage(() => {
+        if (model.getLanguageId() === CustomEditorLanguage.FlinkSQL) {
+          buildFlinkSQLConfiguration(monaco);
+        }
+      });
+    });
+    buildMonarchTokensProvider(monaco);
+  });
 }
diff --git a/dinky-web/src/components/CustomEditor/languages/index.tsx b/dinky-web/src/components/CustomEditor/languages/index.tsx
@@ -44,7 +44,6 @@ export function LoadCustomEditorLanguage(
   if (canLoadLanguage(monaco, CustomEditorLanguage.FlinkSQL)) {
     FlinkSQLLanguage(monaco, registerCompletion);
   }
-  console.log(canLoadLanguage(monaco, CustomEditorLanguage.JavaLog));
   if (canLoadLanguage(monaco, CustomEditorLanguage.JavaLog)) {
     LogLanguage(monaco);
   }

diff --git a/dinky-web/src/components/CustomEditor/languages/javalog/index.tsx b/dinky-web/src/components/CustomEditor/languages/javalog/index.tsx
@@ -25,7 +25,14 @@ export function LogLanguage(monaco: Monaco | undefined) {
   // Register a new language
   monaco?.languages.register({
     id: CustomEditorLanguage.JavaLog,
-    extensions: ['.log'],
+    extensions: [],
+    mimetypes: [
+      'text/x-java-log',
+      'text/x-javalog',
+      'text/x-java-source',
+      'text/x-java',
+      'text/java'
+    ],
     aliases: ['javalog', 'Javalog', 'jl', 'log']
   });
 

diff --git a/dinky-web/src/locales/en-US/global.ts b/dinky-web/src/locales/en-US/global.ts
@@ -210,5 +210,22 @@ export default {
   'global.month.september': 'September',
   'global.month.october': 'October',
   'global.month.november': 'November',
-  'global.month.december': 'December'
+  'global.month.december': 'December',
+
+  // job status
+  'global.job.status.initiating': 'Initializing',
+  'global.job.status.success': 'Success',
+  'global.job.status.created': 'Created',
+  'global.job.status.running': 'Running',
+  'global.job.status.failing': 'Failing',
+  'global.job.status.failed': 'Failed',
+  'global.job.status.cancelling': 'Cancelling',
+  'global.job.status.canceled': 'Canceled',
+  'global.job.status.finished': 'Finished',
+  'global.job.status.restarting': 'Restarting',
+  'global.job.status.suspended': 'Suspended',
+  'global.job.status.reconciling': 'Reconciling',
+  'global.job.status.reconnecting': 'Reconnecting',
+  'global.job.status.unknown': 'Unknown',
+  'global.job.status.failed-tip': 'Failed to submit to the cluster, unable to get the task name'
 };