processors

README

PROCESSORS

Documentation

Index

• func AddFields(fields map[string]interface{}, data *mxj.Map)
• func AddTags(tags []string, data *mxj.Map)
• func Dynamic(str *string, fields *mxj.Map)
• func NormalizeNestedPath(path string) string
• func RemoveFields(fields []string, data *mxj.Map)
• func RemoveTags(tags []string, data *mxj.Map)
• func SetType(typevalue string, data *mxj.Map)
• type Base
  • func (b *Base) B() *Base
  • func (b *Base) Configure(ctx ProcessorContext, conf map[string]interface{}) error
  • func (b *Base) ConfigureAndValidate(ctx ProcessorContext, conf map[string]interface{}, rawVal interface{}) error
  • func (b *Base) Doc() *doc.Processor
  • func (b *Base) MaxConcurent() int
  • func (b *Base) Receive(e IPacket) error
  • func (b *Base) SetPipelineUUID(uuid string)
  • func (b *Base) SetProcessorIdentifiers(name, label string)
  • func (b *Base) Start(e IPacket) error
  • func (b *Base) Stop(e IPacket) error
  • func (b *Base) Tick(e IPacket) error
• type CommonOptions
  • func (c *CommonOptions) ProcessCommonOptions(data *mxj.Map)
• type ICodec
• type IPacket
• type IStore
• type Logger
• type Memory
• type PacketBuilder
• type PacketSender
• type Processor
• type ProcessorContext
• type SinceDB
  • func NewSinceDB(sdboptions *SinceDBOptions) *SinceDB
  • func (s *SinceDB) Close() error
  • func (s *SinceDB) Ressource(ressource string) ([]byte, error)
  • func (s *SinceDB) RessourceOffset(ressource string) (int, error)
  • func (s *SinceDB) SetRessource(ressource string, v []byte) error
  • func (s *SinceDB) SetRessourceOffset(ressource string, offset int) error
• type SinceDBOptions
• type WebHook

Functions

func AddFields

func AddFields(fields map[string]interface{}, data *mxj.Map)

func AddTags

func AddTags(tags []string, data *mxj.Map)

func Dynamic

func Dynamic(str *string, fields *mxj.Map)

Dynamic includes field value in place of %{key.path} When no field is not found replace with ""

func NormalizeNestedPath

func NormalizeNestedPath(path string) string

func RemoveFields

func RemoveFields(fields []string, data *mxj.Map)

func RemoveTags

func RemoveTags(tags []string, data *mxj.Map)

func SetType

func SetType(typevalue string, data *mxj.Map)

Types

type Base

type Base struct {
	Send                  PacketSender
	NewPacket             PacketBuilder
	Logger                Logger
	Memory                Memory
	WebHook               WebHook
	Store                 IStore
	ConfigWorkingLocation string
	DataLocation          string
	PipelineUUID          string
	Name                  string
	Label                 string
}

func (*Base) B

func (b *Base) B() *Base

B returns the Base Processor

func (*Base) Configure

func (b *Base) Configure(ctx ProcessorContext, conf map[string]interface{}) error

func (*Base) ConfigureAndValidate

func (b *Base) ConfigureAndValidate(ctx ProcessorContext, conf map[string]interface{}, rawVal interface{}) error

func (*Base) Doc

func (b *Base) Doc() *doc.Processor

func (*Base) MaxConcurent

func (b *Base) MaxConcurent() int

func (*Base) Receive

func (b *Base) Receive(e IPacket) error

func (*Base) SetPipelineUUID

func (b *Base) SetPipelineUUID(uuid string)

func (*Base) SetProcessorIdentifiers

func (b *Base) SetProcessorIdentifiers(name, label string)

func (*Base) Start

func (b *Base) Start(e IPacket) error

func (*Base) Stop

func (b *Base) Stop(e IPacket) error

func (*Base) Tick

func (b *Base) Tick(e IPacket) error

type CommonOptions

