新增 作业上下线自动提交和停止任务

dlink-admin/src/main/java/com/dlink/controller/TaskController.java

@@ -143,15 +143,15 @@ public class TaskController {
      */
     @GetMapping(value = "/onLineTask")
     public Result onLineTask(@RequestParam Integer id) {
-        return Result.succeed(taskService.onLineTask(id),"操作成功");
+        return taskService.onLineTask(id);
     }
 
     /**
      *  下线任务
      */
     @GetMapping(value = "/offLineTask")
-    public Result offLineTask(@RequestParam Integer id) {
-        return Result.succeed(taskService.offLineTask(id),"操作成功");
+    public Result offLineTask(@RequestParam Integer id,@RequestParam String type) {
+        return taskService.offLineTask(id,type);
     }
 
     /**

dlink-admin/src/main/java/com/dlink/job/Job2MysqlHandler.java

@@ -74,6 +74,7 @@ public class Job2MysqlHandler implements JobHandler {
     @Override
     public boolean success() {
         Job job = JobContextHolder.getJob();
+        Integer taskId = job.getJobConfig().getTaskId();
         History history = new History();
         history.setId(job.getId());
         if (job.isUseGateway() && Asserts.isNullString(job.getJobId())) {
@@ -96,7 +97,7 @@ public class Job2MysqlHandler implements JobHandler {
         if (job.isUseGateway()) {
             cluster = clusterService.registersCluster(Cluster.autoRegistersCluster(job.getJobManagerAddress(),
                     job.getJobId(), job.getJobConfig().getJobName() + LocalDateTime.now(), job.getType().getLongValue(),
-                    job.getJobConfig().getClusterConfigurationId(), job.getJobConfig().getTaskId()));
+                    job.getJobConfig().getClusterConfigurationId(), taskId));
             if (Asserts.isNotNull(cluster)) {
                 clusterId = cluster.getId();
             }
@@ -118,13 +119,15 @@ public class Job2MysqlHandler implements JobHandler {
                 JobInstance jobInstance = history.buildJobInstance();
                 jobInstance.setHistoryId(job.getId());
                 jobInstance.setClusterId(clusterId);
-                jobInstance.setTaskId(job.getJobConfig().getTaskId());
+                jobInstance.setTaskId(taskId);
                 jobInstance.setName(job.getJobConfig().getJobName());
                 jobInstance.setJid(jid);
+                jobInstance.setStep(job.getJobConfig().getStep());
                 jobInstance.setStatus(JobStatus.INITIALIZING.getValue());
                 jobInstanceService.save(jobInstance);
+                job.setJobInstanceId(jobInstance.getId());
                 Task task = new Task();
-                task.setId(jobInstance.getTaskId());
+                task.setId(taskId);
                 task.setJobInstanceId(jobInstance.getId());
                 taskService.updateById(task);
                 JobHistory jobHistory = new JobHistory();

dlink-admin/src/main/java/com/dlink/model/JobInstance.java

@@ -31,6 +31,8 @@ public class JobInstance implements Serializable {
 
     private Integer taskId;
 
+    private Integer step;
+
     private Integer clusterId;
 
     private String jid;

dlink-admin/src/main/java/com/dlink/model/Task.java

@@ -104,7 +104,7 @@ public class Task extends SuperEntity {
         for (Map<String, String> item : config) {
             map.put(item.get("key"), item.get("value"));
         }
-        return new JobConfig(type, false, false, useRemote, clusterId, clusterConfigurationId, jarId, getId(),
+        return new JobConfig(type, step, false, false, useRemote, clusterId, clusterConfigurationId, jarId, getId(),
                 alias, fragment, statementSet, batchModel, checkPoint, parallelism, savePointStrategy, savePointPath, map);
     }
 

dlink-admin/src/main/java/com/dlink/service/TaskService.java

@@ -22,6 +22,8 @@ public interface TaskService extends ISuperService<Task> {
 
     JobResult submitTask(Integer id);
 
+    JobResult submitTaskToOnline(Integer id);
+
     JobResult restartTask(Integer id);
 
     List<SqlExplainResult> explainTask(Integer id);
@@ -40,9 +42,9 @@ public interface TaskService extends ISuperService<Task> {
 
     boolean developTask(Integer id);
 
-    boolean onLineTask(Integer id);
+    Result onLineTask(Integer id);
 
-    boolean offLineTask(Integer id);
+    Result offLineTask(Integer id, String type);
 
     boolean cancelTask(Integer id);
 

dlink-admin/src/main/resources/mapper/JobInstanceMapper.xml

@@ -18,6 +18,9 @@
             <if test='param.type!=null and param.type!=""'>
                 and dh.type = #{param.type}
             </if>
+            <if test='param.step!=null and param.step!=""'>
+                and a.step = #{param.step}
+            </if>
             <if test='param.name!=null and param.name!=""'>
                 and a.name like "%${param.name}%"
             </if>

dlink-core/src/main/java/com/dlink/job/Job.java

@@ -20,6 +20,7 @@ import java.util.List;
 @Setter
 public class Job {
     private Integer id;
+    private Integer jobInstanceId;
     private JobConfig jobConfig;
     private String jobManagerAddress;
     private JobStatus status;
@@ -59,6 +60,6 @@ public class Job {
     }
 
     public JobResult getJobResult() {
-        return new JobResult(id, jobConfig, jobManagerAddress, status, statement, jobId, error, result, startTime, endTime);
+        return new JobResult(id, jobInstanceId, jobConfig, jobManagerAddress, status, statement, jobId, error, result, startTime, endTime);
     }
 }

dlink-core/src/main/java/com/dlink/job/JobConfig.java

@@ -27,6 +27,8 @@ public class JobConfig {
 
     // flink run mode
     private String type;
+    // task JobLifeCycle
+    private Integer step;
     private boolean useResult;
     private boolean useChangeLog;
     private boolean useAutoCancel;
@@ -126,11 +128,12 @@ public class JobConfig {
         this.maxRowNum = maxRowNum;
     }
 
-    public JobConfig(String type,boolean useResult, boolean useSession, boolean useRemote, Integer clusterId,
+    public JobConfig(String type, Integer step,boolean useResult, boolean useSession, boolean useRemote, Integer clusterId,
                      Integer clusterConfigurationId, Integer jarId, Integer taskId, String jobName, boolean useSqlFragment,
                      boolean useStatementSet,boolean useBatchModel,Integer checkpoint, Integer parallelism, Integer savePointStrategyValue,
                      String savePointPath,Map<String,String> config) {
         this.type = type;
+        this.step = step;
         this.useResult = useResult;
         this.useSession = useSession;
         this.useRemote = useRemote;

dlink-core/src/main/java/com/dlink/job/JobResult.java

@@ -24,6 +24,7 @@ public class JobResult {
     private boolean success;
     private String statement;
     private String jobId;
+    private Integer jobInstanceId;
     private String error;
     private IResult result;
     private LocalDateTime startTime;
@@ -32,8 +33,9 @@ public class JobResult {
     public JobResult() {
     }
 
-    public JobResult(Integer id, JobConfig jobConfig, String jobManagerAddress, Job.JobStatus status, String statement, String jobId, String error, IResult result, LocalDateTime startTime, LocalDateTime endTime) {
+    public JobResult(Integer id, Integer jobInstanceId, JobConfig jobConfig, String jobManagerAddress, Job.JobStatus status, String statement, String jobId, String error, IResult result, LocalDateTime startTime, LocalDateTime endTime) {
         this.id = id;
+        this.jobInstanceId = jobInstanceId;
         this.jobConfig = jobConfig;
         this.jobManagerAddress = jobManagerAddress;
         this.status = status;

dlink-doc/sql/dlink.sql

@@ -313,6 +313,7 @@ create table dlink_job_instance
         primary key,
     name                 varchar(255) null comment '作业实例名',
     task_id              int         null comment 'taskID',
+    step              int         null comment '生命周期',
     cluster_id           int         null comment '集群ID',
     jid                  varchar(50) null comment 'FlinkJobId',
     status               varchar(50) null comment '实例状态',

dlink-doc/sql/dlink_history.sql

@@ -629,5 +629,10 @@ ALTER TABLE `dlink_task`
     ADD COLUMN `job_instance_id` BIGINT NULL COMMENT '任务实例ID' AFTER `step`;
 ALTER TABLE `dlink_task`
     ADD COLUMN `alert_group_id` BIGINT NULL COMMENT '报警组ID' AFTER `env_id`;
+-- ----------------------------
+-- 0.6.0-SNAPSHOT 2022-03-13
+-- ----------------------------
+ALTER TABLE `dlink_job_instance`
+    ADD COLUMN `step` INT NULL COMMENT '生命周期' AFTER `task_id`;
 
 SET FOREIGN_KEY_CHECKS = 1;

dlink-web/src/components/Studio/StudioEvent/DDL.ts

@@ -237,8 +237,8 @@ export function onLineTask(id: number) {
   return getData('api/task/onLineTask',{id});
 }
 /*--- 下线作业 ---*/
-export function offLineTask(id: number) {
-  return getData('api/task/offLineTask',{id});
+export function offLineTask(id: number, type: string) {
+  return getData('api/task/offLineTask',{id, type});
 }
 /*--- 注销作业 ---*/
 export function cancelTask(id: number) {

dlink-web/src/components/Studio/StudioMenu/index.tsx

@@ -110,6 +110,7 @@ const StudioMenu = (props: any) => {
     result.then(res => {
       notification.close(taskKey);
       if (res.datas.success) {
+        props.changeTaskJobInstance(current.task.id,res.datas.jobInstanceId);
         message.success('执行成功');
       } else {
         message.error('执行失败');
@@ -154,6 +155,7 @@ const StudioMenu = (props: any) => {
         const res = await postDataArray('/api/task/submit', [task.id]);
         notification.close(taskKey);
         if (res.datas[0].success) {
+          props.changeTaskJobInstance(current.task.id,res.datas[0].jobInstanceId);
           message.success('异步提交成功');
         } else {
           message.success('异步提交失败');
@@ -315,27 +317,51 @@ const StudioMenu = (props: any) => {
       onOk: async () => {
         const res = onLineTask(current.task.id);
         res.then((result) => {
-          result.datas && props.changeTaskStep(current.task.id,TASKSTEPS.ONLINE);
           if(result.code == CODE.SUCCESS) {
+            props.changeTaskStep(current.task.id,TASKSTEPS.ONLINE);
             message.success(`上线作业【${current.task.alias}】成功`);
+          }else {
+            message.error(`上线作业【${current.task.alias}】失败，原因：\n${result.msg}`);
+          }
+        });
+      }
+    });
+  };
+
+  const handleCancelTask = (type: string) => {
+    Modal.confirm({
+      title: '停止作业',
+      content: `确定停止作业【${current.task.alias}】吗？`,
+      okText: '确认',
+      cancelText: '取消',
+      onOk: async () => {
+        const res = offLineTask(current.task.id,type);
+        res.then((result) => {
+          if(result.code == CODE.SUCCESS) {
+            props.changeTaskJobInstance(current.task.id,0);
+            message.success(`停止作业【${current.task.alias}】成功`);
+          }else {
+            message.error(`停止作业【${current.task.alias}】失败，原因：\n${result.msg}`);
           }
         });
       }
     });
   };
 
-  const toOffLineTask = () => {
+  const toOffLineTask = (type: string) => {
     Modal.confirm({
       title: '下线作业',
       content: `确定下线作业【${current.task.alias}】吗？`,
       okText: '确认',
       cancelText: '取消',
       onOk: async () => {
-        const res = offLineTask(current.task.id);
+        const res = offLineTask(current.task.id,type);
         res.then((result) => {
-          result.datas && props.changeTaskStep(current.task.id,TASKSTEPS.RELEASE);
           if(result.code == CODE.SUCCESS) {
+            props.changeTaskStep(current.task.id,TASKSTEPS.RELEASE);
             message.success(`下线作业【${current.task.alias}】成功`);
+          }else {
+            message.error(`下线作业【${current.task.alias}】失败，原因：\n${result.msg}`);
           }
         });
       }
@@ -505,6 +531,14 @@ const StudioMenu = (props: any) => {
                 onClick={onGetStreamGraph}
               />
             </Tooltip>)}
+            { current.task.jobInstanceId&&(current.task.jobInstanceId != 0) ?
+              <Tooltip title="停止">
+                <Button
+                  type="text"
+                  icon={<PauseCircleTwoTone />}
+                  onClick={()=>handleCancelTask('canceljob')}
+                />
+              </Tooltip>:<>
               {(!current.task.dialect||current.task.dialect === DIALECT.FLINKSQL||isSql( current.task.dialect )) &&(
                 <Tooltip title="执行当前的 SQL">
                   <Button
@@ -523,6 +557,8 @@ const StudioMenu = (props: any) => {
                   />
                 </Tooltip>
               </>)}
+              </>
+            }
             <Divider type="vertical"/>
             {current.task.step == TASKSTEPS.DEVELOP ?
               <Tooltip title="发布，发布后将无法修改">
@@ -551,7 +587,7 @@ const StudioMenu = (props: any) => {
               <Button
                 type="text"
                 icon={<PauseCircleTwoTone />}
-                onClick={toOffLineTask}
+                onClick={()=>toOffLineTask('cancel')}
               />
             </Tooltip>:undefined
           }{(current.task.step != TASKSTEPS.ONLINE && current.task.step != TASKSTEPS.CANCEL) ?
@@ -651,6 +687,11 @@ const mapDispatchToProps = (dispatch: Dispatch)=>({
     payload: {
       id,step
     },
+  }),changeTaskJobInstance:(id: number, jobInstanceId: number)=>dispatch({
+    type: "Studio/changeTaskStep",
+    payload: {
+      id,jobInstanceId
+    },
   }),
 });
 

dlink-web/src/pages/DevOps/JobInfo/index.tsx

@@ -13,7 +13,8 @@ import moment from "moment";
 import BaseInfo from "@/pages/DevOps/JobInfo/BaseInfo";
 import Config from "@/pages/DevOps/JobInfo/Config";
 import JobStatus, {isStatusDone} from "@/components/Common/JobStatus";
-import {cancelJob, restartJob, savepointJob} from "@/components/Studio/StudioEvent/DDL";
+import {cancelJob, offLineTask, restartJob} from "@/components/Studio/StudioEvent/DDL";
+import {CODE} from "@/components/Common/crud";
 
 const {Link} = Typography;
 
@@ -65,7 +66,7 @@ const JobInfo = (props: any) => {
           if (!job?.cluster?.id) return;
           const res = cancelJob(job?.cluster?.id, job?.instance?.jid);
           res.then((result) => {
-            if (result.datas == true) {
+            if (result.code == CODE.SUCCESS) {
               message.success(key+"成功");
               handleGetJobInfoDetail();
             } else {
@@ -83,9 +84,9 @@ const JobInfo = (props: any) => {
       cancelText: '取消',
       onOk: async () => {
         if (!job?.cluster?.id) return;
-        const res = savepointJob(job?.cluster?.id, job?.instance?.jid,key,key,job?.instance?.taskId);
+        const res = offLineTask(job?.instance?.taskId,key);
         res.then((result) => {
-          if (result.datas == true) {
+          if (result.code == CODE.SUCCESS) {
             message.success(key+"成功");
             handleGetJobInfoDetail();
           } else {
@@ -106,7 +107,7 @@ const JobInfo = (props: any) => {
         if (!job?.cluster?.id) return;
         const res = restartJob(job?.instance?.taskId);
         res.then((result) => {
-          if (result.datas.success == true) {
+          if (result.code == CODE.SUCCESS) {
             message.success("重新上线成功");
           } else {
             message.error("重新上线失败");
@@ -127,9 +128,9 @@ const JobInfo = (props: any) => {
           FlinkWebUI
         </Link></Button>);
     }
-    buttons.push(<Button key="autorestart" type="primary" onClick={handleRestart}>重新上线</Button>);
+    buttons.push(<Button key="autorestart" type="primary" onClick={handleRestart}>重新{job?.instance?.step == 5?'上线':'启动'}</Button>);
     if(!isStatusDone(job?.instance?.status as string)){
-      buttons.push(<Button key="autostop" type="primary" danger onClick={()=>{handleSavepoint('cancel')}}>下线</Button>);
+      buttons.push(<Button key="autostop" type="primary" danger onClick={()=>{handleSavepoint('cancel')}}>{job?.instance?.step == 5?'下线':'智能停止'}</Button>);
       buttons.push(<Dropdown
         key="dropdown"
         trigger={['click']}

dlink-web/src/pages/DevOps/JobInstanceTable/index.tsx

@@ -10,7 +10,7 @@ import type { ProColumns } from '@ant-design/pro-table';
 import ProTable from "@ant-design/pro-table";
 import {JobInstanceTableListItem} from "@/pages/DevOps/data";
 import moment from 'moment';
-import {RUN_MODE} from "@/components/Studio/conf";
+import {RUN_MODE, TASKSTEPS} from "@/components/Studio/conf";
 import JobStatus from "@/components/Common/JobStatus";
 
 const url = '/api/jobInstance';
@@ -24,6 +24,34 @@ const JobInstanceTable = (props: any) => {
       title: "作业名",
       dataIndex: "name",
       sorter: true,
+    },{
+      title: "生命周期",
+      dataIndex: "step",
+      sorter: true,
+      valueType: 'radio',
+      valueEnum: {
+        '': {text: '全部', status: 'ALL'},
+        1: {
+          text: '已创建',
+          status: TASKSTEPS.CREATE,
+        },
+        2: {
+          text: '开发中',
+          status: TASKSTEPS.DEVELOP,
+        },
+        4: {
+          text: '已发布',
+          status: TASKSTEPS.RELEASE,
+        },
+        5: {
+          text: '已上线',
+          status: TASKSTEPS.ONLINE,
+        },
+        0: {
+          text: '未知',
+          status: TASKSTEPS.UNKNOWN,
+        },
+      },
     },{
       title: "运行模式",
       dataIndex: "type",

dlink-web/src/pages/DevOps/data.d.ts

@@ -7,6 +7,7 @@ export type JobInstanceTableListItem = {
   id: number,
   name: string,
   taskId: number,
+  step: number,
   clusterId: number,
   clusterAlias: string,
   type: string,

dlink-web/src/pages/DevOps/service.ts

@@ -4,10 +4,10 @@ export function getStatusCount() {
   return getData("api/jobInstance/getStatusCount");
 }
 
-export function getJobInfoDetail(id:number) {
+export function getJobInfoDetail(id: number) {
   return getData("api/jobInstance/getJobInfoDetail",{id});
 }
 
-export function refreshJobInfoDetail(id:number) {
+export function refreshJobInfoDetail(id: number) {
   return getData("api/jobInstance/refreshJobInfoDetail",{id});
 }

dlink-web/src/pages/FlinkSqlStudio/model.ts

@@ -80,6 +80,7 @@ export type TaskType = {
   databaseName?: string,
   jarId?: number,
   envId?: number,
+  jobInstanceId?: number,
   note?: string,
   enabled?: boolean,
   createTime?: Date,
@@ -189,6 +190,7 @@ export type ModelType = {
     saveEnv: Reducer<StateType>;
     saveChart: Reducer<StateType>;
     changeTaskStep: Reducer<StateType>;
+    changeTaskJobInstance: Reducer<StateType>;
   };
 };
 
@@ -514,6 +516,23 @@ const Model: ModelType = {
         tabs: {...newTabs},
       };
     },
+    changeTaskJobInstance(state, {payload}) {
+      const newTabs = state.tabs;
+      let newCurrent = state.current;
+      for (let i = 0; i < newTabs.panes.length; i++) {
+        if (newTabs.panes[i].task.id == payload.id) {
+          newTabs.panes[i].task.jobInstanceId = payload.jobInstanceId;
+          if(newCurrent.key == newTabs.panes[i].key){
+            newCurrent = newTabs.panes[i];
+          }
+        }
+      }
+      return {
+        ...state,
+        current: {...newCurrent},
+        tabs: {...newTabs},
+      };
+    },
   },
 };
 

dlink-web/src/pages/Welcome.tsx

@@ -737,6 +737,9 @@ export default (): React.ReactNode => {
                 <li>
                   <Link>优化 IDEA调试时的依赖配置</Link>
                 </li>
+                <li>
+                  <Link>新增 作业上下线自动提交和停止任务</Link>
+                </li>
               </ul>
             </Paragraph>
           </Timeline.Item>