[Feature] [admin] 2 new APIs in openapi: 1. create catalogues ,task and send it to DolphinScheduer 2. save task content

dinky-admin/src/main/java/org/dinky/configure/AppConfig.java

@@ -32,9 +32,7 @@
 import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;
 import org.springframework.web.servlet.i18n.CookieLocaleResolver;
 
-import cn.dev33.satoken.exception.StopMatchException;
 import cn.dev33.satoken.interceptor.SaInterceptor;
-import cn.dev33.satoken.router.SaRouter;
 import cn.dev33.satoken.stp.StpUtil;
 
 /**
@@ -75,16 +73,12 @@ public void addInterceptors(InterceptorRegistry registry) {
         registry.addInterceptor(localeChangeInterceptor());
         // 注册Sa-Token的路由拦截器
         registry.addInterceptor(new SaInterceptor(handler -> {
-                    SaRouter.match("/openapi/**", r -> {
-                        if (!StpUtil.isLogin()) {
-                            StpUtil.switchTo(BaseConstant.ADMIN_ID);
-                        }
-                    });
                     if (!StpUtil.isLogin()) {
-                        throw new StopMatchException();
+                        StpUtil.switchTo(BaseConstant.ADMIN_ID);
                     }
+                    StpUtil.checkLogin();
                 }))
-                .addPathPatterns("/api/**", "/openapi/**")
+                .addPathPatterns("/api/**", "/openapi/createTaskAndSend2Ds", "/openapi/saveTask")
                 .excludePathPatterns("/api/login", "/api/ldap/ldapEnableStatus", "/download/**", "/druid/**");
 
         registry.addInterceptor(new TenantInterceptor())

dinky-admin/src/main/java/org/dinky/controller/APIController.java

@@ -22,6 +22,7 @@
 import org.dinky.DinkyVersion;
 import org.dinky.data.annotations.Log;
 import org.dinky.data.dto.APISavePointTaskDTO;
+import org.dinky.data.dto.CreatingCatalogueTaskDTO;
 import org.dinky.data.dto.TaskDTO;
 import org.dinky.data.dto.TaskSubmitDto;
 import org.dinky.data.enums.BusinessType;
@@ -33,11 +34,13 @@
 import org.dinky.gateway.enums.SavePointType;
 import org.dinky.gateway.result.SavePointResult;
 import org.dinky.job.JobResult;
+import org.dinky.service.APIService;
 import org.dinky.service.JobInstanceService;
 import org.dinky.service.TaskService;
 
 import java.util.List;
 
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
@@ -65,8 +68,12 @@
 public class APIController {
 
     private final TaskService taskService;
+
     private final JobInstanceService jobInstanceService;
 
+    @Autowired
+    private final APIService apiService;
+
     @GetMapping("/version")
     @ApiOperation(value = "Query Service Version", notes = "Query Dinky Service Version Number")
     public Result<String> getVersionInfo() {
@@ -202,4 +209,19 @@ public Result getTaskLineage(@RequestParam Integer id) {
         taskService.initTenantByTaskId(id);
         return Result.succeed(taskService.getTaskLineage(id), Status.QUERY_SUCCESS);
     }
+
+    @PostMapping("/createTaskAndSend2Ds")
+    @ApiOperation("Create Catalogues & Task and Send to DolphinScheduler")
+    @Log(title = "Create Catalogues & Task and Send to DolphinScheduler", businessType = BusinessType.OTHER)
+    public Result createTaskAndSend2Ds(@RequestBody CreatingCatalogueTaskDTO dto) {
+        return Result.succeed(apiService.createTaskAndSend2Ds(dto));
+    }
+
+    @PostMapping("/saveTask")
+    @ApiOperation("Save a task")
+    @Log(title = "Save a task", businessType = BusinessType.OTHER)
+    public Result saveTask(@RequestBody TaskDTO dto) {
+        apiService.saveTask(dto);
+        return Result.succeed();
+    }
 }

dinky-admin/src/main/java/org/dinky/data/dto/CreatingCatalogueTaskDTO.java

@@ -0,0 +1,53 @@
+/*
+ *
+ *  Licensed to the Apache Software Foundation (ASF) under one or more
+ *  contributor license agreements.  See the NOTICE file distributed with
+ *  this work for additional information regarding copyright ownership.
+ *  The ASF licenses this file to You under the Apache License, Version 2.0
+ *  (the "License"); you may not use this file except in compliance with
+ *  the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ */
+
+package org.dinky.data.dto;
+
+import org.dinky.scheduler.model.DinkyTaskRequest;
+
+import java.util.List;
+
+import lombok.Data;
+
+@Data
+public class CreatingCatalogueTaskDTO {
+    /**
+     * 目录名称列表
+     * 例子：["catalogue1", "catalogue2"]
+     * 数组第 0 个元素为根目录，依次下一个元素是上一个元素的子目录
+     * 目录不存在会新建，已存在就保持原来的目录
+     */
+    private List<String> catalogueNames;
+    /**
+     * 作业类型：FlinkSql、Mysql 等，详见 dinky 创建作业时的作业类型下拉菜单
+     */
+    private String type;
+    /**
+     * 任务信息
+     * 例子：{"name": "test", "note": "作业描述", "statement": "sql 语句", "type": "kubernetes-session", "clusterId": 36}
+     * 例子只列出了部分属性，其他属性请参考 TaskDTO 类
+     */
+    private TaskDTO task;
+    /**
+     * Dinky 推送时的作业配置
+     * 例子： {"delayTime": 0, "taskPriority": "MEDIUM", "failRetryInterval": 2, "failRetryTimes": 3, "flag": "YES" }
+     * 例子只列出了部分属性，其他属性请参考 DinkyTaskRequest 类
+     */
+    private DinkyTaskRequest jobConfig;
+}

