Open separate captures for each interface and add timeout for opening captures (#188)

Author: orishoshan

Diff for: .github/workflows/golangci-lint.yml

@@ -18,7 +18,7 @@ jobs:
     steps:
       - uses: actions/setup-go@v3
         with:
-          go-version: '1.21'
+          go-version: '1.21.3'
       - uses: actions/checkout@v3
       - name: Install dependencies
         run: sudo apt update && sudo apt install libpcap-dev # required for the linter to be able to lint github.com/google/gopacket

Diff for: src/sniffer/pkg/collectors/dnssniffer.go

@@ -1,6 +1,7 @@
 package collectors
 
 import (
+	"context"
 	"github.com/google/gopacket"
 	"github.com/google/gopacket/layers"
 	"github.com/google/gopacket/pcap"
@@ -9,6 +10,8 @@ import (
 	"github.com/otterize/network-mapper/src/sniffer/pkg/ipresolver"
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/viper"
+	"net"
+	"sync"
 	"time"
 )
 
@@ -39,18 +42,82 @@ func NewDNSSniffer(resolver ipresolver.IPResolver) *DNSSniffer {
 	return &s
 }
 
-func (s *DNSSniffer) CreateDNSPacketStream() (chan gopacket.Packet, error) {
-	handle, err := pcap.OpenLive("any", 0, true, pcap.BlockForever)
+type PacketChannelCombiner struct {
+	Channels     []chan gopacket.Packet
+	combined     chan gopacket.Packet
+	combinedOnce sync.Once
+}
+
+func NewPacketChannelCombiner(channels []chan gopacket.Packet) *PacketChannelCombiner {
+	return &PacketChannelCombiner{
+		Channels: channels,
+	}
+}
+
+func (p *PacketChannelCombiner) Packets() chan gopacket.Packet {
+	p.combinedOnce.Do(func() {
+		p.combined = make(chan gopacket.Packet)
+		for _, c := range p.Channels {
+			go func(channel chan gopacket.Packet) {
+				for packet := range channel {
+					p.combined <- packet
+				}
+			}(c)
+		}
+	})
+	return p.combined
+}
+
+func (s *DNSSniffer) CreatePacketChannelForInterface(iface net.Interface) (result chan gopacket.Packet, err error) {
+	doneCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	go func() {
+		defer cancel()
+		handle, openLiveErr := pcap.OpenLive(iface.Name, 0, true, pcap.BlockForever)
+		if openLiveErr != nil {
+			err = errors.Wrap(openLiveErr)
+			return
+		}
+		bpfErr := handle.SetBPFFilter("udp port 53")
+		if bpfErr != nil {
+			err = errors.Wrap(bpfErr)
+			return
+		}
+
+		packetSource := gopacket.NewPacketSource(handle, handle.LinkType())
+		result = packetSource.Packets()
+		return
+	}()
+	<-doneCtx.Done()
+	if errors.Is(doneCtx.Err(), context.DeadlineExceeded) {
+		return nil, errors.Errorf("timed out starting capture on interface '%s': %w", iface.Name, doneCtx.Err())
+	}
 	if err != nil {
-		return nil, errors.Wrap(err)
+		return nil, errors.Errorf("failed to start capture on interface '%s': %w", iface.Name, err)
 	}
-	err = handle.SetBPFFilter("udp port 53")
+	return result, nil
+}
+
+func (s *DNSSniffer) CreateDNSPacketStream() (chan gopacket.Packet, error) {
+	interfaceList, err := net.Interfaces()
 	if err != nil {
 		return nil, errors.Wrap(err)
 	}
 
-	packetSource := gopacket.NewPacketSource(handle, handle.LinkType())
-	return packetSource.Packets(), nil
+	packetChans := make([]chan gopacket.Packet, 0)
+	for _, iface := range interfaceList {
+		logrus.Debugf("Starting capture on interface '%s'", iface.Name)
+		packetChannel, err := s.CreatePacketChannelForInterface(iface)
+		if err != nil {
+			logrus.WithError(err).Errorf("failed to open packet channel for interface '%s', skipping", iface.Name)
+			continue
+		}
+		packetChans = append(packetChans, packetChannel)
+	}
+	if len(packetChans) == 0 {
+		return nil, errors.New("no captures opened successfully")
+	}
+	return NewPacketChannelCombiner(packetChans).Packets(), nil
 }
 
 func (s *DNSSniffer) HandlePacket(packet gopacket.Packet) {