flow

Documentation

Index

• Constants
• func Accumulate(r *ebpf.BpfFlowMetrics, src *ebpf.BpfFlowMetrics)
• func Decorate(agentIP net.IP, ifaceNamer InterfaceNamer) func(in <-chan []*Record, out chan<- []*Record)
• func Dedupe(expireTime time.Duration, justMark bool) func(in <-chan []*Record, out chan<- []*Record)
• func IP(ia IPAddr) net.IP
• func IntEncodeV4(ia [net.IPv6len]uint8) uint32
• type Accounter
  • func NewAccounter(maxEntries int, evictTimeout time.Duration, clock func() time.Time, ...) *Accounter
  • func (c *Accounter) Account(in <-chan *RawRecord, out chan<- []*Record)
• type CapacityLimiter
  • func (c *CapacityLimiter) Limit(in <-chan []*Record, out chan<- []*Record)
• type Direction
• type HumanBytes
• type IPAddr
  • func (ia *IPAddr) MarshalJSON() ([]byte, error)
• type InterfaceNamer
• type MacAddr
  • func (m *MacAddr) MarshalJSON() ([]byte, error)
  • func (m *MacAddr) String() string
• type MapTracer
  • func NewMapTracer(fetcher mapFetcher, evictionTimeout time.Duration) *MapTracer
  • func (m *MapTracer) Flush()
  • func (m *MapTracer) TraceLoop(ctx context.Context) node.StartFunc[[]*Record]
• type RawRecord
  • func ReadFrom(reader io.Reader) (*RawRecord, error)
• type Record
  • func NewRecord(key ebpf.BpfFlowId, metrics ebpf.BpfFlowMetrics, currentTime time.Time, ...) *Record
• type RingBufTracer
  • func NewRingBufTracer(reader ringBufReader, flusher mapFlusher, logTimeout time.Duration) *RingBufTracer
  • func (m *RingBufTracer) TraceLoop(ctx context.Context) node.StartFunc[*RawRecord]

Constants

const (
	DirectionIngress = uint8(0)
	DirectionEgress  = uint8(1)
)

Values according to field 61 in https://www.iana.org/assignments/ipfix/ipfix.xhtml

const IPv6Type = 0x86DD

IPv6Type value as defined in IEEE 802: https://www.iana.org/assignments/ieee-802-numbers/ieee-802-numbers.xhtml

const MacLen = 6

Functions

func Accumulate

func Accumulate(r *ebpf.BpfFlowMetrics, src *ebpf.BpfFlowMetrics)

func Decorate

func Decorate(agentIP net.IP, ifaceNamer InterfaceNamer) func(in <-chan []*Record, out chan<- []*Record)

Decorate adds to the flows extra metadata fields that are not directly fetched by eBPF: - The interface name (corresponding to the interface index in the flow). - The IP address of the agent host.

func Dedupe

func Dedupe(expireTime time.Duration, justMark bool) func(in <-chan []*Record, out chan<- []*Record)

Dedupe receives flows and filters these belonging to duplicate interfaces. It will forward the flows from the first interface coming to it, until that flow expires in the cache (no activity for it during the expiration time) The justMark argument tells that the deduper should not drop the duplicate flows but set their Duplicate field.

func IP

func IP(ia IPAddr) net.IP

IP returns the net.IP equivalent object

func IntEncodeV4

func IntEncodeV4(ia [net.IPv6len]uint8) uint32

IntEncodeV4 encodes an IPv4 address as an integer (in network encoding, big endian). It assumes that the passed IP is already IPv4. Otherwise it would just encode the last 4 bytes of an IPv6 address

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. The accounting process is usually done at kernel-space. This type reimplements it at userspace for the edge case where packets are submitted directly via ring-buffer because the kernel-side accounting map is full.

func NewAccounter

func NewAccounter(
	maxEntries int, evictTimeout time.Duration,
	clock func() time.Time,
	monoClock func() 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 *RawRecord, 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 CapacityLimiter

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

CapacityLimiter forwards the flows between two nodes but checks the status of the destination node's buffered channel. If it is already full, it drops the incoming flow and periodically will log a message about the number of lost flows.

func (*CapacityLimiter) Limit

func (c *CapacityLimiter) Limit(in <-chan []*Record, out chan<- []*Record)

type Direction

type Direction uint8

type HumanBytes

type HumanBytes uint64

type IPAddr

type IPAddr [net.IPv6len]uint8

IPAddr encodes v4 and v6 IPs with a fixed length. IPv4 addresses are encoded as IPv6 addresses with prefix ::ffff/96 as described in https://datatracker.ietf.org/doc/html/rfc4038#section-4.2 (same behavior as Go's net.IP type)

func (*IPAddr) MarshalJSON

func (ia *IPAddr) MarshalJSON() ([]byte, error)

type InterfaceNamer

type InterfaceNamer func(ifIndex int) 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 MapTracer

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

MapTracer accesses a mapped source of flows (the eBPF PerCPU HashMap), deserializes it into a flow Record structure, and performs the accumulation of each perCPU-record into a single flow

func NewMapTracer

func NewMapTracer(fetcher mapFetcher, evictionTimeout time.Duration) *MapTracer

func (*MapTracer) Flush

func (m *MapTracer) Flush()

Flush forces reading (and removing) all the flows from the source eBPF map and sending the entries to the next stage in the pipeline

func (*MapTracer) TraceLoop

func (m *MapTracer) TraceLoop(ctx context.Context) node.StartFunc[[]*Record]

type RawRecord

type RawRecord ebpf.BpfFlowRecordT

record structure as parsed from eBPF

func ReadFrom

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

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

type Record

type Record struct {
	RawRecord
	// TODO: redundant field from RecordMetrics. Reorganize structs
	TimeFlowStart time.Time
	TimeFlowEnd   time.Time
	Interface     string
	// Duplicate tells whether this flow has another duplicate so it has to be excluded from
	// any metrics' aggregation (e.g. bytes/second rates between two pods).
	// The reason for this field is that the same flow can be observed from multiple interfaces,
	// so the agent needs to choose only a view of the same flow and mark the others as
	// "exclude from aggregation". Otherwise rates, sums, etc... values would be multiplied by the
	// number of interfaces this flow is observed from.
	Duplicate bool

	// AgentIP provides information about the source of the flow (the Agent that traced it)
	AgentIP net.IP
}

Record contains accumulated metrics from a flow

func NewRecord

func NewRecord(
	key ebpf.BpfFlowId,
	metrics ebpf.BpfFlowMetrics,
	currentTime time.Time,
	monotonicCurrentTime uint64,
) *Record

type RingBufTracer

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

RingBufTracer receives single-packet flows via ringbuffer (usually, these that couldn't be added in the eBPF kernel space due to the map being full or busy) and submits them to the userspace Aggregator map

func NewRingBufTracer

func NewRingBufTracer(
	reader ringBufReader, flusher mapFlusher, logTimeout time.Duration,
) *RingBufTracer

func (*RingBufTracer) TraceLoop

func (m *RingBufTracer) TraceLoop(ctx context.Context) node.StartFunc[*RawRecord]