add arp_packet plugin (#182)

* add plugin arp stat

* add conf for arp

* remove lock

* fetch device ip

* update var to instance

* refector code

* refector code

* add RWlock

* Update arp_packet.go

* Update arp_packet.go

Co-authored-by: ulricqin <ulricqin@qq.com>

agent/agent.go

@@ -8,6 +8,7 @@ import (
 	"flashcat.cloud/categraf/traces"
 
 	// auto registry
+	_ "flashcat.cloud/categraf/inputs/arp_packet"
 	_ "flashcat.cloud/categraf/inputs/conntrack"
 	_ "flashcat.cloud/categraf/inputs/cpu"
 	_ "flashcat.cloud/categraf/inputs/disk"

conf/input.arp_packet/arp_packet.toml

@@ -0,0 +1,5 @@
+# # collect interval
+# interval = 15
+
+[[instances]]
+#eth_device="ens192"

go.mod

@@ -156,6 +156,7 @@ require (
 	github.com/google/go-cmp v0.5.8 // indirect
 	github.com/google/go-querystring v1.1.0 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
+	github.com/google/gopacket v1.1.19 // indirect
 	github.com/google/uuid v1.3.0 // indirect
 	github.com/googleapis/enterprise-certificate-proxy v0.1.0 // indirect
 	github.com/googleapis/gax-go/v2 v2.4.0 // indirect

go.sum

@@ -523,6 +523,8 @@ github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/
 github.com/google/gofuzz v1.1.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/gofuzz v1.2.0 h1:xRy4A+RhZaiKjJ1bPfwQ8sedCA+YS2YcCHW6ec7JMi0=
 github.com/google/gofuzz v1.2.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
+github.com/google/gopacket v1.1.19 h1:ves8RnFZPGiFnTS0uPQStjwru6uO6h+nlr9j6fL7kF8=
+github.com/google/gopacket v1.1.19/go.mod h1:iJ8V8n6KS+z2U1A8pUwu8bW5SyEMkXJB8Yo/Vo+TKTo=
 github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=
 github.com/google/martian/v3 v3.0.0/go.mod h1:y5Zk1BBys9G+gd6Jrk0W3cC1+ELVxBWuIGO+w/tUAp0=
 github.com/google/martian/v3 v3.1.0/go.mod h1:y5Zk1BBys9G+gd6Jrk0W3cC1+ELVxBWuIGO+w/tUAp0=

inputs/arp_packet/arp_packet.go

@@ -0,0 +1,155 @@
+package arp_packet
+
+import (
+	"errors"
+	"fmt"
+	"log"
+	"net"
+	"sync"
+	"time"
+
+	"flashcat.cloud/categraf/config"
+	"flashcat.cloud/categraf/inputs"
+	"flashcat.cloud/categraf/types"
+	"github.com/google/gopacket"
+	"github.com/google/gopacket/layers"
+	"github.com/google/gopacket/pcap"
+)
+
+const inputName = "arp_packet"
+
+type ArpPacket struct {
+	config.PluginConfig
+	Instances []*Instance `toml:"instances"`
+}
+
+func init() {
+	inputs.Add(inputName, func() inputs.Input {
+		return &ArpPacket{}
+	})
+}
+
+func (r *ArpPacket) GetInstances() []inputs.Instance {
+	ret := make([]inputs.Instance, len(r.Instances))
+	for i := 0; i < len(r.Instances); i++ {
+		ret[i] = r.Instances[i]
+	}
+	return ret
+}
+
+type Instance struct {
+	config.InstanceConfig
+	Ethdevice string `toml:"eth_device"`
+
+	EthHandle    *pcap.Handle
+	LocalIP      string
+	reqARP       uint64
+	resARP       uint64
+	snapshot_len int32
+	promiscuous  bool
+	timeout      time.Duration
+
+	mutex sync.RWMutex
+}
+
+func (ins *Instance) GetInterfaceIpv4Addr(interfaceName string) (addr string, err error) {
+	var (
+		ief      *net.Interface
+		addrs    []net.Addr
+		ipv4Addr net.IP
+	)
+	if ief, err = net.InterfaceByName(interfaceName); err != nil { // get interface
+		return "", err
+	}
+	if addrs, err = ief.Addrs(); err != nil { // get addresses
+		return "", err
+	}
+	for _, addr := range addrs { // get ipv4 address
+		if ipv4Addr = addr.(*net.IPNet).IP.To4(); ipv4Addr != nil {
+			break
+		}
+	}
+	if ipv4Addr == nil {
+		return "", errors.New(fmt.Sprintf("interface %s don't have an ipv4 address", interfaceName))
+	}
+	return ipv4Addr.String(), nil
+}
+func (ins *Instance) Init() error {
+	if len(ins.Ethdevice) == 0 {
+		return types.ErrInstancesEmpty
+	}
+	var err error
+	ins.LocalIP, err = ins.GetInterfaceIpv4Addr(ins.Ethdevice)
+	if err != nil {
+		log.Println("E!", err)
+		return types.ErrInstancesEmpty
+	}
+	ins.snapshot_len = 1024
+	ins.promiscuous = false
+	ins.timeout = 30 * time.Second
+	// Open device
+	ins.EthHandle, err = pcap.OpenLive(ins.Ethdevice, ins.snapshot_len, ins.promiscuous, ins.timeout)
+	if err != nil {
+		log.Println("E!", err)
+		return types.ErrInstancesEmpty
+	}
+	go ins.arpStat()
+	log.Println("I! start arp stat")
+	return nil
+}
+func (ins *Instance) Gather(slist *types.SampleList) {
+	tags := map[string]string{"sourceAddr": ins.LocalIP}
+	fields := make(map[string]interface{})
+	ins.mutex.RLock()
+	fields["request_num"] = ins.reqARP
+	fields["response_num"] = ins.resARP
+	ins.mutex.RUnlock()
+	slist.PushSamples(inputName, fields, tags)
+}
+
+func (ins *Instance) arpStat() {
+	var filter string = "arp"
+	ins.EthHandle.SetBPFFilter(filter)
+
+	defer ins.EthHandle.Close()
+	// Use the handle as a packet source to process all packets
+	packetSource := gopacket.NewPacketSource(ins.EthHandle, ins.EthHandle.LinkType())
+
+	for {
+		select {
+		case p := <-packetSource.Packets():
+			arp := p.Layer(layers.LayerTypeARP).(*layers.ARP)
+			if arp.Operation == 2 {
+				macs := net.HardwareAddr(arp.SourceHwAddress)
+				macd := net.HardwareAddr(arp.DstHwAddress)
+				var sip, dip net.IP
+				sip = arp.SourceProtAddress
+				sourceAddr := sip.String()
+				dip = arp.DstProtAddress
+				if sourceAddr == ins.LocalIP {
+					log.Println("I! ARPResp: SourceProtAddress:", sourceAddr, " mac:", macs)
+					log.Println("I! ARPResp: DstProtAddress:", dip.String(), " mac:", macd)
+					ins.mutex.Lock()
+					ins.resARP++
+					ins.mutex.Unlock()
+
+				}
+
+			} else if arp.Operation == 1 {
+				macs := net.HardwareAddr(arp.SourceHwAddress)
+				macd := net.HardwareAddr(arp.DstHwAddress)
+				var sip, dip net.IP
+				sip = arp.SourceProtAddress
+				sourceAddr := sip.String()
+				dip = arp.DstProtAddress
+				if sourceAddr == ins.LocalIP {
+					log.Println("I! ARPReq: SourceProtAddress:", sourceAddr, " mac:", macs)
+					log.Println("I! ARPReq: DstProtAddress:", dip.String(), " mac:", macd)
+					ins.mutex.Lock()
+					ins.reqARP++
+					ins.mutex.Unlock()
+				}
+			}
+		}
+	}
+}