feat(scheduler): add support for scheduler #26
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
baifuyu
suggested changes
Dec 17, 2023
| public static final String LT = "lt"; | ||
|
|
||
| /** merge two bean by discovering differences */ | ||
| public static <M> M merge(M dest, M orig) { |
There was a problem hiding this comment.
这个方法不是SchedulerUtils？而应该放在BeanUtils呢？
There was a problem hiding this comment.
调度的本地查询merge逻辑，只有为目标值空才会覆盖，不适合放到BeanUtils
| descriptor.getWriteMethod().invoke(dest, originalValue); | ||
| } | ||
| } catch (Exception e) { | ||
| log.error("merge bean exception", e); |
There was a problem hiding this comment.
这个merge失败要抛异常的，不然异常吞了上游要得到非预期结果
There was a problem hiding this comment.
因为入参都是同一类型M，这里反射不会抛出异常。
| } | ||
|
|
||
| /** Limit remark. sub String To Length */ | ||
| public static String setRemarkLimit(String oldRemark, StringBuffer appendRemark) { |
There was a problem hiding this comment.
这个方法命名感觉也不太对。太定制化了是不是不需要放在common/util？直接scheduler包那里使用就ok了
There was a problem hiding this comment.
静态的工具方法统一放在SchedulerUtils.java类中，供Scheduler统一调用是合理的吧
| } | ||
|
|
||
| /** content compare key */ | ||
| public static boolean compare(Object content, Object key, String type) { |
There was a problem hiding this comment.
这个方法太定制化了，不像是common/util呢。key/content有可能是 Date或者String类型。
There was a problem hiding this comment.
Scheduler用于对象属性比较的，提供给Scheduler的本地查询用，所以统一放在SchedulerUtils中
| } | ||
|
|
||
| /** get local ips */ | ||
| public static List<String> getLocalIps() { |
There was a problem hiding this comment.
这个方法NetworkAddressUtils里面已经有了
| } | ||
| executeInstance(instance, runningTasks); | ||
| } catch (Exception e) { | ||
| log.error("execute instance error id:", id, e); |
| if (instances.containsKey(type)) { | ||
| return instances.get(type); | ||
| } | ||
| ThreadPoolExecutor instanceExecutor = |
There was a problem hiding this comment.
直接使用ThreadUtils里面的方法创建线程池，里面方法不满足的直接改。别再自己手动创建线程池了。
| import org.springframework.stereotype.Component; | ||
|
|
||
| /** Scheduler Common Value */ | ||
| @Component |
There was a problem hiding this comment.
类名是否改为SchedulerConfig更好？另外往上挪一级吧。放在common里面有点奇怪
| /** Scheduler Common Service */ | ||
| @Service | ||
| @Slf4j | ||
| public class SchedulerCommonService { |
There was a problem hiding this comment.
叫SchedulerCommonService太奇怪了。其实也不是通用服务，这里东西有些是状态相关，有些又是调度实例相关
There was a problem hiding this comment.
这里的方法是对各模块复用的方法进行的抽取 统一放到了一个地方，Scheduler各个地方能进行复用
|
|
||
| /** Scheduler Job Service implementation class: Add, delete, update, and query Jobs */ | ||
| @Service | ||
| public class LocalSchedulerJobServiceImpl implements SchedulerJobService { |
There was a problem hiding this comment.
这里也需要有个工厂，根据不同的metadata生成不同的Local/Db等service。而不应该直接打@service注解
There was a problem hiding this comment.
已经修改为@ConditionalOnProperty的实现
baifuyu
reviewed
Dec 18, 2023
| * or implied. | ||
| */ | ||
|
|
||
| /** Alipay.com Inc. Copyright (c) 2004-2021 All Rights Reserved. */ |
| private Date estimateFinishTime; | ||
|
|
||
| /** remark */ | ||
| private String remark; |
leywar
requested changes
Dec 26, 2023
| /** Scheduler Service:submit,execute,delete and other scheduler interfaces */ | ||
| public interface SchedulerService { | ||
| /** submit job */ | ||
| SchedulerJob submitJob(SchedulerJob job); |
There was a problem hiding this comment.
comments cannot be the same as function name
| List<SchedulerInstance> searchInstances(SchedulerInstance query); | ||
|
|
||
| /** search Tasks */ | ||
| List<SchedulerTask> searchTasks(SchedulerTask query); |
| import org.springframework.stereotype.Service; | ||
| import org.springframework.util.Assert; | ||
|
|
||
| /** Scheduler Service implementation class */ |
...n/java/com/antgroup/openspg/server/core/scheduler/service/api/impl/SchedulerServiceImpl.java
Show resolved
Hide resolved
| /** set Instance To Finish */ | ||
| public void setInstanceFinish( | ||
| SchedulerInstance instance, InstanceStatus instanceStatus, TaskStatus taskStatus) { | ||
| Long finish = 100L; |
| updateInstance.setFinishTime(finishTime); | ||
| Long updateNum = schedulerInstanceService.update(updateInstance); | ||
| Assert.isTrue(updateNum > 0, "update instance failed " + updateInstance); | ||
| stopRunningTasks(instance); |
There was a problem hiding this comment.
add a blank line between line 66 and 67
| AsyncTaskExecute jobAsyncTask = (AsyncTaskExecute) jobTask; | ||
| TaskExecuteContext context = new TaskExecuteContext(job, instance, task); | ||
| jobAsyncTask.stop(context, task.getResource()); | ||
| } else { |
There was a problem hiding this comment.
add continue after line 87 can reduce 1 level code indentation.
| private void checkInstanceRunning(SchedulerJob job) { | ||
| SchedulerInstance query = new SchedulerInstance(); | ||
| query.setJobId(job.getId()); | ||
| query.setStartCreateTime(DateUtils.addDays(new Date(), -1)); |
| if (jobId.equals(instance.getJobId())) { | ||
| instanceList.add(instance.getId()); | ||
| } | ||
| } |
| } | ||
|
|
||
| /** Merge instance pre-check */ | ||
| public TaskStatus processByMerge(TaskExecuteContext context) { |
There was a problem hiding this comment.
this comment does not have any valid information
leywar
approved these changes
Dec 30, 2023
| } | ||
| } | ||
|
|
||
| /** lock task */ |
There was a problem hiding this comment.
comments cannot be the same with function name.
| return true; | ||
| } | ||
|
|
||
| /** unlock task */ |
| nextNodes.forEach(node -> startNextNode(context, instance.getTaskDag(), node)); | ||
| } | ||
|
|
||
| /** start next node */ |
There was a problem hiding this comment.
comments cannot be the same with function name
| } | ||
|
|
||
| /** check all nodes is finished */ | ||
| private boolean checkAllNodesFinished(SchedulerTask task, List<TaskExecuteDag.Node> nodes) { |
| return true; | ||
| } | ||
|
|
||
| /** set instance to finished */ |
| String submit(TaskExecuteContext context); | ||
|
|
||
| /** get task Status */ | ||
| TaskStatus getStatus(TaskExecuteContext context, String resource); |
| /** | ||
| * step 1: create Period Job to submit step 2: query Jobs and Instances step 3: offline Job step | ||
| * 4: online Job step 5: execute Job step 6: trigger first Instance until it ends step 7: trigger | ||
| * second Instance until it ends step 8: get tasks step 9: delete Job |
There was a problem hiding this comment.
Code formatting will automatically indent
| SchedulerJob job = new SchedulerJob(); | ||
| job.setProjectId(0L); | ||
| job.setName("Test Period Job"); | ||
| job.setCreateUser("test"); |
| /** | ||
| * step 1: create RealTime Job to submit step 2: query Jobs and Instances step 3: offline Job step | ||
| * 4: online Job step 5: trigger Instance step 6: get tasks step 7: delete Job; | ||
| */ |
There was a problem hiding this comment.
Code formatting will automatically indent
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
license-eye has checked 1594 files.
| Valid | Invalid | Ignored | Fixed |
|---|---|---|---|
| 1529 | 39 | 26 | 0 |
Click to see the invalid file list
- server/common/model/src/main/java/com/antgroup/openspg/server/common/model/exception/SchedulerException.java
- server/common/model/src/main/java/com/antgroup/openspg/server/common/model/scheduler/SchedulerEnum.java
- server/core/scheduler/model/pom.xml
- server/core/scheduler/model/src/main/java/com/antgroup/openspg/server/core/scheduler/model/service/SchedulerInstance.java
- server/core/scheduler/model/src/main/java/com/antgroup/openspg/server/core/scheduler/model/service/SchedulerJob.java
- server/core/scheduler/model/src/main/java/com/antgroup/openspg/server/core/scheduler/model/service/SchedulerTask.java
- server/core/scheduler/model/src/main/java/com/antgroup/openspg/server/core/scheduler/model/task/TaskExecuteContext.java
- server/core/scheduler/model/src/main/java/com/antgroup/openspg/server/core/scheduler/model/task/TaskExecuteDag.java
- server/core/scheduler/service/pom.xml
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/api/SchedulerService.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/api/impl/SchedulerServiceImpl.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/common/SchedulerCommonService.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/config/SchedulerConfig.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/engine/SchedulerExecuteService.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/engine/impl/SchedulerExecuteServiceImpl.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/handler/SchedulerHandler.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/handler/impl/local/LocalSchedulerHandler.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/metadata/SchedulerInstanceService.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/metadata/SchedulerJobService.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/metadata/SchedulerTaskService.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/metadata/impl/local/LocalSchedulerInstanceServiceImpl.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/metadata/impl/local/LocalSchedulerJobServiceImpl.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/metadata/impl/local/LocalSchedulerTaskServiceImpl.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/task/TaskExecute.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/task/TaskExecuteTemplate.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/task/async/AsyncTaskExecute.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/task/async/AsyncTaskExecuteTemplate.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/task/sync/SyncTaskExecute.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/task/sync/SyncTaskExecuteTemplate.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/translate/Translate.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/translate/TranslatorFactory.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/utils/SchedulerUtils.java
- server/test/src/test/java/com/antgroup/openspg/test/kgschema/SPGSchemaFacadeTest.groovy
- server/test/src/test/java/com/antgroup/openspg/test/scheduler/SchedulerServiceImplTest.java
- server/test/src/test/java/com/antgroup/openspg/test/scheduler/task/LocalExampleAsyncTaskMock.java
- server/test/src/test/java/com/antgroup/openspg/test/scheduler/task/LocalExampleSyncTaskMock.java
- server/test/src/test/java/com/antgroup/openspg/test/scheduler/translate/LocalExampleTranslateMock.java
- server/test/src/test/resources/config/application.properties
- server/test/src/test/resources/spring/spring-common.xml
Use this command to fix any missing license headers
```bash
docker run -it --rm -v $(pwd):/github/workspace apache/skywalking-eyes header fix
</details>
There was a problem hiding this comment.
license-eye has checked 1594 files.
| Valid | Invalid | Ignored | Fixed |
|---|---|---|---|
| 1529 | 39 | 26 | 0 |
Click to see the invalid file list
- server/common/model/src/main/java/com/antgroup/openspg/server/common/model/exception/SchedulerException.java
- server/common/model/src/main/java/com/antgroup/openspg/server/common/model/scheduler/SchedulerEnum.java
- server/core/scheduler/model/pom.xml
- server/core/scheduler/model/src/main/java/com/antgroup/openspg/server/core/scheduler/model/service/SchedulerInstance.java
- server/core/scheduler/model/src/main/java/com/antgroup/openspg/server/core/scheduler/model/service/SchedulerJob.java
- server/core/scheduler/model/src/main/java/com/antgroup/openspg/server/core/scheduler/model/service/SchedulerTask.java
- server/core/scheduler/model/src/main/java/com/antgroup/openspg/server/core/scheduler/model/task/TaskExecuteContext.java
- server/core/scheduler/model/src/main/java/com/antgroup/openspg/server/core/scheduler/model/task/TaskExecuteDag.java
- server/core/scheduler/service/pom.xml
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/api/SchedulerService.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/api/impl/SchedulerServiceImpl.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/common/SchedulerCommonService.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/config/SchedulerConfig.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/engine/SchedulerExecuteService.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/engine/impl/SchedulerExecuteServiceImpl.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/handler/SchedulerHandler.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/handler/impl/local/LocalSchedulerHandler.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/metadata/SchedulerInstanceService.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/metadata/SchedulerJobService.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/metadata/SchedulerTaskService.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/metadata/impl/local/LocalSchedulerInstanceServiceImpl.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/metadata/impl/local/LocalSchedulerJobServiceImpl.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/metadata/impl/local/LocalSchedulerTaskServiceImpl.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/task/TaskExecute.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/task/TaskExecuteTemplate.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/task/async/AsyncTaskExecute.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/task/async/AsyncTaskExecuteTemplate.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/task/sync/SyncTaskExecute.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/task/sync/SyncTaskExecuteTemplate.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/translate/Translate.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/translate/TranslatorFactory.java
- server/core/scheduler/service/src/main/java/com/antgroup/openspg/server/core/scheduler/service/utils/SchedulerUtils.java
- server/test/src/test/java/com/antgroup/openspg/test/kgschema/SPGSchemaFacadeTest.groovy
- server/test/src/test/java/com/antgroup/openspg/test/scheduler/SchedulerServiceImplTest.java
- server/test/src/test/java/com/antgroup/openspg/test/scheduler/task/LocalExampleAsyncTaskMock.java
- server/test/src/test/java/com/antgroup/openspg/test/scheduler/task/LocalExampleSyncTaskMock.java
- server/test/src/test/java/com/antgroup/openspg/test/scheduler/translate/LocalExampleTranslateMock.java
- server/test/src/test/resources/config/application.properties
- server/test/src/test/resources/spring/spring-common.xml
Use this command to fix any missing license headers
```bash
docker run -it --rm -v $(pwd):/github/workspace apache/skywalking-eyes header fix
</details>
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
init Scheduler Model
init Scheduler Api and impl
init Scheduler Task and sync/async impl
init Scheduler Metadata and Local impl
init Scheduler Handler and impl
init Scheduler Translate and impl
Scheduler engine impl