463 lines
20 KiB
C#
463 lines
20 KiB
C#
using Fleet.Identity;
|
||
using System.Net;
|
||
using System.Text.Json;
|
||
using Fleet.Core;
|
||
using Fleet.Hosting;
|
||
using Fleet.Link;
|
||
using Fleet.Rpc;
|
||
using MoShiro.Measurement;
|
||
using MoShiro.Protocol;
|
||
|
||
namespace MoShiro.Agent;
|
||
|
||
/// <summary>
|
||
/// 节点的组装点：与 Backend 的链路、监听侧（数据面和点对点监听，<c>listen.enabled</c> 为 false 时不启动）、拨号侧（Backend 下发的拨号目标）都在这里接起来。
|
||
/// 没有全局状态，测试可以在同一进程里启动多个。
|
||
/// </summary>
|
||
public sealed class AgentApp : IDisposable, IAsyncDisposable
|
||
{
|
||
public const string Version = MoShiro.Protocol.BuildInfo.Version;
|
||
|
||
private readonly AgentConfig _config;
|
||
private readonly bool _listening;
|
||
private readonly bool? _ipv6Supported;
|
||
private readonly KnockGate _knock;
|
||
private readonly TrustAnchor _trust;
|
||
private readonly RestartController _restart;
|
||
private readonly RestartController? _ownedRestart;
|
||
private readonly LoadedNodeIdentity _identity;
|
||
private readonly TimeProvider _time;
|
||
private readonly INodeLog _log;
|
||
private readonly INodeLogSource? _logSource;
|
||
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 RestartUpdater? _update;
|
||
private readonly OpenWrtUpdater? _nativeUpdate;
|
||
private readonly DialerHost _dialer;
|
||
private readonly BackendUplink _uplink;
|
||
private readonly QuicDatagramTransport _datagrams;
|
||
private readonly TcpAcceptor _tcp;
|
||
private readonly PeerListener _peers;
|
||
private readonly PeerHub _hub;
|
||
private readonly ChunkAssembler _chunks;
|
||
private readonly TaskManager _tasks;
|
||
private readonly TestRunner _tests;
|
||
private readonly FleetLink _link;
|
||
private int _disposed;
|
||
|
||
/// <param name="dataDirectory">配置目录（容器里的 <c>/etc/moshiro</c>）：宿主机更新器和节点在这里交换清单、请求和状态。</param>
|
||
public AgentApp(AgentConfig config, LoadedNodeIdentity identity, string dataDirectory, TimeProvider time, INodeLog log, RestartController? restart = null)
|
||
{
|
||
_ownedRestart = restart is null ? new RestartController() : null;
|
||
_restart = restart ?? _ownedRestart!;
|
||
_update = RestartUpdater.FromEnvironment("agent", _restart.RequestNow, time, log);
|
||
_nativeUpdate = OpenWrtUpdater.FromEnvironment(time, log);
|
||
_config = config;
|
||
_listening = config.Listen.Enabled;
|
||
_identity = identity;
|
||
_time = time;
|
||
_log = log;
|
||
_logSource = log as INodeLogSource;
|
||
_clock = new MonoClock(time);
|
||
_uplink = new BackendUplink(time);
|
||
|
||
IPAddress bind = IPAddress.Parse(config.Listen.Bind);
|
||
if (bind.Equals(IPAddress.IPv6Any) && !System.Net.Sockets.Socket.OSSupportsIPv6)
|
||
{
|
||
bind = IPAddress.Any;
|
||
}
|
||
|
||
_ipv6Supported = _listening ? bind.AddressFamily == System.Net.Sockets.AddressFamily.InterNetworkV6 : null;
|
||
_admission = new AgentAdmission(_streams);
|
||
QuicSettings quic = config.Quic.ToSettings();
|
||
if (quic.Validate() is { } invalid)
|
||
{
|
||
throw new InvalidOperationException("agent.json quic: " + invalid);
|
||
}
|
||
|
||
_datagrams = new QuicDatagramTransport(_sessions, _admission, _clock, _counters.Data, log, bind, config.Listen.QuicPorts, identity.Certificate, quic);
|
||
// 预握手敲门：控制端口和数据端口共用一个关卡（限速和去重一起算）；名册就是期望状态里的 peers
|
||
_trust = TrustAnchor.Load(config.Fleet.IdentityDir, log);
|
||
_dialer = new DialerHost(new DialEnv(identity.PeerIdentity, identity.NodeId.Value, config.Limits, quic, new DialDirectory(new ConfigStore(dataDirectory), dataDirectory, log, _trust), _clock, time, log), time, log);
|
||
_knock = new KnockGate(identity.PeerIdentity, RosterKeys, time);
|
||
_tcp = new TcpAcceptor(bind, config.Listen.TcpPorts, identity.Certificate, _sessions, _counters.Data, _knock);
|
||
|
||
// 监听方积木的环境：和拨号方的拨号方共用同一套积木，区别只在这里装配的部分（准入策略、TCP 监听）
|
||
var env = new BlockEnv
|
||
{
|
||
Clock = _clock,
|
||
Log = log,
|
||
Session = () => null, // 每个拨号方会话各用自己绑定的环境（BlockEnv.WithSession）
|
||
Datagrams = _datagrams,
|
||
TcpListen = _tcp,
|
||
Admission = _admission,
|
||
Counters = _counters.Data,
|
||
MaxPacketsPerSecond = config.Limits.MaxTotalPps,
|
||
MaxBulkMbps = config.Limits.MaxTotalMbps,
|
||
};
|
||
_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, () => _datagrams.BoundPorts, () => _tcp.BoundPorts, _trust, acceptLineTasks: _listening);
|
||
_hub = new PeerHub(_sessions, _clock, _chunks, _tasks, _uplink, time, log);
|
||
_tests = new TestRunner(_tasks, _sessions, _streams, env, _uplink, config.Limits.MaxConcurrentTests, transport => transport == "quic_datagram" ? _datagrams.BoundPorts : _tcp.BoundPorts, time, log);
|
||
|
||
_tasks.OnLineTasksChanged = (dialer, set) => _ = _hub.PushToAsync(dialer, set).AsTask();
|
||
_tasks.OnPeersChanged = _hub.CloseNotIn;
|
||
_hub.OnDialerLost = dialer => _tests.DialerLost(dialer);
|
||
_admission.OnEndpointChanged = dialer => _ = _uplink.EmitEventAsync(AgentEventKinds.EndpointChanged, dialer.DatagramRemoteEndPoint?.ToString()).AsTask();
|
||
|
||
_peers = new PeerListener(
|
||
new PeerListenerOptions
|
||
{
|
||
Listen = new IPEndPoint(bind, config.Listen.ControlPort),
|
||
Certificate = identity.Certificate,
|
||
LookupPeerKey = _tasks.LookupPeerKey,
|
||
Knock = _knock,
|
||
},
|
||
_hub.OnSessionAsync,
|
||
log,
|
||
time);
|
||
|
||
_link = BuildLink();
|
||
_uplink.Attach(_link);
|
||
}
|
||
|
||
/// <summary>数据面和控制端口都开始监听之后完成（配置里的端口为 0 时，这里能读到实际端口）。</summary>
|
||
public Task Ready => _ready.Task;
|
||
|
||
public void Dispose()
|
||
{
|
||
if (Interlocked.Exchange(ref _disposed, 1) != 0)
|
||
{
|
||
return;
|
||
}
|
||
|
||
_dialer.Dispose();
|
||
_datagrams.Dispose();
|
||
_ownedRestart?.Dispose();
|
||
}
|
||
|
||
public async ValueTask DisposeAsync()
|
||
{
|
||
if (Interlocked.Exchange(ref _disposed, 1) != 0)
|
||
{
|
||
return;
|
||
}
|
||
|
||
_dialer.Dispose();
|
||
await _datagrams.DisposeAsync().ConfigureAwait(false);
|
||
_ownedRestart?.Dispose();
|
||
}
|
||
|
||
internal StreamRegistry Streams => _streams;
|
||
|
||
public IReadOnlyList<int> QuicPorts => _datagrams.BoundPorts;
|
||
|
||
public IReadOnlyList<int> TcpPorts => _tcp.BoundPorts;
|
||
|
||
public int ControlPort => _peers.Port;
|
||
|
||
/// <summary>节点当前采用的操作员集合版本（没有固定信任根时是 0）。</summary>
|
||
public int TrustVersion => _trust.Version;
|
||
|
||
public AgentCounters Counters => _counters;
|
||
|
||
public FleetLink Link => _link;
|
||
|
||
/// <summary>拨号侧（测试和状态查看用）。</summary>
|
||
public DialerHost Dialer => _dialer;
|
||
|
||
/// <summary>对等面（拨号方连进来的会话）的路由器，测试用。</summary>
|
||
public RpcRouter<PeerContext> PeerRouter => _hub.Router;
|
||
|
||
/// <summary>期望状态里是否已经有这个拨号方的公钥（测试用：握手会按它校验）。</summary>
|
||
public bool IsPeerKnown(string dialerId, string publicKeyBase64) =>
|
||
_tasks.LookupPeerKey(dialerId) is { } key && Convert.ToBase64String(key) == publicKeyBase64;
|
||
|
||
public TestRunner Tests => _tests;
|
||
|
||
public async Task RunAsync(CancellationToken ct)
|
||
{
|
||
_log.Info("moshiro agent starting", MoShiro.Protocol.BuildInfo.Fields);
|
||
using CancellationTokenSource linked = CancellationTokenSource.CreateLinkedTokenSource(ct);
|
||
Task quic = Task.CompletedTask;
|
||
Task tcp = Task.CompletedTask;
|
||
Task peers = Task.CompletedTask;
|
||
if (_listening)
|
||
{
|
||
// 数据面和控制端口先开始监听，再连接 Backend：hello 里上报的是实际端口
|
||
_datagrams.Bind();
|
||
_tcp.Bind();
|
||
quic = _datagrams.RunAsync(linked.Token);
|
||
tcp = _tcp.RunAsync(linked.Token);
|
||
peers = _peers.RunAsync(linked.Token);
|
||
await Task.WhenAny(_peers.Started.Task, peers);
|
||
if (peers.IsFaulted)
|
||
{
|
||
await linked.CancelAsync();
|
||
await Task.WhenAll(quic, tcp, peers); // 抛出监听失败的原因（端口被占用等）
|
||
}
|
||
|
||
_log.Info("agent listening", ("control", ControlPort), ("quic", string.Join(',', QuicPorts)), ("tcp", string.Join(',', TcpPorts)));
|
||
}
|
||
else
|
||
{
|
||
_log.Info("agent is dial-only: listening is disabled");
|
||
}
|
||
|
||
_ready.TrySetResult();
|
||
try
|
||
{
|
||
await Task.WhenAll(quic, tcp, peers, _link.RunAsync(linked.Token), _dialer.RunAsync(linked.Token), ReportHostAsync(linked.Token), _chunks.RunTrafficReportingAsync(linked.Token));
|
||
}
|
||
finally
|
||
{
|
||
await _tasks.StopAsync();
|
||
_chunks.FlushTraffic();
|
||
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,
|
||
AllowedQuicPorts = _datagrams.BoundPorts,
|
||
AllowedTcpPorts = _tcp.BoundPorts,
|
||
},
|
||
_listening ? BlockCatalog.AgentKinds : BlockCatalog.TargetKinds);
|
||
|
||
private FleetLink BuildLink()
|
||
{
|
||
var router = new RpcRouter<LinkContext>(RpcSide.Node, MoShiroMethodTable.Table);
|
||
HashSet<string>? allowed = _config.Policy.AllowMethods is { } list ? [.. list] : null;
|
||
bool Allow(string method) => allowed is null || allowed.Contains(method);
|
||
|
||
if (Allow(MoShiroMethods.TestRun))
|
||
{
|
||
router.On(
|
||
MoShiroMethods.TestRun,
|
||
MoShiroJsonContext.Default.TestRunParams,
|
||
MoShiroJsonContext.Default.TestRunResult,
|
||
(_, run, _) => ValueTask.FromResult(_tests.Start(run)));
|
||
}
|
||
|
||
if (Allow(MoShiroMethods.TestCancel))
|
||
{
|
||
router.On(
|
||
MoShiroMethods.TestCancel,
|
||
MoShiroJsonContext.Default.TestCancelParams,
|
||
MoShiroJsonContext.Default.TestRunResult,
|
||
(_, cancel, _) => ValueTask.FromResult(Result<TestRunResult>.Ok(new TestRunResult { Accepted = _tests.Cancel(cancel.TestId) })));
|
||
}
|
||
|
||
if (Allow(MoShiroMethods.UpdateOffer))
|
||
{
|
||
router.OnNotify(MoShiroMethods.UpdateOffer, MoShiroJsonContext.Default.UpdateOfferParams, (_, offer, _) =>
|
||
{
|
||
_update?.Offer(offer);
|
||
_nativeUpdate?.Offer();
|
||
return ValueTask.CompletedTask;
|
||
});
|
||
}
|
||
|
||
if (Allow(MoShiroMethods.Schedule))
|
||
{
|
||
router.OnNotify(MoShiroMethods.Schedule, MoShiroJsonContext.Default.ScheduleParams, (_, schedule, _) =>
|
||
{
|
||
if (!_restart.SetRemote(schedule.RestartAt))
|
||
{
|
||
_log.Warn("ignoring an invalid restart schedule from the panel");
|
||
}
|
||
|
||
return ValueTask.CompletedTask;
|
||
});
|
||
}
|
||
|
||
if (Allow(MoShiroMethods.UpdateApply))
|
||
{
|
||
router.OnNotify(MoShiroMethods.UpdateApply, MoShiroJsonContext.Default.UpdateOfferParams, (_, _, _) =>
|
||
{
|
||
_update?.RequestApply();
|
||
_nativeUpdate?.RequestApply();
|
||
return ValueTask.CompletedTask;
|
||
});
|
||
}
|
||
|
||
if (Allow(MoShiroMethods.DialerDisconnect))
|
||
{
|
||
router.On(
|
||
MoShiroMethods.DialerDisconnect,
|
||
MoShiroJsonContext.Default.DialerDisconnectParams,
|
||
MoShiroJsonContext.Default.TestRunResult,
|
||
async (_, request, _) => Result<TestRunResult>.Ok(new TestRunResult { Accepted = await _hub.DisconnectAsync(request.DialerId.Value) }));
|
||
}
|
||
|
||
if (Allow(MoShiroMethods.LogsFetch))
|
||
{
|
||
router.On(
|
||
MoShiroMethods.LogsFetch,
|
||
MoShiroJsonContext.Default.LogsFetchParams,
|
||
MoShiroJsonContext.Default.LogsFetchResult,
|
||
(_, request, _) =>
|
||
{
|
||
NodeLogPage page = _logSource?.Read(request.AfterId, request.Limit, request.MinLevel) ?? new NodeLogPage([], 0);
|
||
return ValueTask.FromResult(Result<LogsFetchResult>.Ok(new LogsFetchResult
|
||
{
|
||
LastId = page.LastId,
|
||
Items = [.. page.Items.Select(e => new LogEntryDto { Id = e.Id, Type = e.Type, AtUtcMs = e.Record.AtUtcMs, Level = e.Record.Level, Message = e.Record.Message, Fields = e.Record.Fields })],
|
||
}));
|
||
});
|
||
}
|
||
|
||
if (Allow(MoShiroMethods.LinkRetry))
|
||
{
|
||
router.OnNotify(MoShiroMethods.LinkRetry, MoShiroJsonContext.Default.LinkRetryParams, (_, request, _) =>
|
||
{
|
||
_dialer.RetryListener(request.ListenerId.Value);
|
||
return ValueTask.CompletedTask;
|
||
});
|
||
}
|
||
|
||
var options = new FleetLinkOptions
|
||
{
|
||
Server = new Uri(_config.Fleet.Server),
|
||
Identity = _identity.Identity,
|
||
ServerVerifier = _trust,
|
||
NodeId = _identity.NodeId,
|
||
NodeVersion = Version,
|
||
MinServerVersion = () => Environment.GetEnvironmentVariable("MOSHIRO_MIN_SERVER_VERSION") is { Length: > 0 } required ? required : _config.Fleet.MinServerVersion,
|
||
Capabilities = [MoShiroCapabilities.Base, MoShiroCapabilities.KindsPrefix + string.Join(',', _listening ? BlockCatalog.AgentKinds : BlockCatalog.TargetKinds)],
|
||
HelloMeta = HelloMeta,
|
||
HeartbeatExtra = HeartbeatExtra,
|
||
ShutdownReason = () => _restart.NowReason == "update" ? "updating" : "stopping",
|
||
};
|
||
var link = new FleetLink(
|
||
options,
|
||
router,
|
||
[
|
||
new StateHandler<TaskSet>(TaskSet.StateType, MoShiroJsonContext.Default.TaskSet, _tasks.ApplyAsync),
|
||
new StateHandler<DialTargets>(DialTargets.StateType, MoShiroJsonContext.Default.DialTargets, (version, targets, _) =>
|
||
{
|
||
_dialer.Directory.Apply(targets);
|
||
return ValueTask.FromResult(new StateSyncResult { AppliedVersion = version });
|
||
}),
|
||
],
|
||
_outbox,
|
||
_time,
|
||
_log);
|
||
|
||
// 链路重新连上之后补报：在线的拨号方都告诉 Backend（Backend 重启或链路抖动期间错过了 dialer_connected）
|
||
link.OnConnected = async (_, _) =>
|
||
{
|
||
foreach (ConnectedDialer dialer in _hub.ConnectedDialers())
|
||
{
|
||
await _uplink.EmitEventAsync(AgentEventKinds.DialerConnected, dialer.Id.Value, dialer);
|
||
}
|
||
};
|
||
return link;
|
||
}
|
||
|
||
/// <summary>敲门关卡的名册：期望状态里的 peers（任务管理器在构造之后才有）。</summary>
|
||
private IReadOnlyList<byte[]> RosterKeys() => _tasks.PeerKeys();
|
||
|
||
private JsonElement? HelloMeta()
|
||
{
|
||
var meta = new NodeHelloMeta
|
||
{
|
||
Owners = [.. _trust.Controllers],
|
||
Ipv6 = Ipv6Diagnostics.Sample(_ipv6Supported, Environment.GetEnvironmentVariable("MOSHIRO_NETWORK_SCOPE"), Environment.GetEnvironmentVariable("HOST_PROC")),
|
||
Platform = MoShiro.Measurement.PlatformDetector.Detect(),
|
||
LocalLimits = new LocalLimits
|
||
{
|
||
MaxTotalMbps = _config.Limits.MaxTotalMbps,
|
||
MaxTotalPps = _config.Limits.MaxTotalPps,
|
||
AllowedTargetCidrs = _config.Limits.AllowedTargetCidrs,
|
||
MaxConcurrentTests = _config.Limits.MaxConcurrentTests,
|
||
},
|
||
Listen = !_listening
|
||
? null
|
||
: new ListenInfo
|
||
{
|
||
TcpPort = ControlPort,
|
||
QuicPorts = [.. QuicPorts],
|
||
Ipv6Supported = _ipv6Supported == true,
|
||
Addresses = [.. LocalAddresses.Sample()],
|
||
},
|
||
};
|
||
return JsonSerializer.SerializeToElement(meta, MoShiroJsonContext.Default.NodeHelloMeta);
|
||
}
|
||
|
||
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;
|
||
NodeLogSummary? logSummary = _logSource?.TakeSummary();
|
||
UpdateStatus? updateStatus = _nativeUpdate?.Status() ?? _update?.Status();
|
||
var extra = new NodeHeartbeatExtra
|
||
{
|
||
RuntimeStatus = updateStatus?.State == "applying" ? NodeRuntimeStatus.Updating : !_listening && _dialer.Directory.Targets.Count == 0 ? NodeRuntimeStatus.Idle : NodeRuntimeStatus.Running,
|
||
Ipv6 = Ipv6Diagnostics.Sample(_ipv6Supported, Environment.GetEnvironmentVariable("MOSHIRO_NETWORK_SCOPE"), Environment.GetEnvironmentVariable("HOST_PROC")),
|
||
Links = _dialer.SnapshotLinks(),
|
||
Dialers = [.. _hub.ConnectedDialers()],
|
||
PpsIn = _lastAtMs == 0 ? 0 : (packetsIn - _lastIn) / seconds,
|
||
PpsOut = _lastAtMs == 0 ? 0 : (packetsOut - _lastOut) / seconds,
|
||
DroppedInvalid = _counters.DroppedInvalid,
|
||
DroppedRateLimited = _counters.DroppedRateLimited,
|
||
QueueBytes = _outbox.PendingBytes,
|
||
Update = updateStatus,
|
||
LogSummary = logSummary is null ? null : new LogSummaryDto
|
||
{
|
||
LastId = logSummary.LastId,
|
||
Info = logSummary.Info,
|
||
Warn = logSummary.Warn,
|
||
Error = logSummary.Error,
|
||
Types = [.. logSummary.Types.Select(t => new LogTypeCountDto { Type = t.Type, Level = t.Level, Count = t.Count })],
|
||
},
|
||
};
|
||
_lastIn = packetsIn;
|
||
_lastOut = packetsOut;
|
||
_lastAtMs = nowMs;
|
||
return JsonSerializer.SerializeToElement(extra, MoShiroJsonContext.Default.NodeHeartbeatExtra);
|
||
}
|
||
}
|