共计 5368 个字符，预计需要花费 14 分钟才能阅读完成。
作者|林俊（万念）
起源|尔达 Erda 公众号
Erda Pipeline 是端点自研、用 Go 编写的一款企业级流水线服务。截至目前，曾经为泛滥行业头部客户提供交付和稳固的服务。
为什么咱们保持自研，而不必 jenkins 等产品呢？在过后，至多有以下几点理由：
- 时至今日，开源社区仍没有一个事实上的流水线规范
- K8s、DC/OS 等的 Job 实现都偏弱，上下文传递等缺失，不满足咱们的需要，更不要说 Flow 了
- 自研能更快地响应业务需要，进行定制化开发
作为根底服务，Pipeline 在 Erda 外部撑持了 CI/CD、快数据平台、自动化测试平台、SRE 运维链路等产品化场景。本文就从几个方面来介绍一下 Pipeline。
为什么会有 Pipeline
这就须要从利用构建开始说起。Pipeline 的前身是 Packer 和 CI。
Packer
Erda 最开始是端点外部应用的 PaaS 平台。从 2017 年开始，Erda 就治理了公司所有的研发我的项目。我的项目下每个利用都逃不开 代码 -> 编译 -> 镜像制作 -> 部署 的规范流程。这个时候咱们开发了 Packer，顾名思义，它是一个专门负责 打包 的组件。用户须要提供 Dockerfile，这在过后还是有着较高学习老本的。
CI
随着 CI/CD（继续集成、继续交付）概念的深入人心，咱们也推出了 Packer 的升级版 CI。同时，基础设施即代码（IaC）的理念也在这里失去了实际：通过 erda.yaml 1.0 语法同时申明利用的微服务架构和构建过程。
在用户体验上，咱们不再间接裸露 Dockerfile，而是把最佳实际以 BuildPack 大礼包的形式给到使用者，使用者甚至不须要申明利用的开发语言和构建形式，就能够通过 BuildPack 的主动探测和辨认，实现 CI/CD 流程。
受限于单容器的运行形式，过后咱们也遇到了一些问题，譬如把 CI 构建过程自定义能力凋谢、构建环境多版本问题等，这些问题在 Pipeline 里都迎刃而解。
Pipeline
明天回过头来看，从 CI 降级到 Pipeline 是一个很天然的过程：因为 CI/CD 自身就是一个很规范的流程，咱们齐全能够形象出一个更通用的流程引擎，这就是 Pipeline。CI/CD 成为了 Pipeline 最开始撑持的场景。
在设计之初，咱们就做了以下改良：
- 对外：通过清晰易用的 pipeline.yaml 语法，升高使用者的上手老本。
- 对内：形象出工作定义，配合 ActionExecutor Plugin Mechenism（工作执行器插件机制），很不便地对接各个单任务执行平台，譬如 DC/OS Metronome、K8s Job、Flink/Spark Job 等。
- 由 Pipeline 提供统一、弱小的流程编排能力。
Pipeline 性能个性
Pipeline 有许多灵便、弱小的性能，譬如：
- 配置即代码，通过 pipeline.yaml 语法形容流程，基于 Stage 语法简化编排复杂度。
- 丰盛的扩大市场，平台内置超过百款开箱即用的 Action，满足大部分日常场景；同时可轻松扩大你本人的 Action。
- 可视化编辑，通过图形界面交互疾速配置流水线。
- 反对嵌套流水线，在流水线级别进行复用，组合出更弱小的流水线。
- 灵便的执行策略，包含串并行、循环、分支策略、超时、人工确认等。
- 反对工作流优先队列，优先级可实时调整，保障高优先级流水线优先执行。
- 多维度的重试机制，反对断点重试、全流程重试。
- 定时流水线，同时提供弱小的定时弥补性能。
- 动静配置，反对
值和文件两种类型，均反对加密存储，确保数据安全性。 - 上下文传递，后置工作能够援用前置工作的
值和文件。 - 凋谢的 OpenAPI 接口，不便第三方零碎疾速接入。
- ······
Pipeline 架构
如上图所示，Pipeline 反对 UI / OPENAPI / CLI 多种形式进行交互。
Pipeline 自身反对程度扩大，保障高可用，还能够将其划分为：服务层、核心层和引擎层。上面咱们具体介绍一下。
服务层
- yaml parser 解析流程定义文件，反对灵便的变量语法。例如上下文值援用：
${{outputs.preTaskName.key}}；配置管理援用：${{configs.key}}等。 - 对接扩大市场获取扩大能力。
核心层
- Cron 守护过程。
- EventManager 形象外部事件发送，应用适配器模式解耦监控指标上报、发送 ws 音讯、反对 webhook 等。
- AOP 扩大点机制（借鉴 Spring），把代码要害节点进行裸露，不便开发同学在不批改外围代码的前提下定制流水线行为。这个能力后续咱们还会凋谢给调用方，包含用户，反对他们去做一些有意思的事件。
目前许多有意思的性能都是通过扩大点机制实现的，譬如自动化测试报告嵌套生成、队列弹出前查看、接口测试 Cookie 放弃等：
引擎层
引擎层包含：
- 流程推进器（Reconciler）
- 优先队列管理器
- 工作执行器插件机制
具体内容在下一节会开展解说。
中间件依赖
咱们尽可能做到简化中间件依赖，使部署更简略。
- 应用 MySQL 做数据长久化。
- 应用 etcd watch 性能实现多实例状态同步以及分布式锁。
- 应用 etcd key ttl 实现数据 defer GC。
流水线是如何被推动的
在引擎侧，pipeline.yaml 被解析为 DAG（Directed Acyclic Graph，有向无环图） 构造后被推动。
换句话说，引擎并不意识、也不关怀 pipeline.yaml 语法，用户侧齐全能够提供多种多样的语法不便不同用户应用，只须要最终能被转换成 Pipeline 简略封装过的 DAG 构造。
Pipeline 级别由推进器 Reconciler 依据 DAG 计算出以后可被推动的工作，每个工作异步去执行推动逻辑。
工作的推动由 TaskFramework 解决，其中形象出 prepare -> create -> start -> queue -> wait 规范步骤。当有须要时也能够很不便地进行规范扩大。
当任意一个工作推动结束时，会再次递归调用 reconcile 办法去反复上述流程，直到流程整体执行结束。
Reconciler 中 通过 DAG 计算可调度工作代码如下：
// getSchedulableTasks return the list of schedulable tasks.
// tasks in list can be schedule concurrently.
func (r *Reconciler) getSchedulableTasks(p *spec.Pipeline, tasks []*spec.PipelineTask) ([]*spec.PipelineTask, error) {
// construct DAG
dagNodes := make([]dag.NamedNode, 0, len(tasks))
for _, task := range tasks {dagNodes = append(dagNodes, task)
}
_dag, err := dag.New(dagNodes,
// pipeline DAG 中目前能够禁用任意节点，即 dag.WithAllowMarkArbitraryNodesAsDone=true
dag.WithAllowMarkArbitraryNodesAsDone(true),
)
if err != nil {return nil, err}
// calculate schedulable nodes according to dag and current done tasks
schedulableNodeFromDAG, err := _dag.GetSchedulable((&spec.PipelineWithTasks{Tasks: tasks}).DoneTasks()...)
if err != nil {return nil, err}
......
}ActionExecutor 插件机制
把简单留给本人，把简略留给他人。
在前文咱们说到：由流水线提供灵便、统一的流程编排能力。它的前提是单个工作的执行曾经被很好的形象了。
在 Pipeline 中，咱们对一个工作执行的形象是 ActionExecutor：
type ActionExecutor interface {Kind() Kind
Name() Name
Create(ctx context.Context, action *spec.PipelineTask) (interface{}, error)
Start(ctx context.Context, action *spec.PipelineTask) (interface{}, error)
Update(ctx context.Context, action *spec.PipelineTask) (interface{}, error)
Exist(ctx context.Context, action *spec.PipelineTask) (created bool, started bool, err error)
Status(ctx context.Context, action *spec.PipelineTask) (apistructs.PipelineStatusDesc, error)
// Optional
Inspect(ctx context.Context, action *spec.PipelineTask) (apistructs.TaskInspect, error)
Cancel(ctx context.Context, action *spec.PipelineTask) (interface{}, error)
Remove(ctx context.Context, action *spec.PipelineTask) (interface{}, error)
}因而，一个执行器只有实现 单个工作 的 创立、启动、更新、状态查问、删除 等根底办法，就能够注册成为一个 ActionExecutor。
失当的工作执行器形象，使得 Batch/Streaming/InMemory Job 的配置和应用形式完全一致，批流一体，对使用者屏蔽底层细节，做到无感知切换。在同一条流水线中，能够混用各种 ActionExecutor。
调度时，Pipeline 依据工作类型和集群信息，将任务调度到对应的工作执行器上。
目前咱们曾经领有许多的 ActionExecutor：
插件化的开发机制，使咱们在将来对接其余工作引擎也变得非常简单，例如对接 Jenkins 成为一个 ActionExecutor。
这里举一个实在的例子：在自动化测试平台里，之前每一个 API 都会启动一个容器去执行，而容器的启停最快也须要数秒，这和 API 接口失常毫秒级的耗时比起来，慢了几个数量级。得益于 ActionExecutor 插件机制，咱们疾速开发了基于内存的 API-Test 工作执行器，很快就解决了这个问题，使用者不须要做任何调整，节俭了很多工夫老本。
更敌对的用户接入层 pipeline.yaml
pipeline.yaml 是 IaC 的一个实际，咱们通过 YAML 格局形容流水线定义，基于 Stage 语法简化编排复杂度。
一个简略的示例如下所示：
version: 1.1
cron: 0 */10 * * * ?
# stage 示意 阶段，多个 stage 串行成为 stages
stages:
# 一个 stage 内蕴含多个 并行 的 Action
- stage:
- git-checkout: # Action 类型
params:
depth: 1
- stage:
- buildpack:
alias: backend
params:
context: ${{dirs.git-checkout}}
resources:
cpu: 0.5
mem: 2048
- custom-script:
image: centos:7
commands: # 反对间接执行命令
- sleep 5
- echo hello world
- cat ${{dirs.git-checkout}}/erda.yml # 这里通过 ${{dirs.git-checkout}} 语法来援用文件以 Pipeline 为技术底座
目前，以 Pipeline 作为技术底座，向上撑持了：
- DevOps CI/CD 场景，包含 Erda 本身的继续集成和 Release 版本公布。
- 快数据平台：工作流编排，批流一体，反对工作流优先级队列，保障高优先级数据工作必须执行。至今已为多家世界 500 强企业和头部客户提供稳固服务。
- 自动化测试平台：测试流程编排，API（出参、断言）、数据银行等不同类型的工作对立编排。
- SRE 集群运维链路。
- 提供有限扩大：基于 ActionExecutor 扩大机制和扩大市场。
开源架构降级
目前，Pipeline 所有代码均已实现开源。咱们正在进行的重构工作包含：
- 应用 Erda-Infra 微服务架构从新梳理功能模块
- Pipeline 平台反对独立部署，UI 主动适配
- 通过 ActionExecutor 插件机制反对使用者本地 Agent，充分利用本地资源
- 在 GitHub 上推出 Erda Cloud Pipeline App，提供收费的 CI 能力
结束语
最初，咱们欢送有更多的同学来应用流水线，不论是代码级的应用，还是通过 Erda Cloud 来体验咱们的服务。
欢送 GitHub 提交 Issue 和 PR！
• Erda Github 地址：https://github.com/erda-project/erda
• Erda Cloud 官网：https://www.erda.cloud/