orchestrator

type Event ¶

type Event struct {
	// Location could be a table name, a topic, an event from some store,
	// or anything really- it is up to both the Input and the Process to
	// agree on what this means
	Location  string    `json:"location"`
	Operation Operation `json:"operation"`
	ID        string    `json:"id"`

	// Trigger is the name or ID of the Input which triggers this
	// process, which can be useful for routing/ flow control in
	// triggers
	Trigger string `json:"trigger"`
}

Event represents basic metadata that each Input provides

func (e Event) JSON() (string, error)

type Input ¶

type Input interface {
	// Handle inputs from this input source, creating Events and
	// streaming down the Event channel
	Handle(context.Context, chan Event) error
	ID() string
}

Input is a simple interface, and exposes a long running process called Handle which is expected to stream Events.

It is the job of the Orchestrator to understand which channel is assigned to which input and to route messages accordingly

type InputConfig ¶

type InputConfig struct {
	Name             string      `toml:"name"`
	Type             string      `toml:"type"`
	ConnectionString string      `toml:"connection_string"`
	Operations       []Operation `toml:"operation"`
}

InputConfig contains the necessary values for coniguring an Input, such as how to connect to the input source, and the operations the input supports

func (ic InputConfig) ID() string

type NewInputFunc ¶ added in v0.1.1

type NewInputFunc func(InputConfig) (Input, error)

NewInputFunc is the suggested function that an Input should be instantiated with and, as such, can be used when creating a registry of Inputs an orchestrator supports when creating Inputs dynamically say from a config file, or from an API.

For instance:

var inputs = map[string]orchestrator.NewInputFunc{
   "postgres": orchestrator.NewPostgresInput,
   "webhook": webhooks.NewWebhookInput,
}
func createInput(cfg orchestrator.InputConfig) (orchestrator.Input, error) {
   return inputs[cfg.Type](cfg)
}

type NewProcessFunc ¶ added in v0.1.1

type NewProcessFunc func(ProcessConfig) (Process, error)

NewProcessFunc is the suggested function that an Process should be instantiated with and, as such, can be used when creating a registry of Processs an orchestrator supports when creating Processs dynamically say from a config file, or from an API.

For instance:

var processs = map[string]orchestrator.NewProcessFunc{
   "docker": orchestrator.NewContainerProcess,
   "webhook": webhooks.NewWebhookProcess,
}
func createProcess(cfg orchestrator.ProcessConfig) (orchestrator.Process, error) {
   return processs[cfg.Type](cfg)
}

type Operation ¶

type Operation uint8

Operation represents one of the basic CRUD operations on a piece of data and can be used in Inputs to do clever things around ignoring certain events

const (
	OperationUnknown Operation = iota
	OperationCreate
	OperationRead
	OperationUpdate
	OperationDelete
)

func (o Operation) MarshalJSON() ([]byte, error)

func (o Operation) MarshalText() (b []byte, err error)

func (o Operation) String() string

func (o *Operation) UnmarshalJSON(b []byte) (err error)

func (o *Operation) UnmarshalText(b []byte) error

type Orchestrator ¶

type Orchestrator struct {
	*dag.DAG

	ErrorChan chan error
	// contains filtered or unexported fields
}

Orchestrator is the workhorse of this package. It:

1. Supervises inputs
2. Manages the lifecycle of processes, which run on events
3. Syncs events from inputs across multiple processes in a DAG

Multiple Orchestrators _can_ be run, like in a cluster, but out of the box doesn't contain any logic to synchronise inputs and/or processes that wont cluster natively (such as the postgres sample input)

func New() *Orchestrator

func (d *Orchestrator) AddInput(ctx context.Context, i Input) (err error)

func (d Orchestrator) AddLink(input Input, process Process) (err error)

func (d Orchestrator) AddProcess(p Process) error

type Process ¶

type Process interface {
	Run(context.Context, Event) (ProcessStatus, error)
	ID() string
}

Process is an interface which processes must implement

The inteface is pretty simple: given a

type ProcessConfig ¶

type ProcessConfig struct {
	Name             string            `toml:"name"`
	Type             string            `toml:"type"`
	ExecutionContext map[string]string `toml:"execution_context"`
}

ProcessConfig contains configuration options for processes, including an unkeyed map[string]string for arbitrary values

func (pc ProcessConfig) ID() string

type ProcessExitStatus ¶

type ProcessExitStatus uint8

ProcessExitStatus represents the final status of a Process

const (
	ProcessUnknown ProcessExitStatus = iota
	ProcessUnstarted
	ProcessSuccess
	ProcessFail
)

type ProcessInterfaceConversionError ¶

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

ProcessInterfaceConversionError returns when trying to load a process from our internal process store returns completely unexpected data

This error represents a huge failure somewhere and should cause a stop-the-world event

func NewTestProcessInterfaceConversionError(input, process string, iface any) ProcessInterfaceConversionError

func (e ProcessInterfaceConversionError) Error() string

type ProcessStatus ¶

type ProcessStatus struct {
	Name   string
	Logs   []string
	Status ProcessExitStatus
}

ProcessStatus contains various bits and pieces a process might return, such as logs and statuscodes and so on

type UnknownProcessError ¶

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

UnknownProcessError returns when an input tries to trigger a process whch doesn't exist

func NewTestUnknownProcessError(input, process string) UnknownProcessError

func (e UnknownProcessError) Error() string