Daniel Redetzke
e01b73d677
A new Live page shows current download and upload per peer, updated every 2 seconds, with the last 2 minutes as a chart and a small chart per peer. The server reads the WireGuard counters every 2 seconds and keeps 2 minutes in memory; GET /api/v1/live serves them, with since= for only the newer steps. The dev simulator now adds traffic in proportion to the time between samples.
108 lines
2.4 KiB
Go
108 lines
2.4 KiB
Go
package main
|
|
|
|
import (
|
|
"log/slog"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
// Speeds keeps the last few minutes of each peer's speed in memory for the
|
|
// Live page. It reads the kernel counters every speedStep, apart from the
|
|
// traffic history in Stats, and never writes to disk.
|
|
|
|
const (
|
|
speedStep = 2 * time.Second
|
|
speedPoints = 60 // 2 minutes
|
|
)
|
|
|
|
// SpeedPoint is one step: per peer ID, download and upload in bits per
|
|
// second, from the peer's point of view.
|
|
type SpeedPoint struct {
|
|
T int64 `json:"t"`
|
|
Peers map[string][2]int64 `json:"peers"`
|
|
}
|
|
|
|
type Speeds struct {
|
|
store *Store
|
|
kernel Kernel
|
|
|
|
mu sync.Mutex
|
|
last map[string][2]int64 // raw rx, tx by public key
|
|
lastAt time.Time
|
|
points []SpeedPoint
|
|
}
|
|
|
|
func newSpeeds(store *Store, kernel Kernel) *Speeds {
|
|
return &Speeds{store: store, kernel: kernel, last: map[string][2]int64{}}
|
|
}
|
|
|
|
func (s *Speeds) sample(now time.Time) {
|
|
cfg := s.store.Get()
|
|
samples, err := s.kernel.Sample(cfg.Server.Interface)
|
|
if err != nil {
|
|
slog.Debug("speed sample failed", "err", err)
|
|
return
|
|
}
|
|
idByKey := map[string]string{}
|
|
for _, p := range cfg.Peers {
|
|
if p.hasKey() {
|
|
idByKey[p.PublicKey] = p.ID
|
|
}
|
|
}
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
secs := now.Sub(s.lastAt).Seconds()
|
|
first := s.lastAt.IsZero()
|
|
cur := map[string][2]int64{}
|
|
pt := SpeedPoint{T: now.Unix(), Peers: map[string][2]int64{}}
|
|
for _, smp := range samples {
|
|
cur[smp.PublicKey] = [2]int64{smp.RxBytes, smp.TxBytes}
|
|
id := idByKey[smp.PublicKey]
|
|
prev, ok := s.last[smp.PublicKey]
|
|
if id == "" || !ok || first {
|
|
continue
|
|
}
|
|
dRx, dTx := smp.RxBytes-prev[0], smp.TxBytes-prev[1]
|
|
if dRx < 0 || dTx < 0 { // counters were reset
|
|
continue
|
|
}
|
|
// Tx is what the server sent: the peer's download.
|
|
pt.Peers[id] = [2]int64{int64(float64(dTx*8) / secs), int64(float64(dRx*8) / secs)}
|
|
}
|
|
s.last, s.lastAt = cur, now
|
|
if first {
|
|
return
|
|
}
|
|
s.points = append(s.points, pt)
|
|
if len(s.points) > speedPoints {
|
|
s.points = s.points[len(s.points)-speedPoints:]
|
|
}
|
|
}
|
|
|
|
// Since returns the points newer than the unix time t, oldest first.
|
|
func (s *Speeds) Since(t int64) []SpeedPoint {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
out := []SpeedPoint{}
|
|
for _, p := range s.points {
|
|
if p.T > t {
|
|
out = append(out, p)
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
func (s *Speeds) Run(stop <-chan struct{}) {
|
|
s.sample(time.Now())
|
|
t := time.NewTicker(speedStep)
|
|
defer t.Stop()
|
|
for {
|
|
select {
|
|
case <-stop:
|
|
return
|
|
case now := <-t.C:
|
|
s.sample(now)
|
|
}
|
|
}
|
|
}
|