Add support for wireguard predown hook

.github/workflows/build.yml

@@ -1,5 +1,11 @@
 name: Build
 on:
+  pull_request:
+    types:
+      - edited
+      - reopened
+      - opened
+      - synchronize
   push:
     branches:
       - main
@@ -59,7 +65,7 @@ jobs:
           path: ./bin/ingest
       - name: Release
         uses: softprops/action-gh-release@v1
-        if: startsWith(github.ref, 'refs/tags/')
+        if: ${{ startsWith(github.ref, 'refs/tags/') && github.event_name != 'pull_request' }}
         with:
           files: ./bin/ingest
       - name: Docker Meta
@@ -94,7 +100,7 @@ jobs:
         id: docker_build_and_push
         uses: docker/build-push-action@v4
         with:
-          push: true
+          push: ${{ github.event_name != 'pull_request' }}
           pull: true
           context: .
           file: Dockerfile

go.mod

@@ -3,6 +3,7 @@ module github.com/xeptore/wireuse
 go 1.20
 
 require (
+	github.com/fsnotify/fsnotify v1.6.0
 	github.com/golang/mock v1.6.0
 	github.com/joho/godotenv v1.5.1
 	github.com/rs/zerolog v1.29.0
@@ -17,6 +18,7 @@ require (
 	github.com/google/go-cmp v0.5.9 // indirect
 	github.com/josharian/native v1.1.0 // indirect
 	github.com/klauspost/compress v1.16.3 // indirect
+	github.com/kr/pretty v0.3.1 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.18 // indirect
 	github.com/mdlayher/genetlink v1.3.1 // indirect
@@ -25,6 +27,7 @@ require (
 	github.com/montanaflynn/stats v0.7.0 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/tidwall/pretty v1.2.1 // indirect
 	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
 	github.com/xdg-go/scram v1.1.2 // indirect
 	github.com/xdg-go/stringprep v1.0.4 // indirect
@@ -34,6 +37,7 @@ require (
 	golang.org/x/sync v0.1.0 // indirect
 	golang.org/x/sys v0.6.0 // indirect
 	golang.org/x/text v0.8.0 // indirect
-	golang.zx2c4.com/wireguard v0.0.0-20230317141804-1417a47c8fa8 // indirect
+	golang.zx2c4.com/wireguard v0.0.0-20230324160507-6f895be10d74 // indirect
+	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -1,7 +1,10 @@
 github.com/coreos/go-systemd/v22 v22.3.3-0.20220203105225-a9a7ef127534/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
+github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/fsnotify/fsnotify v1.6.0 h1:n+5WquG0fcWoWp6xPWfHdbskMCQaFnG6PfBrh1Ky4HY=
+github.com/fsnotify/fsnotify v1.6.0/go.mod h1:sl3t1tCWJFWoRz9R8WJCbQihKKwmorjAbSClcnxKAGw=
 github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
 github.com/golang/mock v1.6.0 h1:ErTB+efbowRARo13NNdxyJji2egdxLGQhRaY+DUumQc=
 github.com/golang/mock v1.6.0/go.mod h1:p6yTPP+5HYm5mzsMV8JkE6ZKdX+/wYM6Hr+LicevLPs=
@@ -18,11 +21,14 @@ github.com/josharian/native v1.1.0/go.mod h1:7X/raswPFr05uY3HiLlYeyQntB6OO7E/d2C
 github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
 github.com/klauspost/compress v1.16.3 h1:XuJt9zzcnaz6a16/OU53ZjWp/v7/42WcR5t2a0PcNQY=
 github.com/klauspost/compress v1.16.3/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
-github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
+github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
+github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
-github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
+github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/mattn/go-colorable v0.1.12/go.mod h1:u5H1YNBxpqRaxsYJYSkiCWKzEfiAb1Gb520KVy5xxl4=
 github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA=
 github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg=
@@ -40,10 +46,13 @@ github.com/mikioh/ipaddr v0.0.0-20190404000644-d465c8ab6721 h1:RlZweED6sbSArvlE9
 github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe/go.mod h1:wL8QJuTMNUDYhXwkmfOly8iTdp5TEcJFWZD2D7SIkUc=
 github.com/montanaflynn/stats v0.7.0 h1:r3y12KyNxj/Sb/iOE46ws+3mS1+MZca1wlHQFPsY/JU=
 github.com/montanaflynn/stats v0.7.0/go.mod h1:etXPPgVO6n31NxCd9KQUMvCM+ve0ruNzt6R8Bnaayow=
+github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/rogpeppe/go-internal v1.9.0 h1:73kH8U+JUqXU8lRuOHeVHaa/SZPifC7BkcraZVejAe8=
+github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
 github.com/rs/xid v1.4.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg=
 github.com/rs/zerolog v1.29.0 h1:Zes4hju04hjbvkVkOhdl2HpZa+0PmVwigmo8XoORE5w=
 github.com/rs/zerolog v1.29.0/go.mod h1:NILgTygv/Uej1ra5XxGf82ZFSLk58MFGAUS2o6usyD0=
@@ -55,8 +64,9 @@ github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/stretchr/testify v1.8.2 h1:+h33VjcLVPDHtOdpUCuF+7gSuG3yGIftsP1YvFihtJ8=
 github.com/stretchr/testify v1.8.2/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
-github.com/tidwall/pretty v1.0.0 h1:HsD+QiTn7sK6flMKIvNmpqz1qrpP3Ps6jOKIKMooyg4=
 github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
+github.com/tidwall/pretty v1.2.1 h1:qjsOFOWWQl+N3RsoF5/ssm1pHmJJwhjlSbZ51I6wMl4=
+github.com/tidwall/pretty v1.2.1/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
 github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c=
 github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
 github.com/xdg-go/scram v1.1.1/go.mod h1:RaEWvsqvNKKvBPvcKeFjrG2cJqOkHTiyTpzz23ni57g=
@@ -106,6 +116,7 @@ golang.org/x/sys v0.0.0-20210927094055-39ccf1dd6fa6/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220908164124-27713097b956/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0 h1:MVltZSvRTcU2ljQOhs94SXPftV6DCNnZViHeQps87pQ=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
@@ -125,13 +136,14 @@ golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8T
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-golang.zx2c4.com/wireguard v0.0.0-20230317141804-1417a47c8fa8 h1:gFnsQLPjrJsKpKJDlqbXWGJLyeFSdnd+8EVpEc+RawM=
-golang.zx2c4.com/wireguard v0.0.0-20230317141804-1417a47c8fa8/go.mod h1:tqur9LnfstdR9ep2LaJT4lFUl0EjlHtge+gAjmsHUG4=
+golang.zx2c4.com/wireguard v0.0.0-20230324160507-6f895be10d74 h1:GYLUo7LqqEYBPNDtEYeQExxaoeV63QN4H0d5PySR1rk=
+golang.zx2c4.com/wireguard v0.0.0-20230324160507-6f895be10d74/go.mod h1:tqur9LnfstdR9ep2LaJT4lFUl0EjlHtge+gAjmsHUG4=
 golang.zx2c4.com/wireguard/wgctrl v0.0.0-20230215201556-9c5414ab4bde h1:ybF7AMzIUikL9x4LgwEmzhXtzRpKNqngme1VGDWz+Nk=
 golang.zx2c4.com/wireguard/wgctrl v0.0.0-20230215201556-9c5414ab4bde/go.mod h1:mQqgjkW8GQQcJQsbBvK890TKqUK1DfKWkuBGbOkuMHQ=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

ingest/Makefile

@@ -1,3 +1,3 @@
 gen:
-	mockgen -source ingest.go -destination mocks/ingest.go -package mocks
+	go run github.com/golang/mock/mockgen@v1.6.0 -source ingest.go -destination mocks/ingest.go -package mocks
 .PHONY: gen

ingest/cmd/main.go

@@ -7,10 +7,12 @@ import (
 	"fmt"
 	"os"
 	"os/signal"
+	"path/filepath"
 	"strings"
 	"syscall"
 	"time"
 
+	"github.com/fsnotify/fsnotify"
 	"github.com/joho/godotenv"
 	"github.com/rs/zerolog"
 	"go.mongodb.org/mongo-driver/bson"
@@ -27,8 +29,8 @@ import (
 )
 
 var (
-	restartMarkFileName string
-	wgDeviceName        string
+	watchDir     string
+	wgDeviceName string
 )
 
 func main() {
@@ -49,15 +51,15 @@ func main() {
 		log.Fatal().Msg("TZ environment variable must be set to UTC")
 	}
 
-	flag.StringVar(&restartMarkFileName, "r", "", "restart-mark file name")
+	flag.StringVar(&watchDir, "w", "", "directory path to watch for wireguard device up and dump files")
 	flag.StringVar(&wgDeviceName, "i", "", "wireguard interface")
 
 	flag.Parse()
 	if nonFlagArgs := flag.Args(); len(nonFlagArgs) > 0 {
 		log.Fatal().Msgf("expected no additional flags, got: %s", strings.Join(nonFlagArgs, ","))
 	}
-	if restartMarkFileName == "" {
-		log.Fatal().Msg("restart-mark file name option is required and cannot be empty")
+	if watchDir == "" {
+		log.Fatal().Msg("directory path to watch for wireguard device up and dump files option is required and cannot be empty")
 	}
 	if wgDeviceName == "" {
 		log.Fatal().Msg("wireguard device name option is required and cannot be empty")
@@ -76,7 +78,9 @@ func main() {
 		log.Fatal().Err(err).Msg("failed to verify database connectivity")
 	}
 	defer func() {
-		if err := client.Disconnect(ctx); err != nil {
+		disconnectCtx, cancel := context.WithTimeout(context.Background(), 4*time.Second)
+		defer cancel()
+		if err := client.Disconnect(disconnectCtx); err != nil {
 			log.Err(err).Msg("failed to disconnect from database")
 			return
 		}
@@ -107,29 +111,72 @@ func main() {
 
 	signals := make(chan os.Signal, 1)
 	signal.Notify(signals, syscall.SIGINT)
-	ctx, cancel := context.WithCancelCause(ctx)
+	runCtx, cancelEngineRun := context.WithCancelCause(ctx)
 	stopSignalErr := errors.New("stop signal received")
 	go func() {
 		<-signals
-		cancel(stopSignalErr)
+		cancelEngineRun(stopSignalErr)
 	}()
 
-	timeTicker := time.NewTicker(5 * time.Second)
-	engineTicker := make(chan struct{})
+	wgUpEvents, wgDownEvents := make(chan ingest.WgUpEvent), make(chan ingest.WgDownEvent)
 	go func() {
+		log = log.With().Str("app", "watcher").Logger()
+		w, err := fsnotify.NewWatcher()
+		if nil != err {
+			log.Fatal().Err(err).Msg("failed to initialize")
+		}
+		if err := w.Add(watchDir); nil != err {
+			log.Fatal().Err(err).Msg("failed to add directory")
+		}
+		log.Debug().Str("dir", watchDir).Msg("started watching directory")
+		for {
+			select {
+			case <-runCtx.Done():
+				log.Info().Msg("existing loop due to context done")
+				return
+			case err, open := <-w.Errors:
+				if !open {
+					log.Info().Msg("exiting loop due to closure")
+					return
+				}
+				if nil != err {
+					log.Error().Err(err).Msg("received error")
+					return
+				}
+			case event, open := <-w.Events:
+				if !open {
+					log.Info().Msg("exiting loop due to closure")
+					return
+				}
+
+				if !event.Has(fsnotify.Create) && !event.Has(fsnotify.Write) {
+					log.Debug().Msg("ignoring event with irrelevant op")
+				}
+
+				if filename := filepath.Base(event.Name); strings.HasSuffix(filename, fmt.Sprintf("%s.up", wgDeviceName)) {
+					wgUpEvents <- ingest.WgUpEvent{}
+				} else if strings.HasSuffix(filename, fmt.Sprintf("%s.down", wgDeviceName)) {
+					wgDownEvents <- ingest.WgDownEvent{ChangedAt: time.Now(), FileName: event.Name}
+				}
+			}
+		}
+	}()
+
+	ticker := make(chan ingest.None)
+	go func() {
+		timeTicker := time.NewTicker(5 * time.Second)
 		for range timeTicker.C {
-			engineTicker <- struct{}{}
+			ticker <- ingest.None{}
 		}
 	}()
 
-	rmf := restartMarkFileReadRemover{}
 	wp := wgPeers{wg}
 	store := storeMongo{collection}
-	engine := ingest.NewEngine(&rmf, &wp, &store, log)
-	if err := engine.Run(ctx, engineTicker, restartMarkFileName); nil != err {
-		if err := ctx.Err(); nil != err {
+	engine := ingest.NewEngine(&wp, &store, log.With().Str("app", "engine").Logger())
+	if err := engine.Run(runCtx, ticker, wgUpEvents, wgDownEvents); nil != err {
+		if err := runCtx.Err(); nil != err {
 			if errors.Is(err, context.Canceled) {
-				if errors.Is(context.Cause(ctx), stopSignalErr) {
+				if errors.Is(context.Cause(runCtx), stopSignalErr) {
 					log.Info().Msg("root context was canceled due to receiving an interrupt signal")
 					return
 				}
@@ -151,12 +198,12 @@ type storeMongo struct {
 	collection *mongo.Collection
 }
 
-func (m *storeMongo) LoadBeforeRestartUsage(ctx context.Context) (map[string]ingest.PeerUsage, error) {
+func (m *storeMongo) LoadUsage(ctx context.Context) (map[string]ingest.PeerUsage, error) {
 	cursor, err := m.collection.Aggregate(ctx, bson.A{
 		bson.M{"$project": bson.M{"lastUsage": bson.M{"$last": "$usage"}, "_id": 0, "publicKey": 1}},
 	})
 	if nil != err {
-		return nil, fmt.Errorf("failed to query before restart last usage data: %v", err)
+		return nil, fmt.Errorf("failed to query last usage data: %v", err)
 	}
 
 	var results []struct {
@@ -210,38 +257,11 @@ func (wg *wgPeers) Usage(ctx context.Context) ([]ingest.PeerUsage, time.Time, er
 
 	out := funcutils.Map(dev.Peers, func(p wgtypes.Peer) ingest.PeerUsage {
 		return ingest.PeerUsage{
-			Upload:    uint(p.TransmitBytes),
-			Download:  uint(p.ReceiveBytes),
+			Upload:    uint(p.ReceiveBytes),
+			Download:  uint(p.TransmitBytes),
 			PublicKey: p.PublicKey.String(),
 		}
 	})
 
 	return out, gatheredAt, nil
 }
-
-type restartMarkFileReadRemover struct{}
-
-func (*restartMarkFileReadRemover) Read(filename string) ([1]byte, error) {
-	file, err := os.Open(restartMarkFileName)
-	if nil != err {
-		return [1]byte{0}, fmt.Errorf("failed to open restart-mark file: %w", err)
-	}
-
-	buf := make([]byte, 1)
-	n, err := file.Read(buf)
-	if nil != err {
-		return [1]byte{0}, fmt.Errorf("failed to read first byte of restart-mark file: %w", err)
-	}
-	if n > 1 {
-		return [1]byte{0}, fmt.Errorf("expected to read at most 1 byte from file read: %d", n)
-	}
-	if n == 0 {
-		return [1]byte{0}, nil
-	}
-
-	return [1]byte{buf[0]}, nil
-}
-
-func (*restartMarkFileReadRemover) Remove(filename string) error {
-	return os.Remove(filename)
-}