provisioning

Documentation

Index

• Variables
• type ConnectorPluginService
• type ConnectorService
• type PipelineService
• type ProcessorService
• type Service
  • func NewService(db database.DB, logger log.CtxLogger, plService PipelineService, ...) *Service
  • func (s *Service) Delete(ctx context.Context, id string) error
  • 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")
	ErrNotProvisionedByConfig = cerrors.New("entity was not provisioned by a config file and therefore can't be mutated by the provisioning service")
)

Types

type ConnectorPluginService

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

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, procService pipeline.ProcessorService, 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 ProcessorService

type ProcessorService interface {
	Get(ctx context.Context, id string) (*processor.Instance, error)
	Create(
		ctx context.Context,
		id string,
		plugin string,
		parent processor.Parent,
		cfg processor.Config,
		p processor.ProvisionType,
		condition string,
	) (*processor.Instance, error)
	MakeRunnableProcessor(ctx context.Context, i *processor.Instance) (*processor.RunnableProcessor, 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,
	connPluginService ConnectorPluginService,
	pipelinesDir string,
) *Service

func (*Service) Delete

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

Delete exposes a way to delete pipelines provisioned using the provisioning 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.