plugins

Documentation

Index

• Variables
• func IsRecoverableError(err error) bool
• func NewClient(ctx context.Context, logger zerolog.Logger, path string) *plugin.Client
• func NewPluginMap(source Source, destination Destination, spec Specifier) map[string]plugin.Plugin
• func NewRecoverableError(err error) error
• func Run(source Source, destination Destination, spec Specifier)
• type Config
• type Connector
• type Destination
  • func DispenseDestination(client *plugin.Client) (Destination, error)
• type DestinationPlugin
  • func (sp *DestinationPlugin) GRPCClient(_ context.Context, _ *plugin.GRPCBroker, c *grpc.ClientConn) (interface{}, error)
  • func (sp *DestinationPlugin) GRPCServer(broker *plugin.GRPCBroker, s *grpc.Server) error
• type Parameter
• type Source
  • func DispenseSource(client *plugin.Client) (Source, error)
• type SourcePlugin
  • func (sp *SourcePlugin) GRPCClient(_ context.Context, _ *plugin.GRPCBroker, c *grpc.ClientConn) (interface{}, error)
  • func (sp *SourcePlugin) GRPCServer(broker *plugin.GRPCBroker, s *grpc.Server) error
• type Specification
• type SpecificationPlugin
  • func (sp *SpecificationPlugin) GRPCClient(_ context.Context, _ *plugin.GRPCBroker, c *grpc.ClientConn) (interface{}, error)
  • func (sp *SpecificationPlugin) GRPCServer(broker *plugin.GRPCBroker, s *grpc.Server) error
• type Specifier
  • func DispenseSpecifier(client *plugin.Client) (Specifier, error)

Variables

var (
	ErrNotRunning     = cerrors.New("plugin is not running")
	ErrAlreadyRunning = cerrors.New("plugin is already running")
)

var (
	Handshake = plugin.HandshakeConfig{
		ProtocolVersion:  1,
		MagicCookieKey:   "CONDUIT_PLUGIN",
		MagicCookieValue: "CONDUIT",
	}
	PluginMap = NewPluginMap(nil, nil, nil)

	// ErrEndData is the return value when a Source has no new records.
	ErrEndData = NewRecoverableError(cerrors.New("ErrEndData"))
)

Functions

func IsRecoverableError

func IsRecoverableError(err error) bool

IsRecoverableError returns true if the error is recoverable, false otherwise.

func NewClient

func NewClient(ctx context.Context, logger zerolog.Logger, path string) *plugin.Client

NewClient creates a new plugin client. The provided context is used to kill the process (by calling os.Process.Kill) if the context becomes done before the plugin completes on its own. Path should point to the plugin executable.

func NewPluginMap

func NewPluginMap(source Source, destination Destination, spec Specifier) map[string]plugin.Plugin

NewPluginMap creates a new plugin map with the supplied source and destination. The return value can be used for field plugin.ServeConfig.Plugins when starting the plugin manually, although it is encouraged to use Run instead.

func NewRecoverableError

func NewRecoverableError(err error) error

NewRecoverableError returns an error that can be recovered from. If a plugin returns this error, then Conduit will try to execute the action again with a backoff retry strategy.

func Run

func Run(source Source, destination Destination, spec Specifier)

Run can be called in the main function of the plugin to start the plugin server with the correct configuration.

Types

type Config

type Config struct {
	Settings map[string]string
}

Config contains the configuration for the Connector.

type Connector

type Connector interface {
	Open(ctx context.Context, cfg Config) error
	Teardown() error
	Validate(cfg Config) error
}

Connector defines our Connector interface.

type Destination

type Destination interface {
	Connector

	// Write is writing records to a Destination.
	// The returned position is the position of the record written.
	// Important note: A Destination stops on first error returned from Write.
	Write(context.Context, record.Record) (record.Position, error)
}

Destination gets notified anytime anything comes out of the stream

func DispenseDestination

