transform

README

transform

Contains interfaces and methods for transforming data between a source and a sink. Substation is designed for processing JSON data, but with processors (see process/) it can support any data format. Each transform must select from both the input and kill channels to prevent goroutine leaks (learn more about goroutine leaks here).

Transform	Description
Process	applies processors to the data
Transfer	applies no transformation to the data

process

Transforms data by applying processors (process/). Processors are enabled by matching conditions (condition/) for every event. This transform iteratively processes individual events in the channel and modifies them before passing the event to the next processor.

Below is an example that shows how a single event is iteratively modified through this transform:

// input event
event: {"hello":"world"}

// insert value "bar" into key "foo"
processor: insert("foo", "bar")

event: {"hello":"world","foo":"bar"}

// insert value "qux" into key "foo"
processor: insert("baz", "qux")

event: {"hello":"world", "foo":"bar", "baz":"qux"}

// concat vaues from "foo" and "baz" into key "foo" with separator "."
processor: concat("foo", ["foo", "baz"], ".")

event: {"hello":"world", "foo":"bar.qux"}

The transform uses this Jsonnet configuration (see config/example/substation_example_kinesis/ for more examples):

{
  type: 'process',
  settings: {
    // processors are defined according to the information in `process/`
    processors: [
      // adds an "event.created" key to the JSON event that contains the time that the processor executed
      {
        settings: {
          options: {
            input_format: 'now',
            output_format: '2006-01-02T15:04:05.000000Z',
          },
          output: {
            key: 'event.created',
          }
        },
        type: 'time',
      },
    ],
  },
}

transfer

Transforms data with no modification, which we refer to as a "transfer." This transform is best used when the integrity of the original data needs to be maintained or if no data processing is needed.

The transform uses this Jsonnet configuration (see config/example/ for more examples):

{
  type: 'transfer',
}

Documentation

Index

• Constants
• type Config
• type Process
  • func (transform *Process) Transform(ctx context.Context, in <-chan []byte, out chan<- []byte, kill chan struct{}) error
• type Transfer
  • func (transform *Transfer) Transform(ctx context.Context, in <-chan []byte, out chan<- []byte, kill chan struct{}) error
• type Transform
  • func Factory(cfg Config) (Transform, error)

Constants

const TransformInvalidFactoryConfig = errors.Error("TransformInvalidFactoryConfig")

TransformInvalidFactoryConfig is used when an unsupported Transform is referenced in Factory.

Types

type Config

type Config struct {
	Type     string
	Settings map[string]interface{}
}

Config contains arbitrary JSON settings for Transforms loaded via mapstructure.

type Process

type Process struct {
	Processors []process.Config `mapstructure:"processors"`
}

Process contains this transform's configuration settings. More information is available in the README.

func (*Process) Transform

func (transform *Process) Transform(ctx context.Context, in <-chan []byte, out chan<- []byte, kill chan struct{}) error

Transform processes a channel of bytes with processors defined in the config.

type Transfer

type Transfer struct{}

Transfer contains this transform's configuration settings. More information is available in the README.

func (*Transfer) Transform

func (transform *Transfer) Transform(ctx context.Context, in <-chan []byte, out chan<- []byte, kill chan struct{}) error

Transform tranfers the input channel of bytes directly to the output channel with no modification.

type Transform

type Transform interface {
	Transform(context.Context, <-chan []byte, chan<- []byte, chan struct{}) error
}

Transform is the interface used by all Substation transforms. Transforms read channels of bytes, can optionally write channels of bytes, and are interruptable via an anonymous struct channel.

func Factory

func Factory(cfg Config) (Transform, error)

Factory loads Transforms from a Config. This is the recommended function for retrieving ready-to-use Transforms.