flow

Documentation

Index

• Constants
• func Decorate(agentIP net.IP, ifaceNamer InterfaceNamer) func(in <-chan []*ebpf.Record, out chan<- []*ebpf.Record)
• func DeduperProvider(dd *Deduper) (pipe.MiddleFunc[[]*ebpf.Record, []*ebpf.Record], error)
• func ProtocolFilterProvider(allowed, excluded []string) pipe.MiddleProvider[[]*ebpf.Record, []*ebpf.Record]
• func ReverseDNSProvider(cfg *ReverseDNS) (pipe.MiddleFunc[[]*ebpf.Record, []*ebpf.Record], error)
• 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) pipe.StartFunc[[]*ebpf.Record]
• type ReverseDNS
  • func (r ReverseDNS) Enabled() bool
• type RingBufTracer
  • func NewRingBufTracer(reader ringBufReader, flusher mapFlusher, logTimeout time.Duration) *RingBufTracer
  • func (m *RingBufTracer) TraceLoop(ctx context.Context) pipe.StartFunc[[]*ebpf.Record]

Constants

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

const (
	ReverseDNSNone        = "none"
	ReverseDNSLocalLookup = "local"
)

Functions

func Decorate

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

Decorate the flows with extra metadata fields that are not directly fetched by eBPF or by any previous pipeline stage (DNS, Kubernetes...): - The interface name (corresponding to the interface index in the flow). - The IP address of the agent host. - If there is no source or destination hostname, the source IP and destination

func DeduperProvider

func DeduperProvider(dd *Deduper) (pipe.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) After passing by the deduper, the ebpf.Record instances loose their IfIndex and Direction fields.

func ProtocolFilterProvider

func ProtocolFilterProvider(allowed, excluded []string) pipe.MiddleProvider[[]*ebpf.Record, []*ebpf.Record]

ProtocolFilterProvider allows selecting which protocols are going to be instrumented. It drops any flow not appearing in the "allowed" list. If the Allowed list is empty, it drops any flow appearing in the "excluded" list.

func ReverseDNSProvider

func ReverseDNSProvider(cfg *ReverseDNS) (pipe.MiddleFunc[[]*ebpf.Record, []*ebpf.Record], error)

Types

type Deduper

type Deduper struct {
	Type       string
	ExpireTime time.Duration
}

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) pipe.StartFunc[[]*ebpf.Record]

type ReverseDNS

type ReverseDNS struct {
	// Type of ReverseDNS. Values are "none" (default) and "local".
	Type string `yaml:"type" env:"BEYLA_NETWORK_REVERSE_DNS_TYPE"`
	// CacheLen specifies the max size of the LRU cache that is checked before
	// performing the name lookup. Default: 256
	CacheLen int `yaml:"cache_len" env:"BEYLA_NETWORK_REVERSE_DNS_CACHE_LEN"`
	// CacheTTL specifies the time-to-live of a cached IP->hostname entry. After the
	// cached entry becomes older than this time, the IP->hostname entry will be looked
	// up again.
	CacheTTL time.Duration `yaml:"cache_expiry" env:"BEYLA_NETWORK_REVERSE_DNS_CACHE_TTL"`
}

ReverseDNS is currently experimental. It is kept disabled by default and will be hidden from the documentation. This means that it does not impact in the overall Beyla performance.

func (ReverseDNS) Enabled

func (r ReverseDNS) Enabled() bool

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) pipe.StartFunc[[]*ebpf.Record]