add per-interface and cumulative network traffic charts (#926)

Co-authored-by: Sven van Ginkel <svenvanginkel@icloud.com>

agent/deltatracker/deltatracker.go

@@ -0,0 +1,111 @@
+package deltatracker
+
+import (
+	"sync"
+
+	"golang.org/x/exp/constraints"
+)
+
+// Numeric is a constraint that permits any integer or floating-point type.
+type Numeric interface {
+	constraints.Integer | constraints.Float
+}
+
+// DeltaTracker is a generic, thread-safe tracker for calculating differences
+// in numeric values over time.
+// K is the key type (e.g., int, string).
+// V is the value type (e.g., int, int64, float32, float64).
+type DeltaTracker[K comparable, V Numeric] struct {
+	mu       sync.RWMutex
+	current  map[K]V
+	previous map[K]V
+}
+
+// NewDeltaTracker creates a new generic tracker.
+func NewDeltaTracker[K comparable, V Numeric]() *DeltaTracker[K, V] {
+	return &DeltaTracker[K, V]{
+		current:  make(map[K]V),
+		previous: make(map[K]V),
+	}
+}
+
+// Set records the current value for a given ID.
+func (t *DeltaTracker[K, V]) Set(id K, value V) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.current[id] = value
+}
+
+// Deltas returns a map of all calculated deltas for the current interval.
+func (t *DeltaTracker[K, V]) Deltas() map[K]V {
+	t.mu.RLock()
+	defer t.mu.RUnlock()
+
+	deltas := make(map[K]V)
+	for id, currentVal := range t.current {
+		if previousVal, ok := t.previous[id]; ok {
+			deltas[id] = currentVal - previousVal
+		} else {
+			deltas[id] = 0
+		}
+	}
+	return deltas
+}
+
+// Delta returns the delta for a single key.
+// Returns 0 if the key doesn't exist or has no previous value.
+func (t *DeltaTracker[K, V]) Delta(id K) V {
+	t.mu.RLock()
+	defer t.mu.RUnlock()
+
+	currentVal, currentOk := t.current[id]
+	if !currentOk {
+		return 0
+	}
+
+	previousVal, previousOk := t.previous[id]
+	if !previousOk {
+		return 0
+	}
+
+	return currentVal - previousVal
+}
+
+// Cycle prepares the tracker for the next interval.
+func (t *DeltaTracker[K, V]) Cycle() {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.previous = t.current
+	t.current = make(map[K]V)
+}
+
+// 	// --- Example 1: Integer values (unchanged) ---
+// 	fmt.Println("--- 🚀 Example with int64 values (PIDs) ---")
+// 	pidTracker := NewDeltaTracker[int, int64]()
+// 	pidTracker.Set(101, 20000)
+// 	pidTracker.Cycle()
+// 	pidTracker.Set(101, 22500)
+// 	fmt.Println("PID Deltas:", pidTracker.Deltas())
+// 	fmt.Println("----------------------------------------")
+
+// 	// --- Example 2: Float values (New!) ---
+// 	fmt.Println("\n--- 🚀 Example with float64 values (CPU Load) ---")
+// 	// Track the 1-minute load average for different servers.
+// 	loadTracker := NewDeltaTracker[string, float64]()
+
+// 	// Minute 1
+// 	loadTracker.Set("server-alpha", 0.74)
+// 	loadTracker.Set("server-beta", 1.15)
+// 	fmt.Println("Minute 1 Loads:", loadTracker.Deltas())
+// 	loadTracker.Cycle()
+
+// 	// Minute 2
+// 	loadTracker.Set("server-alpha", 0.68) // Load decreased
+// 	loadTracker.Set("server-beta", 1.55)  // Load increased
+// 	loadTracker.Set("server-gamma", 0.25) // New server
+
+// 	minute2Deltas := loadTracker.Deltas()
+// 	fmt.Println("Minute 2 Load Deltas:", minute2Deltas)
+// 	fmt.Printf("Change in alpha's load: %.2f\n", minute2Deltas["server-alpha"])
+// 	fmt.Printf("Change in beta's load:  %.2f\n", minute2Deltas["server-beta"])
+// 	fmt.Println("----------------------------------------")

