The highest tagged major version is
README
¶
Machine is a library for creating data workflows. These workflows can be either very concise or quite complex, even allowing for cycles for flows that need retry or self healing mechanisms. It supports opentelemetry spans and metrics out of the box and supports building dynamic pipelines using native go plugins and hashicorp or yaegi based plugins by using the providers here.
Installation
Add the primary library to your project
go get -u github.com/whitaker-io/machine
Components is a repository of different vertex implementations
Data is a library for getting and setting values in a map[string]interface{}
Documentation
Example
Redis Subscription with basic receive -> process -> send Stream
// logStore allows for running a cluster and handles communication
var logStore machine.LogStore
// pool is a redigo Pool for a redis cluster to read the stream from
// see also the Google Pub/Sub, Kafka, and SQS implementations
var pool *redigo.Pool
// publisher is a machine.Publisher used for sending data outside of the Stream
var publisher machine.Publisher
redisStream := redis.New(pool, logger)
// NewPipe creates a pipe in which you can run multiple streams
// the id is the instance identifier for the cluster
p := NewPipe(uuid.New().String(), nil, logStore, fiber.Config{
ReadTimeout: time.Second,
WriteTimeout: time.Second,
BodyLimit: 4 * 1024 * 1024,
DisableKeepalive: true,
})
// StreamSubscription takes an instance of machine.Subscription
// and a time interval in which to read
// the id here needs to be the same for all the nodes for the clustering to work
builder := p.StreamSubscription("unique_stream_id", redisStream, 5*time.Millisecond,
&Option{FIFO: boolP(false)},
&Option{Injectable: boolP(true)},
&Option{Metrics: boolP(true)},
&Option{Span: boolP(false)},
&Option{BufferSize: intP(0)},
).Builder()
builder.Map("unique_id2",
func(m Data) error {
var err error
// ...do some processing
return err
},
).
Publish("unique_id3", publisher)
// Run requires a context, the port to run the fiber.App,
// and the timeout for graceful shutdown
if err := p.Run(context.Background(), ":5000", 10 * time.Second); err != nil {
// Run will return an error in the case that
// one of the paths is not terminated (i.e. missing a Transmit)
panic(err)
}
🤝 Contributing
Contributions, issues and feature requests are welcome.
Feel free to check issues page if you want to contribute.
Check the contributing guide.
Author
👤 Jonathan Whitaker
- Twitter: @io_whitaker
- Github: @jonathan-whitaker
Show your support
Please ⭐️ this repository if this project helped you!
License
Machine is provided under the MIT License.
The MIT License (MIT)
Copyright (c) 2020 Jonathan Whitaker
Documentation
¶
Index ¶
- func RegisterPluginProvider(name string, p PluginProvider)
- type Applicative
- type Builder
- type Cluster
- type Error
- type Fold
- type Fork
- type ForkRule
- type HTTPStream
- type InjectionCallback
- type Log
- type Option
- type Packet
- type PluginDefinition
- type PluginProvider
- type Publisher
- type Retriever
- type Stream
- type StreamSerialization
- type Subscription
- type VertexSerialization
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func RegisterPluginProvider ¶ added in v0.11.0
func RegisterPluginProvider(name string, p PluginProvider)
RegisterPluginProvider function for registering a PluginProvider to be used for loading VertexProviders
Types ¶
type Applicative ¶ added in v0.3.0
Applicative is a function that is applied on an individual basis for each Packet in the payload. The data may be modified or checked for correctness. Any resulting error is combined with current errors in the wrapping Packet.
type Builder ¶
type Builder interface {
Map(id string, a Applicative) Builder
FoldLeft(id string, f Fold) Builder
FoldRight(id string, f Fold) Builder
Fork(id string, f Fork) (Builder, Builder)
Loop(id string, x Fork) (loop, out Builder)
Publish(id string, s Publisher)
}
Builder is the interface provided for creating a data processing stream.
type Cluster ¶ added in v0.16.0
type Cluster interface {
Join(id string, callback InjectionCallback) error
Write(logs ...*Log)
Leave(id string) error
}
Cluster is an interface for allowing a distributed cluster of workers It requires 3 methods Join, Write, and Leave.
Join is called during the run method of the Stream and it is used for accouncing membership to the cluster with an identifier and a callback for injection
Injection is how work can be restarted in the system and is the responsibility of the implementation to decide when and how it is reinitiated.
Write is called at the beginning of every vertex and provides the current payload about to be run. The implementation is considered the owner of that payload and may modify the data in special known circumstances if need be.
Leave is called during a graceful termination and any errors are logged.
type Error ¶ added in v0.16.0
type Error struct {
Err error
StreamID string
VertexID string
VertexType string
Packets []*Packet
Time time.Time
}
Error type for wrapping errors coming from the Stream
type Fold ¶ added in v0.5.0
Fold is a function used to combine a payload into a single Packet. It may be used with either a Fold Left or Fold Right operation, which starts at the corresponding side and moves through the payload. The returned instance of data.Data is used as the aggregate in the subsequent call.
type Fork ¶ added in v0.5.0
Fork is a function for splitting the payload into 2 separate paths. Default Forks for duplication and error checking are provided by ForkDuplicate and ForkError respectively.
var ( // ForkDuplicate is a Fork that creates a deep copy of the // payload and sends it down both branches. ForkDuplicate Fork = func(payload []*Packet) (a, b []*Packet) { payload2 := []*Packet{} buf := &bytes.Buffer{} enc, dec := gob.NewEncoder(buf), gob.NewDecoder(buf) _ = enc.Encode(payload) _ = dec.Decode(&payload2) for i, packet := range payload { payload2[i].span = packet.span payload2[i].spanCtx = packet.spanCtx } return payload, payload2 } // ForkError is a Fork that splits machine.Packets based on // the presence of an error. Errors are sent down the // right side path. ForkError Fork = func(payload []*Packet) (s, f []*Packet) { s = []*Packet{} f = []*Packet{} for _, packet := range payload { if len(packet.Errors) > 0 { f = append(f, packet) } else { s = append(s, packet) } } return s, f } )
type ForkRule ¶ added in v0.5.0
ForkRule is a function that can be converted to a Fork via the Handler method allowing for Forking based on the contents of the data.
type HTTPStream ¶ added in v0.16.0
type HTTPStream interface {
Stream
Handler() fiber.Handler
}
HTTPStream is a Stream that also provides a fiber.Handler for receiving data
func LoadHTTP ¶ added in v0.16.4
func LoadHTTP(serialization *StreamSerialization) (HTTPStream, error)
LoadHTTP method loads an HTTPStream based on the StreamSerialization
func NewHTTPStream ¶ added in v0.16.0
func NewHTTPStream(id string, opts ...*Option) HTTPStream
NewHTTPStream a method that creates a Stream which takes in data through a fiber.Handler
func NewWebsocketStream ¶ added in v0.16.0
func NewWebsocketStream(id string, opts ...*Option) HTTPStream
NewWebsocketStream a method that creates a Stream which takes in data through a fiber.Handler that runs a websocket
type InjectionCallback ¶ added in v0.7.0
type InjectionCallback func(logs ...*Log)
InjectionCallback is a function provided to the LogStore Join method so that the cluster may restart work that has been dropped by one of the workers. Injections will only be processed for vertices that have the Injectable option set to true, which is the default.
type Log ¶ added in v0.7.0
type Log struct {
StreamID string `json:"stream_id"`
VertexID string `json:"vertex_id"`
VertexType string `json:"vertex_type"`
State string `json:"state"`
Packet *Packet `json:"packet"`
When time.Time `json:"when"`
}
Log type for holding the data that is recorded from the streams and sent to the LogStore instance
type Option ¶ added in v0.2.0
type Option struct {
// DeepCopy uses encoding/gob to create a deep copy of the payload
// before the processing to ensure concurrent map exceptions cannot
// happen. Is fairly resource intensive so use with caution.
// Default: false
DeepCopy *bool
// FIFO controls the processing order of the payloads
// If set to true the system will wait for one payload
// to be processed before starting the next.
// Default: false
FIFO *bool
// Injectable controls whether the vertex accepts injection calls
// if set to false the data will be logged and processing will not
// take place.
// Default: true
Injectable *bool
// BufferSize sets the buffer size on the edge channels between the
// vertices, this setting can be useful when processing large amounts
// of data with FIFO turned on.
// Default: 0
BufferSize *int
// Span controls whether opentelemetry spans are created for tracing
// Packets processed by the system.
// Default: true
Span *bool
// Metrics controls whether opentelemetry metrics are recorded for
// Packets processed by the system.
// Default: true
Metrics *bool
}
Option type for holding machine settings.
type Packet ¶
type Packet struct {
ID string `json:"id"`
Data data.Data `json:"data"`
Errors map[string]error `json:"errors"`
// contains filtered or unexported fields
}
Packet type that holds information traveling through the machine.
type PluginDefinition ¶ added in v0.11.0
type PluginDefinition struct {
// Type is the name of the PluginProvider to use.
Type string `json:"type" mapstructure:"type"`
// Payload is the location, script, etc provided to load the plugin.
// Depends on the PluginProvider.
Payload string `json:"payload" mapstructure:"payload"`
// Symbol is the name of the symbol to be loaded from the plugin.
Symbol string `json:"symbol" mapstructure:"symbol"`
// Attributes are a map[string]interface{} of properties to be used with the PluginProvider.
Attributes map[string]interface{} `json:"attributes" mapstructure:"attributes"`
}
PluginDefinition type for declaring the path and symbol for a golang plugin containing the Provider
type PluginProvider ¶ added in v0.11.0
type PluginProvider interface {
Load(*PluginDefinition) (interface{}, error)
}
PluginProvider interface for providing a way of loading plugins must return one of the following types:
Subscription Retriever Applicative Fold Fork Publisher
type Retriever ¶ added in v0.3.0
Retriever is a function that provides data to a generic Stream must stop when the context receives a done signal.
type Stream ¶ added in v0.5.0
type Stream interface {
ID() string
Run(ctx context.Context, gracePeriod time.Duration, clusters ...Cluster) error
InjectionCallback(ctx context.Context) func(logs ...*Log)
Builder() Builder
Errors() chan error
}
Stream is a representation of a data stream and its associated logic. Creating a new Stream is handled by the appropriately named NewStream function.
The Builder method is the entrypoint into creating the data processing flow. All branches of the Stream are required to end in either a Publish or a Link in order to be considered valid.
func Load ¶ added in v0.16.0
func Load(serialization *StreamSerialization) (Stream, error)
Load method loads a stream based on the StreamSerialization
func NewStream ¶ added in v0.5.0
NewStream is a function for creating a new Stream. It takes an id, a Retriever function, and a list of Options that can override the defaults and set new defaults for the subsequent vertices in the Stream.
func NewSubscriptionStream ¶ added in v0.16.0
func NewSubscriptionStream(id string, sub Subscription, interval time.Duration, opts ...*Option) Stream
NewSubscriptionStream creates a Stream from the provider Subscription and pulls data continuously after an interval amount of time
type StreamSerialization ¶ added in v0.11.0
type StreamSerialization struct {
// Type type of stream to create.
//
// For root serializations valid values are 'http', 'subscription', or 'stream'.
Type string `json:"type,omitempty" mapstructure:"type,omitempty"`
// Interval is the duration in nanoseconds between pulls in a 'subscription' Type. It is only read
// if the Type is 'subscription'.
Interval time.Duration `json:"interval,omitempty" mapstructure:"interval,omitempty"`
// Options are a slice of machine.Option https://godoc.org/github.com/whitaker-io/machine#Option
Options []*Option `json:"options,omitempty" mapstructure:"options,omitempty"`
*VertexSerialization
}
StreamSerialization config based definition for a stream
func (*StreamSerialization) MarshalJSON ¶ added in v0.12.4
func (s *StreamSerialization) MarshalJSON() ([]byte, error)
MarshalJSON implementation to marshal json
func (*StreamSerialization) MarshalYAML ¶ added in v0.11.0
func (s *StreamSerialization) MarshalYAML() (interface{}, error)
MarshalYAML implementation to marshal yaml
func (*StreamSerialization) UnmarshalJSON ¶ added in v0.12.4
func (s *StreamSerialization) UnmarshalJSON(data []byte) error
UnmarshalJSON implementation to unmarshal json
func (*StreamSerialization) UnmarshalYAML ¶ added in v0.11.0
func (s *StreamSerialization) UnmarshalYAML(unmarshal func(interface{}) error) error
UnmarshalYAML implementation to unmarshal yaml
type Subscription ¶ added in v0.7.0
Subscription is an interface for creating a pull based stream. It requires 2 methods Read and Close.
Read is called when the interval passes and the resulting payload is sent down the Stream.
Close is called during a graceful termination and any errors are logged.
type VertexSerialization ¶ added in v0.11.0
type VertexSerialization struct {
// ID unique identifier for the stream.
ID string `json:"id,omitempty" mapstructure:"id,omitempty"`
// Provider Plugin information to load
Provider *PluginDefinition `json:"provider,omitempty" mapstructure:"provider,omitempty"`
// contains filtered or unexported fields
}
VertexSerialization config based definition for a stream vertex
Source Files
Directories