core

Documentation

Index

• Constants
• func CfgAuthOpts(cfg *viper.Viper) *shttp.AuthenticationOpts
• func Main(defaultCfgFile string)
• func NewClassifySubnet(cfg *viper.Viper) (interface{}, error)
• func NewCompressGzip(cfg *viper.Viper) (interface{}, error)
• func NewCompressNone(cfg *viper.Viper) (interface{}, error)
• func NewEncodeCSV(cfg *viper.Viper) (interface{}, error)
• func NewEncodeJSON(cfg *viper.Viper) (interface{}, error)
• func NewFilterSubnet(cfg *viper.Viper) (interface{}, error)
• func NewStoreBuffered(cfg *viper.Viper) (interface{}, error)
• func NewStoreDirect(cfg *viper.Viper) (interface{}, error)
• func NewSubscriber(pipeline *Pipeline, cfg *viper.Viper) (*websocket.StructSpeaker, error)
• func NewTransformNone(cfg *viper.Viper) (interface{}, error)
• func NewWriteS3(cfg *viper.Viper) (interface{}, error)
• func NewWriteStdout(cfg *viper.Viper) (interface{}, error)
• func SubscriberRun(s *websocket.StructSpeaker)
• type Classifier
• type Compressor
• type EncodeCSV
  • func (e *EncodeCSV) Encode(in interface{}) ([]byte, error)
• type EncodeJSON
  • func (e *EncodeJSON) Encode(in interface{}) ([]byte, error)
• type Encoder
• type Filterer
• type Handler
• type HandlersMap
  • func (m HandlersMap) Init(cfg *viper.Viper, phase string) (interface{}, error)
  • func (m HandlersMap) Register(name string, handler Handler, isDefault bool)
• type Pipeline
  • func NewPipeline(cfg *viper.Viper) (*Pipeline, error)
  • func (p *Pipeline) OnStructMessage(c ws.Speaker, msg *ws.StructMessage)
• type Storer
• type Tag
• type Transformer
• type Writer

Constants

const CfgRoot = "pipeline."

CfgRoot configuration root path

Functions

func CfgAuthOpts

func CfgAuthOpts(cfg *viper.Viper) *shttp.AuthenticationOpts

CfgAuthOpts creates the auth options form configuration

func Main

func Main(defaultCfgFile string)

Main entry point for exporters

func NewClassifySubnet

func NewClassifySubnet(cfg *viper.Viper) (interface{}, error)

NewClassifySubnet returns a new classify, based on the given cluster net masks

func NewCompressGzip

func NewCompressGzip(cfg *viper.Viper) (interface{}, error)

NewCompressGzip create an encode object

func NewCompressNone

func NewCompressNone(cfg *viper.Viper) (interface{}, error)

NewCompressNone create an encode object

func NewEncodeCSV

func NewEncodeCSV(cfg *viper.Viper) (interface{}, error)

NewEncodeCSV creates an encode object

func NewEncodeJSON

func NewEncodeJSON(cfg *viper.Viper) (interface{}, error)

NewEncodeJSON creates an encode object

func NewFilterSubnet

func NewFilterSubnet(cfg *viper.Viper) (interface{}, error)

NewFilterSubnet returns a new filter based on config

func NewStoreBuffered

func NewStoreBuffered(cfg *viper.Viper) (interface{}, error)

NewStoreBuffered returns a new storage interface for storing flows to object store

func NewStoreDirect

func NewStoreDirect(cfg *viper.Viper) (interface{}, error)

NewStoreDirect returns a new storage interface for storing flows to object store

func NewSubscriber

func NewSubscriber(pipeline *Pipeline, cfg *viper.Viper) (*websocket.StructSpeaker, error)

NewSubscriber returns a new flow subscriber writing to object store

func NewTransformNone

func NewTransformNone(cfg *viper.Viper) (interface{}, error)

NewTransformNone create a new transform

func NewWriteS3

func NewWriteS3(cfg *viper.Viper) (interface{}, error)

NewWriteS3 creates a new S3-compatible object storage client

func NewWriteStdout

func NewWriteStdout(cfg *viper.Viper) (interface{}, error)

NewWriteStdout returns a new storage interface for storing flows to object store

func SubscriberRun

func SubscriberRun(s *websocket.StructSpeaker)

SubscriberRun runs the subscriber under main

Types

type Classifier

type Classifier interface {
	GetFlowTag(fl *flow.Flow) Tag
}

Classifier exposes the interface for tag based classification

type Compressor

type Compressor interface {
	Compress(b []byte) (*bytes.Buffer, error)
}

Compressor exposes the interface for compressesing encoded flows

type EncodeCSV

type EncodeCSV struct {
}

EncodeCSV encoder encodes flows as CSV rows

func (*EncodeCSV) Encode

func (e *EncodeCSV) Encode(in interface{}) ([]byte, error)

Encode implements Encoder interface

type EncodeJSON

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

EncodeJSON encoder encodes flows as a JSON array

func (*EncodeJSON) Encode

func (e *EncodeJSON) Encode(in interface{}) ([]byte, error)

Encode implements Encoder interface

type Encoder

type Encoder interface {
	Encode(in interface{}) ([]byte, error)
}

Encoder exposes the interface for encoding flows

type Filterer

type Filterer interface {
	IsExcluded(tag Tag) bool
}

Filterer exposes the interface for tag based filtering

type Handler

type Handler = func(cfg *viper.Viper) (interface{}, error)

Handler used for creating a phase handler from configuration

type HandlersMap

type HandlersMap map[string]Handler

HandlersMap a map of handlers

var (
	TransformerHandlers HandlersMap
	ClassifierHandlers  HandlersMap
	FiltererHandlers    HandlersMap
	EncoderHandlers     HandlersMap
	CompressorHandlers  HandlersMap
	StorerHandlers      HandlersMap
	WriterHandlers      HandlersMap
)

Global set of handlers

func (HandlersMap) Init

func (m HandlersMap) Init(cfg *viper.Viper, phase string) (interface{}, error)

Init creates resource from config

func (HandlersMap) Register

func (m HandlersMap) Register(name string, handler Handler, isDefault bool)

Register associates a handler with its' label

type Pipeline

type Pipeline struct {
	sync.Mutex

	Transformer Transformer
	Classifier  Classifier
	Filterer    Filterer
	Encoder     Encoder
	Compressor  Compressor
	Storer      Storer
	Writer      Writer
}

Pipeline manager

func NewPipeline

func NewPipeline(cfg *viper.Viper) (*Pipeline, error)

NewPipeline defines the pipeline elements

func (*Pipeline) OnStructMessage

func (p *Pipeline) OnStructMessage(c ws.Speaker, msg *ws.StructMessage)

OnStructMessage is triggered when WS server sends us a message.

type Storer

type Storer interface {
	StoreFlows(flows map[Tag][]interface{}) error
	SetPipeline(p *Pipeline)
}

Storer interface of a store object

type Tag

type Tag string

Tag represents the flow classification

type Transformer

type Transformer interface {
	// Transform transforms a flow before being stored
	Transform(f *flow.Flow) interface{}
}

Transformer allows generic transformations of a flow

type Writer

type Writer interface {
	Write(bucket, objectKey, data, contentType, contentEncoding string, metadata map[string]*string) error
}

Writer allows uploading objects to an object storage service