@@ -451,14 +457,19 @@ onMounted(async () => {
{{peer.Endpoint.Value}} |
- {{ $t('interfaces.peer-connected') }}
+
|
- {{ humanFileSize(peers.Statistics(peer.Identifier).BytesReceived) }} / {{ humanFileSize(peers.Statistics(peer.Identifier).BytesTransmitted) }}
+
+
+ {{ humanFileSize(peers.TrafficStats(peer.Identifier).Received) }}/s
+ {{ humanFileSize(peers.TrafficStats(peer.Identifier).Transmitted) }}/s
+
+
|
diff --git a/go.mod b/go.mod
index 41130f4..8a3ad3e 100644
--- a/go.mod
+++ b/go.mod
@@ -12,6 +12,7 @@ require (
github.com/go-playground/validator/v10 v10.30.1
github.com/go-webauthn/webauthn v0.15.0
github.com/google/uuid v1.6.0
+ github.com/gorilla/websocket v1.5.3
github.com/prometheus-community/pro-bing v0.7.0
github.com/prometheus/client_golang v1.23.2
github.com/stretchr/testify v1.11.1
diff --git a/go.sum b/go.sum
index 93ad84f..c60bc20 100644
--- a/go.sum
+++ b/go.sum
@@ -130,6 +130,8 @@ 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/gorilla/securecookie v1.1.1/go.mod h1:ra0sb63/xPlUeL+yeDciTfxMRAA+MP+HVt/4epWDjd4=
github.com/gorilla/sessions v1.2.1/go.mod h1:dk2InVEVJ0sfLlnXv9EAgkf6ecYs/i80K/zI+bUmuGM=
+github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
+github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
github.com/hashicorp/go-uuid v1.0.2/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
github.com/hashicorp/go-uuid v1.0.3 h1:2gKiV6YVmrJ1i2CKKa9obLvRieoRGviZFL26PcT/Co8=
github.com/hashicorp/go-uuid v1.0.3/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
diff --git a/internal/app/api/core/middleware/logging/writer.go b/internal/app/api/core/middleware/logging/writer.go
index 4e3c42f..d77a7f1 100644
--- a/internal/app/api/core/middleware/logging/writer.go
+++ b/internal/app/api/core/middleware/logging/writer.go
@@ -1,6 +1,8 @@
package logging
import (
+ "bufio"
+ "net"
"net/http"
)
@@ -38,6 +40,12 @@ func (w *writerWrapper) Write(data []byte) (int, error) {
return n, err
}
+// Hijack wraps the Hijack method of the ResponseWriter and returns the hijacked connection.
+// This is required for websockets to work.
+func (w *writerWrapper) Hijack() (net.Conn, *bufio.ReadWriter, error) {
+ return http.NewResponseController(w.ResponseWriter).Hijack()
+}
+
// newWriterWrapper returns a new writerWrapper that wraps the given http.ResponseWriter.
// It initializes the StatusCode to http.StatusOK.
func newWriterWrapper(w http.ResponseWriter) *writerWrapper {
diff --git a/internal/app/api/v0/handlers/endpoint_websocket.go b/internal/app/api/v0/handlers/endpoint_websocket.go
new file mode 100644
index 0000000..5dcc35a
--- /dev/null
+++ b/internal/app/api/v0/handlers/endpoint_websocket.go
@@ -0,0 +1,100 @@
+package handlers
+
+import (
+ "context"
+ "net/http"
+ "strings"
+ "sync"
+
+ "github.com/go-pkgz/routegroup"
+ "github.com/gorilla/websocket"
+
+ "github.com/h44z/wg-portal/internal/app"
+ "github.com/h44z/wg-portal/internal/config"
+ "github.com/h44z/wg-portal/internal/domain"
+)
+
+type WebsocketEventBus interface {
+ Subscribe(topic string, fn any) error
+ Unsubscribe(topic string, fn any) error
+}
+
+type WebsocketEndpoint struct {
+ authenticator Authenticator
+ bus WebsocketEventBus
+
+ upgrader websocket.Upgrader
+}
+
+func NewWebsocketEndpoint(cfg *config.Config, auth Authenticator, bus WebsocketEventBus) *WebsocketEndpoint {
+ return &WebsocketEndpoint{
+ authenticator: auth,
+ bus: bus,
+ upgrader: websocket.Upgrader{
+ ReadBufferSize: 1024,
+ WriteBufferSize: 1024,
+ CheckOrigin: func(r *http.Request) bool {
+ origin := r.Header.Get("Origin")
+ return strings.HasPrefix(origin, cfg.Web.ExternalUrl)
+ },
+ },
+ }
+}
+
+func (e WebsocketEndpoint) GetName() string {
+ return "WebsocketEndpoint"
+}
+
+func (e WebsocketEndpoint) RegisterRoutes(g *routegroup.Bundle) {
+ g.With(e.authenticator.LoggedIn()).HandleFunc("GET /ws", e.handleWebsocket())
+}
+
+// wsMessage represents a message sent over websocket to the frontend
+type wsMessage struct {
+ Type string `json:"type"` // either "peer_stats" or "interface_stats"
+ Data any `json:"data"` // domain.TrafficDelta
+}
+
+func (e WebsocketEndpoint) handleWebsocket() http.HandlerFunc {
+ return func(w http.ResponseWriter, r *http.Request) {
+ conn, err := e.upgrader.Upgrade(w, r, nil)
+ if err != nil {
+ return
+ }
+ defer conn.Close()
+
+ ctx, cancel := context.WithCancel(r.Context())
+ defer cancel()
+
+ writeMutex := sync.Mutex{}
+ writeJSON := func(msg wsMessage) error {
+ writeMutex.Lock()
+ defer writeMutex.Unlock()
+ return conn.WriteJSON(msg)
+ }
+
+ peerStatsHandler := func(status domain.TrafficDelta) {
+ _ = writeJSON(wsMessage{Type: "peer_stats", Data: status})
+ }
+ interfaceStatsHandler := func(status domain.TrafficDelta) {
+ _ = writeJSON(wsMessage{Type: "interface_stats", Data: status})
+ }
+
+ _ = e.bus.Subscribe(app.TopicPeerStatsUpdated, peerStatsHandler)
+ defer e.bus.Unsubscribe(app.TopicPeerStatsUpdated, peerStatsHandler)
+ _ = e.bus.Subscribe(app.TopicInterfaceStatsUpdated, interfaceStatsHandler)
+ defer e.bus.Unsubscribe(app.TopicInterfaceStatsUpdated, interfaceStatsHandler)
+
+ // Keep connection open until client disconnects or context is cancelled
+ go func() {
+ for {
+ if _, _, err := conn.ReadMessage(); err != nil {
+ cancel()
+ return
+ }
+ }
+ }()
+
+ <-ctx.Done()
+ }
+}
diff --git a/internal/app/eventbus.go b/internal/app/eventbus.go
index d411aa6..4e265fe 100644
--- a/internal/app/eventbus.go
+++ b/internal/app/eventbus.go
@@ -26,6 +26,7 @@ const TopicUserEnabled = "user:enabled"
const TopicInterfaceCreated = "interface:created"
const TopicInterfaceUpdated = "interface:updated"
const TopicInterfaceDeleted = "interface:deleted"
+const TopicInterfaceStatsUpdated = "interface:stats:updated"
// endregion interface-events
@@ -37,6 +38,7 @@ const TopicPeerUpdated = "peer:updated"
const TopicPeerInterfaceUpdated = "peer:interface:updated"
const TopicPeerIdentifierUpdated = "peer:identifier:updated"
const TopicPeerStateChanged = "peer:state:changed"
+const TopicPeerStatsUpdated = "peer:stats:updated"
// endregion peer-events
diff --git a/internal/app/wireguard/statistics.go b/internal/app/wireguard/statistics.go
index 78ca6eb..bbc0764 100644
--- a/internal/app/wireguard/statistics.go
+++ b/internal/app/wireguard/statistics.go
@@ -121,15 +121,25 @@ func (c *StatisticsCollector) collectInterfaceData(ctx context.Context) {
"error", err)
continue
}
+ now := time.Now()
err = c.db.UpdateInterfaceStatus(ctx, in.Identifier,
func(i *domain.InterfaceStatus) (*domain.InterfaceStatus, error) {
- i.UpdatedAt = time.Now()
+ td := domain.CalculateTrafficDelta(
+ string(in.Identifier),
+ i.UpdatedAt, now,
+ i.BytesTransmitted, physicalInterface.BytesUpload,
+ i.BytesReceived, physicalInterface.BytesDownload,
+ )
+ i.UpdatedAt = now
i.BytesReceived = physicalInterface.BytesDownload
i.BytesTransmitted = physicalInterface.BytesUpload
// Update prometheus metrics
go c.updateInterfaceMetrics(*i)
+ // Publish stats update event
+ c.bus.Publish(app.TopicInterfaceStatsUpdated, td)
+
return i, nil
})
if err != nil {
@@ -172,6 +182,7 @@ func (c *StatisticsCollector) collectPeerData(ctx context.Context) {
slog.Warn("failed to fetch peers for data collection", "interface", in.Identifier, "error", err)
continue
}
+ now := time.Now()
for _, peer := range peers {
var connectionStateChanged bool
var newPeerStatus domain.PeerStatus
@@ -184,8 +195,15 @@ func (c *StatisticsCollector) collectPeerData(ctx context.Context) {
lastHandshake = &peer.LastHandshake
}
+ td := domain.CalculateTrafficDelta(
+ string(peer.Identifier),
+ p.UpdatedAt, now,
+ p.BytesTransmitted, peer.BytesDownload,
+ p.BytesReceived, peer.BytesUpload,
+ )
+
// calculate if session was restarted
- p.UpdatedAt = time.Now()
+ p.UpdatedAt = now
p.LastSessionStart = getSessionStartTime(*p, peer.BytesUpload, peer.BytesDownload,
lastHandshake)
p.BytesReceived = peer.BytesUpload // store bytes that where uploaded from the peer and received by the server
@@ -195,7 +213,8 @@ func (c *StatisticsCollector) collectPeerData(ctx context.Context) {
p.CalcConnected()
if wasConnected != p.IsConnected {
- slog.Debug("peer connection state changed", "peer", peer.Identifier, "connected", p.IsConnected)
+ slog.Debug("peer connection state changed",
+ "peer", peer.Identifier, "connected", p.IsConnected)
connectionStateChanged = true
newPeerStatus = *p // store new status for event publishing
}
@@ -203,6 +222,9 @@ func (c *StatisticsCollector) collectPeerData(ctx context.Context) {
// Update prometheus metrics
go c.updatePeerMetrics(ctx, *p)
+ // Publish stats update event
+ c.bus.Publish(app.TopicPeerStatsUpdated, td)
+
return p, nil
})
if err != nil {
diff --git a/internal/domain/statistics.go b/internal/domain/statistics.go
index cbc987d..aa205e8 100644
--- a/internal/domain/statistics.go
+++ b/internal/domain/statistics.go
@@ -61,3 +61,25 @@ func (r PingerResult) AverageRtt() time.Duration {
}
return total / time.Duration(len(r.Rtts))
}
+
+type TrafficDelta struct {
+ EntityId string `json:"EntityId"` // Either peerId or interfaceId
+ BytesReceivedPerSecond uint64 `json:"BytesReceived"`
+ BytesTransmittedPerSecond uint64 `json:"BytesTransmitted"`
+}
+
+func CalculateTrafficDelta(id string, oldTime, newTime time.Time, oldTx, newTx, oldRx, newRx uint64) TrafficDelta {
+ timeDiff := uint64(newTime.Sub(oldTime).Seconds())
+ if timeDiff == 0 {
+ return TrafficDelta{
+ EntityId: id,
+ BytesReceivedPerSecond: 0,
+ BytesTransmittedPerSecond: 0,
+ }
+ }
+ return TrafficDelta{
+ EntityId: id,
+ BytesReceivedPerSecond: (newRx - oldRx) / timeDiff,
+ BytesTransmittedPerSecond: (newTx - oldTx) / timeDiff,
+ }
+}
|