pipeline

Documentation

Index

• type ActivityState
  • func (as ActivityState) String() string
• type Analyzer
  • func NewAnalyzer(dRegistry registry.Registry) Analyzer
• type ComponentGraph
  • func NewComponentGraph() ComponentGraph
• type EtlStore
  • func NewEtlStore() EtlStore
• type Manager
  • func NewManager(ctx context.Context, analyzer Analyzer, cRegistry registry.Registry, ...) Manager
• type Pipeline
  • func NewPipeline(cfg *core.PipelineConfig, pUUID core.PUUID, comps []component.Component) (Pipeline, error)

Types

type ActivityState

type ActivityState uint8

const (
	INACTIVE ActivityState = iota
	ACTIVE
	CRASHED
	TERMINATED
)

func (ActivityState) String

func (as ActivityState) String() string

type Analyzer

type Analyzer interface {
	Mergable(p1 Pipeline, p2 Pipeline) bool
}

Analyzer ... Interface for analyzing pipelines

func NewAnalyzer

func NewAnalyzer(dRegistry registry.Registry) Analyzer

NewAnalyzer ... Initializer

type ComponentGraph

type ComponentGraph interface {
	ComponentExists(cUIID core.CUUID) bool
	GetComponent(cUIID core.CUUID) (component.Component, error)
	AddEdge(cUUID1, cUUID2 core.CUUID) error
	AddComponent(cUIID core.CUUID, comp component.Component) error
	AddComponents(cSlice []component.Component) error

	Edges() map[core.CUUID][]core.CUUID // Useful for testing

	// TODO(#23): Manager DAG Component Removal Support
	RemoveEdge(_, _ core.CUUID) error
	RemoveComponent(_ core.CUUID) error
}

ComponentGraph ...

func NewComponentGraph

func NewComponentGraph() ComponentGraph

NewComponentGraph ... Initializer

type EtlStore

type EtlStore interface {
	AddComponentLink(cID core.CUUID, pID core.PUUID)
	AddPipeline(id core.PUUID, pl Pipeline)
	ActiveCount() int
	GetAllPipelines() []Pipeline
	GetExistingPipelinesByPID(pPID core.PipelinePID) []core.PUUID
	GetPUUIDs(cID core.CUUID) ([]core.PUUID, error)
	GetPipelineFromPUUID(pUUID core.PUUID) (Pipeline, error)
}

TODO(#48): Pipeline Analysis Functionality EtlStore ... Interface used to define all etl storage based functions

func NewEtlStore

func NewEtlStore() EtlStore

NewEtlStore ... Initializer

type Manager

type Manager interface {
	InferComponent(cc *core.ClientConfig, cUUID core.CUUID, pUUID core.PUUID,
		register *core.DataRegister) (component.Component, error)
	GetStateKey(rt core.RegisterType) (*core.StateKey, bool, error)
	CreateDataPipeline(cfg *core.PipelineConfig) (core.PUUID, bool, error)
	RunPipeline(pID core.PUUID) error
	ActiveCount() int

	core.Subsystem
}

Manager ... ETL manager interface

func NewManager

func NewManager(ctx context.Context, analyzer Analyzer, cRegistry registry.Registry,
	store EtlStore, dag ComponentGraph,
	eo chan core.HeuristicInput) Manager

NewManager ... Initializer

type Pipeline

type Pipeline interface {
	Config() *core.PipelineConfig
	Components() []component.Component
	UUID() core.PUUID
	State() ActivityState

	Close() error
	Run(wg *sync.WaitGroup)
	AddEngineRelay(engineChan chan core.HeuristicInput) error
}

Pipeline ... Pipeline interface

func NewPipeline

func NewPipeline(cfg *core.PipelineConfig, pUUID core.PUUID, comps []component.Component) (Pipeline, error)

NewPipeline ... Initializer