container_discovery/internal/discoverer/ipv4/port_udp.go

package ipv4

import (
	"encoding/json"
	"fmt"
	"net"
	"strconv"
	"sync"
	"sync/atomic"
	"time"

	"git.loafle.net/commons/logging-go"
	occc "git.loafle.net/overflow/commons-go/core/constants"
	occu "git.loafle.net/overflow/commons-go/core/util"
	ocmd "git.loafle.net/overflow/commons-go/model/discovery"
	"git.loafle.net/overflow/container_discovery/internal/pcap"

	"git.loafle.net/overflow/container_discovery/internal/matcher"
	"github.com/google/gopacket"
	"github.com/google/gopacket/layers"
)

func scanPortUDP(host *ocmd.Host, dp *ocmd.DiscoverPort, resultChan chan interface{}, errChan chan error, stopChan chan struct{}, wg *sync.WaitGroup) {
	defer func() {
		wg.Done()
	}()

	ps, err := pcap.RetainScanner(host.Zone)
	if nil != err {
		errChan <- fmt.Errorf("Discovery: Cannot retain pcap instance %v", err)
		return
	}
	defer func() {
		pcap.ReleaseScanner(host.Zone)
	}()

	udpChan := ps.OpenUDP(host.IPV4)
	defer func() {
		ps.CloseUDP(host.IPV4, udpChan)
	}()

	timerStopped := make(chan struct{})
	go func() {
		ports := make(map[int]*ocmd.Port)

		var delay atomic.Value
		delay.Store(false)
		ticker := time.NewTicker(time.Second * 3)

		for {
			select {
			case packet, ok := <-udpChan:
				if !ok {
					logging.Logger().Debugf("Discovery: udp channel is closed")
					return
				}
				delay.Store(true)
				if p := handlePacketUDP(host, dp, ports, packet); nil != p {
					resultChan <- p
				}
			case <-ticker.C:
				if false == delay.Load().(bool) {
					ticker.Stop()
					timerStopped <- struct{}{}
					return
				}
				delay.Store(false)
			case <-stopChan:
				return
			}
		}
	}()

	if err := sendUDP(host, dp, stopChan); nil != err {
		errChan <- err
		return
	}

	select {
	case <-stopChan:
		return
	case <-timerStopped:
		return
	}
}

func sendUDP(host *ocmd.Host, dp *ocmd.DiscoverPort, stopChan chan struct{}) error {
	ip := net.ParseIP(host.IPV4)
	if nil == ip {
		return fmt.Errorf("Discovery: IP(%s) of host is not valid", host.IPV4)
	}

	ms := matcher.GetUDPMatchers()

	conn, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: 0})
	if err != nil {
		return err
	}
	defer func() {
		conn.Close()
	}()

	for indexI := 0; indexI < len(ms); indexI++ {
		m := ms[indexI]

	Loop:
		for portNumber := dp.FirstScanRange; portNumber < dp.LastScanRange; portNumber++ {
			if nil != dp.ExcludePorts {
				for _, exPortNumber := range dp.ExcludePorts {
					if portNumber == exPortNumber {
						continue Loop
					}
				}
			}

			if !m.IsSend(portNumber) {
				continue
			}

			addr := &net.UDPAddr{IP: ip.To4(), Port: portNumber}
			for i := 0; i < m.PacketCount(); i++ {
				p := m.Packet(i)
				if _, err := conn.WriteToUDP(p.Buffer, addr); err != nil {
					logging.Logger().Errorf("Discovery: UDP write error %v", err)
				}
			}

			timer := time.NewTimer(time.Microsecond * 200)

			select {
			case <-stopChan:
				return nil
			case <-timer.C:
			}
		}
	}

	return nil
}

func handlePacketUDP(host *ocmd.Host, dp *ocmd.DiscoverPort, ports map[int]*ocmd.Port, packet gopacket.Packet) *ocmd.Port {
	ipLayer := packet.Layer(layers.LayerTypeIPv4)

	if ipLayer.(*layers.IPv4).SrcIP.String() == host.Zone.IPV4 {
		return nil
	}

	if net := packet.NetworkLayer(); net == nil {
	} else if udpLayer := packet.Layer(layers.LayerTypeUDP); udpLayer == nil {
	} else if udp, ok := udpLayer.(*layers.UDP); ok {

		srcIP := ipLayer.(*layers.IPv4).SrcIP
		port := int(udp.SrcPort)
		logging.Logger().Debugf("Discovery: IP of UDP(%d) src %v", port, srcIP)
		if _, ok := ports[port]; ok || !dp.Contains(port) {
			return nil
		}

		p := &ocmd.Port{
			PortType:       occc.PortTypeUDP,
			PortNumber:     json.Number(strconv.Itoa(port)),
			UDPLayer:       udpLayer,
			DiscoveredDate: occu.NowPtr(),
		}
		p.Host = host
		ports[port] = p

		return p
	}

	return nil
}