pipelinex

README

pipelinex

This is a programming library that can execute CICD tasks and can support Docker, Kubernetes, SSH, and local machines as execution backends.

TODO

• status store (现在是和日志强绑定的，需要库中提供接口来实现状态的读取和保存)
• 有向无环图
• k8s ssh持久连接需要一个连接做多个事情，如果这些事情是串行的

Documentation

Index

• Constants
• type Adapter
• type Bridge
• type DGAGraph
  • func NewDGAGraph() *DGAGraph
  • func (dga *DGAGraph) AddEdge(src, dest Node)
  • func (dga *DGAGraph) AddVertex(node Node)
  • func (dga *DGAGraph) BFS() []string
  • func (dga *DGAGraph) CycelCheck() bool
  • func (dga *DGAGraph) Nodes() map[string]Node
• type DGANode
  • func NewDGANode(id, groupId, status string) *DGANode
  • func (dgaNode *DGANode) Get(key string) string
  • func (dgaNode *DGANode) GroupId() string
  • func (dgaNode *DGANode) Id() string
  • func (dgaNode *DGANode) Set(key string, value interface{})
  • func (dgaNode *DGANode) Status() string
• type Event
• type Executor
• type Graph
• type Listener
• type ListeningFn
• type Metadata
• type Node
• type Pipeline
  • func NewPipeline(ctx context.Context) Pipeline
• type PipelineImpl
  • func (p *PipelineImpl) Cancel()
  • func (p *PipelineImpl) Done() <-chan struct{}
  • func (p *PipelineImpl) GetGraph() Graph
  • func (p *PipelineImpl) Id() string
  • func (p *PipelineImpl) Listening(fn Listener)
  • func (p *PipelineImpl) Metadata() Metadata
  • func (p *PipelineImpl) Notify()
  • func (p *PipelineImpl) Run(ctx context.Context) error
  • func (p *PipelineImpl) SetGraph(graph Graph)
  • func (p *PipelineImpl) Status() string
• type Runtime

Constants

const (
	StatusRunning   = "RUNNING"
	StatusFailed    = "FAILED"
	StatusSuccess   = "SUCCESS"
	StatusTerminate = "ABORTED"
	StatusPaused    = "PAUSED"
	StatusUnknown   = "UNKNOWN"
)

Types

type Adapter

type Adapter interface {
	// Config 适配器配置
	Config(config map[string]interface{}) Adapter
	// Conn 连接到环境中
	Conn(ctx context.Context) interface{}
}

type Bridge

type Bridge interface {
	// Transfer 传输需要执行的数据，并且反回执行的结果
	Transfer(ctx context.Context, in chan<- interface{}, out <-chan interface{})
}

type DGAGraph

type DGAGraph struct {
	// contains filtered or unexported fields
}

func NewDGAGraph

func NewDGAGraph() *DGAGraph

func (*DGAGraph) AddEdge

func (dga *DGAGraph) AddEdge(src, dest Node)

AddEdge 添加边

func (*DGAGraph) AddVertex

func (dga *DGAGraph) AddVertex(node Node)

AddVertex 添加顶点

func (*DGAGraph) BFS

func (dga *DGAGraph) BFS() []string

BFS 广度有限搜索返回搜索序列

func (*DGAGraph) CycelCheck

func (dga *DGAGraph) CycelCheck() bool

CycelCheck 循环检查序列

func (*DGAGraph) Nodes

func (dga *DGAGraph) Nodes() map[string]Node

type DGANode

type DGANode struct {
	// contains filtered or unexported fields
}

func NewDGANode

func NewDGANode(id, groupId, status string) *DGANode

func (*DGANode) Get

func (dgaNode *DGANode) Get(key string) string

func (*DGANode) GroupId

func (dgaNode *DGANode) GroupId() string

func (*DGANode) Id

func (dgaNode *DGANode) Id() string

func (*DGANode) Set

func (dgaNode *DGANode) Set(key string, value interface{})

func (*DGANode) Status

func (dgaNode *DGANode) Status() string

type Event

type Event string

流水线事件

