plugins

Documentation

Index

• Variables
• func BigQueryTableFQNToURN(fqn string) (string, error)
• func BigQueryURN(projectID, datasetID, tableID string) string
• func CaraMLStoreURN(scope, project, featureTable string) string
• func GetLog() log.Logger
• func KafkaServersToScope(servers string) string
• func KafkaURN(bootstrapServers, topic string) string
• func NewRetryError(err error) error
• func SetLog(l log.Logger)
• type BaseExtractor
  • func NewBaseExtractor(info Info, configRef interface{}) BaseExtractor
  • func (p *BaseExtractor) Init(ctx context.Context, config Config) error
  • func (p *BaseExtractor) Validate(config Config) error
• type BasePlugin
  • func NewBasePlugin(info Info, configRef interface{}) BasePlugin
  • func (p *BasePlugin) Info() Info
  • func (p *BasePlugin) Init(ctx context.Context, config Config) error
  • func (p *BasePlugin) Validate(config Config) error
• type Config
• type ConfigError
• type Emit
• type Extractor
• type Info
• type InvalidConfigError
  • func (err InvalidConfigError) Error() string
  • func (err InvalidConfigError) HasError() bool
• type NotFoundError
  • func (err NotFoundError) Error() string
• type Plugin
• type PluginType
• type Processor
• type RetryError
  • func (e RetryError) Error() string
  • func (e RetryError) Unwrap() error
• type Syncer

Variables

var ErrEmptyURNScope = errors.New("urn scope is required to generate unique urn")

Functions

func BigQueryTableFQNToURN

func BigQueryTableFQNToURN(fqn string) (string, error)

func BigQueryURN

func BigQueryURN(projectID, datasetID, tableID string) string

func CaraMLStoreURN

func CaraMLStoreURN(scope, project, featureTable string) string

func GetLog

func GetLog() log.Logger

GetLog returns the logger

func KafkaServersToScope

func KafkaServersToScope(servers string) string

func KafkaURN

func KafkaURN(bootstrapServers, topic string) string

func NewRetryError

func NewRetryError(err error) error

func SetLog

func SetLog(l log.Logger)

SetLog sets the logger

Types

type BaseExtractor

type BaseExtractor struct {
	ScopeNotRequired bool
	BasePlugin
}

func NewBaseExtractor

func NewBaseExtractor(info Info, configRef interface{}) BaseExtractor

func (*BaseExtractor) Init

func (p *BaseExtractor) Init(ctx context.Context, config Config) error

Init will be called once before running the plugin. This is where you want to initiate any client or test any connection to external service.

func (*BaseExtractor) Validate

func (p *BaseExtractor) Validate(config Config) error

Validate checks if the given options is valid for the plugin.

type BasePlugin

type BasePlugin struct {
	UrnScope  string
	RawConfig map[string]interface{}
	// contains filtered or unexported fields
}

func NewBasePlugin

func NewBasePlugin(info Info, configRef interface{}) BasePlugin

func (*BasePlugin) Info

func (p *BasePlugin) Info() Info

Info returns plugin's information.

func (*BasePlugin) Init

func (p *BasePlugin) Init(ctx context.Context, config Config) error

Init will be called once before running the plugin. This is where you want to initiate any client or test any connection to external service.

func (*BasePlugin) Validate

func (p *BasePlugin) Validate(config Config) error

Validate checks if the given options is valid for the plugin.

type Config

type Config struct {
	URNScope  string
	RawConfig map[string]interface{}
}

type ConfigError

type ConfigError struct {
	Key     string
	Message string
}

ConfigError contains fields to check error

type Emit

type Emit func(models.Record)

type Extractor

type Extractor interface {
	Plugin
	Extract(ctx context.Context, emit Emit) (err error)
}

Extractor is a plugin that extracts data from a source.

type Info

type Info struct {
	Description  string   `yaml:"description"`
	SampleConfig string   `yaml:"sample_config"`
	Tags         []string `yaml:"tags"`
	Summary      string   `yaml:"summary"`
}

Info represents the meta.yaml file of a plugin.

type InvalidConfigError

type InvalidConfigError struct {
	Type       PluginType
	PluginName string
	Errors     []ConfigError
}

InvalidConfigError is returned when a plugin's configuration is invalid.

func (InvalidConfigError) Error

func (err InvalidConfigError) Error() string

func (InvalidConfigError) HasError

func (err InvalidConfigError) HasError() bool

type NotFoundError

type NotFoundError struct {
	Type PluginType
	Name string
}

NotFoundError contains fields required to checks for a missing plugin.

func (NotFoundError) Error

func (err NotFoundError) Error() string

type Plugin

type Plugin interface {
	// Info returns plugin's information.
	Info() Info

	// Validate checks if the given options is valid for the plugin.
	Validate(config Config) error

	// Init will be called once before running the plugin.
	// This is where you want to initiate any client or test any connection to external service.
	Init(ctx context.Context, config Config) error
}

type PluginType

type PluginType string

PluginType is the type of plugin.

const (
	PluginTypeExtractor PluginType = "extractor"
	PluginTypeProcessor PluginType = "processor"
	PluginTypeSink      PluginType = "sink"
)

PluginType names

type Processor

type Processor interface {
	Plugin
	Process(ctx context.Context, src models.Record) (dst models.Record, err error)
}

Processor are the functions that are executed on the extracted data.

type RetryError

type RetryError struct {
	Err error
}

RetryError is an error signalling that retry is needed for the operation.

func (RetryError) Error

func (e RetryError) Error() string

func (RetryError) Unwrap

func (e RetryError) Unwrap() error

type Syncer

type Syncer interface {
	Plugin
	Sink(ctx context.Context, batch []models.Record) (err error)

	// Close will be called once after everything is done
	Close() error
}

Syncer is a plugin that can be used to sync data from one source to another.