justme
48ffa56285
A pair of nodes now has a single line, dialed by the non-listening side or by the smaller id when both listen, so RTT is measured once. The board draws it undirected and colours it by RTT and loss; a manual setting applies to the pair.
169 lines
9.3 KiB
C#
169 lines
9.3 KiB
C#
using Fleet.Core;
|
|
using Fleet.Server;
|
|
using Microsoft.AspNetCore.Builder;
|
|
using Microsoft.AspNetCore.Http;
|
|
using Microsoft.AspNetCore.Routing;
|
|
using MoShiro.Backend.Application;
|
|
using MoShiro.Protocol;
|
|
|
|
namespace MoShiro.Backend.Api;
|
|
|
|
/// <summary>
|
|
/// 线路开关：哪些（拨号方，监听方）要连，一对节点一条线路。自动连接的总开关是设置 <c>fleet.auto_link</c>（不监听的节点拨监听节点；两个都监听时 id 较小的一方拨）；
|
|
/// 这里看每一对的结果，并且手动强制开、关或取消手动设置。把一个节点拖到另一个监听节点上建立线路，就是 <c>PUT</c> 一个 <c>enabled: true</c>。
|
|
/// </summary>
|
|
public static class LinkEndpoints
|
|
{
|
|
private const string Tag = "admin-links";
|
|
|
|
public static void Map(RouteGroupBuilder group)
|
|
{
|
|
group.MapGet("/links", ListAsync)
|
|
.Api("links_list", "Every possible dialer/listener pair: whether it is linked, the manual override and whether it is connected; plus the nodes and which of them listen", Tag).Returns<LinkMatrixDto>().ProducesProblems(401, 403)
|
|
.Requires(Capability.FleetRead);
|
|
group.MapPut("/links/{dialer_id}/{listener_id}", SetAsync)
|
|
.Api("links_set", "Connect a node to a listening node (enabled: true), force the link off (false), or give it back to the automatic setting (null)", Tag).Returns<LinkDto>().ProducesProblems(400, 401, 403, 404, 409, 422)
|
|
.Requires(Capability.FleetWrite);
|
|
group.MapPost("/links/{dialer_id}/{listener_id}/retry", RetryAsync)
|
|
.Api("links_retry", "Enable and immediately retry the dialer-to-listener connection", Tag).Returns<LinkRetryDto>().ProducesProblems(401, 403, 404, 409, 422)
|
|
.Requires(Capability.FleetWrite);
|
|
}
|
|
|
|
private static async Task<IResult> ListAsync(AgentService agents, LinkService links, LinkRuntime runtime, TimeProvider time, CancellationToken ct)
|
|
{
|
|
LinkSnapshot snapshot = await links.SnapshotAsync(ct);
|
|
AgentView[] nodes = [.. (await agents.ListAsync(ct)).Where(n => n.Agent.Status == NodeStatus.Active).OrderBy(n => n.Agent.Name, StringComparer.Ordinal)];
|
|
var items = new List<LinkDto>();
|
|
for (int i = 0; i < nodes.Length; i++)
|
|
{
|
|
for (int j = i + 1; j < nodes.Length; j++)
|
|
{
|
|
// 一对节点一条线路：连着的那个方向，没连就是默认方向；两个都不监听的一对没有线路
|
|
(AgentView Dialer, AgentView Listener)? pair =
|
|
snapshot.IsLinked(nodes[i].Agent, nodes[j].Agent) ? (nodes[i], nodes[j])
|
|
: snapshot.IsLinked(nodes[j].Agent, nodes[i].Agent) ? (nodes[j], nodes[i])
|
|
: LinkSnapshot.IsDefaultDialer(nodes[i].Agent, nodes[j].Agent) ? (nodes[i], nodes[j])
|
|
: LinkSnapshot.IsDefaultDialer(nodes[j].Agent, nodes[i].Agent) ? (nodes[j], nodes[i])
|
|
: null;
|
|
if (pair is { } p)
|
|
{
|
|
items.Add(ToDto(snapshot, p.Dialer, p.Listener, runtime, time));
|
|
}
|
|
}
|
|
}
|
|
|
|
LinkNodeDto[] nodeItems = [.. nodes.Select(n => new LinkNodeDto(n.Agent.Id.Value, n.Agent.Name, n.Agent.Listens, n.Online, n.RuntimeStatus))];
|
|
return TypedResults.Json(new LinkMatrixDto(snapshot.Auto, nodeItems, items), ApiJsonContext.Default.LinkMatrixDto);
|
|
}
|
|
|
|
private static async Task<IResult> SetAsync(HttpContext http, AgentService agents, LinkService links, LinkRuntime runtime, TimeProvider time, string dialer_id, string listener_id, SetLinkRequest request, CancellationToken ct)
|
|
{
|
|
(AgentView? dialer, AgentView? listener, IResult? problem) = await ResolveAsync(http, agents, dialer_id, listener_id, request.Enabled == true, ct);
|
|
if (problem is not null)
|
|
{
|
|
return problem;
|
|
}
|
|
|
|
await links.SetAsync(dialer_id, listener_id, request.Enabled, ct);
|
|
await agents.AfterAgentSetChangedAsync(ct); // 重新下发：拨号方的拨号目标、监听方允许的拨号方
|
|
if (request.Enabled == true)
|
|
{
|
|
await agents.NotifyLinkRetryAsync(dialer!.Agent.Id, listener!.Agent.Id, ct);
|
|
}
|
|
|
|
return TypedResults.Json(ToDto(await links.SnapshotAsync(ct), dialer!, listener!, runtime, time), ApiJsonContext.Default.LinkDto);
|
|
}
|
|
|
|
private static async Task<IResult> RetryAsync(HttpContext http, AgentService agents, LinkService links, LinkRuntime runtime, TimeProvider time, string dialer_id, string listener_id, CancellationToken ct)
|
|
{
|
|
(AgentView? dialer, AgentView? listener, IResult? problem) = await ResolveAsync(http, agents, dialer_id, listener_id, requireListening: true, ct);
|
|
if (problem is not null)
|
|
{
|
|
return problem;
|
|
}
|
|
|
|
await links.SetAsync(dialer_id, listener_id, true, ct);
|
|
await agents.AfterAgentSetChangedAsync(ct);
|
|
bool queued = await agents.NotifyLinkRetryAsync(dialer!.Agent.Id, listener!.Agent.Id, ct);
|
|
LinkDto link = ToDto(await links.SnapshotAsync(ct), dialer, listener, runtime, time);
|
|
return TypedResults.Json(new LinkRetryDto(queued, link), ApiJsonContext.Default.LinkRetryDto, statusCode: StatusCodes.Status202Accepted);
|
|
}
|
|
|
|
/// <summary>两端都得是存在且激活的节点，不能连自己；开线路时监听方必须在监听。</summary>
|
|
private static async Task<(AgentView? Dialer, AgentView? Listener, IResult? Problem)> ResolveAsync(HttpContext http, AgentService agents, string dialerId, string listenerId, bool requireListening, CancellationToken ct)
|
|
{
|
|
if (dialerId == listenerId)
|
|
{
|
|
return (null, null, Problems.Create(http, MoShiroErrors.SelfLink, "a node cannot connect to itself"));
|
|
}
|
|
|
|
if (await agents.GetAsync(new NodeId(dialerId), ct) is not { } dialer)
|
|
{
|
|
return (null, null, Problems.Create(http, MoShiroErrors.AgentNotFound, "dialer not found"));
|
|
}
|
|
|
|
if (await agents.GetAsync(new NodeId(listenerId), ct) is not { } listener)
|
|
{
|
|
return (null, null, Problems.Create(http, MoShiroErrors.AgentNotFound, "listener not found"));
|
|
}
|
|
|
|
if (requireListening && !listener.Agent.Listens)
|
|
{
|
|
return (null, null, Problems.Create(http, MoShiroErrors.NotListening, "the listener does not accept connections: only nodes that listen can be connected to"));
|
|
}
|
|
|
|
return (dialer, listener, null);
|
|
}
|
|
|
|
private static LinkDto ToDto(LinkSnapshot snapshot, AgentView dialer, AgentView listener, LinkRuntime runtime, TimeProvider time)
|
|
{
|
|
bool linked = snapshot.IsLinked(dialer.Agent, listener.Agent);
|
|
ReportedLink? reported = runtime.Get(dialer.Agent.Id.Value, listener.Agent.Id.Value, IpVersions.V4);
|
|
LinkHealth? health = reported?.Health;
|
|
bool fresh = reported is not null && time.GetUtcNow().ToUnixTimeMilliseconds() - reported.AtMs <= 30_000;
|
|
string state;
|
|
string? reason = null;
|
|
if (!linked) { state = "disabled"; reason = "线路策略已关闭"; }
|
|
else if (dialer.RuntimeStatus != NodeRuntimeStatus.Running) { state = "dialer_unavailable"; reason = $"拨号节点状态：{dialer.RuntimeStatus}"; }
|
|
else if (listener.RuntimeStatus != NodeRuntimeStatus.Running) { state = "listener_unavailable"; reason = $"监听节点状态：{listener.RuntimeStatus}"; }
|
|
else if (health is null || !fresh)
|
|
{
|
|
state = "waiting_report";
|
|
reason = health is null ? "拨号节点尚未上报此线路的状态" : "拨号节点的线路状态上报已过期";
|
|
}
|
|
else if (health.Control.State == ControlState.Backoff)
|
|
{
|
|
state = "control_down";
|
|
reason = health.Control.LastError is { } fault ? $"与监听节点失联：{fault}" : "与监听节点失联，等待自动重试";
|
|
}
|
|
else if (health.Control.State != ControlState.Up || listener.Dialers.All(d => d.Id != dialer.Agent.Id.Value))
|
|
{
|
|
state = "control_connecting";
|
|
reason = "正在建立到监听节点的控制连接";
|
|
}
|
|
else if (health.Task.Quic.State == QuicState.Ready)
|
|
{
|
|
state = "connected";
|
|
}
|
|
else if (health.Task.Quic.State == QuicState.Retrying)
|
|
{
|
|
state = "task_retrying";
|
|
reason = health.Task.Quic.LastError is { } fault ? $"控制连接正常，QUIC 数据通道重试中：{fault}" : "控制连接正常，QUIC 数据通道重试中";
|
|
}
|
|
else
|
|
{
|
|
state = "task_connecting";
|
|
reason = "控制连接正常，正在建立 QUIC 数据通道";
|
|
}
|
|
|
|
bool inBackoff = health?.Control.State == ControlState.Backoff;
|
|
return new LinkDto(dialer.Agent.Id.Value, dialer.Agent.Name, listener.Agent.Id.Value, listener.Agent.Name, linked,
|
|
snapshot.PairOverride(dialer.Agent.Id.Value, listener.Agent.Id.Value), state == "connected", dialer.RuntimeStatus, listener.RuntimeStatus,
|
|
state, reason, (inBackoff ? health!.Control.LastError : health?.Task.Quic.LastError)?.ToString(),
|
|
health is null ? 0 : inBackoff ? health.Control.RetryAttempt : health.Task.Quic.RetryAttempt,
|
|
health is null ? null : inBackoff ? health.Control.RetryInMs : health.Task.Quic.RetryInMs);
|
|
}
|
|
}
|
|
|
|
public sealed record LinkRetryDto(bool Queued, LinkDto Link);
|