flow

Documentation

Index

• Constants
• type Accounter
  • func NewAccounter(maxEntries, evictBufferLen int, evictTimeout time.Duration) *Accounter
  • func (c *Accounter) Account(in <-chan *Record, out chan<- []*Record)
• type DataLink
• type Direction
  • func (d Direction) MarshalJSON() ([]byte, error)
• type HumanBytes
  • func (b HumanBytes) String() string
• type MacAddr
  • func (m *MacAddr) MarshalJSON() ([]byte, error)
  • func (m *MacAddr) String() string
• type Network
• type RawIP
  • func (i RawIP) MarshalJSON() ([]byte, error)
  • func (i RawIP) String() string
• type Record
  • func ReadFrom(reader io.Reader) (*Record, error)
  • func (r *Record) Accumulate(src *Record)
• type Transport
• type TransportProtocol
  • func (p TransportProtocol) MarshalJSON() ([]byte, error)
  • func (p TransportProtocol) String() string

Constants

const MacLen = 6

Types

type Accounter

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

Accounter accumulates flows metrics in memory and eventually evicts them via an evictor channel. It is not safe for concurrent access so if at some point you need to parallelize also the accounting process you should either: (1) reimplement the Accounter by using a concurrent map implementation and performing the properties operations atomically; or (2) spinup parallel accounters that evict the flow records to another accounter that consolidates data from all (some kind of map-reduce).

func NewAccounter

func NewAccounter(maxEntries, evictBufferLen int, evictTimeout time.Duration) *Accounter

NewAccounter creates a new Accounter. The cache has no limit and it's assumed that eviction is done by the caller.

func (*Accounter) Account

func (c *Accounter) Account(in <-chan *Record, out chan<- []*Record)

Account runs in a new goroutine. It reads all the records from the input channel and accumulate their metrics internally. Once the metrics have reached their max size or the eviction times out, it evicts all the accumulated flows by the returned channel.

type DataLink

type DataLink struct {
	SrcMac MacAddr
	DstMac MacAddr
}

type Direction

type Direction uint8

func (Direction) MarshalJSON

func (d Direction) MarshalJSON() ([]byte, error)

type HumanBytes

type HumanBytes uint64

func (HumanBytes) String

func (b HumanBytes) String() string

type MacAddr

type MacAddr [MacLen]uint8

func (*MacAddr) MarshalJSON

func (m *MacAddr) MarshalJSON() ([]byte, error)

func (*MacAddr) String

func (m *MacAddr) String() string

type Network

type Network struct {
	SrcAddr RawIP
	DstAddr RawIP
}

type RawIP

type RawIP uint32

TODO: support IPv6

func (RawIP) MarshalJSON

func (i RawIP) MarshalJSON() ([]byte, error)

func (RawIP) String

func (i RawIP) String() string

type Record

type Record struct {
	TimeFlowStart time.Time
	TimeFlowEnd   time.Time
	Interface     string
	Packets       int
	// contains filtered or unexported fields
}

Record contains accumulated metrics from a flow

func ReadFrom

func ReadFrom(reader io.Reader) (*Record, error)

ReadFrom reads a Record from a binary source, in LittleEndian order

func (*Record) Accumulate

func (r *Record) Accumulate(src *Record)

type Transport

type Transport struct {
	SrcPort  uint16
	DstPort  uint16
	Protocol TransportProtocol `json:"Proto"`
}

type TransportProtocol

type TransportProtocol uint8

func (TransportProtocol) MarshalJSON

func (p TransportProtocol) MarshalJSON() ([]byte, error)

func (TransportProtocol) String

func (p TransportProtocol) String() string