108 lines
3.5 KiB
C#
108 lines
3.5 KiB
C#
using Fleet.Core;
|
|
using Fleet.Rpc;
|
|
|
|
namespace MoShiro.Agent;
|
|
|
|
/// <summary>Aggregates per-task Agent → Backend chunk sizes into low-volume diagnostic logs.</summary>
|
|
internal sealed class TaskTrafficReporter
|
|
{
|
|
private static readonly TimeSpan s_window = TimeSpan.FromMinutes(1);
|
|
|
|
private readonly object _gate = new();
|
|
private readonly INodeLog _log;
|
|
private readonly TimeProvider _time;
|
|
private readonly Dictionary<(string Task, string Kind), Totals> _totals = [];
|
|
private long _windowStarted;
|
|
|
|
public TaskTrafficReporter(INodeLog log, TimeProvider time)
|
|
{
|
|
_log = log;
|
|
_time = time;
|
|
_windowStarted = time.GetTimestamp();
|
|
}
|
|
|
|
public void Record(string task, string kind, long records, int uncompressedBytes, int compressedBytes)
|
|
{
|
|
lock (_gate)
|
|
{
|
|
var key = (task, kind);
|
|
if (!_totals.TryGetValue(key, out Totals? totals))
|
|
{
|
|
totals = new Totals();
|
|
_totals.Add(key, totals);
|
|
}
|
|
|
|
totals.Chunks++;
|
|
totals.Records += records;
|
|
totals.UncompressedBytes += uncompressedBytes;
|
|
totals.CompressedBytes += compressedBytes;
|
|
}
|
|
}
|
|
|
|
public async Task RunAsync(CancellationToken ct)
|
|
{
|
|
using var timer = new PeriodicTimer(s_window, _time);
|
|
try
|
|
{
|
|
while (await timer.WaitForNextTickAsync(ct))
|
|
{
|
|
Flush();
|
|
}
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
// The final snapshot is flushed by the Agent shutdown path.
|
|
}
|
|
}
|
|
|
|
public void Flush()
|
|
{
|
|
List<Snapshot> completed;
|
|
long windowSeconds;
|
|
lock (_gate)
|
|
{
|
|
if (_totals.Count == 0)
|
|
{
|
|
_windowStarted = _time.GetTimestamp();
|
|
return;
|
|
}
|
|
|
|
TimeSpan elapsed = _time.GetElapsedTime(_windowStarted);
|
|
windowSeconds = Math.Max(1, (long)elapsed.TotalSeconds);
|
|
completed = [.. _totals.Select(pair => new Snapshot(pair.Key.Task, pair.Key.Kind, pair.Value.Chunks, pair.Value.Records, pair.Value.UncompressedBytes, pair.Value.CompressedBytes))];
|
|
_totals.Clear();
|
|
_windowStarted = _time.GetTimestamp();
|
|
}
|
|
|
|
foreach (Snapshot item in completed)
|
|
{
|
|
double savedPercent = item.UncompressedBytes == 0
|
|
? 0
|
|
: 100d * (item.UncompressedBytes - item.CompressedBytes) / item.UncompressedBytes;
|
|
long queuedBytes = item.CompressedBytes + (long)item.Chunks * OutboxFrame.HeaderBytes;
|
|
_log.Info(
|
|
"task upload traffic",
|
|
("link", "agent-backend"),
|
|
("task", item.Task),
|
|
("kind", item.Kind),
|
|
("chunks", item.Chunks),
|
|
("records", item.Records),
|
|
("raw_bytes", item.UncompressedBytes),
|
|
("compressed_bytes", item.CompressedBytes),
|
|
("queued_bytes", queuedBytes),
|
|
("saved_pct", Math.Round(savedPercent, 1)),
|
|
("window_s", windowSeconds));
|
|
}
|
|
}
|
|
|
|
private sealed class Totals
|
|
{
|
|
public long Chunks;
|
|
public long Records;
|
|
public long UncompressedBytes;
|
|
public long CompressedBytes;
|
|
}
|
|
|
|
private sealed record Snapshot(string Task, string Kind, long Chunks, long Records, long UncompressedBytes, long CompressedBytes);
|
|
}
|