provisioning

Documentation

Index

• Variables
• type ConnectorService
• type PipelineService
• type PluginService
• type ProcessorService
• type Service
  • func NewService(db database.DB, logger log.CtxLogger, plService PipelineService, ...) *Service
  • func (s *Service) Export(ctx context.Context, pipelineID string) (config.Pipeline, error)
  • func (s *Service) Import(ctx context.Context, newConfig config.Pipeline) error
  • func (s *Service) Init(ctx context.Context) error

Variables

var (
	ErrDuplicatedPipelineID = cerrors.New("duplicated pipeline ID")
)

Types

type ConnectorService

type ConnectorService interface {
	Get(ctx context.Context, id string) (*connector.Instance, error)
	Create(ctx context.Context, id string, t connector.Type, plugin string, pipelineID string, c connector.Config, p connector.ProvisionType) (*connector.Instance, error)
	Update(ctx context.Context, id string, c connector.Config) (*connector.Instance, error)
	Delete(ctx context.Context, id string, dispenserFetcher connector.PluginDispenserFetcher) error

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

type PipelineService

type PipelineService interface {
	Get(ctx context.Context, id string) (*pipeline.Instance, error)
	List(ctx context.Context) map[string]*pipeline.Instance
	Create(ctx context.Context, id string, cfg pipeline.Config, p pipeline.ProvisionType) (*pipeline.Instance, error)
	Update(ctx context.Context, pipelineID string, cfg pipeline.Config) (*pipeline.Instance, error)
	Delete(ctx context.Context, pipelineID string) error
	UpdateDLQ(ctx context.Context, pipelineID string, cfg pipeline.DLQ) (*pipeline.Instance, error)

	Start(ctx context.Context, connFetcher pipeline.ConnectorFetcher, procFetcher pipeline.ProcessorFetcher, pluginFetcher pipeline.PluginDispenserFetcher, pipelineID string) error
	Stop(ctx context.Context, pipelineID string, force bool) error

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

type PluginService

type PluginService interface {
	NewDispenser(ctx log.CtxLogger, name string) (plugin.Dispenser, error)
}

type ProcessorService

type ProcessorService interface {
	Get(ctx context.Context, id string) (*processor.Instance, error)
	Create(ctx context.Context, id string, procType string, parent processor.Parent, cfg processor.Config, p processor.ProvisionType) (*processor.Instance, error)
	Update(ctx context.Context, id string, cfg processor.Config) (*processor.Instance, error)
	Delete(ctx context.Context, id string) error
}

type Service

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

func NewService

func NewService(
	db database.DB,
	logger log.CtxLogger,
	plService PipelineService,
	connService ConnectorService,
	procService ProcessorService,
	pluginService PluginService,
	pipelinesDir string,
) *Service

func (*Service) Export

func (s *Service) Export(ctx context.Context, pipelineID string) (config.Pipeline, error)

Export takes a pipeline ID and exports its configuration. It either returns the exported configuration or pipeline.ErrInstanceNotFound, any other error points towards a corrupted state.

func (*Service) Import

func (s *Service) Import(ctx context.Context, newConfig config.Pipeline) error

Import takes a pipeline config and imports it into Conduit.

func (*Service) Init

func (s *Service) Init(ctx context.Context) error

Init provision pipelines defined in pipelinePath directory. should initialize pipeline service before calling this function, and all pipelines should be stopped.