Skip to content

Commit 7707e11

Browse files
AarC10AarC10
authored
PCAP Capture + Graceful Shutdown (#63)
1 parent bf923f1 commit 7707e11

9 files changed

Lines changed: 230 additions & 35 deletions

File tree

‎.github/workflows/go.yml‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,9 @@ jobs:
1919
with:
2020
go-version: '1.21.3'
2121

22+
- name: Install libpcap-dev
23+
run: sudo apt update && sudo apt-get install -y libpcap-dev
24+
2225
- name: Build
2326
run: go build -v ./...
2427

‎.github/workflows/golangci-lint.yml‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,10 @@ jobs:
1818
- uses: actions/setup-go@v6
1919
with:
2020
go-version: stable
21+
22+
- name: Install libpcap-dev
23+
run: sudo apt update && sudo apt-get install -y libpcap-dev
24+
2125
- name: golangci-lint
2226
uses: golangci/golangci-lint-action@1e7e51e771db61008b38414a730f564565cf7c20
2327
with:

‎cmd/Containerfile‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
FROM golang:1.25-alpine AS build
22

3-
RUN apk add --update --no-cache build-base
3+
RUN apk add --update --no-cache build-base libpcap-dev
44

55
WORKDIR /go/src/app
66

@@ -22,7 +22,7 @@ RUN --mount=type=cache,id=gsw_mod,target=/go/pkg/mod \
2222

2323
RUN --mount=type=cache,id=gsw_mod,target=/go/pkg/mod \
2424
--mount=type=cache,id=gsw_build,target=/root/.cache/go-build \
25-
for binary in grafana_live live_setup mem_view ipc_benchmark telem_view udp_command mqtt_producer; do \
25+
for binary in grafana_live live_setup mem_view ipc_benchmark telem_view udp_command mqtt_producer pkt_cap; do \
2626
CGO_ENABLED=1 GOOS=linux go build \
2727
-trimpath \
2828
-o /build-output/bin/$binary \

‎cmd/gsw_service.go‎

Lines changed: 28 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ import (
77
"fmt"
88
"os"
99
"os/signal"
10+
"sync"
1011
"syscall"
1112

1213
"github.com/AarC10/GSW-V2/lib/db"
@@ -77,14 +78,16 @@ func telemetryConfigInitialize(config *viper.Viper) (func(), error) {
7778
}
7879

7980
// decomInitialize starts decommutation goroutines for each telemetry packet
80-
func decomInitialize(ctx context.Context) map[int]chan []byte {
81+
func decomInitialize(ctx context.Context, wg *sync.WaitGroup) map[int]chan []byte {
8182
channelMap := make(map[int]chan []byte)
8283

8384
for _, packet := range proc.GswConfig.TelemetryPackets {
8485
finalOutputChannel := make(chan []byte)
8586
channelMap[packet.Port] = finalOutputChannel
8687

88+
wg.Add(1)
8789
go func(packet tlm.TelemetryPacket, ch chan []byte) {
90+
defer wg.Done()
8891
err := proc.TelemetryPacketWriter(ctx, packet, finalOutputChannel, *shmDir)
8992
if err != nil && !errors.Is(err, context.Canceled) {
9093
logger.Error("error initializing packet writer", zap.Error(err))
@@ -96,19 +99,19 @@ func decomInitialize(ctx context.Context) map[int]chan []byte {
9699
return channelMap
97100
}
98101

99-
func dbInitialize(ctx context.Context, channelMap map[int]chan []byte, host string, port int) error {
102+
func dbInitialize(ctx context.Context, channelMap map[int]chan []byte, host string, port int, wg *sync.WaitGroup) error {
100103
dbHandler := db.InfluxDBV1Handler{}
101-
err := dbHandler.Initialize(host, port)
102-
if err != nil {
104+
if err := dbHandler.Initialize(host, port); err != nil {
103105
return err
104106
}
105107

106108
for _, packet := range proc.GswConfig.TelemetryPackets {
109+
wg.Add(1)
107110
go func(packet tlm.TelemetryPacket, ch chan []byte) {
108-
proc.DatabaseWriter(&dbHandler, packet, ch)
111+
defer wg.Done()
112+
proc.DatabaseWriter(ctx, &dbHandler, packet, ch)
109113
}(packet, channelMap[packet.Port])
110114
}
111-
112115
return nil
113116
}
114117

@@ -146,49 +149,48 @@ func main() {
146149
flag.Parse()
147150
logger.InitLogger()
148151

149-
// Read gsw_service config
150152
config, profilingPort := readConfig()
151153

152-
ctx, cancel := context.WithCancel(context.Background())
153-
defer cancel()
154-
155154
if profilingPort != 0 {
156155
initProfiling(profilingPort)
157156
}
158157

158+
telemetryConfigCleanup, err := telemetryConfigInitialize(config)
159+
if err != nil {
160+
logger.Fatal("Exiting GSW...")
161+
return
162+
}
163+
defer telemetryConfigCleanup()
164+
165+
ctx, cancel := context.WithCancel(context.Background())
166+
defer cancel()
167+
159168
// Setup signal handling
160169
sigs := make(chan os.Signal, 1)
161170
signal.Notify(sigs, syscall.SIGTERM, syscall.SIGINT)
162-
163171
go func() {
164172
sig := <-sigs
165173
logger.Debug("Received signal", zap.String("signal", sig.String()))
166174
cancel()
167175
}()
168176

169-
telemetryConfigCleanup, err := telemetryConfigInitialize(config)
170-
if err != nil {
171-
logger.Panic("Exiting GSW...")
172-
return
173-
}
174-
defer telemetryConfigCleanup()
177+
var wg sync.WaitGroup
178+
179+
// Start decom writers
180+
channelMap := decomInitialize(ctx, &wg)
175181

176-
channelMap := decomInitialize(ctx)
182+
// Start DB writers
177183
if config.IsSet("database_host_name") && config.IsSet("database_port_number") {
178-
err = dbInitialize(ctx, channelMap, config.GetString("database_host_name"), config.GetInt("database_port_number"))
179-
if err != nil {
184+
if err = dbInitialize(ctx, channelMap, config.GetString("database_host_name"), config.GetInt("database_port_number"), &wg); err != nil {
180185
logger.Warn("DB Initialization failed, telemetry packets will not be published to the database", zap.Error(err))
181186
}
182187
} else {
183188
logger.Warn("database_host_name or database_port_number is not set, telemetry packets will not be published to the database")
184189
}
185190

186-
// Wait for context cancellation or signal handling
191+
// Wait for shutdown signal
187192
<-ctx.Done()
188193
logger.Info("Shutting down GSW...")
189-
190-
for i, channel := range channelMap {
191-
<-channel
192-
logger.Info("channel closed", zap.Int("port", i))
193-
}
194+
wg.Wait()
195+
logger.Info("GSW stopped")
194196
}

‎cmd/pkt_cap/README.md‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
# Prerequisites
2+
libpcap-dev is required to build the project.
3+
4+
### Ubuntu
5+
```bash
6+
sudo apt-get install libpcap-dev
7+
```
8+
9+
### Arch Linux
10+
```bash
11+
sudo pacman -S libpcap
12+
```
13+
14+
### Fedora
15+
```bash
16+
sudo dnf install libpcap-devel
17+
```
18+
19+
### Alpine
20+
```bash
21+
sudo apk add libpcap-dev
22+
```
23+
24+
# Installation
25+
Packet capture requires elevated privileges to access network interfaces. You can either run the program with `sudo` or set the necessary capabilities on the executable.
26+
```bash
27+
go build -o pkt_cap cmd/pkt_cap/main.go
28+
sudo setcap 'cap_net_raw,cap_net_admin=eip' ./pcap
29+
```

‎cmd/pkt_cap/main.go‎

Lines changed: 140 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,140 @@
1+
package main
2+
3+
import (
4+
"bufio"
5+
"context"
6+
"flag"
7+
"fmt"
8+
"os"
9+
"os/signal"
10+
"strings"
11+
"sync"
12+
"syscall"
13+
"time"
14+
15+
"github.com/AarC10/GSW-V2/lib/logger"
16+
"github.com/AarC10/GSW-V2/proc"
17+
"github.com/google/gopacket"
18+
"github.com/google/gopacket/pcap"
19+
"github.com/google/gopacket/pcapgo"
20+
"go.uber.org/zap"
21+
)
22+
23+
var shmDir = flag.String("shm", "/dev/shm", "directory to use for shared memory")
24+
25+
func getFilter() (string, error) {
26+
if len(proc.GswConfig.TelemetryPackets) == 0 {
27+
return "", fmt.Errorf("no telemetry packets configured")
28+
}
29+
30+
ports := make([]string, 0, len(proc.GswConfig.TelemetryPackets))
31+
for _, packet := range proc.GswConfig.TelemetryPackets {
32+
ports = append(ports, fmt.Sprintf("udp port %d", packet.Port))
33+
}
34+
35+
return strings.Join(ports, " or "), nil
36+
}
37+
38+
func createOutputFile() (*os.File, error) {
39+
timestamp := time.Now().Format("2006-01-02_15-04-05")
40+
if err := os.MkdirAll("captures", 0755); err != nil {
41+
return nil, fmt.Errorf("creating captures directory: %w", err)
42+
}
43+
44+
filename := fmt.Sprintf("captures/%s_%s.pcap", proc.GswConfig.Name, timestamp)
45+
return os.Create(filename)
46+
}
47+
48+
func capture(ctx context.Context) error {
49+
filter, err := getFilter()
50+
if err != nil {
51+
return fmt.Errorf("building capture filter: %w", err)
52+
}
53+
54+
snaplen := uint32(65535)
55+
handle, err := pcap.OpenLive("any", int32(snaplen), true, 100*time.Millisecond)
56+
if err != nil {
57+
return fmt.Errorf("opening pcap handle: %w", err)
58+
}
59+
defer handle.Close()
60+
61+
if err := handle.SetBPFFilter(filter); err != nil {
62+
return fmt.Errorf("setting BPF filter: %w", err)
63+
}
64+
65+
pcapFile, err := createOutputFile()
66+
if err != nil {
67+
return fmt.Errorf("creating output file: %w", err)
68+
}
69+
defer func() {
70+
if err := pcapFile.Close(); err != nil {
71+
logger.Error("failed closing pcap file", zap.Error(err))
72+
}
73+
}()
74+
75+
bufferedFile := bufio.NewWriterSize(pcapFile, 128*1024)
76+
defer func() {
77+
if err := bufferedFile.Flush(); err != nil {
78+
logger.Error("failed flushing buffered writer", zap.Error(err))
79+
}
80+
}()
81+
82+
pcapWriter := pcapgo.NewWriterNanos(bufferedFile)
83+
if err := pcapWriter.WriteFileHeader(snaplen, handle.LinkType()); err != nil {
84+
return fmt.Errorf("writing pcap file header: %w", err)
85+
}
86+
87+
logger.Info("network capture started",
88+
zap.String("filter", filter),
89+
zap.String("file", pcapFile.Name()),
90+
)
91+
92+
go func() {
93+
<-ctx.Done()
94+
handle.Close()
95+
}()
96+
97+
packetSource := gopacket.NewPacketSource(handle, handle.LinkType())
98+
for packet := range packetSource.Packets() {
99+
if err := pcapWriter.WritePacket(packet.Metadata().CaptureInfo, packet.Data()); err != nil {
100+
logger.Error("failed writing packet", zap.Error(err))
101+
}
102+
}
103+
104+
return ctx.Err()
105+
}
106+
func main() {
107+
flag.Parse()
108+
logger.InitLogger()
109+
110+
configData, err := proc.ReadTelemetryConfigFromShm(*shmDir)
111+
if err != nil {
112+
logger.Fatal("couldn't read config from gsw", zap.Error(err))
113+
}
114+
if _, err = proc.ParseConfigBytes(configData); err != nil {
115+
logger.Fatal("couldn't parse gsw config", zap.Error(err))
116+
}
117+
118+
ctx, cancel := context.WithCancel(context.Background())
119+
defer cancel()
120+
121+
sigs := make(chan os.Signal, 1)
122+
signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
123+
go func() {
124+
sig := <-sigs
125+
logger.Info("received signal", zap.String("signal", sig.String()))
126+
cancel()
127+
}()
128+
129+
var wg sync.WaitGroup
130+
wg.Add(1)
131+
go func() {
132+
defer wg.Done()
133+
if err := capture(ctx); err != nil {
134+
logger.Error("network capture error", zap.Error(err))
135+
}
136+
}()
137+
138+
wg.Wait()
139+
logger.Info("network capture stopped")
140+
}

‎go.mod‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ toolchain go1.24.9
77
require (
88
github.com/eclipse/paho.mqtt.golang v1.5.1
99
github.com/gdamore/tcell/v2 v2.7.1
10+
github.com/google/gopacket v1.1.19
1011
github.com/gorilla/websocket v1.5.3
1112
github.com/joho/godotenv v1.5.1
1213
github.com/rivo/tview v0.0.0-20250330220935-949945f8d922

‎go.sum‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,8 @@ github.com/gdamore/tcell/v2 v2.7.1 h1:TiCcmpWHiAU7F0rA2I3S2Y4mmLmO9KHxJ7E1QhYzQb
1414
github.com/gdamore/tcell/v2 v2.7.1/go.mod h1:dSXtXTSK0VsW1biw65DZLZ2NKr7j0qP/0J7ONmsraWg=
1515
github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
1616
github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
17+
github.com/google/gopacket v1.1.19 h1:ves8RnFZPGiFnTS0uPQStjwru6uO6h+nlr9j6fL7kF8=
18+
github.com/google/gopacket v1.1.19/go.mod h1:iJ8V8n6KS+z2U1A8pUwu8bW5SyEMkXJB8Yo/Vo+TKTo=
1719
github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
1820
github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
1921
github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4=
@@ -78,11 +80,15 @@ go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN8
7880
go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8=
7981
go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
8082
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
83+
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
8184
golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
8285
golang.org/x/exp v0.0.0-20230905200255-921286631fa9 h1:GoHiUyI/Tp2nVkLI2mCxVkOjsbSXD66ic0XW0js0R9g=
8386
golang.org/x/exp v0.0.0-20230905200255-921286631fa9/go.mod h1:S2oDrQGGwySpoQPVqRShND87VCbxmc6bL1Yd2oYrm6k=
87+
golang.org/x/lint v0.0.0-20200302205851-738671d3881b/go.mod h1:3xt1FjdF8hUf6vQPIChWIBhFzV8gjjsPE/fR3IyQdNY=
88+
golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
8489
golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
8590
golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
91+
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
8692
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
8793
golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
8894
golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
@@ -95,6 +101,7 @@ golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
95101
golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug=
96102
golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI=
97103
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
104+
golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
98105
golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
99106
golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
100107
golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
@@ -118,9 +125,11 @@ golang.org/x/text v0.29.0 h1:1neNs90w9YzJ9BocxfsQNHKuAT4pkghyXc4nhZ6sJvk=
118125
golang.org/x/text v0.29.0/go.mod h1:7MhJOA9CD2qZyOKYazxdYMF85OwPdEr9jTtBpO7ydH4=
119126
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
120127
golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
128+
golang.org/x/tools v0.0.0-20200130002326-2f3ba24bd6e7/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
121129
golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
122130
golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU=
123131
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
132+
golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
124133
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
125134
gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
126135
gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

0 commit comments

Comments
 (0)