开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> 系统运维 -> 开源基于docker的任务调度器pipeline，比`quartzs` 更强大的分布式任务调度器 -> 正文阅读

[系统运维]开源基于docker的任务调度器pipeline，比`quartzs` 更强大的分布式任务调度器

pipeline 分布式任务调度器

目标：基于docker的布式任务调度器, 比quartzs,xxl-job 更强大的分布式任务调度器。

可以将要执行的任务打包为docker镜像，或者选择已有镜像，自定义脚本程序，通过pipeline框架来实现调度。

开源地址： https://github.com/jadepeng/docker-pipeline

架构

pipeline master 中心节点，管理和调度任务
pipeline agent 执行任务的节点，接收到任务后，调用docker执行pipeline任务

功能特性 && TODO List

分布式框架，高可用，服务注册与状态维护
Agent执行任务
rolling日志接口
运行老版本pipeline任务
支持定时执行任务(固定周期和cron表达式)
快速创建任务，支持python、node等脚本程序直接执行
- python、java等基础镜像
- 快速docker镜像任务API
- 快速创建脚本任务
根据资源配额（内存、CPU）调度任务, 运行任务需要指定资源配额
agent 增加label标识，调度时可以调度到指定label的agent，比如gpu=true
增加任务管理web, 管理提交任务、查询运行日志等
- 复用腾讯bk-job 网页
- 修改bk-job前端，适配pipeline

进展

2021.07.31

支持定时执行任务(固定周期和cron表达式)
增加分布式mongodb锁，多master时，同时只能有一个master schedule任务

2021.07.28

新增运行老版本pipeline任务能力
增加日志接口

2021.07.27

引入bk-job的ui，待修改

2021.07.21

Master 调用 agent执行任务
agnet 启动docker执行任务

2021.07.19

基于jhipster搭建框架
分布式实现

数据结构

一个pipeline 任务：

支持多个pipelineTask
一个pipelineTask 包含多个Step

@Data
public class Pipeline {

    @Id
    private String id;

    private String name;

    @JSONField(name = "pipeline")
    private List<PipelineTask> pipelineTasks = new ArrayList<>();

    private List<Network> networks = Lists.newArrayList(new Network());

    private List<Volume> volumes = Lists.newArrayList(new Volume());

    private String startNode;

    /**
     * 调度类型：
     *      1） CRON， 设置cronExpression
     *      2） FIX_RATE， 设置fixRateInSeconds
     */
    private ScheduleType scheduleType = ScheduleType.NONE;

    /**
     * CRON表达式，在scheduleType=CRON 时生效
     */
    private String cronExpression;

    /**
     * 固定周期运行，比如每隔多少s,在scheduleType=FIX_RATE 时生效
     */
    private int fixRateInSeconds;

    /**
     * 是否需要调度,为true时，才调度
     */
    @Indexed
    private boolean enableTrigger;

    private long lastTriggerTime;

    @Indexed
    private long nextTriggerTime;

    /**
     * 执行超时时间
     */
    private int executorTimeout;

    /**
     * 重试次数
     */
    private int executorFailRetryCount;

    /**
     * 内存限制
     */
    private String memory;

    /**
     * CPU 限制
     */
    private String cpu;

    @Data
    @Builder
    public static class PipelineTask {

        /**
         * 名称
         */
        String name;

        /**
         * 别名
         */
        String alias;

        /**
         * 依赖的pipelines，必须依赖的执行完成才能运行该PipelineTask
         */
        List<String> dependencies;

        /**
         * 任务步骤，顺序执行
         */
        List<Step> steps;
    }

    @Data
    public static class Network {
        String name = "pipeline_default";
        String driver = "bridge";
    }

    @Data
    public static class Volume {
        String name = "pipeline_default";
        String driver = "local";
    }

    @Data
    public static class StepNetwork {
        private String name;
        private List<String> aliases = Lists.newArrayList("default");

        public StepNetwork(String name) {
            this.name = name;
        }
    }

}

举例：

