Something went wrong. Try again.
router flow monitor
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818package export
import ( "bytes" "errors" "fmt" "maps" "net" "net/netip" "sync" "syscall" "time"
"github.com/charmbracelet/log" "github.com/vmware/go-ipfix/pkg/entities" "github.com/vmware/go-ipfix/pkg/registry" "golang.org/x/sys/unix" "ysun.co/rfm/collector" "ysun.co/rfm/config")
var loadIPFIXRegistry sync.Once
var ( ipfixCommonFields = []string{ "sourceTransportPort", "destinationTransportPort", "protocolIdentifier", "ingressInterface", "egressInterface", "flowDirection", "flowStartMilliseconds", "flowEndMilliseconds", "packetDeltaCount", "octetDeltaCount", "flowEndReason", "samplingProbability", } ipfixIPv4Fields = []string{ "sourceIPv4Address", "destinationIPv4Address", } ipfixIPv6Fields = []string{ "sourceIPv6Address", "destinationIPv6Address", })
const ( ipfixIPv4TemplateID uint16 = 256 ipfixIPv6TemplateID uint16 = 257
// ipfixSetHeaderLen is the set id plus set length ipfixSetHeaderLen = 4 // ipfixTemplateHeaderLen is the template id plus field count ipfixTemplateHeaderLen = 4 // ipfixFieldSpecLen is one iana field specifier, id plus length ipfixFieldSpecLen = 4
// ipfixGatherLimit caps how many queued records one gather collects // before the sender packs them, so a burst is split into messages // without waiting for the flush interval ipfixGatherLimit = 512
ipfixDialBackoffMin = time.Second ipfixDialBackoffMax = 30 * time.Second)
// IPFIXStats counts what the exporter did since it was createdtype IPFIXStats struct { // Connected reports whether a socket to the collector is open Connected bool // Dials and DialErrors count socket setup attempts and failures Dials uint64 DialErrors uint64 // Messages and Records count what went out on the wire Messages uint64 Records uint64 // QueueDropped counts records refused because the queue was full QueueDropped uint64 // Unsent counts records dropped because no socket was available, the // records refused after Close among them Unsent uint64 // EncodeErrors counts records that could not be turned into a data record EncodeErrors uint64 // SendFailed counts the records lost with messages that failed to send SendFailed uint64 // SendErrors counts failed sends by errno name, "other" when not an errno SendErrors map[string]uint64}
// Failures counts the lost records, every reason once, so it sums the// dropped record counters// failed dials and failed messages lose nothing beyond the records counted// as unsent and as failed sendsfunc (s IPFIXStats) Failures() uint64 { return s.QueueDropped + s.Unsent + s.EncodeErrors + s.SendFailed}
// IPFIXExporter sends flow records to a single UDP IPFIX collector// ExportFlow only queues a record, a sender goroutine packs queued records// into messages up to the configured size, refreshes templates, and sends// the socket is dialed lazily and re-dialed with backoff after a failure, so// the agent starts and keeps counting while the collector or the local bind// address is unavailable, and every loss is counted in Statstype IPFIXExporter struct { queue chan collector.ExportedFlow flushReq chan chan struct{} stop chan struct{} done chan struct{} closeOnce sync.Once
// mu guards the socket, the endpoints read by isOwnExportFlowLocked, // closed and stats mu sync.Mutex // closed is set by Close before the sender drains the queue for the // last time, ExportFlow queues under mu only while it is clear, so no // record lands in a queue nobody drains closed bool conn *net.UDPConn localAddr netip.Addr localPort uint16 collectorAddr netip.Addr collectorPort uint16 remote *net.UDPAddr localBind *net.UDPAddr nextDial time.Time dialBackoff time.Duration dial func(local, remote *net.UDPAddr) (*net.UDPConn, error) stats IPFIXStats
// sender state, touched by the sender goroutine only observationDomainID uint32 seqNumber uint32 templateRefreshTimeout time.Duration flushInterval time.Duration maxMessageSize int nowFunc func() time.Time buf bytes.Buffer ipv4 ipfixTemplate ipv6 ipfixTemplate}
// ipfixTemplate caches one template with its field lengths and refresh statetype ipfixTemplate struct { id uint16 names []string elements []*entities.InfoElement recordLen int setLen int sentAt time.Time}
// NewIPFIX creates an IPFIX exporter for a single configured collectorfunc NewIPFIX(cfg config.IPFIXConfig) (*IPFIXExporter, error) { cfg = cfg.WithDefaults() if !cfg.Enabled() { return nil, fmt.Errorf("ipfix exporter requires a collector host or port") }
loadIPFIXRegistry.Do(registry.LoadRegistry)
ipv4, err := newIPFIXTemplate(ipfixIPv4TemplateID, ipfixIPv4Fields, ipfixCommonFields) if err != nil { return nil, err } ipv6, err := newIPFIXTemplate(ipfixIPv6TemplateID, ipfixIPv6Fields, ipfixCommonFields) if err != nil { return nil, err }
remote, err := net.ResolveUDPAddr("udp", cfg.Addr()) if err != nil { return nil, err } collectorAddr, ok := netip.AddrFromSlice(remote.IP) if !ok { return nil, fmt.Errorf("collector address %q is not a valid ip address", remote.IP.String()) }
var localBind *net.UDPAddr if cfg.Bind.Enabled() { localBind, err = net.ResolveUDPAddr("udp", cfg.Bind.Addr()) if err != nil { return nil, fmt.Errorf("resolve ipfix bind address %q: %w", cfg.Bind.Addr(), err) } }
e := &IPFIXExporter{ queue: make(chan collector.ExportedFlow, cfg.QueueSize), flushReq: make(chan chan struct{}), stop: make(chan struct{}), done: make(chan struct{}), collectorAddr: collectorAddr.Unmap(), collectorPort: uint16(remote.Port), remote: remote, localBind: localBind, dial: func(local, remote *net.UDPAddr) (*net.UDPConn, error) { return net.DialUDP("udp", local, remote) }, observationDomainID: cfg.ObservationDomainID, templateRefreshTimeout: cfg.TemplateRefresh, flushInterval: cfg.FlushInterval, maxMessageSize: cfg.MaxMessageSize, nowFunc: time.Now, ipv4: ipv4, ipv6: ipv6, } e.stats.SendErrors = make(map[string]uint64)
// a failed first dial is not fatal, the sender retries with backoff so // a bind address that appears after boot still gets its exporter if err := e.ensureConn(time.Now()); err != nil { log.Warn("ipfix collector not reachable yet", "addr", cfg.Addr(), "err", err) }
go e.run() return e, nil}
func ipfixFieldNames(family, common []string) []string { names := make([]string, 0, len(family)+len(common)) names = append(names, family...) names = append(names, common...) return names}
func newIPFIXTemplate(id uint16, family, common []string) (ipfixTemplate, error) { names := ipfixFieldNames(family, common)
t := ipfixTemplate{ id: id, names: names, elements: make([]*entities.InfoElement, len(names)), } for i, name := range names { ie, err := registry.GetInfoElement(name, registry.IANAEnterpriseID) if err != nil { return ipfixTemplate{}, fmt.Errorf("ipfix info element %q: %w", name, err) } t.elements[i] = ie t.recordLen += int(ie.Len) } t.setLen = ipfixSetHeaderLen + ipfixTemplateHeaderLen + len(names)*ipfixFieldSpecLen return t, nil}
// Stats returns a snapshot of the exporter countersfunc (e *IPFIXExporter) Stats() IPFIXStats { e.mu.Lock() defer e.mu.Unlock()
s := e.stats s.Connected = e.conn != nil s.SendErrors = make(map[string]uint64, len(e.stats.SendErrors)) maps.Copy(s.SendErrors, e.stats.SendErrors) return s}
// Close sends what is still queued and closes the socket, later records are// refused and counted as unsentfunc (e *IPFIXExporter) Close() error { e.closeOnce.Do(func() { e.mu.Lock() e.closed = true e.mu.Unlock() close(e.stop) }) <-e.done
e.mu.Lock() defer e.mu.Unlock() if e.conn == nil { return nil } err := e.conn.Close() e.conn = nil return err}
// Flush sends every queued record now and returns once they went outfunc (e *IPFIXExporter) Flush() error { req := make(chan struct{}) select { case e.flushReq <- req: <-req return nil case <-e.done: return errors.New("ipfix exporter closed") }}
// ExportFlow queues a completed flow for export// it returns an error when the queue is full, the exporter is closed or the// record has no packets or an estimate below them, the record is then lost// and countedfunc (e *IPFIXExporter) ExportFlow(flow collector.ExportedFlow) error { e.mu.Lock() defer e.mu.Unlock()
if e.closed { e.stats.Unsent++ return errors.New("ipfix exporter closed") } // the sampling probability of a record is its packets over its estimate, // and the collector scales every sampled packet by a rate of at least 1 if flow.Packets == 0 || flow.EstPackets < flow.Packets { e.stats.EncodeErrors++ return fmt.Errorf("ipfix record of %d packets with an estimate of %d", flow.Packets, flow.EstPackets) } if e.isOwnExportFlowLocked(flow) { return nil }
select { case e.queue <- flow: return nil default: e.stats.QueueDropped++ return errors.New("ipfix queue full") }}
// run is the sender goroutine// records gather for up to flushInterval after the first one arrives, or// until ipfixGatherLimit records are waiting, then go out as messagesfunc (e *IPFIXExporter) run() { defer close(e.done)
var ( pending []collector.ExportedFlow timer *time.Timer timerC <-chan time.Time ) stopTimer := func() { if timer != nil { timer.Stop() timer = nil timerC = nil } }
for { select { case flow := <-e.queue: pending = append(pending, flow) if timer == nil { timer = time.NewTimer(e.flushInterval) timerC = timer.C } if len(pending) >= ipfixGatherLimit { stopTimer() e.sendAll(pending) pending = pending[:0] } case <-timerC: timer = nil timerC = nil e.sendAll(pending) pending = pending[:0] case req := <-e.flushReq: stopTimer() pending = e.drain(pending) e.sendAll(pending) pending = pending[:0] close(req) case <-e.stop: stopTimer() pending = e.drain(pending) e.sendAll(pending) return } }}
// drain moves every record already queued into pending without blockingfunc (e *IPFIXExporter) drain(pending []collector.ExportedFlow) []collector.ExportedFlow { for { select { case flow := <-e.queue: pending = append(pending, flow) default: return pending } }}
// ipfixMessage accumulates records for one datagramtype ipfixMessage struct { ipv4 []collector.ExportedFlow ipv6 []collector.ExportedFlow withIPv4 bool withIPv6 bool}
// sendAll packs pending records into messages no larger than maxMessageSize// and sends them, one message can carry an ipv4 and an ipv6 data setfunc (e *IPFIXExporter) sendAll(pending []collector.ExportedFlow) { if len(pending) == 0 { return }
now := e.nowFunc() if err := e.ensureConn(now); err != nil { e.mu.Lock() e.stats.Unsent += uint64(len(pending)) e.mu.Unlock() return }
// templates that are due go into the first message that carries their // family, so their space is reserved in the budget until they are sent ipv4Due := e.templateDue(&e.ipv4, now) ipv6Due := e.templateDue(&e.ipv6, now)
var msg ipfixMessage for i, flow := range pending { isIPv6, err := flowIsIPv6(flow) if err != nil { e.mu.Lock() e.stats.EncodeErrors++ e.mu.Unlock() continue }
if e.messageLen(msg, isIPv6, ipv4Due, ipv6Due) > e.maxMessageSize && e.messageRecords(msg) > 0 { sent4, sent6 := e.send(msg, now, ipv4Due, ipv6Due) ipv4Due = ipv4Due && !sent4 ipv6Due = ipv6Due && !sent6 msg = ipfixMessage{}
// a failed send that closed the socket leaves the rest of the // batch without one, those records are unsent rather than // failed sends one message at a time e.mu.Lock() if e.conn == nil { e.stats.Unsent += uint64(len(pending) - i) e.mu.Unlock() return } e.mu.Unlock() } if isIPv6 { msg.ipv6 = append(msg.ipv6, flow) msg.withIPv6 = true } else { msg.ipv4 = append(msg.ipv4, flow) msg.withIPv4 = true } } if e.messageRecords(msg) > 0 { e.send(msg, now, ipv4Due, ipv6Due) }}
func (e *IPFIXExporter) messageRecords(msg ipfixMessage) int { return len(msg.ipv4) + len(msg.ipv6)}
// messageLen is the wire size of msg with one more record of the given// family added, templates counted when they are duefunc (e *IPFIXExporter) messageLen(msg ipfixMessage, addIPv6, ipv4Due, ipv6Due bool) int { n4, n6 := len(msg.ipv4), len(msg.ipv6) if addIPv6 { n6++ } else { n4++ } size := entities.MsgHeaderLength if n4 > 0 { size += ipfixSetHeaderLen + n4*e.ipv4.recordLen if ipv4Due { size += e.ipv4.setLen } } if n6 > 0 { size += ipfixSetHeaderLen + n6*e.ipv6.recordLen if ipv6Due { size += e.ipv6.setLen } } return size}
func (e *IPFIXExporter) templateDue(t *ipfixTemplate, now time.Time) bool { return t.sentAt.IsZero() || now.Sub(t.sentAt) >= e.templateRefreshTimeout}
// send encodes and sends one message, it reports which templates went outfunc (e *IPFIXExporter) send(msg ipfixMessage, now time.Time, ipv4Due, ipv6Due bool) (sent4, sent6 bool) { var sets []entities.Set var dataRecords uint32
if len(msg.ipv4) > 0 && ipv4Due { set, err := entities.MakeTemplateSet(e.ipv4.id, e.ipv4.elements) if err != nil { e.countEncodeError(uint64(e.messageRecords(msg))) return false, false } sets = append(sets, set) sent4 = true } if len(msg.ipv6) > 0 && ipv6Due { set, err := entities.MakeTemplateSet(e.ipv6.id, e.ipv6.elements) if err != nil { e.countEncodeError(uint64(e.messageRecords(msg))) return false, false } sets = append(sets, set) sent6 = true } for _, family := range []struct { flows []collector.ExportedFlow tmpl *ipfixTemplate isV6 bool }{{msg.ipv4, &e.ipv4, false}, {msg.ipv6, &e.ipv6, true}} { if len(family.flows) == 0 { continue } set := entities.NewSet(false) if err := set.PrepareSet(entities.Data, family.tmpl.id); err != nil { e.countEncodeError(uint64(len(family.flows))) continue } for _, flow := range family.flows { if err := set.AddRecord(e.dataElements(flow, family.isV6), family.tmpl.id); err != nil { e.countEncodeError(1) continue } dataRecords++ } if set.GetNumberOfRecords() > 0 { sets = append(sets, set) } } if dataRecords == 0 { return false, false }
if err := e.write(now, sets, dataRecords); err != nil { e.countSendError(err, uint64(dataRecords), now) return false, false }
if sent4 { e.ipv4.sentAt = now } if sent6 { e.ipv6.sentAt = now } e.mu.Lock() e.stats.Messages++ e.stats.Records += uint64(dataRecords) e.mu.Unlock() return sent4, sent6}
func (e *IPFIXExporter) write(now time.Time, sets []entities.Set, dataRecords uint32) error { msgLen := entities.MsgHeaderLength for _, set := range sets { set.UpdateLenInHeader() msgLen += set.GetSetLength() } if msgLen > entities.MaxSocketMsgSize { return fmt.Errorf("message size %d exceeds max socket buffer size", msgLen) }
msg := entities.NewMessage(false) msg.SetVersion(10) msg.SetObsDomainID(e.observationDomainID) msg.SetMessageLen(uint16(msgLen)) msg.SetExportTime(uint32(now.Unix())) msg.SetSequenceNum(e.seqNumber)
e.buf.Reset() e.buf.Grow(msgLen) e.buf.Write(msg.GetMsgHeader()) for _, set := range sets { e.buf.Write(set.GetHeaderBuffer()) b := e.buf.AvailableBuffer() for _, record := range set.GetRecords() { var err error b, err = record.AppendToBuffer(b) if err != nil { return err } } e.buf.Write(b) }
e.mu.Lock() conn := e.conn e.mu.Unlock() if conn == nil { return errors.New("ipfix socket not connected") }
written, err := conn.Write(e.buf.Bytes()) if err != nil { return err } if written != msgLen { return fmt.Errorf("short udp write: wrote %d bytes, want %d", written, msgLen) } e.seqNumber += dataRecords return nil}
func (e *IPFIXExporter) countEncodeError(records uint64) { e.mu.Lock() e.stats.EncodeErrors += records e.mu.Unlock()}
// countSendError classifies err by errno and schedules a re-dial when the// socket itself is unusablefunc (e *IPFIXExporter) countSendError(err error, records uint64, now time.Time) { name := errnoName(err)
e.mu.Lock() e.stats.SendErrors[name]++ e.stats.SendFailed += records total := e.stats.SendErrors[name] if socketUnusable(err) && e.conn != nil { _ = e.conn.Close() e.conn = nil e.scheduleDialLocked(now) } e.mu.Unlock()
// the first failure of each kind is worth a log line, later ones are // visible as counters if total == 1 { log.Error("ipfix send", "errno", name, "records", records, "err", err) }}
// errnoName returns the symbolic errno behind err, or "other"func errnoName(err error) string { var errno syscall.Errno if errors.As(err, &errno) { if name := unix.ErrnoName(errno); name != "" { return name } return errno.Error() } return "other"}
// socketUnusable reports whether a send error means the socket must be// re-dialed rather than retried, an icmp error such as ECONNREFUSED or a// firewall verdict such as EPERM keeps the socketfunc socketUnusable(err error) bool { var errno syscall.Errno if !errors.As(err, &errno) { return false } switch errno { case unix.ENETUNREACH, unix.EHOSTUNREACH, unix.ENETDOWN, unix.EADDRNOTAVAIL, unix.EPIPE, unix.ENOTCONN, unix.EBADF, unix.EINVAL: return true } return false}
// ensureConn dials the collector when no socket is open and the backoff// allows another attempt// the dial runs without the lock so a slow dial never stalls ExportFlowfunc (e *IPFIXExporter) ensureConn(now time.Time) error { e.mu.Lock() if e.conn != nil { e.mu.Unlock() return nil } if now.Before(e.nextDial) { e.mu.Unlock() return errors.New("ipfix socket not connected, retry pending") } e.stats.Dials++ dial, localBind, remote := e.dial, e.localBind, e.remote e.mu.Unlock()
conn, err := dial(localBind, remote) if err != nil { e.dialFailed(now) return fmt.Errorf("dial ipfix collector %q: %w", remote.String(), err) }
local, ok := conn.LocalAddr().(*net.UDPAddr) if !ok { _ = conn.Close() e.dialFailed(now) return fmt.Errorf("unexpected local address type %T", conn.LocalAddr()) } localAddr, ok := netip.AddrFromSlice(local.IP) if !ok { _ = conn.Close() e.dialFailed(now) return fmt.Errorf("local address %q is not a valid ip address", local.IP.String()) }
e.mu.Lock() e.conn = conn e.localAddr = localAddr.Unmap() e.localPort = uint16(local.Port) e.dialBackoff = 0 e.nextDial = time.Time{} e.mu.Unlock()
// a new socket is a new transport session, rfc 7011 scopes templates to // it, so they go out again before any data set, the sender goroutine // owns the template state and is the one dialing, or NewIPFIX is before // it started e.ipv4.sentAt = time.Time{} e.ipv6.sentAt = time.Time{} return nil}
func (e *IPFIXExporter) dialFailed(now time.Time) { e.mu.Lock() e.stats.DialErrors++ e.scheduleDialLocked(now) e.mu.Unlock()}
// scheduleDialLocked doubles the backoff up to the cap// it must be called with mu heldfunc (e *IPFIXExporter) scheduleDialLocked(now time.Time) { if e.dialBackoff == 0 { e.dialBackoff = ipfixDialBackoffMin } else { e.dialBackoff = min(e.dialBackoff*2, ipfixDialBackoffMax) } e.nextDial = now.Add(e.dialBackoff)}
func (e *IPFIXExporter) dataElements(flow collector.ExportedFlow, isIPv6 bool) []entities.InfoElementWithValue { t := &e.ipv4 if isIPv6 { t = &e.ipv6 }
srcAddr := netip.AddrFrom16(flow.SrcAddr).Unmap() dstAddr := netip.AddrFrom16(flow.DstAddr).Unmap()
var ingressIf uint32 var egressIf uint32 if flow.Dir == 0 { ingressIf = flow.Ifindex } else { egressIf = flow.Ifindex }
elements := make([]entities.InfoElementWithValue, 0, len(t.names)) for i, name := range t.names { ie := t.elements[i] switch name { case "sourceIPv4Address", "sourceIPv6Address": elements = append(elements, entities.NewIPAddressInfoElement(ie, net.IP(srcAddr.AsSlice()))) case "destinationIPv4Address", "destinationIPv6Address": elements = append(elements, entities.NewIPAddressInfoElement(ie, net.IP(dstAddr.AsSlice()))) case "sourceTransportPort": elements = append(elements, entities.NewUnsigned16InfoElement(ie, flow.SrcPort)) case "destinationTransportPort": elements = append(elements, entities.NewUnsigned16InfoElement(ie, flow.DstPort)) case "protocolIdentifier": elements = append(elements, entities.NewUnsigned8InfoElement(ie, flow.Proto)) case "ingressInterface": elements = append(elements, entities.NewUnsigned32InfoElement(ie, ingressIf)) case "egressInterface": elements = append(elements, entities.NewUnsigned32InfoElement(ie, egressIf)) case "flowDirection": elements = append(elements, entities.NewUnsigned8InfoElement(ie, flow.Dir)) case "flowStartMilliseconds": elements = append(elements, entities.NewDateTimeMillisecondsInfoElement(ie, uint64(flow.Start/int64(time.Millisecond)))) case "flowEndMilliseconds": elements = append(elements, entities.NewDateTimeMillisecondsInfoElement(ie, uint64(flow.End/int64(time.Millisecond)))) case "packetDeltaCount": elements = append(elements, entities.NewUnsigned64InfoElement(ie, flow.Packets)) case "octetDeltaCount": // rfc 7012 counts ip header and payload here, the wire bytes // with the l2 header stay with the prometheus counters elements = append(elements, entities.NewUnsigned64InfoElement(ie, flow.Octets)) case "flowEndReason": elements = append(elements, entities.NewUnsigned8InfoElement(ie, flow.EndReason)) case "samplingProbability": // records carry the share of wire packets they stand for, which // tracks runtime rate changes elements = append(elements, entities.NewFloat64InfoElement(ie, flow.SamplingProbability())) } } return elements}
func flowIsIPv6(flow collector.ExportedFlow) (bool, error) { src := netip.AddrFrom16(flow.SrcAddr).Unmap() dst := netip.AddrFrom16(flow.DstAddr).Unmap() if src.Is4() != dst.Is4() { return false, fmt.Errorf("flow address families do not match") } return src.Is6(), nil}
// isOwnExportFlowLocked reports whether flow is this exporter's own udp// stream to the collector, which must not be exported again// it must be called with mu heldfunc (e *IPFIXExporter) isOwnExportFlowLocked(flow collector.ExportedFlow) bool { if flow.Proto != 17 || e.conn == nil { return false } if flow.SrcPort != e.localPort || flow.DstPort != e.collectorPort { return false } src := netip.AddrFrom16(flow.SrcAddr).Unmap() dst := netip.AddrFrom16(flow.DstAddr).Unmap() return src == e.localAddr && dst == e.collectorAddr}