orchestrator

package
v0.2.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: May 12, 2022 License: Apache-2.0 Imports: 10 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrInvalidProcessorParentType     = cerrors.New("invalid processor parent type")
	ErrPipelineHasProcessorsAttached  = cerrors.New("pipeline has processors attached")
	ErrPipelineHasConnectorsAttached  = cerrors.New("pipeline has connectors attached")
	ErrConnectorHasProcessorsAttached = cerrors.New("connector has processors attached")
)

Functions

This section is empty.

Types

type ConnectorOrchestrator 

type ConnectorOrchestrator base

func (*ConnectorOrchestrator) Create 

func (c *ConnectorOrchestrator) Create(
	ctx context.Context,
	t connector.Type,
	config connector.Config,
) (connector.Connector, error)

func (*ConnectorOrchestrator) Delete 

func (c *ConnectorOrchestrator) Delete(ctx context.Context, id string) error

func (*ConnectorOrchestrator) Get 

func (c *ConnectorOrchestrator) Get(ctx context.Context, id string) (connector.Connector, error)

func (*ConnectorOrchestrator) List 

func (c *ConnectorOrchestrator) List(ctx context.Context) map[string]connector.Connector

func (*ConnectorOrchestrator) Update 

func (c *ConnectorOrchestrator) Update(ctx context.Context, id string, config connector.Config) (connector.Connector, error)

func (*ConnectorOrchestrator) Validate added in v0.2.1 

func (c *ConnectorOrchestrator) Validate(
	ctx context.Context,
	t connector.Type,
	config connector.Config,
) error

type ConnectorService 

type ConnectorService interface {
	List(ctx context.Context) map[string]connector.Connector
	Get(ctx context.Context, id string) (connector.Connector, error)
	Create(ctx context.Context, id string, t connector.Type, c connector.Config) (connector.Connector, error)
	Delete(ctx context.Context, id string) error
	Update(ctx context.Context, id string, c connector.Config) (connector.Connector, error)

	AddProcessor(ctx context.Context, connectorID string, processorID string) (connector.Connector, error)
	RemoveProcessor(ctx context.Context, connectorID string, processorID string) (connector.Connector, error)
}

type Orchestrator 

type Orchestrator struct {
	Processors *ProcessorOrchestrator
	Pipelines  *PipelineOrchestrator
	Connectors *ConnectorOrchestrator
	Plugins    *PluginOrchestrator
}

func NewOrchestrator 

func NewOrchestrator(
	db database.DB,
	logger log.CtxLogger,
	pipelines PipelineService,
	connectors ConnectorService,
	processors ProcessorService,
	plugins PluginService,
) *Orchestrator

type PipelineOrchestrator 

type PipelineOrchestrator base

func (*PipelineOrchestrator) Create 

func (s *PipelineOrchestrator) Create(ctx context.Context, cfg pipeline.Config) (*pipeline.Instance, error)

func (*PipelineOrchestrator) Delete 

func (s *PipelineOrchestrator) Delete(ctx context.Context, id string) error

func (*PipelineOrchestrator) Get 

func (s *PipelineOrchestrator) Get(ctx context.Context, id string) (*pipeline.Instance, error)

func (*PipelineOrchestrator) List 

func (s *PipelineOrchestrator) List(ctx context.Context) map[string]*pipeline.Instance

func (*PipelineOrchestrator) Start 

func (s *PipelineOrchestrator) Start(ctx context.Context, id string) error

func (*PipelineOrchestrator) Stop 

func (s *PipelineOrchestrator) Stop(ctx context.Context, id string) error

func (*PipelineOrchestrator) Update 

func (s *PipelineOrchestrator) Update(ctx context.Context, id string, cfg pipeline.Config) (*pipeline.Instance, error)

type PipelineService 

type PipelineService interface {
	Start(ctx context.Context, connFetcher pipeline.ConnectorFetcher, procFetcher pipeline.ProcessorFetcher, pipeline *pipeline.Instance) error
	// Stop initiates a graceful shutdown of the given pipeline and sets its status to the provided status.
	// The method does not wait for the pipeline (and its nodes) to actually stop,
	// because there still might be some in-flight messages.
	Stop(ctx context.Context, pipeline *pipeline.Instance) error

	List(ctx context.Context) map[string]*pipeline.Instance
	Get(ctx context.Context, id string) (*pipeline.Instance, error)
	Create(ctx context.Context, id string, cfg pipeline.Config) (*pipeline.Instance, error)
	Update(ctx context.Context, pl *pipeline.Instance, cfg pipeline.Config) (*pipeline.Instance, error)
	Delete(ctx context.Context, pl *pipeline.Instance) error

	AddConnector(ctx context.Context, pl *pipeline.Instance, connectorID string) (*pipeline.Instance, error)
	RemoveConnector(ctx context.Context, pl *pipeline.Instance, connectorID string) (*pipeline.Instance, error)
	AddProcessor(ctx context.Context, pl *pipeline.Instance, processorID string) (*pipeline.Instance, error)
	RemoveProcessor(ctx context.Context, pl *pipeline.Instance, processorID string) (*pipeline.Instance, error)
}

type PluginOrchestrator added in v0.2.0 

type PluginOrchestrator base

func (*PluginOrchestrator) List added in v0.2.0 

func (ps *PluginOrchestrator) List(ctx context.Context) (map[string]plugin.Specification, error)

type PluginService added in v0.2.0 

type PluginService interface {
	List(ctx context.Context) (map[string]plugin.Specification, error)
	NewDispenser(logger log.CtxLogger, name string) (plugin.Dispenser, error)
	ValidateSourceConfig(ctx context.Context, d plugin.Dispenser, settings map[string]string) error
	ValidateDestinationConfig(ctx context.Context, d plugin.Dispenser, settings map[string]string) error
}

type ProcessorOrchestrator 

type ProcessorOrchestrator base

func (*ProcessorOrchestrator) Create 

func (p *ProcessorOrchestrator) Create(
	ctx context.Context,
	name string,
	t processor.Type,
	parent processor.Parent,
	cfg processor.Config,
) (*processor.Instance, error)

func (*ProcessorOrchestrator) Delete 

func (p *ProcessorOrchestrator) Delete(ctx context.Context, id string) error

func (*ProcessorOrchestrator) Get 

func (p *ProcessorOrchestrator) Get(ctx context.Context, id string) (*processor.Instance, error)

func (*ProcessorOrchestrator) List 

func (p *ProcessorOrchestrator) List(ctx context.Context) map[string]*processor.Instance

func (*ProcessorOrchestrator) Update 

func (p *ProcessorOrchestrator) Update(ctx context.Context, id string, cfg processor.Config) (*processor.Instance, error)

type ProcessorService 

type ProcessorService interface {
	List(ctx context.Context) map[string]*processor.Instance
	Get(ctx context.Context, id string) (*processor.Instance, error)
	Create(ctx context.Context, id string, name string, t processor.Type, parent processor.Parent, cfg processor.Config) (*processor.Instance, error)
	Update(ctx context.Context, id string, cfg processor.Config) (*processor.Instance, error)
	Delete(ctx context.Context, id string) error
}

Source Files

View all Source files

Directories

Path Synopsis
Package mock is a generated GoMock package.
 Package mock is a generated GoMock package.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.