Skip to content
This repository was archived by the owner on Feb 19, 2025. It is now read-only.

Commit 4e2a3f7

Browse files
lspgnlspgn
committed
Add reuse port to have multiple threads consumming from socket
* Expands reuseport to have multiple threads consuming within GoFlow (it was only possible for multiple processes and would work on sFlow)
1 parent 8b0a2cc commit 4e2a3f7

9 files changed

Lines changed: 96 additions & 76 deletions

File tree

‎cmd/cnetflow/cnetflow.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@ var (
2020

2121
Addr = flag.String("addr", "", "NetFlow/IPFIX listening address")
2222
Port = flag.Int("port", 2055, "NetFlow/IPFIX listening port")
23-
Reuse = flag.Bool("reuse", false, "Enable so_reuseport for NetFlow/IPFIX listening port")
23+
Reuse = flag.Int("reuse", 0, "Enable so_reuseport for NetFlow/IPFIX listening port")
2424

2525
Workers = flag.Int("workers", 1, "Number of NetFlow workers")
2626
LogLevel = flag.String("loglevel", "info", "Log level")
@@ -86,7 +86,7 @@ func main() {
8686
}
8787
log.WithFields(log.Fields{
8888
"Type": "NetFlow"}).
89-
Infof("Listening on UDP %v:%v", *Addr, *Port)
89+
Infof("Listening on UDP %s:%d (reuse: %d, workers: %d)", *Addr, *Port, *Reuse, *Workers)
9090

9191
err := s.FlowRoutine(*Workers, *Addr, *Port, *Reuse)
9292
if err != nil {

‎cmd/cnflegacy/cnflegacy.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@ var (
2020

2121
Addr = flag.String("addr", "", "NetFlow v5 listening address")
2222
Port = flag.Int("port", 2055, "NetFlow v5 listening port")
23-
Reuse = flag.Bool("reuse", false, "Enable so_reuseport for NetFlow v5 listening port")
23+
Reuse = flag.Int("reuse", 0, "Enable so_reuseport for NetFlow v5 listening port")
2424

2525
Workers = flag.Int("workers", 1, "Number of NetFlow v5 workers")
2626
LogLevel = flag.String("loglevel", "info", "Log level")
@@ -84,7 +84,7 @@ func main() {
8484
}
8585
log.WithFields(log.Fields{
8686
"Type": "NetFlowLegacy"}).
87-
Infof("Listening on UDP %v:%v", *Addr, *Port)
87+
Infof("Listening on UDP %s:%d (reuse: %d, workers: %d)", *Addr, *Port, *Reuse, *Workers)
8888

8989
err := s.FlowRoutine(*Workers, *Addr, *Port, *Reuse)
9090
if err != nil {

‎cmd/csflow/csflow.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@ var (
2020

2121
Addr = flag.String("addr", "", "sFlow listening address")
2222
Port = flag.Int("port", 6343, "sFlow listening port")
23-
Reuse = flag.Bool("reuse", false, "Enable so_reuseport for sFlow listening port")
23+
Reuse = flag.Int("reuse", 0, "Enable so_reuseport for sFlow listening port")
2424

2525
Workers = flag.Int("workers", 1, "Number of sFlow workers")
2626
LogLevel = flag.String("loglevel", "info", "Log level")
@@ -84,7 +84,7 @@ func main() {
8484
}
8585
log.WithFields(log.Fields{
8686
"Type": "sFlow"}).
87-
Infof("Listening on UDP %v:%v", *Addr, *Port)
87+
Infof("Listening on UDP %s:%d (reuse: %d, workers: %d)", *Addr, *Port, *Reuse, *Workers)
8888

8989
err := s.FlowRoutine(*Workers, *Addr, *Port, *Reuse)
9090
if err != nil {

‎cmd/goflow/goflow.go‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -22,17 +22,17 @@ var (
2222
SFlowEnable = flag.Bool("sflow", true, "Enable sFlow")
2323
SFlowAddr = flag.String("sflow.addr", "", "sFlow listening address")
2424
SFlowPort = flag.Int("sflow.port", 6343, "sFlow listening port")
25-
SFlowReuse = flag.Bool("sflow.reuserport", false, "Enable so_reuseport for sFlow")
25+
SFlowReuse = flag.Int("sflow.reuse", 0, "Enable so_reuseport for sFlow")
2626

2727
NFLEnable = flag.Bool("nfl", true, "Enable NetFlow v5")
2828
NFLAddr = flag.String("nfl.addr", "", "NetFlow v5 listening address")
2929
NFLPort = flag.Int("nfl.port", 2056, "NetFlow v5 listening port")
30-
NFLReuse = flag.Bool("nfl.reuserport", false, "Enable so_reuseport for NetFlow v5")
30+
NFLReuse = flag.Int("nfl.reuse", 0, "Enable so_reuseport for NetFlow v5")
3131

3232
NFEnable = flag.Bool("nf", true, "Enable NetFlow/IPFIX")
3333
NFAddr = flag.String("nf.addr", "", "NetFlow/IPFIX listening address")
3434
NFPort = flag.Int("nf.port", 2055, "NetFlow/IPFIX listening port")
35-
NFReuse = flag.Bool("nf.reuserport", false, "Enable so_reuseport for NetFlow/IPFIX")
35+
NFReuse = flag.Int("nf.reuse", 0, "Enable so_reuseport for NetFlow/IPFIX")
3636

3737
Workers = flag.Int("workers", 1, "Number of workers per collector")
3838
LogLevel = flag.String("loglevel", "info", "Log level")
@@ -115,7 +115,7 @@ func main() {
115115
go func() {
116116
log.WithFields(log.Fields{
117117
"Type": "sFlow"}).
118-
Infof("Listening on UDP %v:%v", *SFlowAddr, *SFlowPort)
118+
Infof("Listening on UDP %s:%d (reuse: %d, workers: %d)", *SFlowAddr, *SFlowPort, *SFlowReuse, *Workers)
119119

120120
err := sSFlow.FlowRoutine(*Workers, *SFlowAddr, *SFlowPort, *SFlowReuse)
121121
if err != nil {
@@ -129,7 +129,7 @@ func main() {
129129
go func() {
130130
log.WithFields(log.Fields{
131131
"Type": "NetFlow"}).
132-
Infof("Listening on UDP %v:%v", *NFAddr, *NFPort)
132+
Infof("Listening on UDP %s:%d (reuse: %d, workers: %d)", *NFAddr, *NFPort, *NFReuse, *Workers)
133133

134134
err := sNF.FlowRoutine(*Workers, *NFAddr, *NFPort, *NFReuse)
135135
if err != nil {
@@ -143,7 +143,7 @@ func main() {
143143
go func() {
144144
log.WithFields(log.Fields{
145145
"Type": "NetFlowLegacy"}).
146-
Infof("Listening on UDP %v:%v", *NFLAddr, *NFLPort)
146+
Infof("Listening on UDP %s:%d (reuse: %d, workers: %d)", *NFLAddr, *NFLPort, *NFLReuse, *Workers)
147147

148148
err := sNFL.FlowRoutine(*Workers, *NFLAddr, *NFLPort, *NFLReuse)
149149
if err != nil {

‎utils/metrics.go‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -13,22 +13,22 @@ var (
1313
Name: "flow_traffic_bytes",
1414
Help: "Bytes received by the application.",
1515
},
16-
[]string{"remote_ip", "remote_port", "local_ip", "local_port", "type"},
16+
[]string{"remote_ip", "remote_port", "local_ip", "local_port", "type", "lane"},
1717
)
1818
MetricTrafficPackets = prometheus.NewCounterVec(
1919
prometheus.CounterOpts{
2020
Name: "flow_traffic_packets",
2121
Help: "Packets received by the application.",
2222
},
23-
[]string{"remote_ip", "remote_port", "local_ip", "local_port", "type"},
23+
[]string{"remote_ip", "remote_port", "local_ip", "local_port", "type", "lane"},
2424
)
2525
MetricPacketSizeSum = prometheus.NewSummaryVec(
2626
prometheus.SummaryOpts{
2727
Name: "flow_traffic_summary_size_bytes",
2828
Help: "Summary of packet size.",
2929
Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
3030
},
31-
[]string{"remote_ip", "remote_port", "local_ip", "local_port", "type"},
31+
[]string{"remote_ip", "remote_port", "local_ip", "local_port", "type", "lane"},
3232
)
3333
DecoderStats = prometheus.NewCounterVec(
3434
prometheus.CounterOpts{

‎utils/netflow.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -349,7 +349,7 @@ func (s *StateNetFlow) InitTemplates() {
349349
s.samplinglock = &sync.RWMutex{}
350350
}
351351

352-
func (s *StateNetFlow) FlowRoutine(workers int, addr string, port int, reuseport bool) error {
352+
func (s *StateNetFlow) FlowRoutine(workers int, addr string, port int, reuseport int) error {
353353
s.InitTemplates()
354354
return UDPRoutine("NetFlow", s.DecodeFlow, workers, addr, port, reuseport, s.Logger)
355355
}

‎utils/nflegacy.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -84,6 +84,6 @@ func (s *StateNFLegacy) DecodeFlow(msg interface{}) error {
8484
return nil
8585
}
8686

87-
func (s *StateNFLegacy) FlowRoutine(workers int, addr string, port int, reuseport bool) error {
87+
func (s *StateNFLegacy) FlowRoutine(workers int, addr string, port int, reuseport int) error {
8888
return UDPRoutine("NetFlowV5", s.DecodeFlow, workers, addr, port, reuseport, s.Logger)
8989
}

‎utils/sflow.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -137,6 +137,6 @@ func (s *StateSFlow) DecodeFlow(msg interface{}) error {
137137
return nil
138138
}
139139

140-
func (s *StateSFlow) FlowRoutine(workers int, addr string, port int, reuseport bool) error {
140+
func (s *StateSFlow) FlowRoutine(workers int, addr string, port int, reuseport int) error {
141141
return UDPRoutine("sFlow", s.DecodeFlow, workers, addr, port, reuseport, s.Logger)
142142
}

‎utils/utils.go‎

Lines changed: 78 additions & 58 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import (
66
"fmt"
77
"net"
88
"strconv"
9+
"sync"
910
"time"
1011

1112
decoder "github.com/cloudflare/goflow/v3/decoders"
@@ -146,7 +147,7 @@ func FlowMessageToJSON(fmsg *flowmessage.FlowMessage) string {
146147
return s
147148
}
148149

149-
func UDPRoutine(name string, decodeFunc decoder.DecoderFunc, workers int, addr string, port int, sockReuse bool, logger Logger) error {
150+
func UDPRoutine(name string, decodeFunc decoder.DecoderFunc, workers int, addr string, port int, sockReuse int, logger Logger) error {
150151
ecb := DefaultErrorCallback{
151152
Logger: logger,
152153
}
@@ -165,73 +166,92 @@ func UDPRoutine(name string, decodeFunc decoder.DecoderFunc, workers int, addr s
165166
Port: port,
166167
}
167168

168-
var udpconn *net.UDPConn
169-
var err error
170-
171-
if sockReuse {
172-
pconn, err := reuseport.ListenPacket("udp", addrUDP.String())
173-
defer pconn.Close()
174-
if err != nil {
175-
return err
176-
}
177-
var ok bool
178-
udpconn, ok = pconn.(*net.UDPConn)
179-
if !ok {
180-
return err
169+
udpconnL := make([]*net.UDPConn, 0)
170+
if sockReuse > 0 {
171+
for i := 0; i < sockReuse; i++ {
172+
pconn, err := reuseport.ListenPacket("udp", addrUDP.String())
173+
defer pconn.Close()
174+
if err != nil {
175+
return err
176+
}
177+
udpconn, ok := pconn.(*net.UDPConn)
178+
if !ok {
179+
return err
180+
}
181+
udpconnL = append(udpconnL, udpconn)
181182
}
182183
} else {
183-
udpconn, err = net.ListenUDP("udp", &addrUDP)
184-
defer udpconn.Close()
184+
udpconn, err := net.ListenUDP("udp", &addrUDP)
185185
if err != nil {
186186
return err
187187
}
188+
udpconnL = append(udpconnL, udpconn)
188189
}
189190

190-
payload := make([]byte, 9000)
191+
routine := func(lane int, udpconn *net.UDPConn) {
192+
payload := make([]byte, 9000)
193+
localIP := addrUDP.IP.String()
194+
if addrUDP.IP == nil {
195+
localIP = ""
196+
}
191197

192-
localIP := addrUDP.IP.String()
193-
if addrUDP.IP == nil {
194-
localIP = ""
195-
}
198+
for {
199+
size, pktAddr, _ := udpconn.ReadFromUDP(payload)
200+
payloadCut := make([]byte, size)
201+
copy(payloadCut, payload[0:size])
196202

197-
for {
198-
size, pktAddr, _ := udpconn.ReadFromUDP(payload)
199-
payloadCut := make([]byte, size)
200-
copy(payloadCut, payload[0:size])
203+
baseMessage := BaseMessage{
204+
Src: pktAddr.IP,
205+
Port: pktAddr.Port,
206+
Payload: payloadCut,
207+
}
208+
processor.ProcessMessage(baseMessage)
201209

202-
baseMessage := BaseMessage{
203-
Src: pktAddr.IP,
204-
Port: pktAddr.Port,
205-
Payload: payloadCut,
210+
MetricTrafficBytes.With(
211+
prometheus.Labels{
212+
"remote_ip": pktAddr.IP.String(),
213+
"remote_port": strconv.Itoa(pktAddr.Port),
214+
"local_ip": localIP,
215+
"local_port": strconv.Itoa(addrUDP.Port),
216+
"type": name,
217+
"lane": strconv.Itoa(lane),
218+
}).
219+
Add(float64(size))
220+
MetricTrafficPackets.With(
221+
prometheus.Labels{
222+
"remote_ip": pktAddr.IP.String(),
223+
"remote_port": strconv.Itoa(pktAddr.Port),
224+
"local_ip": localIP,
225+
"local_port": strconv.Itoa(addrUDP.Port),
226+
"type": name,
227+
"lane": strconv.Itoa(lane),
228+
}).
229+
Inc()
230+
MetricPacketSizeSum.With(
231+
prometheus.Labels{
232+
"remote_ip": pktAddr.IP.String(),
233+
"remote_port": strconv.Itoa(pktAddr.Port),
234+
"local_ip": localIP,
235+
"local_port": strconv.Itoa(addrUDP.Port),
236+
"type": name,
237+
"lane": strconv.Itoa(lane),
238+
}).
239+
Observe(float64(size))
206240
}
207-
processor.ProcessMessage(baseMessage)
208-
209-
MetricTrafficBytes.With(
210-
prometheus.Labels{
211-
"remote_ip": pktAddr.IP.String(),
212-
"remote_port": strconv.Itoa(pktAddr.Port),
213-
"local_ip": localIP,
214-
"local_port": strconv.Itoa(addrUDP.Port),
215-
"type": name,
216-
}).
217-
Add(float64(size))
218-
MetricTrafficPackets.With(
219-
prometheus.Labels{
220-
"remote_ip": pktAddr.IP.String(),
221-
"remote_port": strconv.Itoa(pktAddr.Port),
222-
"local_ip": localIP,
223-
"local_port": strconv.Itoa(addrUDP.Port),
224-
"type": name,
225-
}).
226-
Inc()
227-
MetricPacketSizeSum.With(
228-
prometheus.Labels{
229-
"remote_ip": pktAddr.IP.String(),
230-
"remote_port": strconv.Itoa(pktAddr.Port),
231-
"local_ip": localIP,
232-
"local_port": strconv.Itoa(addrUDP.Port),
233-
"type": name,
234-
}).
235-
Observe(float64(size))
236241
}
242+
243+
wg := &sync.WaitGroup{}
244+
for i := range udpconnL {
245+
wg.Add(1)
246+
lane := i
247+
udpconn := udpconnL[i]
248+
go func() {
249+
defer wg.Done()
250+
routine(lane, udpconn)
251+
udpconn.Close()
252+
}()
253+
}
254+
255+
wg.Wait()
256+
return nil
237257
}

0 commit comments

Comments
 (0)