Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

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

Draft
wants to merge 35 commits into
base: dev
Choose a base branch
from
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
35 commits
Select commit Hold shift + click to select a range
3548432
feat: 新增从 DS 获取任务组接口
Apr 16, 2024
8b93776
feat: 调用 DolphinScheduler 接口 ProcessClient.createOrUpdateProcessDefin…
Apr 17, 2024
5cfc1dd
feat: 注册任务组类型
Apr 16, 2024
195a610
feat: 任务组联调
Apr 17, 2024
ceab412
fix:dinky推送任务字段dolphinTaskDefinition做非空处理
Apr 18, 2024
bafce4a
feat: dinky-任务推送增加回显数据操作
Apr 22, 2024
a59d965
chore: resolve conflict caused by cherry-pick
Apr 23, 2024
c8fa711
style: remove print code & add license
Apr 23, 2024
e26587b
style: remove print code & add license
Apr 23, 2024
c47efb0
Merge branch 'dev' of github.com:Yangxiaotian/dinky into dev
Apr 24, 2024
689450b
refactor: Remove new classes, and use existed classes
Apr 24, 2024
913e872
feat: 新增任务组内优先级
Apr 24, 2024
0e41186
feat: 增加组内优先级选项
Apr 24, 2024
375dcb9
feat: 创建目录和任务,往 DolphinScheduler 推送
May 15, 2024
0d2d506
feat: 创建目录和任务,往 DolphinScheduler 推送
May 15, 2024
2d65f93
fix: 创建任务接口推送任务时参数设错
May 15, 2024
78008c4
fix: token 已被冻结
May 16, 2024
3b65e36
feat: 新增保存任务的接口
May 20, 2024
20aa8d7
fix: /openapi 接口去掉 sa-token 机制,验证有报错
May 20, 2024
d76abf0
fix: 参数中的 json 字符串换成标准的 Java 类
May 21, 2024
fe4d4e3
fix: 把 /openapi/createTaskAndSend2Ds、saveTask 两个接口加 sa-token 机制
May 27, 2024
3a403f9
chore: 冲突解决
Jun 5, 2024
7d339fe
chore: add license info
Jun 5, 2024
349b7b9
chore: add license info
Jun 5, 2024
f0937cd
chore: code style
Jun 5, 2024
a916971
chore: code style
Jun 5, 2024
3a39e59
chore: code style
Jun 5, 2024
9cc5d5b
chore: code style
Jun 5, 2024
df1d27b
chore: package path changed
Jun 5, 2024
c725eeb
chore: code style
Jun 5, 2024
c2160a8
chore: package path changed
Jun 5, 2024
78c6445
chore: package path changed
Jun 5, 2024
ca9db19
fix: 低版本 jdk 不支持 List.of
Jun 5, 2024
15c6a2b
fix: sa-token 校验机制写的有问题
Jun 7, 2024
a824e34
fix: sa-token 校验机制写的有问题
Jun 11, 2024
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 22 additions & 0 deletions dinky-admin/src/main/java/org/dinky/controller/APIController.java
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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() {
Expand Down Expand Up @@ -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();
}
}
Original file line number Diff line number Diff line change
@@ -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;
}
9 changes: 8 additions & 1 deletion dinky-admin/src/main/java/org/dinky/service/APIService.java
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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);
}
}
Original file line number Diff line number Diff line change
@@ -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;
}
}
12 changes: 6 additions & 6 deletions dinky-common/src/main/java/org/dinky/data/model/Schema.java
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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() {}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,8 @@ export const transformPushDolphinParams = (
dolphinTaskDefinition: DolphinTaskDefinition,
pushDolphinParams: PushDolphinParams,
toFormValues: boolean
) => {
) => {

if (toFormValues && dolphinTaskDefinition) {
const transformValue: PushDolphinParams = {
...pushDolphinParams,
Expand Down
Loading