{
    "_id" : "29103d5e4a77409b9f6050eea8110bb3",
    "name" : "docker image pipeline",
    "pipelineTasks" : [ 
        {
            "name" : "docker image pipeline",
            "steps" : [ 
                {
                    "name" : "defaultJob",
                    "image" : "java-pipeline:1.0.1",
                    "workingDir" : "/workspace",
                    "environment" : {},
                    "networks" : [ 
                        {
                            "name" : "pipeline_default",
                            "aliases" : [ 
                                "default"
                            ]
                        }
                    ],
                    "onSuccess" : false,
                    "authConfig" : {}
                }
            ]
        }
    ],
    "networks" : [ 
        {
            "name" : "pipeline_default",
            "driver" : "bridge"
        }
    ],
    "volumes" : [ 
        {
            "name" : "pipeline_default",
            "driver" : "local"
        }
    ],
    "cronExpression" : "0 0 * * * ?",
    "fixRateInSeconds" : 0,
    "scheduleType" : "CRON",
    "enableTrigger" : true,
    "lastTriggerTime" : 1627744509047,
    "nextTriggerTime" : 1627747200000,
    "executorTimeout" : 0,
    "executorFailRetryCount" : 0,
    "isAvailable" : 1,
    "runningPipelines" : [],
    "finishedPipeliens" : [],
    "created_by" : "admin",
    "created_date" : "2021-07-20T04:33:16.477Z",
    "last_modified_by" : "system",
    "last_modified_date" : "2021-07-31T15:15:09.048Z"
}

使用说明

安装部署

编译

使用mvn编译

mvn package -DskipTests

部署master

根据需要，修改master的prod配置文件application-prod.yml。

包含kafka配置，server端口，mongodb地址,jwt secret配置。

mongodb 会自动新建collection和初始化数据，无需手动导入数据。

kafka:
    producer:
      bootstrap-servers: 127.0.0.1:9092
      retries: 3
      batch-size: 2000
      buffer-memory: 33554432
    consumer:
      group-id: consumer-pipeline
      auto-offset-reset: earliest
      enable-auto-commit: true
      bootstrap-servers: 172.31.161.38:9092

server:
  port: 8080

spring:
  data:
    mongodb:
      uri: mongodb://127.0.0.1:28017
      database: pipeline

jhipster:
  security:
    authentication:
      jwt:
        base64-secret:

注意master的jwt secret需要和agent的保持一致。

配置好后，启动：

nohup java -jar pipeline-master-$version.jar --spring.profiles.active=prod &

可以将application-prod.yml放在和jar包同一目录。

部署agent

根据需要，修改master的prod配置文件application-prod.yml。

包含：

eureka的defaultZone，配置master的地址
端口
docker地址
- docker-tls-verify: 是否启动tls验证
- docker-cert-path：启动tls验证的ca证书
- pipeline-log-path: 运行日志存储路径


eureka:
  instance:
    prefer-ip-address: true
  client:
    service-url:
      defaultZone: http://admin:${jhipster.registry.password}@127.0.0.1:8080/eureka/

server:
  port: 8081

application:
  docker-server: 
  docker-tls-verify: true
  docker-cert-path: /mnt/parastor/pipeline/ca/
  pipeline-log-path: /mnt/parastor/pipeline/logs/


jhipster:
  security:
    authentication:
      jwt:
        base64-secret:

执行老版本任务

POST /api/pipelines/exec-old


Body：

