Add tests to packet poller (#184)

* Add tests to packet poller

* disable LayersPacket error

* resolve issue 856

* Update tests

---------

Co-authored-by: bogdan.balan1 <[email protected]>
Co-authored-by: Ilya Gavrilov <[email protected]>

Author: 3 people

pkg/bpf/tls_stream.go

@@ -170,7 +170,8 @@ func (t *TlsStream) writeLayers(timestamp uint64, cgroupId uint64, direction uin
 
 		pkt, parseErr := t.poller.layerParser.CreatePacket(bufBytes, cgroupId, unixpacket.PacketDirection(direction), ci, decodeOptions)
 		if parseErr != nil {
-			log.Error().Err(parseErr).Msg("DecodingLayerParser failed")
+			// gopacket.NewPacket is recovers in case of errors, so we can return nil
+			log.Debug().Err(parseErr).Msg("DecodingLayerParser failed")
 			return
 		}
 		t.stats.PacketsGot++

pkg/decodedpacket/decoded_packet.go

@@ -491,3 +491,5 @@ func (p *DecodedPacket) SetErrorLayer(l gopacket.ErrorLayer) {
 		}
 	}
 }
+
+func DetectIPVersion(data []byte) int { return detectIPVersion(data) }

pkg/poller/packets/packets_poller.go

@@ -22,6 +22,12 @@ import (
 	"github.com/rs/zerolog/log"
 )
 
+type perfReader interface {
+	ReadInto(r *perf.Record) error
+	Close() error
+	SetDeadline(t time.Time)
+}
+
 // Buffer pool for pktBuffer objects to avoid large allocations
 var pktBufferPool = sync.Pool{
 	New: func() interface{} {
@@ -118,7 +124,7 @@ type PacketsPoller struct {
 	pktsMaps []map[uint64]*pktBuffer // one map per CPU
 	maxCPUs  int
 	// Original fields
-	chunksReader    *perf.Reader
+	chunksReader    perfReader
 	gopacketWriter  bpf.GopacketWriter
 	rawPacketWriter rawpacket.RawPacketWriter
 	receivedPackets uint64
@@ -342,6 +348,7 @@ func (p *PacketsPoller) handlePktChunk(chunk *pktBuffer) (bool, error) {
 
 func (p *PacketsPoller) writePacket(pktBuf *pktBuffer, ptr *tracerPacketsData) (bool, error) {
 	if p.gopacketWriter == nil {
+		pktBufferPool.Put(pktBuf)
 		return false, nil
 	}
 
@@ -379,9 +386,11 @@ func (p *PacketsPoller) writePacket(pktBuf *pktBuffer, ptr *tracerPacketsData) (
 
 	packet, parseErr := pktBuf.layerParser.CreatePacket(pkt, ptr.CgroupID, unixpacket.PacketDirection(ptr.Direction), ci, decodeOptions)
 	if parseErr != nil {
-		log.Error().Err(parseErr).Msg("DecodingLayerParser failed")
+		log.Debug().Err(parseErr).Msg("DecodingLayerParser failed")
 		p.stats.PacketsError++
-		return false, parseErr
+		pktBufferPool.Put(pktBuf)
+		// gopacket.NewPacket is recovers in case of errors, so we can return nil
+		return false, nil
 	}
 	p.stats.PacketsGot++
 	p.stats.BytesProcessed += uint64(len(pkt))