dinky-admin/src/main/java/org/dinky/service/APIService.java

@@ -19,9 +19,16 @@
 
 package org.dinky.service;
 
+import org.dinky.data.dto.CreatingCatalogueTaskDTO;
+import org.dinky.data.dto.TaskDTO;
+
 /**
  * APIService
  *
  * @since 2021/12/11 21:45
  */
-public interface APIService {}
+public interface APIService {
+    Integer createTaskAndSend2Ds(CreatingCatalogueTaskDTO dto);
+
+    void saveTask(TaskDTO dto);
+}

dinky-admin/src/main/java/org/dinky/service/impl/APIServiceImpl.java

@@ -19,8 +19,19 @@
 
 package org.dinky.service.impl;
 
+import org.dinky.data.dto.CatalogueTaskDTO;
+import org.dinky.data.dto.CreatingCatalogueTaskDTO;
+import org.dinky.data.dto.TaskDTO;
+import org.dinky.data.enums.JobLifeCycle;
+import org.dinky.data.model.Catalogue;
+import org.dinky.data.model.Task;
+import org.dinky.scheduler.model.DinkyTaskRequest;
 import org.dinky.service.APIService;
+import org.dinky.service.SchedulerService;
+import org.dinky.service.TaskService;
+import org.dinky.service.catalogue.CatalogueService;
 
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
 import lombok.RequiredArgsConstructor;
@@ -34,4 +45,73 @@
 @Service
 @RequiredArgsConstructor
 @Slf4j
