recipe

Documentation

Index

• Variables
• type InvalidRecipeError
  • func (err InvalidRecipeError) Error() string
• type Monitor
• type ProcessorRecipe
• type Reader
  • func NewReader() *Reader
  • func (r *Reader) Read(path string) (recipe Recipe, err error)
  • func (r *Reader) ReadDir(path string) (recipes []Recipe, err error)
• type Recipe
• type Run
• type RunTaskError
  • func (e RunTaskError) Error() string
• type Runner
  • func NewRunner(extractorFactory *extractor.Factory, processorFactory *processor.Factory, ...) *Runner
  • func (r *Runner) Run(recipe Recipe) (err error)
  • func (r *Runner) RunMultiple(recipes []Recipe) (faileds []string, err error)
• type SinkRecipe
• type SourceRecipe
• type Task

Variables

var (
	TaskTypeExtract = "extract"
	TaskTypeProcess = "process"
	TaskTypeSink    = "sink"
)

var (
	TaskStatusReady    = "ready"
	TaskStatusComplete = "complete"
	TaskStatusFailed   = "failed"
)

Types

type InvalidRecipeError

type InvalidRecipeError struct {
	Message string
}

func (InvalidRecipeError) Error

func (err InvalidRecipeError) Error() string

type Monitor

type Monitor interface {
	RecordRun(recipe Recipe, durationInMs int, success bool)
}

type ProcessorRecipe

type ProcessorRecipe struct {
	Name   string                 `yaml:"name" validate:"required"`
	Config map[string]interface{} `yaml:"config"`
}

type Reader

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

func NewReader

func NewReader() *Reader

func (*Reader) Read

func (r *Reader) Read(path string) (recipe Recipe, err error)

func (*Reader) ReadDir

func (r *Reader) ReadDir(path string) (recipes []Recipe, err error)

type Recipe

type Recipe struct {
	Name       string            `yaml:"name" validate:"required"`
	Source     SourceRecipe      `yaml:"source" validate:"required"`
	Sinks      []SinkRecipe      `yaml:"sinks" validate:"required,min=1"`
	Processors []ProcessorRecipe `yaml:"processors"`
}

type Run

type Run struct {
	Recipe Recipe        `json:"recipe"`
	Tasks  []Task        `json:"tasks"`
	Data   []interface{} `json:"-"`
}

type RunTaskError

type RunTaskError struct {
	Task Task
	Err  error
}

func (RunTaskError) Error

func (e RunTaskError) Error() string

type Runner

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

func NewRunner

func NewRunner(
	extractorFactory *extractor.Factory,
	processorFactory *processor.Factory,
	sinkFactory *sinks.Factory,
	monitor Monitor) *Runner

func (*Runner) Run

func (r *Runner) Run(recipe Recipe) (err error)

func (*Runner) RunMultiple

func (r *Runner) RunMultiple(recipes []Recipe) (faileds []string, err error)

type SinkRecipe

type SinkRecipe struct {
	Name         string                 `yaml:"name" validate:"required"`
	BaseSinkName string                 `yaml:"base_sink_name"`
	Config       map[string]interface{} `yaml:"config"`
}

type SourceRecipe

type SourceRecipe struct {
	Type   string                 `yaml:"type" validate:"required"`
	Config map[string]interface{} `yaml:"config"`
}

type Task

type Task struct {
	Type   string                 `json:"type"`
	Name   string                 `json:"name"`
	Config map[string]interface{} `json:"config"`
	Status string                 `json:"status"`
}