agent/deltatracker/deltatracker_test.go

@@ -0,0 +1,201 @@
+package deltatracker
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNewDeltaTracker(t *testing.T) {
+	tracker := NewDeltaTracker[string, int]()
+	assert.NotNil(t, tracker)
+	assert.Empty(t, tracker.current)
+	assert.Empty(t, tracker.previous)
+}
+
+func TestSet(t *testing.T) {
+	tracker := NewDeltaTracker[string, int]()
+	tracker.Set("key1", 10)
+
+	tracker.mu.RLock()
+	defer tracker.mu.RUnlock()
+
+	assert.Equal(t, 10, tracker.current["key1"])
+}
+
+func TestDeltas(t *testing.T) {
+	tracker := NewDeltaTracker[string, int]()
+
+	// Test with no previous values
+	tracker.Set("key1", 10)
+	tracker.Set("key2", 20)
+
+	deltas := tracker.Deltas()
+	assert.Equal(t, 0, deltas["key1"])
+	assert.Equal(t, 0, deltas["key2"])
+
+	// Cycle to move current to previous
+	tracker.Cycle()
+
+	// Set new values and check deltas
+	tracker.Set("key1", 15) // Delta should be 5 (15-10)
+	tracker.Set("key2", 25) // Delta should be 5 (25-20)
+	tracker.Set("key3", 30) // New key, delta should be 0
+
+	deltas = tracker.Deltas()
+	assert.Equal(t, 5, deltas["key1"])
+	assert.Equal(t, 5, deltas["key2"])
+	assert.Equal(t, 0, deltas["key3"])
+}
+
+func TestCycle(t *testing.T) {
+	tracker := NewDeltaTracker[string, int]()
+
+	tracker.Set("key1", 10)
+	tracker.Set("key2", 20)
+
+	// Verify current has values
+	tracker.mu.RLock()
+	assert.Equal(t, 10, tracker.current["key1"])
+	assert.Equal(t, 20, tracker.current["key2"])
+	assert.Empty(t, tracker.previous)
+	tracker.mu.RUnlock()
+
+	tracker.Cycle()
+
+	// After cycle, previous should have the old current values
+	// and current should be empty
+	tracker.mu.RLock()
+	assert.Empty(t, tracker.current)
+	assert.Equal(t, 10, tracker.previous["key1"])
+	assert.Equal(t, 20, tracker.previous["key2"])
+	tracker.mu.RUnlock()
+}
+
+func TestCompleteWorkflow(t *testing.T) {
+	tracker := NewDeltaTracker[string, int]()
+
+	// First interval
+	tracker.Set("server1", 100)
+	tracker.Set("server2", 200)
+
+	// Get deltas for first interval (should be zero)
+	firstDeltas := tracker.Deltas()
+	assert.Equal(t, 0, firstDeltas["server1"])
+	assert.Equal(t, 0, firstDeltas["server2"])
+
+	// Cycle to next interval
+	tracker.Cycle()
+
+	// Second interval
+	tracker.Set("server1", 150) // Delta: 50
+	tracker.Set("server2", 180) // Delta: -20
+	tracker.Set("server3", 300) // New server, delta: 300
+
+	secondDeltas := tracker.Deltas()
+	assert.Equal(t, 50, secondDeltas["server1"])
+	assert.Equal(t, -20, secondDeltas["server2"])
+	assert.Equal(t, 0, secondDeltas["server3"])
+}
+
+func TestDeltaTrackerWithDifferentTypes(t *testing.T) {
+	// Test with int64
+	intTracker := NewDeltaTracker[string, int64]()
+	intTracker.Set("pid1", 1000)
+	intTracker.Cycle()
+	intTracker.Set("pid1", 1200)
+	intDeltas := intTracker.Deltas()
+	assert.Equal(t, int64(200), intDeltas["pid1"])
+
+	// Test with float64
+	floatTracker := NewDeltaTracker[string, float64]()
+	floatTracker.Set("cpu1", 1.5)
+	floatTracker.Cycle()
+	floatTracker.Set("cpu1", 2.7)
+	floatDeltas := floatTracker.Deltas()
+	assert.InDelta(t, 1.2, floatDeltas["cpu1"], 0.0001)
+
+	// Test with int keys
+	pidTracker := NewDeltaTracker[int, int64]()
+	pidTracker.Set(101, 20000)
+	pidTracker.Cycle()
+	pidTracker.Set(101, 22500)
+	pidDeltas := pidTracker.Deltas()
+	assert.Equal(t, int64(2500), pidDeltas[101])
+}
+
+func TestDelta(t *testing.T) {
+	tracker := NewDeltaTracker[string, int]()
+
+	// Test getting delta for non-existent key
+	result := tracker.Delta("nonexistent")
+	assert.Equal(t, 0, result)
+
+	// Test getting delta for key with no previous value
+	tracker.Set("key1", 10)
+	result = tracker.Delta("key1")
+	assert.Equal(t, 0, result)
+
+	// Cycle to move current to previous
+	tracker.Cycle()
+
+	// Test getting delta for key with previous value
+	tracker.Set("key1", 15)
+	result = tracker.Delta("key1")
+	assert.Equal(t, 5, result)
+
+	// Test getting delta for key that exists in previous but not current
+	result = tracker.Delta("key1")
+	assert.Equal(t, 5, result) // Should still return 5
+
+	// Test getting delta for key that exists in current but not previous
+	tracker.Set("key2", 20)
+	result = tracker.Delta("key2")
+	assert.Equal(t, 0, result)
+}
+
+func TestDeltaWithDifferentTypes(t *testing.T) {
+	// Test with int64
+	intTracker := NewDeltaTracker[string, int64]()
+	intTracker.Set("pid1", 1000)
+	intTracker.Cycle()
+	intTracker.Set("pid1", 1200)
+	result := intTracker.Delta("pid1")
+	assert.Equal(t, int64(200), result)
+
+	// Test with float64
+	floatTracker := NewDeltaTracker[string, float64]()
+	floatTracker.Set("cpu1", 1.5)
+	floatTracker.Cycle()
+	floatTracker.Set("cpu1", 2.7)
+	floatResult := floatTracker.Delta("cpu1")
+	assert.InDelta(t, 1.2, floatResult, 0.0001)
+
+	// Test with int keys
+	pidTracker := NewDeltaTracker[int, int64]()
+	pidTracker.Set(101, 20000)
+	pidTracker.Cycle()
+	pidTracker.Set(101, 22500)
+	pidResult := pidTracker.Delta(101)
+	assert.Equal(t, int64(2500), pidResult)
+}
+
+func TestDeltaConcurrentAccess(t *testing.T) {
+	tracker := NewDeltaTracker[string, int]()
+
+	// Set initial values
+	tracker.Set("key1", 10)
+	tracker.Set("key2", 20)
+	tracker.Cycle()
+
+	// Set new values
+	tracker.Set("key1", 15)
+	tracker.Set("key2", 25)
+
+	// Test concurrent access safety
+	result1 := tracker.Delta("key1")
+	result2 := tracker.Delta("key2")
+
+	assert.Equal(t, 5, result1)
+	assert.Equal(t, 5, result2)
+}

