redis, latency processor work, minor main.go cleanup, remove Greet, ScoredOffer.Latency

2025-12-09 20:09:02 -06:00 · 2024-12-16 22:57:14 -06:00
parent 0659047a9b
commit 6660538ff9
13 changed files with 193 additions and 58 deletions
--- a/api/go.mod
+++ b/api/go.mod
@@ -2,11 +2,16 @@ module xevion.dev/vastly/api

 go 1.23.3

-require go.uber.org/zap v1.27.0
+require (
+	github.com/prometheus-community/pro-bing v0.5.0
+	go.uber.org/zap v1.27.0
+)

 require (
+	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/google/uuid v1.6.0 // indirect
-	github.com/prometheus-community/pro-bing v0.5.0 // indirect
+	github.com/redis/go-redis/v9 v9.7.0 // indirect
 	go.uber.org/multierr v1.10.0 // indirect
 	golang.org/x/net v0.32.0 // indirect
 	golang.org/x/sync v0.10.0 // indirect
--- a/api/go.sum
+++ b/api/go.sum
@@ -1,7 +1,21 @@
+github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
+github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+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/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+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/prometheus-community/pro-bing v0.5.0 h1:Fq+4BUXKIvsPtXUY8K+04ud9dkAuFozqGmRAyNUpffY=
 github.com/prometheus-community/pro-bing v0.5.0/go.mod h1:1joR9oXdMEAcAJJvhs+8vNDvTg5thfAZcRFhcUozG2g=
+github.com/redis/go-redis/v9 v9.7.0 h1:HhLSs+B6O021gwzl+locl0zEDnyNkxMtf/Z3NNBMa9E=
+github.com/redis/go-redis/v9 v9.7.0/go.mod h1:f6zhXITC7JUJIlPEiBOTXxJgPLdZcA93GewI7inzyWw=
+github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
+github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
+go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
+go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 go.uber.org/multierr v1.10.0 h1:S0h4aNzvfcFsC3dRF1jLoaov7oRaKqRGC/pUEJ2yvPQ=
 go.uber.org/multierr v1.10.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
 go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8=
@@ -12,3 +26,5 @@ golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ=
 golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA=
 golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
--- a/api/latency.go
+++ b/api/latency.go
@@ -1,18 +1,105 @@
 package api

-import probing "github.com/prometheus-community/pro-bing"
+import (
+	"errors"
+	"net"
+	"time"
+
+	probing "github.com/prometheus-community/pro-bing"
+	"github.com/redis/go-redis/v9"
+	"go.uber.org/zap"
+)
+
+type LatencyRequest struct {
+	RequestTime int64
+	Ip          net.IPAddr
+}
+
+type LatencyQueue struct {
+	processChannel chan LatencyRequest
+	stopChannel    chan bool
+	logger         *zap.SugaredLogger
+	redis          *redis.Client
+	pinger         *probing.Pinger
+	handlerChannel chan<- PingResult
+}
+
+func NewLatencyQueue() *LatencyQueue {
+	logger, _ := zap.NewDevelopment()
+	pinger, err := probing.NewPinger("127.0.0.1")
+	if err != nil {
+		logger.Fatal("Failed to create pinger")
+	}
+	pinger.Count = 1
+	pinger.Interval = time.Millisecond * 850
+	return &LatencyQueue{
+		processChannel: make(chan LatencyRequest, 1024),
+		logger:         logger.Sugar(),
+		redis:          redis.NewClient(&redis.Options{}),
+		pinger:         pinger,
+	}
+}
+
+type PingRequest struct {
+	Ip net.IPAddr
+}

 type PingResult struct {
-	Ip      string
-	Latency float64
+	Ip      net.IPAddr
+	Latency int64
 }

-func GetLatency(ip string) (float64, error) {
-	pinger, err := probing.NewPinger(ip)
-	if err != nil {
-		return 0, err
+func (l *LatencyQueue) QueuePing(ip string) error {
+	// Parse the IP
+	parsedIp := net.ParseIP(ip)
+	if parsedIp == nil {
+		return errors.New("Invalid IP address")
 	}

-	pinger
-	return 0, nil
+	// Create the request
+	request := LatencyRequest{
+		RequestTime: time.Now().Unix(),
+		Ip:          net.IPAddr{IP: parsedIp},
+	}
+
+	// Add the request to the queue
+	l.processChannel <- request
+
+	return nil
+}
+
+func (l *LatencyQueue) Start() {
+	for {
+		select {
+		case request := <-l.processChannel:
+
+			ip := request.Ip.String()
+			l.pinger.SetIPAddr(&request.Ip)
+
+			// Process the request
+			err := l.pinger.Run()
+			if err != nil {
+				l.logger.Errorf("Failed to ping %s: %s", ip, err)
+				continue
+			}
+
+			// Get the results
+			results := l.pinger.Statistics()
+			if (l.handlerChannel) != nil {
+				l.handlerChannel <- PingResult{
+					Ip:      request.Ip,
+					Latency: results.AvgRtt.Milliseconds(),
+				}
+			}
+
+		case <-l.stopChannel:
+			return
+		}
+	}
+}
+
+func (l *LatencyQueue) Kill() error {
+	l.stopChannel <- true
+	err := l.redis.Close()
+	return err
 }
--- a/api/score.go
+++ b/api/score.go
@@ -18,6 +18,7 @@ type ScoredOffer struct {
 	Offer   Offer
 	Score   float64
 	Reasons []ScoreReason
+	Latency *int32
 }

 var (