intermediate

Documentation

Index

• type AggregationElements
• type AggregationFlowRecord
• type AggregationInput
• type AggregationProcess
  • func InitAggregationProcess(input AggregationInput) (*AggregationProcess, error)
  • func (a *AggregationProcess) AggregateMsgByFlowKey(message *entities.Message) error
  • func (a *AggregationProcess) DeleteFlowKeyFromMap(flowKey FlowKey)
  • func (a *AggregationProcess) ForAllRecordsDo(callback FlowKeyRecordMapCallBack) error
  • func (a *AggregationProcess) Start()
  • func (a *AggregationProcess) Stop()
• type FlowKey
• type FlowKeyRecordMapCallBack

Types

type AggregationElements

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

type AggregationFlowRecord

type AggregationFlowRecord struct {
	Record entities.Record
	// ReadyToSend is an indicator that we received all required records for the
	// given flow, i.e., records from source and destination nodes for the case
	// inter-node flow and record from the node for the case of intra-node flow.
	ReadyToSend bool
	// IsActive is a flag that indicates whether the flow is active or not. If
	// aggregation process stop receiving flows from collector process, we deem
	// the flow as inactive.
	IsActive bool
}

type AggregationInput

type AggregationInput struct {
	MessageChan       chan *entities.Message
	WorkerNum         int
	CorrelateFields   []string
	AggregateElements *AggregationElements
}

type AggregationProcess

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

func InitAggregationProcess

func InitAggregationProcess(input AggregationInput) (*AggregationProcess, error)

InitAggregationProcess takes in message channel (e.g. from collector) as input channel, workerNum(number of workers to process message) and correlateFields (fields to be correlated and filled).

func (*AggregationProcess) AggregateMsgByFlowKey

func (a *AggregationProcess) AggregateMsgByFlowKey(message *entities.Message) error

AggregateMsgByFlowKey gets flow key from records in message and stores in cache

func (*AggregationProcess) DeleteFlowKeyFromMap

func (a *AggregationProcess) DeleteFlowKeyFromMap(flowKey FlowKey)

func (*AggregationProcess) ForAllRecordsDo

func (a *AggregationProcess) ForAllRecordsDo(callback FlowKeyRecordMapCallBack) error

ForAllRecordsDo takes in callback function to process the operations to flowkey->records pairs in the map

func (*AggregationProcess) Start

func (a *AggregationProcess) Start()

func (*AggregationProcess) Stop

func (a *AggregationProcess) Stop()

type FlowKey

type FlowKey struct {
	SourceAddress      string
	DestinationAddress string
	Protocol           uint8
	SourcePort         uint16
	DestinationPort    uint16
}

type FlowKeyRecordMapCallBack

type FlowKeyRecordMapCallBack func(key FlowKey, record AggregationFlowRecord) error