engine

Documentation

Index

• type EngineConfig
  • func DefaultConfig() *EngineConfig
• type ExecutionPhase
• type ExecutionPlan
• type ExecutionTask
• type Executor
• type MetricsCollector
• type Option
  • func WithCleanupInterval(interval time.Duration) Option
  • func WithDefaultContextTTL(ttl time.Duration) Option
  • func WithExecutionTimeout(timeout time.Duration) Option
  • func WithMaxConcurrentWorkflows(max int) Option
  • func WithMaxPoolSize(size int) Option
  • func WithMetricsEnabled(enabled bool) Option
  • func WithPoolSize(size int) Option
  • func WithTracingEnabled(enabled bool) Option
• type WorkflowEngine
  • func NewWorkflowEngine(opts ...Option) *WorkflowEngine
  • func (e *WorkflowEngine) Cleanup()
  • func (e *WorkflowEngine) DeregisterWorkflow(ctx context.Context, workflowID string, graceful bool) error
  • func (e *WorkflowEngine) ExecuteWorkflow(ctx context.Context, workflowID string, serialID string, params map[string]any) error
  • func (e *WorkflowEngine) GetMetrics(workflowID string) (*WorkflowMetrics, bool)
  • func (e *WorkflowEngine) GetNodeResult(workflowID, serialID, nodeID string) (*core.NodeResult, bool)
  • func (e *WorkflowEngine) GetPoolStats() map[string]any
  • func (e *WorkflowEngine) GetWorkflowStatus(workflowID string) (WorkflowStatus, bool)
  • func (e *WorkflowEngine) ListWorkflows() []string
  • func (e *WorkflowEngine) RegisterWorkflow(ctx context.Context, def *core.WorkflowDef) error
• type WorkflowMetrics
• type WorkflowNode
• type WorkflowStatus

Types

type EngineConfig

type EngineConfig struct {
	InitialPoolSize        int           // 初始池大小
	MaxPoolSize            int           // 最大池大小
	ExecutionTimeout       time.Duration // 执行超时
	CleanupInterval        time.Duration // 清理间隔
	DefaultContextTTL      time.Duration // 上下文TTL
	EnableMetrics          bool          // 启用指标
	EnableTracing          bool          // 启用跟踪
	MaxConcurrentWorkflows int           // 最大并发工作流
}

func DefaultConfig

func DefaultConfig() *EngineConfig

DefaultConfig 返回默认配置

type ExecutionPhase

type ExecutionPhase struct {
	Nodes []*WorkflowNode
}

ExecutionPhase 执行阶段

type ExecutionPlan

type ExecutionPlan struct {
	Phases []ExecutionPhase
}

ExecutionPlan 执行计划

type ExecutionTask

type ExecutionTask struct {
	NodeID     string
	Component  components.Component
	Context    *core.ExecutionContext
	Engine     *WorkflowEngine
	Executor   *Executor
	WorkflowID string
	SerialID   string
	Sw         *sync.WaitGroup
}

ExecutionTask 表示一个执行任务

type Executor

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

type MetricsCollector

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

MetricsCollector 指标收集器

type Option

type Option func(*WorkflowEngine)

Option 是工作流引擎的配置选项

func WithCleanupInterval

func WithCleanupInterval(interval time.Duration) Option

WithCleanupInterval 设置清理间隔

func WithDefaultContextTTL

func WithDefaultContextTTL(ttl time.Duration) Option

WithDefaultContextTTL 设置上下文TTL

func WithExecutionTimeout

func WithExecutionTimeout(timeout time.Duration) Option

WithExecutionTimeout 设置执行超时

func WithMaxConcurrentWorkflows

func WithMaxConcurrentWorkflows(max int) Option

WithMaxConcurrentWorkflows 设置最大并发工作流

func WithMaxPoolSize

func WithMaxPoolSize(size int) Option

WithMaxPoolSize 设置最大池大小

func WithMetricsEnabled

func WithMetricsEnabled(enabled bool) Option

WithMetricsEnabled 启用指标

func WithPoolSize

func WithPoolSize(size int) Option

WithPoolSize 设置池大小

func WithTracingEnabled

func WithTracingEnabled(enabled bool) Option

WithTracingEnabled 启用跟踪

type WorkflowEngine

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

WorkflowEngine 工作流引擎

func NewWorkflowEngine

func NewWorkflowEngine(opts ...Option) *WorkflowEngine

NewWorkflowEngine 创建新的工作流引擎

func (*WorkflowEngine) Cleanup

func (e *WorkflowEngine) Cleanup()

Cleanup 释放资源

func (*WorkflowEngine) DeregisterWorkflow

func (e *WorkflowEngine) DeregisterWorkflow(ctx context.Context, workflowID string, graceful bool) error

DeregisterWorkflow 注销工作流

func (*WorkflowEngine) ExecuteWorkflow

func (e *WorkflowEngine) ExecuteWorkflow(ctx context.Context, workflowID string, serialID string, params map[string]any) error

ExecuteWorkflow 执行工作流

func (*WorkflowEngine) GetMetrics

func (e *WorkflowEngine) GetMetrics(workflowID string) (*WorkflowMetrics, bool)

GetMetrics 获取指定工作流的指标

func (*WorkflowEngine) GetNodeResult

func (e *WorkflowEngine) GetNodeResult(workflowID, serialID, nodeID string) (*core.NodeResult, bool)

GetNodeResult 获取指定工作流节点的执行结果

func (*WorkflowEngine) GetPoolStats

func (e *WorkflowEngine) GetPoolStats() map[string]any

GetPoolStats 获取池统计信息

func (*WorkflowEngine) GetWorkflowStatus

func (e *WorkflowEngine) GetWorkflowStatus(workflowID string) (WorkflowStatus, bool)

GetWorkflowStatus 获取工作流状态

func (*WorkflowEngine) ListWorkflows

func (e *WorkflowEngine) ListWorkflows() []string

ListWorkflows 列出所有工作流

func (*WorkflowEngine) RegisterWorkflow

func (e *WorkflowEngine) RegisterWorkflow(ctx context.Context, def *core.WorkflowDef) error

RegisterWorkflow 注册工作流

type WorkflowMetrics

type WorkflowMetrics struct {
	ActiveExecutions     int64
	CompletedExecutions  int64
	FailedExecutions     int64
	AverageExecutionTime float64
	LastExecutionTime    time.Time
}

WorkflowMetrics 工作流指标

type WorkflowNode

type WorkflowNode struct {
	*core.NodeDefinition
	Component components.Component
}

WorkflowNode 工作流节点

type WorkflowStatus

type WorkflowStatus string

WorkflowStatus 表示工作流的状态

const (
	WorkflowStatusActive    WorkflowStatus = "active"
	WorkflowStatusDeploying WorkflowStatus = "deploying"
	WorkflowStatusShutdown  WorkflowStatus = "shutdown"
)