{
	"networks":[
		{
			"driver":"bridge",
			"name":"pipeline_network_3eac4b36209a41e58a5f22dd403fee50"
		}
	],
	"pipeline":[
		{
			"alias":"Word",
			"dependencies":[],
			"name":"pipeline_task_3eac4b36209a41e58a5f22dd403fee50_1",
			"nextPipelines":[],
			"steps":[
				{
					"alias":"Word",
					"auth_config":{},
					"command":[
						"echo $CI_SCRIPT | base64 -d | /bin/bash -e"
					],
					"entrypoint":[
						"/bin/bash",
						"-c"
					],
					"environment":{
						"CI_SCRIPT":"CmlmIFsgLW4gIiRDSV9ORVRSQ19NQUNISU5FIiBdOyB0aGVuCmNhdCA8PEVPRiA+ICRIT01FLy5uZXRyYwptYWNoaW5lICRDSV9ORVRSQ19NQUNISU5FCmxvZ2luICRDSV9ORVRSQ19VU0VSTkFNRQpwYXNzd29yZCAkQ0lfTkVUUkNfUEFTU1dPUkQKRU9GCmNobW9kIDA2MDAgJEhPTUUvLm5ldHJjCmZpCnVuc2V0IENJX05FVFJDX1VTRVJOQU1FCnVuc2V0IENJX05FVFJDX1BBU1NXT1JECnVuc2V0IENJX1NDUklQVAplY2hvICsgamF2YSAtY3AgL2RhdGF2b2x1bWUvcGRmX3RvX3dvcmQvcGRmYm94X3V0aWwtMS4wLVNOQVBTSE9ULmphciBjb20uaWZseXRlay5pbmRleGVyLlJ1bm5lciAtLWlucHV0UERGIC9kYXRhdm9sdW1lL2V4dHJhY3QvZjkyYzJhNzViYWU4NGJiMDg4MzIwNWRiM2YyZGFlNzkvcGRmL2VjNWMwYjk0M2QwYjRmNDI5MzcyMmE1ZGRjNjFlNjZkL0hTNy5wZGYgLS1vdXRwdXRXb3JkIC9kYXRhdm9sdW1lL2V4dHJhY3QvZjkyYzJhNzViYWU4NGJiMDg4MzIwNWRiM2YyZGFlNzkvcGRmVG9Xb3JkL2VjNWMwYjk0M2QwYjRmNDI5MzcyMmE1ZGRjNjFlNjZkLyAtLXNjaGVtYUlucHV0UGF0aCAvZGF0YXZvbHVtZS9leHRyYWN0L2ticWEvZjkyYzJhNzViYWU4NGJiMDg4MzIwNWRiM2YyZGFlNzkgLS1lbnRpdHlJbmRleFBhdGggL2RhdGF2b2x1bWUvZXh0cmFjdC9mOTJjMmE3NWJhZTg0YmIwODgzMjA1ZGIzZjJkYWU3OS9wZGZUb1dvcmQvZW50aXR5IC0tZmllbGRJbmRleFBhdGggL2RhdGF2b2x1bWUvZXh0cmFjdC9mOTJjMmE3NWJhZTg0YmIwODgzMjA1ZGIzZjJkYWU3OS9wZGZUb1dvcmQvZmllbGQgLS10eXBlIGx1Y2VuZSAtLW91dHB1dCAvZGF0YXZvbHVtZS9leHRyYWN0L2Y5MmMyYTc1YmFlODRiYjA4ODMyMDVkYjNmMmRhZTc5L3BkZlRvV29yZC9lYzVjMGI5NDNkMGI0ZjQyOTM3MjJhNWRkYzYxZTY2ZC9lbnRpdHlJbmZvLnR4dApqYXZhIC1jcCAvZGF0YXZvbHVtZS9wZGZfdG9fd29yZC9wZGZib3hfdXRpbC0xLjAtU05BUFNIT1QuamFyIGNvbS5pZmx5dGVrLmluZGV4ZXIuUnVubmVyIC0taW5wdXRQREYgL2RhdGF2b2x1bWUvZXh0cmFjdC9mOTJjMmE3NWJhZTg0YmIwODgzMjA1ZGIzZjJkYWU3OS9wZGYvZWM1YzBiOTQzZDBiNGY0MjkzNzIyYTVkZGM2MWU2NmQvSFM3LnBkZiAtLW91dHB1dFdvcmQgL2RhdGF2b2x1bWUvZXh0cmFjdC9mOTJjMmE3NWJhZTg0YmIwODgzMjA1ZGIzZjJkYWU3OS9wZGZUb1dvcmQvZWM1YzBiOTQzZDBiNGY0MjkzNzIyYTVkZGM2MWU2NmQvIC0tc2NoZW1hSW5wdXRQYXRoIC9kYXRhdm9sdW1lL2V4dHJhY3Qva2JxYS9mOTJjMmE3NWJhZTg0YmIwODgzMjA1ZGIzZjJkYWU3OSAtLWVudGl0eUluZGV4UGF0aCAvZGF0YXZvbHVtZS9leHRyYWN0L2Y5MmMyYTc1YmFlODRiYjA4ODMyMDVkYjNmMmRhZTc5L3BkZlRvV29yZC9lbnRpdHkgLS1maWVsZEluZGV4UGF0aCAvZGF0YXZvbHVtZS9leHRyYWN0L2Y5MmMyYTc1YmFlODRiYjA4ODMyMDVkYjNmMmRhZTc5L3BkZlRvV29yZC9maWVsZCAtLXR5cGUgbHVjZW5lIC0tb3V0cHV0IC9kYXRhdm9sdW1lL2V4dHJhY3QvZjkyYzJhNzViYWU4NGJiMDg4MzIwNWRiM2YyZGFlNzkvcGRmVG9Xb3JkL2VjNWMwYjk0M2QwYjRmNDI5MzcyMmE1ZGRjNjFlNjZkL2VudGl0eUluZm8udHh0Cg=="
					},
					"image":"registry.iflyresearch.com/aimind/java:v1.0.0",
					"name":"pipeline_task_3eac4b36209a41e58a5f22dd403fee50_1",
					"networks":[
						{
							"aliases":[
								"default"
							],
							"name":"pipeline_network_3eac4b36209a41e58a5f22dd403fee50"
						}
					],
					"on_success":true,
					"volumes":[
						"pipeline_default:/aimind",
						"/mnt/parastor/aimind/shared/:/share",
						"/mnt/parastor/aimind/pipeline-jobs/2021/07/26/3eac4b36209a41e58a5f22dd403fee50:/workspace",
						"/mnt/parastor/aimind/datavolumes/carmaster:/datavolume"
					],
					"working_dir":"/workspace"
				}
			]
		}
	],
	"volumes":[
		{
			"driver":"local",
			"name":"pipeline_default"
		}
	]
}

