287 lines
11 KiB
C#
287 lines
11 KiB
C#
using System.Net;
|
||
using System.Text.Json;
|
||
using Fleet.Core;
|
||
using Fleet.Hosting;
|
||
using Fleet.Link;
|
||
using Fleet.Rpc;
|
||
using Monitor.Measurement;
|
||
using Monitor.Protocol;
|
||
|
||
namespace Monitor.Agent;
|
||
|
||
/// <summary>Agent 的组装点：数据面、点对点监听、与 Backend 的链路都在这里接起来。没有全局状态，测试可以在同一进程里启动多个。</summary>
|
||
public sealed class AgentApp : IDisposable
|
||
{
|
||
public const string Version = "0.1.0";
|
||
|
||
private readonly AgentConfig _config;
|
||
private readonly LoadedNodeIdentity _identity;
|
||
private readonly TimeProvider _time;
|
||
private readonly INodeLog _log;
|
||
private readonly MonoClock _clock;
|
||
private readonly AgentCounters _counters = new();
|
||
private readonly SessionRegistry _sessions = new();
|
||
private readonly StreamRegistry _streams = new();
|
||
private readonly TaskCompletionSource _ready = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||
private readonly AgentAdmission _admission;
|
||
private readonly Outbox _outbox = new(64 << 20);
|
||
private readonly BackendUplink _uplink;
|
||
private readonly UdpMux _udp;
|
||
private readonly TcpAcceptor _tcp;
|
||
private readonly PeerListener _peers;
|
||
private readonly PeerHub _hub;
|
||
private readonly TaskManager _tasks;
|
||
private readonly TestRunner _tests;
|
||
private readonly FleetLink _link;
|
||
|
||
public AgentApp(AgentConfig config, LoadedNodeIdentity identity, TimeProvider time, INodeLog log)
|
||
{
|
||
_config = config;
|
||
_identity = identity;
|
||
_time = time;
|
||
_log = log;
|
||
_clock = new MonoClock(time);
|
||
_uplink = new BackendUplink(time);
|
||
|
||
IPAddress bind = IPAddress.Parse(config.Listen.Bind);
|
||
_admission = new AgentAdmission(_streams);
|
||
_udp = new UdpMux(_sessions, _admission, _clock, _counters.Data, log, bind, config.Listen.UdpPorts);
|
||
_tcp = new TcpAcceptor(bind, config.Listen.TcpPorts, identity.Certificate, _sessions, _counters.Data);
|
||
|
||
// 监听方积木的环境：和 Probe 的拨号方共用同一套积木，区别只在这里装配的部分（准入策略、TCP 监听）
|
||
var env = new BlockEnv
|
||
{
|
||
Clock = _clock,
|
||
Log = log,
|
||
Session = () => null, // 每个 Probe 会话各用自己绑定的环境（BlockEnv.WithSession）
|
||
Udp = _udp,
|
||
TcpListen = _tcp,
|
||
Admission = _admission,
|
||
Counters = _counters.Data,
|
||
MaxPacketsPerSecond = config.Limits.MaxTotalPps,
|
||
MaxBulkMbps = config.Limits.MaxTotalMbps,
|
||
};
|
||
var chunks = new ChunkAssembler(_streams, _clock, _counters, _outbox, identity.NodeId, time, log);
|
||
var targets = new TargetRunners(chunks, identity.NodeId, _clock, time, log);
|
||
_tasks = new TaskManager(CreateGuard, _streams, env, targets, () => _udp.BoundPorts, () => _tcp.BoundPorts);
|
||
_hub = new PeerHub(_sessions, _clock, chunks, _tasks, _uplink, time, log);
|
||
_tests = new TestRunner(_tasks, _sessions, _streams, env, _uplink, config.Limits.MaxConcurrentTests, transport => transport == "udp" ? _udp.BoundPorts : _tcp.BoundPorts, time, log);
|
||
|
||
_tasks.OnLineTasksChanged = (probe, set) => _ = _hub.PushToAsync(probe, set).AsTask();
|
||
_tasks.OnPeersChanged = _hub.CloseNotIn;
|
||
_hub.OnProbeLost = probe => _tests.ProbeLost(probe);
|
||
_admission.OnEndpointChanged = probe => _ = _uplink.EmitEventAsync(AgentEventKinds.EndpointChanged, probe.UdpEndpoint?.ToString()).AsTask();
|
||
|
||
_peers = new PeerListener(
|
||
new PeerListenerOptions
|
||
{
|
||
Listen = new IPEndPoint(bind, config.Listen.ControlPort),
|
||
Certificate = identity.Certificate,
|
||
LookupPeerKey = _tasks.LookupPeerKey,
|
||
},
|
||
_hub.OnSessionAsync,
|
||
log,
|
||
time);
|
||
|
||
_link = BuildLink();
|
||
_uplink.Attach(_link);
|
||
}
|
||
|
||
/// <summary>数据面和控制端口都开始监听之后完成（配置里的端口为 0 时，这里能读到实际端口）。</summary>
|
||
public Task Ready => _ready.Task;
|
||
|
||
public void Dispose() => _udp.Dispose();
|
||
|
||
internal StreamRegistry Streams => _streams;
|
||
|
||
public IReadOnlyList<int> UdpPorts => _udp.BoundPorts;
|
||
|
||
public IReadOnlyList<int> TcpPorts => _tcp.BoundPorts;
|
||
|
||
public int ControlPort => _peers.Port;
|
||
|
||
public AgentCounters Counters => _counters;
|
||
|
||
public FleetLink Link => _link;
|
||
|
||
/// <summary>期望状态里是否已经有这个 Probe 的公钥（测试用：握手会按它校验）。</summary>
|
||
public bool IsPeerKnown(string probeId, string publicKeyBase64) =>
|
||
_tasks.LookupPeerKey(probeId) is { } key && Convert.ToBase64String(key) == publicKeyBase64;
|
||
|
||
public TestRunner Tests => _tests;
|
||
|
||
public async Task RunAsync(CancellationToken ct)
|
||
{
|
||
using CancellationTokenSource linked = CancellationTokenSource.CreateLinkedTokenSource(ct);
|
||
// 数据面和控制端口先开始监听，再连接 Backend：hello 里上报的是实际端口
|
||
_udp.Bind();
|
||
_tcp.Bind();
|
||
Task udp = _udp.RunAsync(linked.Token);
|
||
Task tcp = _tcp.RunAsync(linked.Token);
|
||
Task peers = _peers.RunAsync(linked.Token);
|
||
await Task.WhenAny(_peers.Started.Task, peers);
|
||
if (peers.IsFaulted)
|
||
{
|
||
await linked.CancelAsync();
|
||
await Task.WhenAll(udp, tcp, peers); // 抛出监听失败的原因（端口被占用等）
|
||
}
|
||
|
||
_ready.TrySetResult();
|
||
_log.Info("agent listening", ("control", ControlPort), ("udp", string.Join(',', UdpPorts)), ("tcp", string.Join(',', TcpPorts)));
|
||
try
|
||
{
|
||
await Task.WhenAll(udp, tcp, peers, _link.RunAsync(linked.Token), ReportHostAsync(linked.Token));
|
||
}
|
||
finally
|
||
{
|
||
await _tasks.StopAsync();
|
||
await linked.CancelAsync();
|
||
}
|
||
}
|
||
|
||
/// <summary>每隔几秒把主机资源占用的快照发给 Backend（尽力而为：链路断开时丢弃，重连之后下一次就补上）。</summary>
|
||
private async Task ReportHostAsync(CancellationToken ct)
|
||
{
|
||
if (!_config.Reporting.HostStats)
|
||
{
|
||
return;
|
||
}
|
||
|
||
var sampler = new HostSampler(_time);
|
||
sampler.Sample(); // 第一次只用来建立 CPU 和网速的基线
|
||
using var timer = new PeriodicTimer(TimeSpan.FromSeconds(Math.Clamp(_config.Reporting.HostIntervalSeconds, 1, 60)), _time);
|
||
try
|
||
{
|
||
while (await timer.WaitForNextTickAsync(ct))
|
||
{
|
||
await _uplink.SendHostAsync(sampler.Sample());
|
||
}
|
||
}
|
||
catch (OperationCanceledException)
|
||
{
|
||
// 停止
|
||
}
|
||
}
|
||
|
||
private BudgetGuard CreateGuard() =>
|
||
new(
|
||
new NodeLimits
|
||
{
|
||
MaxTotalMbps = _config.Limits.MaxTotalMbps,
|
||
MaxTotalPps = _config.Limits.MaxTotalPps,
|
||
AllowedTargetCidrs = _config.Limits.AllowedTargetCidrs,
|
||
MaxConcurrentTests = _config.Limits.MaxConcurrentTests,
|
||
AllowedUdpPorts = _udp.BoundPorts,
|
||
AllowedTcpPorts = _tcp.BoundPorts,
|
||
},
|
||
BlockCatalog.AgentKinds);
|
||
|
||
private FleetLink BuildLink()
|
||
{
|
||
var router = new RpcRouter<LinkContext>();
|
||
HashSet<string>? allowed = _config.Policy.AllowMethods is { } list ? [.. list] : null;
|
||
bool Allow(string method) => allowed is null || allowed.Contains(method);
|
||
|
||
if (Allow(MonitorMethods.TestRun))
|
||
{
|
||
router.On(
|
||
MonitorMethods.TestRun,
|
||
MonitorJsonContext.Default.TestRunParams,
|
||
MonitorJsonContext.Default.TestRunResult,
|
||
(_, run, _) => ValueTask.FromResult(_tests.Start(run)));
|
||
}
|
||
|
||
if (Allow(MonitorMethods.TestCancel))
|
||
{
|
||
router.On(
|
||
MonitorMethods.TestCancel,
|
||
MonitorJsonContext.Default.TestCancelParams,
|
||
MonitorJsonContext.Default.TestRunResult,
|
||
(_, cancel, _) => ValueTask.FromResult(Result<TestRunResult>.Ok(new TestRunResult { Accepted = _tests.Cancel(cancel.TestId) })));
|
||
}
|
||
|
||
if (Allow(MonitorMethods.ProbeDisconnect))
|
||
{
|
||
router.On(
|
||
MonitorMethods.ProbeDisconnect,
|
||
MonitorJsonContext.Default.ProbeDisconnectParams,
|
||
MonitorJsonContext.Default.TestRunResult,
|
||
async (_, request, _) => Result<TestRunResult>.Ok(new TestRunResult { Accepted = await _hub.DisconnectAsync(request.ProbeId.Value) }));
|
||
}
|
||
|
||
if (Allow(MonitorMethods.ProbeAgentsChanged))
|
||
{
|
||
router.OnNotify(
|
||
MonitorMethods.ProbeAgentsChanged,
|
||
MonitorJsonContext.Default.ProbeAgentsResult,
|
||
(_, list, _) => _hub.RelayAgentsAsync(list));
|
||
}
|
||
|
||
var options = new FleetLinkOptions
|
||
{
|
||
Server = new Uri(_config.Fleet.Server),
|
||
Identity = _identity.Identity,
|
||
NodeId = _identity.NodeId,
|
||
NodeVersion = Version,
|
||
Capabilities = [MonitorCapabilities.Base, MonitorCapabilities.KindsPrefix + string.Join(',', BlockCatalog.AgentKinds)],
|
||
HelloMeta = HelloMeta,
|
||
HeartbeatExtra = HeartbeatExtra,
|
||
};
|
||
var link = new FleetLink(options, router, [new StateHandler<TaskSet>(TaskSet.StateType, MonitorJsonContext.Default.TaskSet, _tasks.ApplyAsync)], _outbox, _time, _log);
|
||
|
||
// 链路重新连上之后补报：在线的 Probe 都告诉 Backend（Backend 重启或链路抖动期间错过了 probe_connected）
|
||
link.OnConnected = async (_, _) =>
|
||
{
|
||
foreach (ConnectedProbe probe in _hub.ConnectedProbes())
|
||
{
|
||
await _uplink.EmitEventAsync(AgentEventKinds.ProbeConnected, probe.Id.Value, probe);
|
||
}
|
||
};
|
||
return link;
|
||
}
|
||
|
||
private JsonElement? HelloMeta()
|
||
{
|
||
var meta = new AgentHelloMeta
|
||
{
|
||
TlsSpki = _identity.File.TlsSpki,
|
||
ProbeTcpPort = ControlPort,
|
||
ProbeUdpPorts = [.. UdpPorts],
|
||
LocalLimits = new LocalLimits
|
||
{
|
||
MaxTotalMbps = _config.Limits.MaxTotalMbps,
|
||
MaxTotalPps = _config.Limits.MaxTotalPps,
|
||
AllowedTargetCidrs = _config.Limits.AllowedTargetCidrs,
|
||
MaxConcurrentTests = _config.Limits.MaxConcurrentTests,
|
||
},
|
||
};
|
||
return JsonSerializer.SerializeToElement(meta, MonitorJsonContext.Default.AgentHelloMeta);
|
||
}
|
||
|
||
private long _lastIn;
|
||
private long _lastOut;
|
||
private long _lastAtMs;
|
||
|
||
private JsonElement? HeartbeatExtra()
|
||
{
|
||
long nowMs = _time.GetUtcNow().ToUnixTimeMilliseconds();
|
||
long seconds = Math.Max(1, (nowMs - _lastAtMs) / 1000);
|
||
long packetsIn = _counters.PacketsIn;
|
||
long packetsOut = _counters.PacketsOut;
|
||
var extra = new AgentHeartbeatExtra
|
||
{
|
||
Probes = [.. _hub.ConnectedProbes()],
|
||
Sessions = _sessions.All.Count,
|
||
PpsIn = _lastAtMs == 0 ? 0 : (packetsIn - _lastIn) / seconds,
|
||
PpsOut = _lastAtMs == 0 ? 0 : (packetsOut - _lastOut) / seconds,
|
||
DroppedInvalid = _counters.DroppedInvalid,
|
||
DroppedRateLimited = _counters.DroppedRateLimited,
|
||
QueueBytes = _outbox.PendingBytes,
|
||
};
|
||
_lastIn = packetsIn;
|
||
_lastOut = packetsOut;
|
||
_lastAtMs = nowMs;
|
||
return JsonSerializer.SerializeToElement(extra, MonitorJsonContext.Default.AgentHeartbeatExtra);
|
||
}
|
||
}
|