core

Documentation

Index

• Constants
• func HTTPHandler(path string, s http.Handler) fnMux
• func Pipelines() map[string]*Pipeline
• func ProcessorsDocs(code string) map[string]*doc.Processor
• func RegisterProcessor(name string, procFact ProcessorFactory)
• func Start(opt Options)
• func Stop() error
• func StopPipeline(Uuid string) error
• func Storage() *store.Store
• type Agent
  • func NewAgent() Agent
  • func Sort(agentConflist map[int]*Agent, sortOrder int) []*Agent
  • func (a *Agent) Processor() processors.Processor
• type Codec
  • func NewCodec(name string) *Codec
  • func (c *Codec) GetName() string
  • func (c *Codec) GetOptions() map[string]interface{}
  • func (c *Codec) GetRole() string
  • func (c *Codec) String() string
• type Logger
  • func Log() *Logger
  • func NewLogger(component string, data map[string]interface{}) *Logger
  • func (p *Logger) Debug(args ...interface{})
  • func (p *Logger) Debugf(format string, args ...interface{})
  • func (p *Logger) Debugln(args ...interface{})
  • func (p *Logger) Error(args ...interface{})
  • func (p *Logger) Errorf(format string, args ...interface{})
  • func (p *Logger) Errorln(args ...interface{})
  • func (p *Logger) Fatal(args ...interface{})
  • func (p *Logger) Fatalf(format string, args ...interface{})
  • func (p *Logger) Fatalln(args ...interface{})
  • func (p *Logger) Info(args ...interface{})
  • func (p *Logger) Infof(format string, args ...interface{})
  • func (p *Logger) Infoln(args ...interface{})
  • func (p *Logger) Panic(args ...interface{})
  • func (p *Logger) Panicf(format string, args ...interface{})
  • func (p *Logger) Panicln(args ...interface{})
  • func (p *Logger) Print(args ...interface{})
  • func (p *Logger) Printf(format string, args ...interface{})
  • func (p *Logger) Println(args ...interface{})
  • func (p *Logger) Warn(args ...interface{})
  • func (p *Logger) Warnf(format string, args ...interface{})
  • func (p *Logger) Warning(args ...interface{})
  • func (p *Logger) Warningf(format string, args ...interface{})
  • func (p *Logger) Warningln(args ...interface{})
  • func (p *Logger) Warnln(args ...interface{})
• type Options
• type Pipeline
  • func GetPipeline(UUID string) (*Pipeline, bool)
  • func NewPipeline() *Pipeline
  • func (p *Pipeline) AddAgent(a Agent) error
  • func (p *Pipeline) Agents() map[int]*Agent
  • func (p *Pipeline) Start() (string, error)
  • func (p *Pipeline) Stop() error
• type Port
• type PortList
  • func (a *PortList) String() string
  • func (a *PortList) StringReversePort() string
• type ProcessorFactory

Constants

const (
	SortInputsFirst = iota + 1
	SortOutputsFirst
)

Sorting order

Functions

func HTTPHandler

func HTTPHandler(path string, s http.Handler) fnMux

func Pipelines

func Pipelines() map[string]*Pipeline

Pipelines returns running core.Pipeline

func ProcessorsDocs

func ProcessorsDocs(code string) map[string]*doc.Processor

ProcessorsDocs returns available ProcessorDoc

func RegisterProcessor

func RegisterProcessor(name string, procFact ProcessorFactory)

RegisterProcessor is called by the processor loader when the program starts each processor give its name and factory func()

func Start

func Start(opt Options)

func Stop

func Stop() error

Stop each pipeline

func StopPipeline

func StopPipeline(Uuid string) error

func Storage

func Storage() *store.Store

TODO : should be unexported

Types

type Agent

type Agent struct {
	ID    int
	Label string

	Done chan bool

	Sources         []string `json:"sources"`
	AgentSources    PortList
	AgentRecipients PortList
	Type            string `json:"type"`
	Schedule        string `json:"schedule"`
	Trace           bool   `json:"trace"`
	PoolSize        int    `json:"pool_size"`
	PipelineName    string
	PipelineUUID    string
	Buffer          int `json:"buffer_size"`
	Options         map[string]interface{}
	Wd              string
	// contains filtered or unexported fields
}

func NewAgent

func NewAgent() Agent

func Sort

