/
pcap.go
91 lines (76 loc) · 2.09 KB
/
pcap.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
package pcap
import (
"os"
"os/signal"
"syscall"
"time"
"github.com/metlo-labs/metlo/ingestors/govxlan/utils"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
)
func init() {
utils.Log.SetLevel(logrus.FatalLevel)
}
// VXCap is one of main components of the package
type Pcap struct {
QueueSize int
captureInterface string
}
const (
// DefaultReceiverQueueSize is default queue size of channel from packet server to packet processor.
DefaultReceiverQueueSize = 1024
)
// New is constructor of VXCap
func New(captureInterface string) *Pcap {
cap := Pcap{
QueueSize: DefaultReceiverQueueSize,
captureInterface: captureInterface,
}
return &cap
}
// Start invokes packet listener and forwards captured packets to processor.
func (x *Pcap) Start(proc Processor) error {
utils.Log.Trace("Setting up processor...")
if err := proc.Setup(); err != nil {
return err
}
// Setup channels
utils.Log.WithFields(logrus.Fields{
"queueSize": x.QueueSize,
"captureInterface": x.captureInterface,
}).Trace("Starting listening on interface...")
queueCh := listenPCAP(x.QueueSize, x.captureInterface)
utils.Log.Trace("Setting up channels")
ticker := time.NewTicker(time.Second)
defer ticker.Stop()
tickerCh := ticker.C
signalCh := make(chan os.Signal, 1)
signal.Notify(signalCh, syscall.SIGTERM)
signal.Notify(signalCh, syscall.SIGINT)
defer signal.Stop(signalCh)
utils.Log.Infof("Starting loop, listening on interface %s", x.captureInterface)
MainLoop:
for {
select {
case q := <-queueCh:
if q.Err != nil {
return errors.Wrap(q.Err, "Failed to receive packet data")
}
if err := proc.Put(q.Pkt); err != nil {
return errors.Wrap(err, "Fail to handle packet")
}
case t := <-tickerCh:
if err := proc.Tick(t); err != nil {
return errors.Wrap(err, "Fail in tick process")
}
case s := <-signalCh:
utils.Log.WithField("signal", s).Warn("Caught signal, Shutting down...")
if err := proc.Shutdown(); err != nil {
return errors.Wrap(err, "Fail in shutdown process")
}
break MainLoop
}
}
utils.Log.Info("Exit normally")
return nil
}