-public class APIServiceImpl implements APIService {}
+public class APIServiceImpl implements APIService {
+    @Autowired
+    private TaskService taskService;
+
+    @Autowired
+    private SchedulerService schedulerService;
+
+    @Autowired
+    private CatalogueService catalogueService;
+
+    /**
+     * 创建目录、任务并推送到 DolphinScheduler
+     *
+     * @param dto CreateCatalogueTaskDTO
+     * @return Integer taskId
+     */
+    @Override
+    public Integer createTaskAndSend2Ds(CreatingCatalogueTaskDTO dto) {
+        int parentId = 0;
+        for (String catalogueName : dto.getCatalogueNames()) {
+            Catalogue catalogue = catalogueService.findByParentIdAndName(parentId, catalogueName);
+            // 目录不存在则创建
+            if (catalogue == null) {
+                catalogue = new Catalogue();
+                catalogue.setName(catalogueName);
+                catalogue.setIsLeaf(false);
+                catalogue.setParentId(parentId);
+                catalogueService.save(catalogue);
+            }
+            parentId = catalogue.getId();
+        }
+        TaskDTO taskDTO = dto.getTask();
+        CatalogueTaskDTO catalogueTaskDTO = new CatalogueTaskDTO();
+        catalogueTaskDTO.setLeaf(false);
+        catalogueTaskDTO.setName(taskDTO.getName());
+        catalogueTaskDTO.setNote(taskDTO.getNote());
+        catalogueTaskDTO.setParentId(parentId);
+        catalogueTaskDTO.setType(dto.getType());
+        catalogueTaskDTO.setTask(taskDTO);
+        // 保存任务
+        Catalogue catalogue = catalogueService.saveOrUpdateCatalogueAndTask(catalogueTaskDTO);
+
+        // 发布任务
+        try {
+            taskService.changeTaskLifeRecyle(catalogue.getTaskId(), JobLifeCycle.PUBLISH);
+        } catch (Exception e) {
+            log.error(e.getMessage());
+            throw new RuntimeException(e);
+        }
+        // 推送任务
+        DinkyTaskRequest dinkyTaskRequest = dto.getJobConfig();
+        dinkyTaskRequest.setTaskId(catalogue.getTaskId() + "");
+        schedulerService.pushAddTask(dinkyTaskRequest);
+
+        return catalogue.getTaskId();
+    }
+
+    /**
+     * 更新任务的名称和 sql
+     * @param dto
+     */
+    @Override
+    public void saveTask(TaskDTO dto) {
+        Task task = new Task();
+        task.setId(dto.getId());
+        task.setName(dto.getName());
+        task.setStatement(dto.getStatement());
+        taskService.save(task);
+    }
+}

dinky-admin/src/test/java/org/dinky/service/impl/TestServiceImplTest.java