func DispenseDestination(client *plugin.Client) (Destination, error)

DispenseDestination will run the plugin executable if it's not already running and dispense the destination plugin.

type DestinationPlugin

type DestinationPlugin struct {
	plugin.NetRPCUnsupportedPlugin
	Impl Destination
}

DestinationPlugin represents a plugin that acts as a destination connector.

func (*DestinationPlugin) GRPCClient

func (sp *DestinationPlugin) GRPCClient(_ context.Context, _ *plugin.GRPCBroker, c *grpc.ClientConn) (interface{}, error)

func (*DestinationPlugin) GRPCServer

func (sp *DestinationPlugin) GRPCServer(broker *plugin.GRPCBroker, s *grpc.Server) error

type Parameter

type Parameter struct {
	// Default is the default value of the parameter, if any.
	Default string
	// Required is whether it must be provided in the Config or not.
	Required bool
	// Description holds a description of the field and how to configure it.
	Description string
}

Parameter is a helper struct for defining plugin Specifications.

type Source

type Source interface {
	Connector

	// Read reads data from a data source and returns the record for the
	// requested position.
	// Note, the Read method returns not the record at the position passed to
	// it, but the next one. If the position is nil, first record is returned.
	// When the position of the first record is passed, a second record is
	// returned etc.
	Read(context.Context, record.Position) (record.Record, error)

	// Ack signals to the source that the message has been successfully
	// processed and can be acknowledged. Sources that don't need to ack the
	// message should return nil.
	Ack(context.Context, record.Position) error
}

Source reads from a source and pipes it into our system.

func DispenseSource

func DispenseSource(client *plugin.Client) (Source, error)

DispenseSource will run the plugin executable if it's not already running and dispense the source plugin.

type SourcePlugin

type SourcePlugin struct {
	plugin.NetRPCUnsupportedPlugin
	Impl Source
}

SourcePlugin represents a plugin that acts as a source connector.

func (*SourcePlugin) GRPCClient

func (sp *SourcePlugin) GRPCClient(_ context.Context, _ *plugin.GRPCBroker, c *grpc.ClientConn) (interface{}, error)

func (*SourcePlugin) GRPCServer

func (sp *SourcePlugin) GRPCServer(broker *plugin.GRPCBroker, s *grpc.Server) error

type Specification

type Specification struct {
	// a brief description of the plugins in this package and what they do
	Summary string
	// Description is a more long form area appropriate for README-like text
	// that the author can provide for documentation about the specified
	// Parameters.
	Description string
	// Version string. Should be prepended with `v` like Go, e.g. `v1.54.3`
	Version string
	// Author declares the entity that created or maintains this plugin.
	Author string
	// DestinationParams and SourceParams is a map of named Parameters that describe
	// how to configure a the plugin's Destination or Source.
	DestinationParams map[string]Parameter
	SourceParams      map[string]Parameter
}

Specification is returned by a plugin when Specify is called. It contains information about the configuration parameters for plugins and allows them to describe their parameters.

type SpecificationPlugin

type SpecificationPlugin struct {
	plugin.NetRPCUnsupportedPlugin
	Impl Specifier
}

This SpecificationPlugin is where we enforce the fulfillment of the Specifier to the Specifications Service in the RPC protocol.

func (*SpecificationPlugin) GRPCClient

func (sp *SpecificationPlugin) GRPCClient(_ context.Context, _ *plugin.GRPCBroker, c *grpc.ClientConn) (interface{}, error)

func (*SpecificationPlugin) GRPCServer

func (sp *SpecificationPlugin) GRPCServer(broker *plugin.GRPCBroker, s *grpc.Server) error

type Specifier

type Specifier interface {
	Specify() (Specification, error)
}

Specifier allows a plugin to return its Specification to any caller.

func DispenseSpecifier

func DispenseSpecifier(client *plugin.Client) (Specifier, error)

DispenseSpecifier will run the plugin executable and return the Specifier if any from that plugin.