feat(core): add network throughput collector

src/Vantage.Core/Collectors/Network/INetworkProvider.cs

@@ -0,0 +1,8 @@
+using Vantage.Core.Models;
+
+namespace Vantage.Core.Collectors.Network;
+
+public interface INetworkProvider
+{
+    Task<NetworkSnapshot> GetSnapshotAsync(HostConfig host, CancellationToken cancellationToken);
+}

src/Vantage.Core/Collectors/Network/InterfaceCounters.cs

@@ -0,0 +1,25 @@
+namespace Vantage.Core.Collectors.Network;
+
+public readonly record struct InterfaceCounters(
+    string Name,
+    long RxBytes,
+    long TxBytes,
+    long RxPackets,
+    long TxPackets)
+{
+    public bool IsLoopback => Name == "lo";
+}
+
+public sealed record NetworkSnapshot(DateTimeOffset CollectedAt, IReadOnlyList<InterfaceCounters> Interfaces)
+{
+    public InterfaceCounters? Find(string name)
+    {
+        foreach (var counters in Interfaces)
+            if (counters.Name == name)
+                return counters;
+        return null;
+    }
+
+    public long TotalRxBytes => Interfaces.Where(i => !i.IsLoopback).Sum(i => i.RxBytes);
+    public long TotalTxBytes => Interfaces.Where(i => !i.IsLoopback).Sum(i => i.TxBytes);
+}

src/Vantage.Core/Collectors/Network/MockNetworkProvider.cs

@@ -0,0 +1,54 @@
+using System.Collections.Concurrent;
+using Vantage.Core.Models;
+
+namespace Vantage.Core.Collectors.Network;
+
+public sealed class MockNetworkProvider : INetworkProvider
+{
+    private sealed class Accumulator
+    {
+        public long Rx;
+        public long Tx;
+        public long RxPackets;
+        public long TxPackets;
+        public DateTimeOffset LastAt;
+        public Random Jitter = new();
+    }
+
+    private readonly ConcurrentDictionary<string, Accumulator> state = new();
+    private readonly Func<DateTimeOffset> clock;
+
+    public MockNetworkProvider(Func<DateTimeOffset>? clock = null)
+    {
+        this.clock = clock ?? (() => DateTimeOffset.UtcNow);
+    }
+
+    public Task<NetworkSnapshot> GetSnapshotAsync(HostConfig host, CancellationToken cancellationToken)
+    {
+        var now = clock();
+        var accumulator = state.GetOrAdd(host.Id, _ => new Accumulator
+        {
+            Rx = 4_000_000_000,
+            Tx = 900_000_000,
+            LastAt = now
+        });
+
+        var seconds = Math.Max(0, (now - accumulator.LastAt).TotalSeconds);
+        var rxRate = 200_000 + accumulator.Jitter.NextDouble() * 2_000_000;
+        var txRate = 80_000 + accumulator.Jitter.NextDouble() * 500_000;
+
+        accumulator.Rx += (long)(rxRate * seconds);
+        accumulator.Tx += (long)(txRate * seconds);
+        accumulator.RxPackets += (long)(rxRate * seconds / 1400);
+        accumulator.TxPackets += (long)(txRate * seconds / 1400);
+        accumulator.LastAt = now;
+
+        var interfaces = new[]
+        {
+            new InterfaceCounters("lo", 12_000_000, 12_000_000, 40_000, 40_000),
+            new InterfaceCounters("eth0", accumulator.Rx, accumulator.Tx, accumulator.RxPackets, accumulator.TxPackets)
+        };
+
+        return Task.FromResult(new NetworkSnapshot(now, interfaces));
+    }
+}

src/Vantage.Core/Collectors/Network/NetDevParser.cs

@@ -0,0 +1,38 @@
+using System.Globalization;
+
+namespace Vantage.Core.Collectors.Network;
+
+public static class NetDevParser
+{
+    public static IReadOnlyList<InterfaceCounters> Parse(string procNetDev)
+    {
+        var interfaces = new List<InterfaceCounters>();
+
+        foreach (var rawLine in procNetDev.Replace("\r\n", "\n").Split('\n'))
+        {
+            var colon = rawLine.IndexOf(':');
+            if (colon <= 0)
+                continue;
+
+            var name = rawLine[..colon].Trim();
+            if (name.Length == 0 || name.Contains('|'))
+                continue;
+
+            var fields = rawLine[(colon + 1)..]
+                .Split(' ', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries);
+            if (fields.Length < 16)
+                continue;
+
+            interfaces.Add(new InterfaceCounters(
+                Name: name,
+                RxBytes: ParseLong(fields[0]),
+                RxPackets: ParseLong(fields[1]),
+                TxBytes: ParseLong(fields[8]),
+                TxPackets: ParseLong(fields[9])));
+        }
+
+        return interfaces;
+    }
+
+    private static long ParseLong(string value) => long.Parse(value, CultureInfo.InvariantCulture);
+}

src/Vantage.Core/Collectors/Network/NetworkThroughput.cs

@@ -0,0 +1,36 @@
+namespace Vantage.Core.Collectors.Network;
+
+public readonly record struct InterfaceThroughput(string Name, double RxBytesPerSecond, double TxBytesPerSecond);
+
+public static class NetworkThroughput
+{
+    public static IReadOnlyList<InterfaceThroughput> Between(NetworkSnapshot previous, NetworkSnapshot current)
+    {
+        var elapsed = (current.CollectedAt - previous.CollectedAt).TotalSeconds;
+        if (elapsed <= 0)
+            return Array.Empty<InterfaceThroughput>();
+
+        var rates = new List<InterfaceThroughput>();
+        foreach (var now in current.Interfaces)
+        {
+            var before = previous.Find(now.Name);
+            if (before is null)
+                continue;
+
+            rates.Add(new InterfaceThroughput(
+                now.Name,
+                Rate(before.Value.RxBytes, now.RxBytes, elapsed),
+                Rate(before.Value.TxBytes, now.TxBytes, elapsed)));
+        }
+
+        return rates;
+    }
+
+    private static double Rate(long before, long after, double elapsedSeconds)
+    {
+        var delta = after - before;
+        if (delta < 0)
+            return 0;
+        return delta / elapsedSeconds;
+    }
+}

src/Vantage.Core/Collectors/Network/SshNetworkProvider.cs

@@ -0,0 +1,24 @@
+using Vantage.Core.Models;
+using Vantage.Core.Ssh;
+
+namespace Vantage.Core.Collectors.Network;
+
+public sealed class SshNetworkProvider : INetworkProvider
+{
+    public const string Command = "cat /proc/net/dev";
+
+    private readonly ISshCommandRunner runner;
+    private readonly Func<DateTimeOffset> clock;
+
+    public SshNetworkProvider(ISshCommandRunner runner, Func<DateTimeOffset>? clock = null)
+    {
+        this.runner = runner;
+        this.clock = clock ?? (() => DateTimeOffset.UtcNow);
+    }
+
+    public async Task<NetworkSnapshot> GetSnapshotAsync(HostConfig host, CancellationToken cancellationToken)
+    {
+        var output = await runner.RunAsync(host, Command, cancellationToken).ConfigureAwait(false);
+        return new NetworkSnapshot(clock(), NetDevParser.Parse(output));
+    }
+}