igeist

Documentation

Index

• type Executor
• type Registry
• type Spec
• type Stream
• type StreamBuilder
• type StreamEntityFactory
• type StreamRegistry
• type Transformer

Types

type Executor

type Executor interface {
	Stream() Stream
	StreamId() string
	Metrics() entity.Metrics
	Run(ctx context.Context, wg *sync.WaitGroup)

	// ProcessEvent returns the processed/persisted event's event/resource ID, together with error, retryable.
	// The implementation varies across different sink types, but the returned resource ID it will be the key
	// used for key lookups of the event (e.g. row-key for BigTable), and will be used as the resource ID when
	// using API as Source.
	ProcessEvent(ctx context.Context, events []entity.Event) entity.EventProcessingResult
	Shutdown()
}

Executor interface required for Stream ETL Executors

type Registry

type Registry interface {
	Fetch(ctx context.Context) error
	Put(ctx context.Context, id string, spec Spec) error
	Get(ctx context.Context, id string) (Spec, error)
	GetAll(ctx context.Context) (map[string]Spec, error)
	Delete(ctx context.Context, id string) error
	Exists(id string) bool
	ExistsSameVersion(specData []byte) (bool, error)
	Validate(spec []byte) (Spec, error)
}

type Spec

type Spec interface {
	Id() string       // Returns Stream Spec ID
	IsDisabled() bool // True if spec is disabled and should not be run by Executor
	Validate() error  // Returns an error if Spec is not according to spec
	JSON() []byte     // Returns spec data as JSON
}

type Stream

type Stream interface {
	Spec() Spec
	Instance() string
	Extractor() entity.Extractor
	Transformer() Transformer
	Loader() entity.Loader
	Publish(ctx context.Context, event []byte) (string, error)
	ExtractFromSink(
		ctx context.Context,
		query entity.ExtractorQuery,
		result *[]*entity.Transformed) (error, bool)
}

type StreamBuilder

type StreamBuilder interface {
	Build(ctx context.Context, spec Spec) (Stream, error)
}

type StreamEntityFactory

type StreamEntityFactory interface {
	SetAdminLoader(loader entity.Loader)
	CreateExtractor(ctx context.Context, spec Spec, instanceId string) (entity.Extractor, error)
	CreateSinkExtractor(ctx context.Context, spec Spec, instanceId string) (entity.Extractor, error)
	CreateTransformer(ctx context.Context, spec Spec) (Transformer, error)
	CreateLoader(ctx context.Context, spec Spec, instanceId string) (entity.Loader, error)
}

type StreamRegistry

type StreamRegistry interface {
	Registry
	Executor
	SetAdminStream(stream Stream)
}

A StreamRegistry is a Registry also managing a stream as an Executor, receiving Specs to be added to registry in an arbitrary ETL stream, e.g. via Kafka or REST API into a db. It is also required to publish admin events when registry is modified (e.g. new stream registrations, etc).

type Transformer

type Transformer interface {
	// Based on input event data, and the transformation rules in the Spec, the Transform() function returns
	// a key-value map where keys are "id" fields from Transform spec, and values are the transformation results.
	// The output from Transform() can contain multiple new events, in case of applied
	// event-split transformations.
	// If Transform() succeeded, but the transformation resulted in no output, e.g. for non-applicable incoming events,
	// the return values are nil, nil (i.e., not regarded as an error).
	Transform(
		ctx context.Context,
		event []byte,
		retryable *bool) ([]*entity.Transformed, error)
}

Transformer interface required for transformer implementations