@@ -0,0 +1,121 @@
+/*
+ *
+ *  Licensed to the Apache Software Foundation (ASF) under one or more
+ *  contributor license agreements.  See the NOTICE file distributed with
+ *  this work for additional information regarding copyright ownership.
+ *  The ASF licenses this file to You under the Apache License, Version 2.0
+ *  (the "License"); you may not use this file except in compliance with
+ *  the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ */
+
+package org.dinky.service.impl;
+
+import org.dinky.Dinky;
+import org.dinky.data.dto.CatalogueTaskDTO;
+import org.dinky.data.dto.CreatingCatalogueTaskDTO;
+import org.dinky.data.dto.TaskDTO;
+import org.dinky.data.enums.JobLifeCycle;
+import org.dinky.data.exception.SqlExplainExcepition;
+import org.dinky.data.model.Catalogue;
+import org.dinky.scheduler.model.DinkyTaskRequest;
+import org.dinky.service.SchedulerService;
+import org.dinky.service.TaskService;
+import org.dinky.service.catalogue.CatalogueService;
+
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+
+import org.jetbrains.annotations.NotNull;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+@RunWith(SpringJUnit4ClassRunner.class)
+@SpringBootTest(classes = Dinky.class)
+public class TestServiceImplTest {
+    @Autowired
+    private CatalogueService catalogueService;
+
+    @Autowired
+    private TaskService taskService;
+
+    @Autowired
+    SchedulerService schedulerService;
+
+    @Test
+    @Ignore
+    public void testCreateCatalogueAndTask() throws SqlExplainExcepition {
+        CreatingCatalogueTaskDTO dto = new CreatingCatalogueTaskDTO();
+        List<String> catalogueNames = new ArrayList<>();
+        catalogueNames.add("test");
+        catalogueNames.add("test2");
+        catalogueNames.add("test3");
+        dto.setCatalogueNames(catalogueNames);
+        int parentId = 0;
+        for (String catalogueName : catalogueNames) {
+            Catalogue catalogue = catalogueService.findByParentIdAndName(parentId, catalogueName);
+            // 目录不存在则创建
+            if (catalogue == null) {
+                catalogue = new Catalogue();
+                catalogue.setName(catalogueName);
+                catalogue.setIsLeaf(false);
+                catalogue.setParentId(parentId);
+                catalogueService.save(catalogue);
+            }
+            parentId = catalogue.getId();
+        }
+        CatalogueTaskDTO catalogueTaskDTO = getCatalogueTaskDTO(parentId);
+        // 新建任务
+        Catalogue catalogue = catalogueService.saveOrUpdateCatalogueAndTask(catalogueTaskDTO);
+
+        // 发布任务
+        taskService.changeTaskLifeRecyle(catalogue.getTaskId(), JobLifeCycle.PUBLISH);
+        DinkyTaskRequest dinkyTaskRequest = new DinkyTaskRequest();
+        dinkyTaskRequest.setTaskId(catalogue.getTaskId() + "");
+        dinkyTaskRequest.setDelayTime(0);
+        dinkyTaskRequest.setFailRetryTimes(3);
+        dinkyTaskRequest.setFailRetryInterval(2);
+        dinkyTaskRequest.setFlag("YES");
+        dinkyTaskRequest.setTaskPriority("MEDIUM");
+        // 推送任务
+        schedulerService.pushAddTask(dinkyTaskRequest);
+    }
+
+    private static @NotNull CatalogueTaskDTO getCatalogueTaskDTO(int parentId) {
+        TaskDTO taskDTO = new TaskDTO();
+        taskDTO.setSavePointStrategy(0);
+        SimpleDateFormat sdf = new SimpleDateFormat("yyyy_MM_dd_hh_mm_ss");
+        taskDTO.setName("Flink 测试任务" + sdf.format(new Date()));
+        taskDTO.setNote("备注信息");
+        taskDTO.setStatement("select now()");
+        taskDTO.setParallelism(5);
+        taskDTO.setEnvId(-1);
+        taskDTO.setStep(1);
+        taskDTO.setAlertGroupId(-1);
+        taskDTO.setType("kubernetes-session");
+        taskDTO.setClusterId(36);
+        CatalogueTaskDTO catalogueTaskDTO = new CatalogueTaskDTO();
+        catalogueTaskDTO.setLeaf(false);
+        catalogueTaskDTO.setName(taskDTO.getName());
+        catalogueTaskDTO.setNote(taskDTO.getNote());
+        catalogueTaskDTO.setParentId(parentId);
+        catalogueTaskDTO.setType("FlinkSql");
+        catalogueTaskDTO.setTask(taskDTO);
+
+        return catalogueTaskDTO;
+    }
+}

dinky-common/src/main/java/org/dinky/data/model/Schema.java

@@ -20,7 +20,7 @@
 package org.dinky.data.model;
 
 import java.io.Serializable;
-import java.util.LinkedList;
+import java.util.ArrayList;
 import java.util.List;
 
 import lombok.Getter;
@@ -38,11 +38,11 @@ public class Schema implements Serializable, Comparable<Schema> {
     private static final long serialVersionUID = 4278304357661271040L;
 
     private String name;
-    private List<Table> tables = new LinkedList<>();
-    private List<String> views = new LinkedList<>();
-    private List<String> functions = new LinkedList<>();
-    private List<String> userFunctions = new LinkedList<>();
-    private List<String> modules = new LinkedList<>();
+    private List<Table> tables = new ArrayList<>();
+    private List<String> views = new ArrayList<>();
+    private List<String> functions = new ArrayList<>();
+    private List<String> userFunctions = new ArrayList<>();
+    private List<String> modules = new ArrayList<>();
 
     /** 需要保留一个空构造方法，否则序列化有问题 */
     public Schema() {}