var (
	//监听事件
	PipelineInit                Event = "pipeline-init"  // 流水线初始化
	PipelineStart               Event = "pipeline-start" // 流水线开始执行
	PipelineFinish              Event = "pipeline-finish"
	PipelineExecutorPrepare     Event = "pipeline-executor-prepare"      // 流水线执行器开始准备
	PipelineExecutorPrepareDone Event = "pipeline-executor-prepare-done" // 流水线执行器准备完毕
	PipelineNodeStart           Event = "pipeline-node-start"
	PipelineNodeFinish          Event = "pipeline-node-finish"
)

type Executor

type Executor interface {
	// Prepare 准备环境
	Prepare(ctx context.Context) error
	// Destruction 销毁环境
	Destruction(ctx context.Context) error
}

Executor 执行器

type Graph

type Graph interface {
	//Nodes
	Nodes() map[string]Node
	//AddVertex 添加顶点
	AddVertex(node Node)
	//AddEdge 添加边
	AddEdge(src, dest string)
	//CycelCheck
	CycelCheck() bool
	//BFS 广度有限搜索
	BFS() []string
}

type Listener

type Listener interface {
	// 处理对应的事件将事件发生的对应的流水线和对应的事件作为参数传入
	Handle(p Pipeline, event Event)
	// 获取当前注册的Event
	Events() []Event
}

我们将整个流水线的运行过程中的事件抽象成对应的Event 这样我们就能再外部监听Event

type ListeningFn

type ListeningFn func(p Pipeline)

PipelineListeningFn 流水线监听函数

type Metadata

type Metadata map[string]interface{}

type Node

type Node interface {
	//ID 获取节点唯一id
	Id() string
	//节点组id
	GroupId() string
	//Status 获取节点状态
	Status() string
	//Get 获取节点属性数据
	Get(key string) string
	// Set 设置节点属性数据
	Set(key string, value interface{})
}

type Pipeline

type Pipeline interface {
	//ID 流水线的id
	Id() string
	//GetGraph 返回图结构
	GetGraph() Graph
	//SetGraph 设置图结构
	SetGraph(graph Graph)
	//Status 返回流水线的整体状态
	Status() string
	//Metadata 返回流水线执行的源数据
	Metadata() Metadata
	//Listening 流水线执行事件监听设置
	Listening(listener Listener)
	//Done流水线是否执行完成
	Done() <-chan struct{}
	//Run执行流水线
	Run(ctx context.Context) error
	//Notify 执行的步骤通知流水线
	Notify()
	//Cancel 取消流水线
	Cancel()
}

func NewPipeline

func NewPipeline(ctx context.Context) Pipeline

type PipelineImpl

type PipelineImpl struct {
	// contains filtered or unexported fields
}

func (*PipelineImpl) Cancel

func (p *PipelineImpl) Cancel()

func (*PipelineImpl) Done

func (p *PipelineImpl) Done() <-chan struct{}

Done流水线是否执行完成

func (*PipelineImpl) GetGraph

func (p *PipelineImpl) GetGraph() Graph

GetGraph 返回图结构

func (*PipelineImpl) Id

func (p *PipelineImpl) Id() string

ID 流水线的id

func (*PipelineImpl) Listening

func (p *PipelineImpl) Listening(fn Listener)

Listening 流水线执行事件监听设置

func (*PipelineImpl) Metadata

func (p *PipelineImpl) Metadata() Metadata

Metadata 返回流水线执行的源数据

func (*PipelineImpl) Notify

func (p *PipelineImpl) Notify()

func (*PipelineImpl) Run

func (p *PipelineImpl) Run(ctx context.Context) error

func (*PipelineImpl) SetGraph

func (p *PipelineImpl) SetGraph(graph Graph)

SetGraph 设置图结构

func (*PipelineImpl) Status

func (p *PipelineImpl) Status() string

Status 返回流水线的整体状态

type Runtime

type Runtime interface {
	//获取流水线状态
	Get(id string) (Pipeline, error)
	//取消运行中的流水线
	Cancel(ctx context.Context, id string) error
	//执行异步流水线
	RunAsync(ctx context.Context, id string, config string, listener Listener) (Pipeline, error)
	//执行同步流水线
	RunSync(ctx context.Context, id string, config string, listener Listener) (Pipeline, error)
	//移除流水线记录
	Rm(id string)
	//runtime已经执行完成
	Done() chan struct{}
	//通知runtime
	Notify(data interface{}) error
	//反回runtime公共
	Ctx() context.Context
	//停止后台处理
	StopBackground()
}

Runtime 运行时