type CommonOptions struct {
	// If this filter is successful, add any arbitrary fields to this event.
	AddField map[string]interface{} `mapstructure:"add_field"`

	// If this filter is successful, add arbitrary tags to the event. Tags can be dynamic
	// and include parts of the event using the %{field} syntax.
	AddTag []string `mapstructure:"add_tag"`

	// Add a type field to all events handled by this input
	Type string `mapstructure:"type"`

	// If this filter is successful, remove arbitrary fields from this event. Example:
	// ` kv {
	// `   remove_field => [ "foo_%{somefield}" ]
	// ` }
	RemoveField []string `mapstructure:"remove_field"`

	// If this filter is successful, remove arbitrary tags from the event. Tags can be dynamic and include parts of the event using the %{field} syntax.
	// Example:
	// ` kv {
	// `   remove_tag => [ "foo_%{somefield}" ]
	// ` }
	// If the event has field "somefield" == "hello" this filter, on success, would remove the tag foo_hello if it is present. The second example would remove a sad, unwanted tag as well.
	RemoveTag []string `mapstructure:"remove_tag"`

	// Log each event produced by the processor (usefull while building or debugging a pipeline)
	Trace bool `mapstructure:"trace"`
}

func (*CommonOptions) ProcessCommonOptions

func (c *CommonOptions) ProcessCommonOptions(data *mxj.Map)

type ICodec

type ICodec interface {
	GetName() string
	GetOptions() map[string]interface{}
	GetRole() string
}

type IPacket

type IPacket interface {
	Message() string
	Fields() *mxj.Map

	SetMessage(string)
	SetFields(map[string]interface{})

	Clone() IPacket
}

type IStore

type IStore interface {
	Get(string, string) ([]byte, error) // Get(key, bucket)
	Set(string, string, []byte) error   // Set(key, bucket, value)
	Delete(string, string) error        // Delete(key, bucket)
	Has(string, string) (bool, error)   // Has(key, bucket)
}

IStore is a key value permanent storage each key/value is saved in a named bucket the default bucket name is "default"

type Logger

type Logger interface {
	Debug(args ...interface{})
	Debugf(format string, args ...interface{})
	Debugln(args ...interface{})

	Error(args ...interface{})
	Errorf(format string, args ...interface{})
	Errorln(args ...interface{})

	Fatal(args ...interface{})
	Fatalf(format string, args ...interface{})
	Fatalln(args ...interface{})

	Info(args ...interface{})
	Infof(format string, args ...interface{})
	Infoln(args ...interface{})

	Panic(args ...interface{})
	Panicf(format string, args ...interface{})
	Panicln(args ...interface{})

	Print(args ...interface{})
	Printf(format string, args ...interface{})
	Println(args ...interface{})

	Warn(args ...interface{})
	Warnf(format string, args ...interface{})
	Warning(args ...interface{})
	Warningf(format string, args ...interface{})
	Warningln(args ...interface{})
	Warnln(args ...interface{})
}

type Memory

type Memory interface {
	Set(name string, value interface{})
	Get(name string) (interface{}, bool)
	Delete(name string)
	Items() map[string]interface{}
}

type PacketBuilder

type PacketBuilder func(map[string]interface{}) IPacket

type PacketSender

type PacketSender func(IPacket, ...int) bool

type Processor

type Processor interface {
	B() *Base
	Configure(ProcessorContext, map[string]interface{}) error
	Start(IPacket) error
	Tick(IPacket) error
	Receive(IPacket) error
	Stop(IPacket) error
	Doc() *doc.Processor
	MaxConcurent() int
	SetPipelineUUID(string)
	SetProcessorIdentifiers(string, string)
}

type ProcessorContext

type ProcessorContext interface {
	Log() Logger
	PacketSender() PacketSender
	PacketBuilder() PacketBuilder
	Memory() Memory
	WebHook() WebHook
	ConfigWorkingLocation() string
	DataLocation() string
	Store() IStore
}

type SinceDB

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

func NewSinceDB

func NewSinceDB(sdboptions *SinceDBOptions) *SinceDB

NewSinceDB loadExisting data from datastore according to the Identifier option.

func (*SinceDB) Close

func (s *SinceDB) Close() error

Save SinceDB Offsets to Storage

func (*SinceDB) Ressource

func (s *SinceDB) Ressource(ressource string) ([]byte, error)

func (*SinceDB) RessourceOffset

func (s *SinceDB) RessourceOffset(ressource string) (int, error)

Retreive SinceDB ressource's offset from Storage

func (*SinceDB) SetRessource

func (s *SinceDB) SetRessource(ressource string, v []byte) error

Update a ressource's offset

func (*SinceDB) SetRessourceOffset

func (s *SinceDB) SetRessourceOffset(ressource string, offset int) error

Update a ressource's offset

type SinceDBOptions

type SinceDBOptions struct {
	Identifier    string
	WriteInterval int
	Storage       IStore
}

type WebHook

type WebHook interface {
	Add(string, http.HandlerFunc)
	AddShort(string, http.HandlerFunc)
	Unregister()
}