成功返回：

{
    "retcode": "000000",
    "desc": "成功",
    "data": {
        "id": "8137f344-f52d-4595-bdbb-425363847b61",
    }
}

可根据id获取日志。

获取job执行日志

GET /api/pipelines/jobLog/{jobid}/

结果：

{
    "retcode": "000000",
    "desc": "成功",
    "data": {
        "currentTask": null,
        "logs": [
            {
                "id": "e76a686f68b64c0783b7721b058be137",
                "jobId": "8137f344-f52d-4595-bdbb-425363847b61",
                "status": "FINISHED",
                "taskName": "pipeline_task_3eac4b36209a41e58a5f22dd403fee50_1",
                "exitedValue": 0,
                "logs": [
                    "proc \"pipeline_task_3eac4b36209a41e58a5f22dd403fee50_1\" started",
                    "pipeline_task_3eac4b36209a41e58a5f22dd403fee50_1:+ java -cp /datavolume/pdf_to_word/pdfbox_util-1.0-SNAPSHOT.jar com.iflytek.indexer.Runner --inputPDF /datavolume/extract/f92c2a75bae84bb0883205db3f2dae79/pdf/ec5c0b943d0b4f4293722a5ddc61e66d/HS7.pdf --outputWord /datavolume/extract/f92c2a75bae84bb0883205db3f2dae79/pdfToWord/ec5c0b943d0b4f4293722a5ddc61e66d/ --schemaInputPath /datavolume/extract/kbqa/f92c2a75bae84bb0883205db3f2dae79 --entityIndexPath /datavolume/extract/f92c2a75bae84bb0883205db3f2dae79/pdfToWord/entity --fieldIndexPath /datavolume/extract/f92c2a75bae84bb0883205db3f2dae79/pdfToWord/field --type lucene --output /datavolume/extract/f92c2a75bae84bb0883205db3f2dae79/pdfToWord/ec5c0b943d0b4f4293722a5ddc61e66d/entityInfo.txt",
                    "proc \"pipeline_task_3eac4b36209a41e58a5f22dd403fee50_1\" exited with status 0"
                ]
            }
        ],
        "exitedValue": 0,
        "status": "FINISHED",
        "pipelineJobSt": 1627477250599,
        "pipelineJobFt": 1627477274299
    }
}

周期任务

如果pipelien需要周期执行,需要配置enableTrigger为true，同时设置按照CRON或者’FIX_RATE` 运行:

FIX_RATE: 固定周期，通过fixRateInSeconds配置周期运行时间

示例：每360秒运行一次：

{
    // pipeline ...
    "pipelineTasks" : [ ],
    "fixRateInSeconds" : 360,
    "scheduleType" : "FIX_RATE",
    "enableTrigger" : true
}

CRON: 按照CRON表达式周期执行，通过cronExpression配置.

示例：每小时开始的时候运行一次：

{
    // pipeline ...
    "pipelineTasks" : [ ],
    "cronExpression" : "0 0 * * * ?",
    "scheduleType" : "CRON",
    "enableTrigger" : true
}