third_codes/PowerJob
1
0
Fork 1
mirror of https://github.com/PowerJob/PowerJob.git synced 2025-07-17 00:00:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

suit for the front end develop(Vue is really good)

Browse Source
This commit is contained in:
tjq 2020-04-13 22:10:34 +08:00
parent 87d4654930
commit eff1f996be
15 changed files with 214 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/ExecuteType.java
View File

@ -12,11 +12,12 @@ import lombok.Getter;
@Getter
@AllArgsConstructor
public enum ExecuteType {
STANDALONE(1),
BROADCAST(2),
MAP_REDUCE(3);
STANDALONE(1, "单机执行"),
BROADCAST(2, "广播执行"),
MAP_REDUCE(3, "MapReduce");
int v;
String des;
public static ExecuteType of(int v) {
for (ExecuteType type : values()) {

2
oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/ProcessorType.java
View File

@ -13,7 +13,7 @@ import lombok.Getter;
@AllArgsConstructor
public enum ProcessorType {
EMBEDDED_JAVA(1, "内置Java对象");
EMBEDDED_JAVA(1, "内置JAVA处理器");
private int v;
private String des;

2
oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/request/ServerScheduleJobReq.java
View File

@ -46,7 +46,7 @@ public class ServerScheduleJobReq implements Serializable {
*/
// 任务级别的参数，相当于类的static变量
private String jobParams;
// 实例级别的参数，相当于类的普通变量
// 实例级别的参数，相当于类的普通变量（API触发专用，从API触发处带入）
private String instanceParams;
// 每台机器的处理线程数上限

20
oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/common/config/WebConfig.java Normal file
View File

@ -0,0 +1,20 @@
package com.github.kfcfans.oms.server.common.config;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.servlet.config.annotation.CorsRegistry;
import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;
/**
* CORS
*
* @author tjq
* @since 2020/4/13
*/
@Configuration
public class WebConfig implements WebMvcConfigurer {
@Override
public void addCorsMappings(CorsRegistry registry) {
registry.addMapping("/**")
.allowedMethods("HEAD", "GET", "PUT", "POST", "DELETE", "PATCH");
}
}

7
oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/persistence/model/JobInfoDO.java
View File

@ -31,8 +31,6 @@ public class JobInfoDO {
private Long appId;
// 任务自带的参数
private String jobParams;
// 任务实例的参数(API触发专用)
private String instanceParams;
/* ************************** 定时参数 ************************** */
// 时间表达式类型（CRON/API/FIX_RATE/FIX_DELAY）
@ -73,6 +71,11 @@ public class JobInfoDO {
// 最低磁盘空间，单位 GB，0代表不限
private double minDiskSpace;
/* ************************** 集群配置 ************************** */
// 指定机器运行，空代表不限，非空则只会使用其中的机器运行（多值逗号分割）
private String designatedWorkers;
// 最大机器数量
private Integer maxWorkerCount;
private Date gmtCreate;
private Date gmtModified;

2
oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/persistence/repository/AppInfoRepository.java
View File

@ -15,6 +15,8 @@ public interface AppInfoRepository extends JpaRepository<AppInfoDO, Long> {
AppInfoDO findByAppName(String appName);
List<AppInfoDO> findByAppNameLike(String condition);
/**
* 根据 currentServer 查询 appId
* 其实只需要 id，处于性能考虑可以直接写SQL只返回ID

4
oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/persistence/repository/InstanceLogRepository.java
View File

@ -37,12 +37,12 @@ public interface InstanceLogRepository extends JpaRepository<InstanceLogDO, Long
*/
@Transactional
@Modifying
@Query(value = "update execute_log set status = ?2, running_times = ?3, actual_trigger_time = ?4, task_tracker_address = ?5, result = ?6, gmt_modified = now() where instance_id = ?1", nativeQuery = true)
@Query(value = "update instance_log set status = ?2, running_times = ?3, actual_trigger_time = ?4, task_tracker_address = ?5, result = ?6, gmt_modified = now() where instance_id = ?1", nativeQuery = true)
int update4Trigger(long instanceId, int status, long runningTimes, long actualTriggerTime, String taskTrackerAddress, String result);
@Modifying
@Transactional
@Query(value = "update execute_log set status = ?2, running_times = ?3, gmt_modified = now() where instance_id = ?1", nativeQuery = true)
@Query(value = "update instance_log set status = ?2, running_times = ?3, gmt_modified = now() where instance_id = ?1", nativeQuery = true)
int update4FrequentJob(long instanceId, int status, long runningTimes);
// 状态检查三兄弟，对应 WAITING_DISPATCH 、 WAITING_WORKER_RECEIVE 和 RUNNING 三阶段

4
oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/persistence/repository/JobInfoRepository.java
View File

@ -18,5 +18,7 @@ public interface JobInfoRepository extends JpaRepository<JobInfoDO, Long> {
List<JobInfoDO> findByAppIdInAndStatusAndTimeExpressionTypeAndNextTriggerTimeLessThanEqual(List<Long> appIds, int status, int timeExpressionType, long time);
Page<JobInfoDO> findByAppId(Long appId, Pageable pageable);
Page<JobInfoDO> findByAppIdAndStatusNot(Long appId, Pageable pageable, int status);
Page<JobInfoDO> findByAppIdAndJobNameLikeAndStatusNot(Long appId, String condition, int status, Pageable pageable);
}

42
oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/service/DispatchService.java
View File

@ -7,13 +7,19 @@ import com.github.kfcfans.oms.server.akka.OhMyServer;
import com.github.kfcfans.oms.server.persistence.model.JobInfoDO;
import com.github.kfcfans.oms.server.persistence.repository.InstanceLogRepository;
import com.github.kfcfans.oms.server.service.ha.WorkerManagerService;
import com.google.common.base.Splitter;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import javax.annotation.Resource;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import static com.github.kfcfans.common.InstanceStatus.*;
@ -32,15 +38,20 @@ public class DispatchService {
private InstanceLogRepository instanceLogRepository;
private static final String EMPTY_RESULT = "";
private static final Splitter commaSplitter = Splitter.on(",");
public void dispatch(JobInfoDO jobInfo, long instanceId, long currentRunningTimes) {
dispatch(jobInfo, instanceId, currentRunningTimes, null);
}
/**
* 将任务从Server派发到Worker（TaskTracker）
* @param jobInfo 任务的元信息
* @param instanceId 任务实例ID
* @param currentRunningTimes 当前运行的次数
* @param instanceParams 实例的运行参数，API触发方式专用
*/
public void dispatch(JobInfoDO jobInfo, long instanceId, long currentRunningTimes) {
public void dispatch(JobInfoDO jobInfo, long instanceId, long currentRunningTimes, String instanceParams) {
Long jobId = jobInfo.getId();
log.info("[DispatchService] start to dispatch job: {}.", jobInfo);
// 查询当前运行的实例数
@ -55,21 +66,42 @@ public class DispatchService {
return;
}
// 获取 Worker
// 获取当前所有可用的Worker
List<String> allAvailableWorker = WorkerManagerService.getSortedAvailableWorker(jobInfo.getAppId(), jobInfo.getMinCpuCores(), jobInfo.getMinMemorySpace(), jobInfo.getMinDiskSpace());
if (CollectionUtils.isEmpty(allAvailableWorker)) {
// 筛选指定的机器
List<String> finalWorkers = Lists.newLinkedList();
if (!StringUtils.isEmpty(jobInfo.getDesignatedWorkers())) {
Set<String> designatedWorkers = Sets.newHashSet(commaSplitter.splitToList(jobInfo.getDesignatedWorkers()));
for (String av : allAvailableWorker) {
if (designatedWorkers.contains(av)) {
finalWorkers.add(av);
}
}
}else {
finalWorkers = allAvailableWorker;
}
if (CollectionUtils.isEmpty(finalWorkers)) {
String clusterStatusDescription = WorkerManagerService.getWorkerClusterStatusDescription(jobInfo.getAppId());
log.warn("[DispatchService] cancel dispatch job(jobId={}) due to no worker available, clusterStatus is {}.", jobId, clusterStatusDescription);
instanceLogRepository.update4Trigger(instanceId, FAILED.getV(), currentRunningTimes, current, RemoteConstant.EMPTY_ADDRESS, SystemInstanceResult.NO_WORKER_AVAILABLE);
return;
}
// 限定集群大小（0代表不限制）
if (jobInfo.getMaxWorkerCount() > 0) {
if (finalWorkers.size() > jobInfo.getMaxWorkerCount()) {
finalWorkers = finalWorkers.subList(0, jobInfo.getMaxWorkerCount());
}
}
// 构造请求
ServerScheduleJobReq req = new ServerScheduleJobReq();
BeanUtils.copyProperties(jobInfo, req);
req.setInstanceParams(instanceParams);
req.setInstanceId(instanceId);
req.setAllWorkerAddress(allAvailableWorker);
req.setAllWorkerAddress(finalWorkers);
req.setExecuteType(ExecuteType.of(jobInfo.getExecuteType()).name());
req.setProcessorType(ProcessorType.of(jobInfo.getProcessorType()).name());

30
oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/web/controller/AppInfoController.java
View File

@ -4,11 +4,12 @@ import com.github.kfcfans.oms.server.persistence.model.AppInfoDO;
import com.github.kfcfans.oms.server.persistence.repository.AppInfoRepository;
import com.github.kfcfans.common.response.ResultDTO;
import com.github.kfcfans.oms.server.web.request.ModifyAppInfoRequest;
import com.google.common.collect.Lists;
import lombok.Data;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.util.CollectionUtils;
import org.springframework.web.bind.annotation.*;
import javax.annotation.Resource;
import java.util.Date;
@ -17,6 +18,7 @@ import java.util.stream.Collectors;
/**
* AppName Controller
* vue axios 的POST请求必须使用 @RequestBody 接收
*
* @author tjq
* @since 2020/4/1
@ -28,8 +30,8 @@ public class AppInfoController {
@Resource
private AppInfoRepository appInfoRepository;
@GetMapping("/save")
public ResultDTO<Void> saveAppInfo(ModifyAppInfoRequest appInfoRequest) {
@PostMapping("/save")
public ResultDTO<Void> saveAppInfo(@RequestBody ModifyAppInfoRequest appInfoRequest) {
AppInfoDO appInfoDO = new AppInfoDO();
BeanUtils.copyProperties(appInfoRequest, appInfoDO);
@ -49,13 +51,25 @@ public class AppInfoController {
}
@GetMapping("/list")
public ResultDTO<List<AppInfoVO>> listAppInfo() {
List<AppInfoVO> result = appInfoRepository.findAll().stream().map(appInfoDO -> {
public ResultDTO<List<AppInfoVO>> listAppInfo(@RequestParam(required = false) String condition) {
List<AppInfoDO> result;
if (StringUtils.isEmpty(condition)) {
result = appInfoRepository.findAll();
}else {
result = appInfoRepository.findByAppNameLike("%" + condition + "%");
}
return ResultDTO.success(convert(result));
}
private static List<AppInfoVO> convert(List<AppInfoDO> data) {
if (CollectionUtils.isEmpty(data)) {
return Lists.newLinkedList();
}
return data.stream().map(appInfoDO -> {
AppInfoVO appInfoVO = new AppInfoVO();
BeanUtils.copyProperties(appInfoDO, appInfoVO);
return appInfoVO;
}).collect(Collectors.toList());
return ResultDTO.success(result);
}
@Data

87
oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/web/controller/JobController.java
View File

@ -16,17 +16,17 @@ import com.github.kfcfans.oms.server.service.DispatchService;
import com.github.kfcfans.oms.server.service.IdGenerateService;
import com.github.kfcfans.oms.server.service.instance.InstanceService;
import com.github.kfcfans.oms.server.web.request.ModifyJobInfoRequest;
import com.github.kfcfans.oms.server.web.request.QueryJobInfoRequest;
import com.github.kfcfans.oms.server.web.response.JobInfoVO;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Sort;
import org.springframework.util.CollectionUtils;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.*;
import javax.annotation.Resource;
import java.util.Date;
@ -56,7 +56,7 @@ public class JobController {
private InstanceLogRepository instanceLogRepository;
@PostMapping("/save")
public ResultDTO<Void> saveJobInfo(ModifyJobInfoRequest request) throws Exception {
public ResultDTO<Void> saveJobInfo(@RequestBody ModifyJobInfoRequest request) throws Exception {
JobInfoDO jobInfoDO = new JobInfoDO();
BeanUtils.copyProperties(request, jobInfoDO);
@ -66,6 +66,12 @@ public class JobController {
jobInfoDO.setExecuteType(ExecuteType.valueOf(request.getExecuteType()).getV());
jobInfoDO.setProcessorType(ProcessorType.valueOf(request.getProcessorType()).getV());
jobInfoDO.setTimeExpressionType(timeExpressionType.getV());
jobInfoDO.setStatus(request.isEnable() ? JobStatus.ENABLE.getV() : JobStatus.STOPPED.getV());
if (jobInfoDO.getMaxWorkerCount() == null) {
jobInfoDO.setMaxInstanceNum(0);
}
// 计算下次调度时间
Date now = new Date();
if (timeExpressionType == TimeExpressionType.CRON) {
@ -110,21 +116,44 @@ public class JobController {
return ResultDTO.success(null);
}
@GetMapping("/list")
public ResultDTO<PageResult<JobInfoVO>> listJobs(Long appId, int index, int pageSize) {
@PostMapping("/list")
public ResultDTO<PageResult<JobInfoVO>> listJobs(@RequestBody QueryJobInfoRequest request) {
Sort sort = Sort.by(Sort.Direction.DESC, "gmtCreate");
PageRequest pageRequest = PageRequest.of(index, pageSize, sort);
Page<JobInfoDO> jobInfoPage = jobInfoRepository.findByAppId(appId, pageRequest);
List<JobInfoVO> jobInfoVOList = jobInfoPage.getContent().stream().map(jobInfoDO -> {
JobInfoVO jobInfoVO = new JobInfoVO();
BeanUtils.copyProperties(jobInfoDO, jobInfoVO);
return jobInfoVO;
}).collect(Collectors.toList());
PageRequest pageRequest = PageRequest.of(request.getIndex(), request.getPageSize(), sort);
Page<JobInfoDO> jobInfoPage;
PageResult<JobInfoVO> pageResult = new PageResult<>(jobInfoPage);
pageResult.setData(jobInfoVOList);
return ResultDTO.success(pageResult);
// 无查询条件，查询全部
if (request.getJobId() == null && StringUtils.isEmpty(request.getKeyword())) {
jobInfoPage = jobInfoRepository.findByAppIdAndStatusNot(request.getAppId(), pageRequest, JobStatus.DELETED.getV());
return ResultDTO.success(convertPage(jobInfoPage));
}
// 有 jobId，直接精确查询
if (request.getJobId() != null) {
Optional<JobInfoDO> jobInfoOpt = jobInfoRepository.findById(request.getJobId());
PageResult<JobInfoVO> result = new PageResult<>();
result.setIndex(0);
result.setPageSize(request.getPageSize());
if (jobInfoOpt.isPresent()) {
result.setTotalItems(1);
result.setTotalPages(1);
result.setData(Lists.newArrayList(convert(jobInfoOpt.get())));
}else {
result.setTotalPages(0);
result.setTotalItems(0);
result.setData(Lists.newLinkedList());
}
return ResultDTO.success(result);
}
// 模糊查询
String condition = "%" + request.getKeyword() + "%";
jobInfoPage = jobInfoRepository.findByAppIdAndJobNameLikeAndStatusNot(request.getAppId(), condition, JobStatus.DELETED.getV(), pageRequest);
return ResultDTO.success(convertPage(jobInfoPage));
}
/**
@ -181,5 +210,29 @@ public class JobController {
});
}
private static PageResult<JobInfoVO> convertPage(Page<JobInfoDO> jobInfoPage) {
List<JobInfoVO> jobInfoVOList = jobInfoPage.getContent().stream().map(JobController::convert).collect(Collectors.toList());
PageResult<JobInfoVO> pageResult = new PageResult<>(jobInfoPage);
pageResult.setData(jobInfoVOList);
return pageResult;
}
private static JobInfoVO convert(JobInfoDO jobInfoDO) {
JobInfoVO jobInfoVO = new JobInfoVO();
BeanUtils.copyProperties(jobInfoDO, jobInfoVO);
TimeExpressionType timeExpressionType = TimeExpressionType.of(jobInfoDO.getTimeExpressionType());
ExecuteType executeType = ExecuteType.of(jobInfoDO.getExecuteType());
ProcessorType processorType = ProcessorType.of(jobInfoDO.getProcessorType());
jobInfoVO.setTimeExpressionType(timeExpressionType.name());
jobInfoVO.setExecuteType(executeType.name());
jobInfoVO.setProcessorType(processorType.name());
jobInfoVO.setEnable(jobInfoDO.getStatus() == JobStatus.ENABLE.getV());
return jobInfoVO;
}
}

7
oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/web/controller/UserInfoController.java
View File

@ -8,10 +8,7 @@ import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.NoArgsConstructor;
import org.springframework.beans.BeanUtils;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.bind.annotation.*;
import javax.annotation.Resource;
import java.util.Date;
@ -32,7 +29,7 @@ public class UserInfoController {
private UserInfoRepository userInfoRepository;
@PostMapping("save")
public ResultDTO<Void> save(ModifyUserInfoRequest request) {
public ResultDTO<Void> save(@RequestBody ModifyUserInfoRequest request) {
UserInfoDO userInfoDO = new UserInfoDO();
BeanUtils.copyProperties(request, userInfoDO);
userInfoDO.setGmtCreate(new Date());

14
oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/web/request/ModifyJobInfoRequest.java
View File

@ -2,6 +2,8 @@ package com.github.kfcfans.oms.server.web.request;
import lombok.Data;
import java.util.Date;
/**
* 创建/修改 JobInfo 请求
* 测试用例快速复制区域：MAP_REDUCE、EMBEDDED_JAVA、CRON、com.github.kfcfans.oms.processors.TestMapReduceProcessor
@ -25,8 +27,6 @@ public class ModifyJobInfoRequest {
private String groupName;
// 任务自带的参数
private String jobParams;
// 任务实例的参数(API触发专用)
private String instanceParams;
/* ************************** 定时参数 ************************** */
// 时间表达式类型（CRON/API/FIX_RATE/FIX_DELAY）
@ -65,5 +65,13 @@ public class ModifyJobInfoRequest {
private double minDiskSpace;
// 1 正常运行，2 停止（不再调度）
private Integer status;
private boolean enable;
private Date gmtCreate;
/* ************************** 集群配置 ************************** */
// 指定机器运行，空代表不限，非空则只会使用其中的机器运行（多值逗号分割）
private String designatedWorkers;
// 最大机器数量
private Integer maxWorkerCount;
}

24
oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/web/request/QueryJobInfoRequest.java Normal file
View File

@ -0,0 +1,24 @@
package com.github.kfcfans.oms.server.web.request;
import lombok.Data;
/**
* 查询任务列表
*
* @author tjq
* @since 2020/4/13
*/
@Data
public class QueryJobInfoRequest {
// 任务所属应用ID
private Long appId;
// 当前页码
private Integer index;
// 页大小
private Integer pageSize;
// 查询条件
private Long jobId;
private String keyword;
}

16
oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/web/response/JobInfoVO.java
View File

@ -24,20 +24,18 @@ public class JobInfoVO {
private Long appId;
// 任务自带的参数
private String jobParams;
// 任务实例的参数(API触发专用)
private String instanceParams;
/* ************************** 定时参数 ************************** */
// 时间表达式类型（CRON/API/FIX_RATE/FIX_DELAY）
private Integer timeExpressionType;
private String timeExpressionType;
// 时间表达式，CRON/NULL/LONG/LONG
private String timeExpression;
/* ************************** 执行方式 ************************** */
// 执行类型，单机/广播/MR
private Integer executeType;
private String executeType;
// 执行器类型，Java/Shell
private Integer processorType;
private String processorType;
// 执行器信息
private String processorInfo;
@ -54,7 +52,7 @@ public class JobInfoVO {
private Integer taskRetryNum;
// 1 正常运行，2 停止（不再调度）
private Integer status;
private boolean enable;
// 下一次调度时间
private Long nextTriggerTime;
@ -68,4 +66,10 @@ public class JobInfoVO {
private Date gmtCreate;
private Date gmtModified;
/* ************************** 集群配置 ************************** */
// 指定机器运行，空代表不限，非空则只会使用其中的机器运行（多值逗号分割）
private String designatedWorkers;
// 最大机器数量
private Integer maxWorkerCount;
}