diff --git a/src/Vantage.Core/Collectors/Network/INetworkProvider.cs b/src/Vantage.Core/Collectors/Network/INetworkProvider.cs new file mode 100644 index 0000000..68dbfc8 --- /dev/null +++ b/src/Vantage.Core/Collectors/Network/INetworkProvider.cs @@ -0,0 +1,8 @@ +using Vantage.Core.Models; + +namespace Vantage.Core.Collectors.Network; + +public interface INetworkProvider +{ + Task GetSnapshotAsync(HostConfig host, CancellationToken cancellationToken); +} diff --git a/src/Vantage.Core/Collectors/Network/InterfaceCounters.cs b/src/Vantage.Core/Collectors/Network/InterfaceCounters.cs new file mode 100644 index 0000000..957d8f2 --- /dev/null +++ b/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 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); +} diff --git a/src/Vantage.Core/Collectors/Network/MockNetworkProvider.cs b/src/Vantage.Core/Collectors/Network/MockNetworkProvider.cs new file mode 100644 index 0000000..ff0b367 --- /dev/null +++ b/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 state = new(); + private readonly Func clock; + + public MockNetworkProvider(Func? clock = null) + { + this.clock = clock ?? (() => DateTimeOffset.UtcNow); + } + + public Task 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)); + } +} diff --git a/src/Vantage.Core/Collectors/Network/NetDevParser.cs b/src/Vantage.Core/Collectors/Network/NetDevParser.cs new file mode 100644 index 0000000..892c16c --- /dev/null +++ b/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 Parse(string procNetDev) + { + var interfaces = new List(); + + 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); +} diff --git a/src/Vantage.Core/Collectors/Network/NetworkThroughput.cs b/src/Vantage.Core/Collectors/Network/NetworkThroughput.cs new file mode 100644 index 0000000..0e3ee67 --- /dev/null +++ b/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 Between(NetworkSnapshot previous, NetworkSnapshot current) + { + var elapsed = (current.CollectedAt - previous.CollectedAt).TotalSeconds; + if (elapsed <= 0) + return Array.Empty(); + + var rates = new List(); + 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; + } +} diff --git a/src/Vantage.Core/Collectors/Network/SshNetworkProvider.cs b/src/Vantage.Core/Collectors/Network/SshNetworkProvider.cs new file mode 100644 index 0000000..95ad708 --- /dev/null +++ b/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 clock; + + public SshNetworkProvider(ISshCommandRunner runner, Func? clock = null) + { + this.runner = runner; + this.clock = clock ?? (() => DateTimeOffset.UtcNow); + } + + public async Task GetSnapshotAsync(HostConfig host, CancellationToken cancellationToken) + { + var output = await runner.RunAsync(host, Command, cancellationToken).ConfigureAwait(false); + return new NetworkSnapshot(clock(), NetDevParser.Parse(output)); + } +}