flow

Documentation

Index

• Constants
• func Decorate(agentIP net.IP, ifaceNamer InterfaceNamer) func(in <-chan []*ebpf.Record, out chan<- []*ebpf.Record)
• func DeduperProvider(dd Deduper) (node.MiddleFunc[[]*ebpf.Record, []*ebpf.Record], error)
• type Accounter
  • func NewAccounter(maxEntries int, evictTimeout time.Duration, clock func() time.Time, ...) *Accounter
  • func (c *Accounter) Account(in <-chan *ebpf.NetFlowRecordT, out chan<- []*ebpf.Record)
• type CapacityLimiter
  • func (c *CapacityLimiter) Limit(in <-chan []*ebpf.Record, out chan<- []*ebpf.Record)
• type Deduper
  • func (d Deduper) Enabled() bool
• type InterfaceNamer
• type MapTracer
  • func NewMapTracer(fetcher mapFetcher, evictionTimeout time.Duration) *MapTracer
  • func (m *MapTracer) Flush()
  • func (m *MapTracer) TraceLoop(ctx context.Context) node.StartFunc[[]*ebpf.Record]
• type RingBufTracer
  • func NewRingBufTracer(reader ringBufReader, flusher mapFlusher, logTimeout time.Duration) *RingBufTracer
  • func (m *RingBufTracer) TraceLoop(ctx context.Context) node.StartFunc[*ebpf.NetFlowRecordT]

Constants

const (
	DeduperNone      = "none"
	DeduperFirstCome = "firstCome"
)

Functions

func Decorate

func Decorate(agentIP net.IP, ifaceNamer InterfaceNamer) func(in <-chan []*ebpf.Record, out chan<- []*ebpf.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 DeduperProvider

func DeduperProvider(dd Deduper) (node.MiddleFunc[[]*ebpf.Record, []*ebpf.Record], error)

DeduperProvider 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.

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 *ebpf.NetFlowRecordT, out chan<- []*ebpf.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 []*ebpf.Record, out chan<- []*ebpf.Record)

type Deduper

type Deduper struct {
	Type       string
	ExpireTime time.Duration
	JustMark   bool
}

func (Deduper) Enabled

func (d Deduper) Enabled() bool

type InterfaceNamer

type InterfaceNamer func(ifIndex int) 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[[]*ebpf.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[*ebpf.NetFlowRecordT]