agent/network.go

@@ -1,13 +1,83 @@
 package agent
 
 import (
+	"fmt"
 	"log/slog"
 	"strings"
 	"time"
 
+	"github.com/henrygd/beszel/internal/entities/system"
 	psutilNet "github.com/shirou/gopsutil/v4/net"
 )
 
+func (a *Agent) updateNetworkStats(systemStats *system.Stats) {
+	// network stats
+	if len(a.netInterfaces) == 0 {
+		// if no network interfaces, initialize again
+		// this is a fix if agent started before network is online (#466)
+		// maybe refactor this in the future to not cache interface names at all so we
+		// don't miss an interface that's been added after agent started in any circumstance
+		a.initializeNetIoStats()
+	}
+
+	if systemStats.NetworkInterfaces == nil {
+		systemStats.NetworkInterfaces = make(map[string][4]uint64, 0)
+	}
+
+	if netIO, err := psutilNet.IOCounters(true); err == nil {
+		msElapsed := uint64(time.Since(a.netIoStats.Time).Milliseconds())
+		a.netIoStats.Time = time.Now()
+		totalBytesSent := uint64(0)
+		totalBytesRecv := uint64(0)
+		netInterfaceDeltaTracker.Cycle()
+		// sum all bytes sent and received
+		for _, v := range netIO {
+			// skip if not in valid network interfaces list
+			if _, exists := a.netInterfaces[v.Name]; !exists {
+				continue
+			}
+			totalBytesSent += v.BytesSent
+			totalBytesRecv += v.BytesRecv
+
+			// track deltas for each network interface
+			netInterfaceDeltaTracker.Set(fmt.Sprintf("%sdown", v.Name), v.BytesRecv)
+			netInterfaceDeltaTracker.Set(fmt.Sprintf("%sup", v.Name), v.BytesSent)
+			upDelta := netInterfaceDeltaTracker.Delta(fmt.Sprintf("%sup", v.Name)) * 1000 / msElapsed
+			downDelta := netInterfaceDeltaTracker.Delta(fmt.Sprintf("%sdown", v.Name)) * 1000 / msElapsed
+			// add interface to systemStats
+			systemStats.NetworkInterfaces[v.Name] = [4]uint64{upDelta, downDelta, v.BytesSent, v.BytesRecv}
+		}
+
+		// add to systemStats
+		var bytesSentPerSecond, bytesRecvPerSecond uint64
+		if msElapsed > 0 {
+			bytesSentPerSecond = (totalBytesSent - a.netIoStats.BytesSent) * 1000 / msElapsed
+			bytesRecvPerSecond = (totalBytesRecv - a.netIoStats.BytesRecv) * 1000 / msElapsed
+		}
+		networkSentPs := bytesToMegabytes(float64(bytesSentPerSecond))
+		networkRecvPs := bytesToMegabytes(float64(bytesRecvPerSecond))
+		// add check for issue (#150) where sent is a massive number
+		if networkSentPs > 10_000 || networkRecvPs > 10_000 {
+			slog.Warn("Invalid net stats. Resetting.", "sent", networkSentPs, "recv", networkRecvPs)
+			for _, v := range netIO {
+				if _, exists := a.netInterfaces[v.Name]; !exists {
+					continue
+				}
+				slog.Info(v.Name, "recv", v.BytesRecv, "sent", v.BytesSent)
+			}
+			// reset network I/O stats
+			a.initializeNetIoStats()
+		} else {
+			systemStats.NetworkSent = networkSentPs
+			systemStats.NetworkRecv = networkRecvPs
+			systemStats.Bandwidth[0], systemStats.Bandwidth[1] = bytesSentPerSecond, bytesRecvPerSecond
+			// update netIoStats
+			a.netIoStats.BytesSent = totalBytesSent
+			a.netIoStats.BytesRecv = totalBytesRecv
+		}
+	}
+}
+
 func (a *Agent) initializeNetIoStats() {
 	// reset valid network interfaces
 	a.netInterfaces = make(map[string]struct{}, 0)

agent/system.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/henrygd/beszel"
 	"github.com/henrygd/beszel/agent/battery"
+	"github.com/henrygd/beszel/agent/deltatracker"
 	"github.com/henrygd/beszel/internal/entities/system"
 
 	"github.com/shirou/gopsutil/v4/cpu"
@@ -18,9 +19,10 @@ import (
 	"github.com/shirou/gopsutil/v4/host"
 	"github.com/shirou/gopsutil/v4/load"
 	"github.com/shirou/gopsutil/v4/mem"
-	psutilNet "github.com/shirou/gopsutil/v4/net"
 )
 
+var netInterfaceDeltaTracker = deltatracker.NewDeltaTracker[string, uint64]()
+
 // Sets initial / non-changing values about the host system
 func (a *Agent) initializeSystemInfo() {
 	a.systemInfo.AgentVersion = beszel.Version
@@ -70,7 +72,7 @@ func (a *Agent) initializeSystemInfo() {
 
 // Returns current info, stats about the host system
 func (a *Agent) getSystemStats() system.Stats {
-	systemStats := system.Stats{}
+	var systemStats system.Stats
 
 	// battery
 	if battery.HasReadableBattery() {
@@ -173,55 +175,7 @@ func (a *Agent) getSystemStats() system.Stats {
 	}
 
 	// network stats
-	if len(a.netInterfaces) == 0 {
-		// if no network interfaces, initialize again
-		// this is a fix if agent started before network is online (#466)
-		// maybe refactor this in the future to not cache interface names at all so we
-		// don't miss an interface that's been added after agent started in any circumstance
-		a.initializeNetIoStats()
-	}
-	if netIO, err := psutilNet.IOCounters(true); err == nil {
-		msElapsed := uint64(time.Since(a.netIoStats.Time).Milliseconds())
-		a.netIoStats.Time = time.Now()
-		totalBytesSent := uint64(0)
-		totalBytesRecv := uint64(0)
-		// sum all bytes sent and received
-		for _, v := range netIO {
-			// skip if not in valid network interfaces list
-			if _, exists := a.netInterfaces[v.Name]; !exists {
-				continue
-			}
-			totalBytesSent += v.BytesSent
-			totalBytesRecv += v.BytesRecv
-		}
-		// add to systemStats
-		var bytesSentPerSecond, bytesRecvPerSecond uint64
-		if msElapsed > 0 {
-			bytesSentPerSecond = (totalBytesSent - a.netIoStats.BytesSent) * 1000 / msElapsed
-			bytesRecvPerSecond = (totalBytesRecv - a.netIoStats.BytesRecv) * 1000 / msElapsed
-		}
-		networkSentPs := bytesToMegabytes(float64(bytesSentPerSecond))
-		networkRecvPs := bytesToMegabytes(float64(bytesRecvPerSecond))
-		// add check for issue (#150) where sent is a massive number
-		if networkSentPs > 10_000 || networkRecvPs > 10_000 {
-			slog.Warn("Invalid net stats. Resetting.", "sent", networkSentPs, "recv", networkRecvPs)
-			for _, v := range netIO {
-				if _, exists := a.netInterfaces[v.Name]; !exists {
-					continue
-				}
-				slog.Info(v.Name, "recv", v.BytesRecv, "sent", v.BytesSent)
-			}
-			// reset network I/O stats
-			a.initializeNetIoStats()
-		} else {
-			systemStats.NetworkSent = networkSentPs
-			systemStats.NetworkRecv = networkRecvPs
-			systemStats.Bandwidth[0], systemStats.Bandwidth[1] = bytesSentPerSecond, bytesRecvPerSecond
-			// update netIoStats
-			a.netIoStats.BytesSent = totalBytesSent
-			a.netIoStats.BytesRecv = totalBytesRecv
-		}
-	}
+	a.updateNetworkStats(&systemStats)
 
 	// temperatures
 	// TODO: maybe refactor to methods on systemStats