cmd

README

cmd

This directory contains applications (apps) used in Substation deployments. Apps are organized by either the infrastructure they are deployed to (e.g., AWS) or the source of the data (e.g., file, http).

Any app that implements the ingest, transform, load (ITL) functionality of Substation is named substation and shares the same configuration file format (see config/ for more information).

app.go

Contains the core Substation application code. This code can be used to create new Substation applications.

design

Substation operates through a system of channels and concurrent goroutines; ingest, transform, and load are handled in separate goroutines and managed by the cmd that invokes the app. Below is a diagram that describes the execution model:

cmd/main loads configuration
cmd/main creates channels
cmd/main executes anonymous goroutine
  - anon goroutine creates waitgroups for sink and transform goroutines
  - anon goroutine executes sink goroutine
  - anon goroutine executes transform goroutines
  - anon goroutine sends data to the tranform channel
  - anon goroutine closes transform channel, waits for transform goroutines to finish
  - anon goroutine closes sink channel, waits for sink goroutine to finish
cmd/main blocks waiting for feedback from non-anonymous goroutines

This execution model was chosen for its ability to support horizontal scaling, high-latency data processing, and efficient delivery of data.

aws/lambda

Contains Substation apps deployed as AWS Lambda functions. More information is available in cmd/aws/lambda/README.md.

file/substation

This app handles ITL for data stored in files. The app can be deployed anywhere, including non-container infrastructure, and is recommended for local testing of Substation configs.

Documentation

Index

• func GetConcurrency() (int, error)
• type Channels
• type Substation
  • func (sub *Substation) Block(ctx context.Context) error
  • func (sub *Substation) CreateChannels(size int)
  • func (sub *Substation) DoneSignal()
  • func (sub *Substation) KillSignal()
  • func (sub *Substation) SendErr(err error)
  • func (sub *Substation) SendTransform(cap config.Capsule)
  • func (sub *Substation) Sink(ctx context.Context, wg *sync.WaitGroup)
  • func (sub *Substation) SinkSignal()
  • func (sub *Substation) Transform(ctx context.Context, wg *sync.WaitGroup)
  • func (sub *Substation) TransformSignal()

Functions

func GetConcurrency

func GetConcurrency() (int, error)

GetConcurrency retrieves a concurrency value from the SUBSTATION_CONCURRENCY environment variable. If the environment variable is missing, then the concurrency value is the number of CPUs on the operating system. In native Substation applications, this value determines the number of transform goroutines; if set to 1, then multi-core processing is not enabled.

Types

type Channels

type Channels struct {
	Done      chan struct{}
	Kill      chan struct{}
	Errs      chan error
	Transform chan config.Capsule
	Sink      chan config.Capsule
}

Channels contains channels used by the app for managing state and sending encapsulated data between goroutines:

- Done: signals that all data processing (ingest, transform, load) is complete; this is always invoked by the Sink goroutine

- Kill: signals that all non-anonymous goroutines should end processing

- Errs: signals that an error occurred from an internal component

- Transform: sends encapsulated data from the handler to the Transform goroutines

- Sink: sends encapsulated data from the Transform goroutines to the Sink goroutine

type Substation

type Substation struct {
	Channels Channels
	Config   cfg
}

Substation is the application core, all data processing and flow happens through Substation.

func (*Substation) Block

func (sub *Substation) Block(ctx context.Context) error

Block blocks the handler from returning until one of these conditions is met:

- the handler request times out (ctx.Done)

- a data processing error occurs

- all data processing is complete

This is usually the final call made by main() in a cmd invoking the app.

func (*Substation) CreateChannels

func (sub *Substation) CreateChannels(size int)

CreateChannels initializes channels used by the app. Non-blocking channels can leak if the caller closes before processing completes; this is most likely to happen if the caller uses context to timeout. To avoid goroutine leaks, set larger buffer sizes.

func (*Substation) DoneSignal

func (sub *Substation) DoneSignal()

DoneSignal closes the Done channel. This signals that all data was sent to a sink. This should only be called by the Sink goroutine.

func (*Substation) KillSignal

func (sub *Substation) KillSignal()

KillSignal closes the Kill channel. This signals all non-anonymous goroutines to stop running. This should always be deferred by the cmd invoking the app.

func (*Substation) SendErr

func (sub *Substation) SendErr(err error)

SendErr puts an error into the Errs channel.

func (*Substation) SendTransform

func (sub *Substation) SendTransform(cap config.Capsule)

SendTransform puts byte data into the Transform channel.

func (*Substation) Sink

func (sub *Substation) Sink(ctx context.Context, wg *sync.WaitGroup)

Sink is the data sink method for the app. Data is input on the Sink channel and sent to the configured sink. The Sink goroutine completes when the Sink channel is closed and all data is flushed.

func (*Substation) SinkSignal

func (sub *Substation) SinkSignal()

SinkSignal closes the Sink channel. This signals that there is no more data to send. This should only be called by the cmd invoking the app.

func (*Substation) Transform

func (sub *Substation) Transform(ctx context.Context, wg *sync.WaitGroup)

Transform is the data transformation method for the app. Data is input on the Transform channel, transformed by a Transform interface (see: internal/transform), and output on the Sink channel. All Transform goroutines complete when the Transform channel is closed and all data is flushed.

func (*Substation) TransformSignal

func (sub *Substation) TransformSignal()

TransformSignal closes the Transform channel. This signals that there is no more incoming data to process. This should only be called by the cmd invoking the app.