func Sort(agentConflist map[int]*Agent, sortOrder int) []*Agent

Sort will return a sorted list of config.Agent, it sorts agents by computing links dependencies between them

use sortOrder param config.SortInputsFirst to get agents which are not waiting events (no sources) firstly (like inputs)

use sortOrder param config.SortOutputsFirst to get agents which are not sources of any other agents firstly (like outputs)

func (*Agent) Processor

func (a *Agent) Processor() processors.Processor

Processor return the agent's processor

type Codec

type Codec struct {
	Name    string
	Role    string //decoder/encoder
	Options map[string]interface{}
}

func NewCodec

func NewCodec(name string) *Codec

func (*Codec) GetName

func (c *Codec) GetName() string

func (*Codec) GetOptions

func (c *Codec) GetOptions() map[string]interface{}

func (*Codec) GetRole

func (c *Codec) GetRole() string

func (*Codec) String

func (c *Codec) String() string

type Logger

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

func Log

func Log() *Logger

Should be unexported

func NewLogger

func NewLogger(component string, data map[string]interface{}) *Logger

func (*Logger) Debug

func (p *Logger) Debug(args ...interface{})

func (*Logger) Debugf

func (p *Logger) Debugf(format string, args ...interface{})

func (*Logger) Debugln

func (p *Logger) Debugln(args ...interface{})

func (*Logger) Error

func (p *Logger) Error(args ...interface{})

func (*Logger) Errorf

func (p *Logger) Errorf(format string, args ...interface{})

func (*Logger) Errorln

func (p *Logger) Errorln(args ...interface{})

func (*Logger) Fatal

func (p *Logger) Fatal(args ...interface{})

func (*Logger) Fatalf

func (p *Logger) Fatalf(format string, args ...interface{})

func (*Logger) Fatalln

func (p *Logger) Fatalln(args ...interface{})

func (*Logger) Info

func (p *Logger) Info(args ...interface{})

func (*Logger) Infof

func (p *Logger) Infof(format string, args ...interface{})

func (*Logger) Infoln

func (p *Logger) Infoln(args ...interface{})

func (*Logger) Panic

func (p *Logger) Panic(args ...interface{})

func (*Logger) Panicf

func (p *Logger) Panicf(format string, args ...interface{})

func (*Logger) Panicln

func (p *Logger) Panicln(args ...interface{})

func (*Logger) Print

func (p *Logger) Print(args ...interface{})

func (*Logger) Printf

func (p *Logger) Printf(format string, args ...interface{})

func (*Logger) Println

func (p *Logger) Println(args ...interface{})

func (*Logger) Warn

func (p *Logger) Warn(args ...interface{})

func (*Logger) Warnf

func (p *Logger) Warnf(format string, args ...interface{})

func (*Logger) Warning

func (p *Logger) Warning(args ...interface{})

func (*Logger) Warningf

func (p *Logger) Warningf(format string, args ...interface{})

func (*Logger) Warningln

func (p *Logger) Warningln(args ...interface{})

func (*Logger) Warnln

func (p *Logger) Warnln(args ...interface{})

type Options

type Options struct {
	Host         string
	HttpHandlers []fnMux
	Debug        bool
	VerboseLog   bool
	LogFile      string
	DataLocation string
	Prometheus   string
}

type Pipeline

type Pipeline struct {
	Uuid  string
	Label string

	ConfigLocation     string
	ConfigHostLocation string
	StartedAt          time.Time

	Description string

	Webhooks   []webhook.Hook
	Schedulers []schedulerJob
	// contains filtered or unexported fields
}

func GetPipeline

func GetPipeline(UUID string) (*Pipeline, bool)

func NewPipeline

func NewPipeline() *Pipeline

func (*Pipeline) AddAgent

func (p *Pipeline) AddAgent(a Agent) error

func (*Pipeline) Agents

func (p *Pipeline) Agents() map[int]*Agent

func (*Pipeline) Start

func (p *Pipeline) Start() (string, error)

Start all agents, begin with last

func (*Pipeline) Stop

func (p *Pipeline) Stop() error

type Port

type Port struct {
	AgentID    int
	PortNumber int
}

type PortList

type PortList []Port

func (*PortList) String

func (a *PortList) String() string

func (*PortList) StringReversePort

func (a *PortList) StringReversePort() string

type ProcessorFactory

type ProcessorFactory func() processors.Processor