12 Commits
Author SHA1 Message Date
justme 49e787a7a5 backend: confirm a registered update by itself after a delay when auto update is switched on [CI]
ci/woodpecker/push/web-build Pipeline was successful
ci/woodpecker/push/verify Pipeline was successful
ci/woodpecker/push/dev-image Pipeline was successful
ci/woodpecker/push/dev-backend-image Pipeline was successful
2026-10-22 10:59:32 +08:00
justme 4e7a8c2ba3 web: draw isolated points as a connected series and mark unrecorded time in the adaptive audit [CI]
ci/woodpecker/push/web-build Pipeline was successful
ci/woodpecker/push/verify Pipeline was successful
ci/woodpecker/push/dev-backend-image Pipeline was canceled
ci/woodpecker/push/dev-image Pipeline was canceled
2026-10-22 10:57:53 +08:00
justme 93abf59afb backend: leave the time before the first rate record unknown in the adaptive audit 2026-10-22 10:57:53 +08:00
justme 20b0f7fbd8 measurement: record the start of adaptive probing as a rate change 2026-10-22 10:57:53 +08:00
justme af385cf688 web: show the adaptive probe-rate audit on the line page [CI]
ci/woodpecker/push/web-build Pipeline was successful
ci/woodpecker/push/verify Pipeline was successful
ci/woodpecker/push/dev-image Pipeline was successful
ci/woodpecker/push/dev-backend-image Pipeline was successful
2026-10-22 10:44:46 +08:00
justme 9ad34acbac backend: store adaptive rate changes as events and serve an audit of them per line 2026-10-22 10:44:46 +08:00
justme d63c3bff74 agent: keep adaptive rate changes apart from the logs and report them in the heartbeat 2026-10-22 10:44:46 +08:00
justme 557b169dae web: make loss and retransmission charts obvious, draw sparse series as points and keep percent axes within 100% [CI]
ci/woodpecker/push/web-build Pipeline was successful
ci/woodpecker/push/verify Pipeline was successful
ci/woodpecker/push/dev-image Pipeline was successful
ci/woodpecker/push/dev-backend-image Pipeline was successful
2026-10-22 10:14:38 +08:00
justme 4a9608ead4 fleet: count only admitted peers against the session limit so silent connections cannot lock the roster out 2026-10-22 10:14:38 +08:00
justme fee236fe56 web: explain the public-only default when an agent refuses a target [CI]
ci/woodpecker/push/web-build Pipeline was successful
ci/woodpecker/push/verify Pipeline was successful
ci/woodpecker/push/dev-image Pipeline was successful
ci/woodpecker/push/dev-backend-image Pipeline was successful
2026-10-22 07:36:29 +08:00
justme 32547032d8 agent: allow only public addresses as single-agent targets unless allowed_target_cidrs says otherwise 2026-10-22 07:36:29 +08:00
justme 8efc16f341 web: group the QUIC and TCP rows of a line card by protocol [CI]
ci/woodpecker/push/web-build Pipeline was successful
ci/woodpecker/push/verify Pipeline was successful
ci/woodpecker/push/dev-image Pipeline was successful
ci/woodpecker/push/dev-backend-image Pipeline was successful
2026-10-22 07:11:33 +08:00
33 changed files with 1192 additions and 46 deletions

No files matched your search

+4 -2
View File
@@ -45,6 +45,7 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable
private readonly PeerListener _peers;
private readonly PeerHub _hub;
private readonly ChunkAssembler _chunks;
private readonly RateChangeLog _rateChanges = new();
private readonly TaskManager _tasks;
private readonly TestRunner _tests;
private readonly FleetLink _link;
@@ -83,7 +84,7 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable
_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);
_dialer = new DialerHost(new DialEnv(identity.PeerIdentity, identity.NodeId.Value, config.Limits, quic, new DialDirectory(new ConfigStore(dataDirectory), dataDirectory, log, _trust), _clock, time, log, _rateChanges), time, log);
_knock = new KnockGate(identity.PeerIdentity, RosterKeys, time);
_tcp = new TcpAcceptor(bind, config.Listen.TcpPorts, identity.Certificate, _sessions, _counters.Data, _knock);
@@ -101,7 +102,7 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable
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);
var targets = new TargetRunners(_chunks, identity.NodeId, _clock, time, log, _rateChanges);
_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);
@@ -445,6 +446,7 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable
DroppedRateLimited = _counters.DroppedRateLimited,
QueueBytes = _outbox.PendingBytes,
Update = updateStatus,
RateChanges = _rateChanges.Take(64),
LogSummary = logSummary is null ? null : new LogSummaryDto
{
LastId = logSummary.LastId,
@@ -98,7 +98,7 @@ public sealed class LimitsSection
[JsonPropertyName("max_total_pps")]
public long MaxTotalPps { get; set; } = 20_000;
/// <summary>默认为空:不允许执行任何目标任务。</summary>
/// <summary>单个 Agent 任务(外部目标)能探测的网段。默认为空:只允许公网地址,内网、回环、链路本地等一律拒绝;写了就只允许列出的网段(可以包含内网)。</summary>
[JsonPropertyName("allowed_target_cidrs")]
public List<string> AllowedTargetCidrs { get; set; } = [];
@@ -11,7 +11,7 @@ using MoShiro.Protocol;
namespace MoShiro.Agent;
/// <summary>所有连接共用的东西。</summary>
public sealed record DialEnv(NodeIdentity Identity, string NodeId, LimitsSection Limits, QuicSettings Quic, DialDirectory Directory, MonoClock Clock, TimeProvider Time, INodeLog Log);
public sealed record DialEnv(NodeIdentity Identity, string NodeId, LimitsSection Limits, QuicSettings Quic, DialDirectory Directory, MonoClock Clock, TimeProvider Time, INodeLog Log, RateChangeLog RateChanges);
/// <summary>
/// 与一个 Agent 的点对点连接,分两层:
@@ -190,6 +190,12 @@ public sealed class DialConnection : IDisposable
Fragments = shipper,
MaxPacketsPerSecond = _env.Limits.MaxTotalPps,
MaxBulkMbps = _env.Limits.MaxTotalMbps,
OnRateChange = change =>
{
change.Listener = Endpoint.NodeId;
change.IpVersion = Endpoint.IpVersion;
_env.RateChanges.Add(change);
},
};
var session = new DialSession(blockEnv, clock, _outbox, _env);
var rpc = new RpcSession<DialSession>(transport, s_router, session, _env.Time);
@@ -0,0 +1,40 @@
using System.Collections.Concurrent;
using MoShiro.Protocol;
namespace MoShiro.Agent;
/// <summary>
/// 自适应探测频率的状态变化,等着随下一次心跳上报给 Backend(Backend 存成事件,用于事后审计)。
/// 和普通日志分开存:日志环形缓冲几分钟到几小时就会被覆盖,而这里的变化要能在几天后回看。有界,满了丢最旧的。
/// </summary>
public sealed class RateChangeLog
{
public const int Capacity = 512;
private readonly ConcurrentQueue<RateChange> _queue = new();
public void Add(RateChange change)
{
_queue.Enqueue(change);
while (_queue.Count > Capacity && _queue.TryDequeue(out _))
{
}
}
/// <summary>取走最多 <paramref name="max"/> 条(最旧的在前);没有返回 null,心跳里就不带这个字段。</summary>
public List<RateChange>? Take(int max)
{
List<RateChange>? taken = null;
while ((taken?.Count ?? 0) < max)
{
if (!_queue.TryDequeue(out RateChange? change))
{
break;
}
(taken ??= []).Add(change);
}
return taken;
}
}
@@ -24,9 +24,11 @@ public sealed class TargetRunners
private readonly MonoClock _clock;
private readonly TimeProvider _time;
private readonly INodeLog _log;
private readonly RateChangeLog? _rateChanges;
public TargetRunners(ChunkAssembler chunks, NodeId agent, MonoClock clock, TimeProvider time, INodeLog log)
public TargetRunners(ChunkAssembler chunks, NodeId agent, MonoClock clock, TimeProvider time, INodeLog log, RateChangeLog? rateChanges = null)
{
_rateChanges = rateChanges;
_chunks = chunks;
_agent = agent;
_clock = clock;
@@ -92,6 +94,7 @@ public sealed class TargetRunners
PeerAddress = address,
Target = spec.Target,
AllowAddress = allow,
OnRateChange = _rateChanges is null ? null : _rateChanges.Add,
Fragments = new LocalSink(this, spec, address?.AddressFamily == System.Net.Sockets.AddressFamily.InterNetworkV6 ? IpVersions.V6 : IpVersions.V4),
};
IBlock block = BlockCatalog.Find(spec.Kind)!.Dialer(env, new BlockSpec(decision.Kind!, decision.Resolved!, 0, port, new FragmentScope(spec.Id, null, null)));
+31 -4
View File
@@ -27,7 +27,15 @@ public sealed class PeerListenerOptions
public int MaxFrameBytes { get; init; } = 8 << 20;
public int MaxSessions { get; init; } = 8;
/// <summary>
/// 同时在线的会话数上限,只数已经通过敲门的对端。每个拨号方每个地址族一个会话,所以要容得下「名册里的节点数 × 2」。
/// 之前这个上限连还没敲门的连接一起数:节点多起来之后,已建立的会话把名额占满,后到的拨号方(哪怕在名册里)一连上就被断开,
/// 而且一直连不上,重试只会得到同样的结果。
/// </summary>
public int MaxSessions { get; init; } = 64;
/// <summary>还没通过敲门的连接数上限(陌生人、扫描器、还在握手的):敲门有读超时,占着名额的时间有限,但它们不能挤占已通过敲门的会话。</summary>
public int MaxPending { get; init; } = 64;
/// <summary>敲门、TLS、挑战应答整个握手的上限。线路差的时候(RTT 好几秒、还丢包)握手本来就要很久,10 秒做不完就会一直连不上;敲门已经挡住了不认识的连接,放宽不会放大被占连接的风险。</summary>
public TimeSpan HandshakeTimeout { get; init; } = PeerTimings.Handshake;
@@ -44,6 +52,7 @@ public sealed class PeerListener
private readonly INodeLog _log;
private readonly TimeProvider _time;
private int _sessions;
private int _pending;
public PeerListener(PeerListenerOptions options, Func<PeerSessionInfo, IRpcTransport, CancellationToken, Task> onSession, INodeLog log, TimeProvider time)
{
@@ -74,9 +83,9 @@ public sealed class PeerListener
while (!ct.IsCancellationRequested)
{
TcpClient client = await listener.AcceptTcpClientAsync(ct);
if (Interlocked.Increment(ref _sessions) > _options.MaxSessions)
if (Interlocked.Increment(ref _pending) > _options.MaxPending)
{
Interlocked.Decrement(ref _sessions);
Interlocked.Decrement(ref _pending);
client.Dispose();
continue;
}
@@ -97,6 +106,7 @@ public sealed class PeerListener
private async Task HandleAsync(TcpClient client, CancellationToken ct)
{
bool admitted = false;
bool counted = false;
string phase = "knock";
long started = _time.GetTimestamp();
EndPoint? remote = client.Client.RemoteEndPoint;
@@ -113,6 +123,15 @@ public sealed class PeerListener
}
admitted = true;
Interlocked.Decrement(ref _pending);
if (Interlocked.Increment(ref _sessions) > _options.MaxSessions)
{
Interlocked.Decrement(ref _sessions);
_log.Warn("peer session limit reached", ("remote", remote), ("limit", _options.MaxSessions));
return;
}
counted = true;
phase = "tls";
_log.Info("peer knock accepted", ("remote", remote), ("phase", phase));
var tls = new SslStream(stream, leaveInnerStreamOpen: false);
@@ -152,7 +171,15 @@ public sealed class PeerListener
finally
{
client.Dispose();
Interlocked.Decrement(ref _sessions);
if (!admitted)
{
Interlocked.Decrement(ref _pending);
}
if (counted)
{
Interlocked.Decrement(ref _sessions);
}
}
}
+61 -1
View File
@@ -303,7 +303,7 @@ public sealed class LinkTests
private readonly CancellationTokenSource _stop = new();
private readonly NodeIdentity _identity;
public PeerFixture(Func<string, byte[]?> lookup, Func<PeerSessionInfo, IRpcTransport, CancellationToken, Task>? onSession = null)
public PeerFixture(Func<string, byte[]?> lookup, Func<PeerSessionInfo, IRpcTransport, CancellationToken, Task>? onSession = null, int maxSessions = 64)
{
Certificate = NodeCertificate.LoadPfx(NodeCertificate.CreateSelfSignedPfx("agent", TimeSpan.FromDays(30), TimeProvider.System));
_identity = NodeIdentity.FromPkcs8(Certificate.GetECDsaPrivateKey()!.ExportPkcs8PrivateKey());
@@ -314,6 +314,7 @@ public sealed class LinkTests
Certificate = Certificate,
LookupPeerKey = lookup,
Knock = new KnockGate(_identity, () => KnockPeers, TimeProvider.System),
MaxSessions = maxSessions,
},
onSession ?? ((_, transport, ct) => KeepOpenAsync(transport, ct)),
NullNodeLog.Instance, TimeProvider.System);
@@ -462,6 +463,65 @@ public sealed class LinkTests
await ok.DisposeAsync();
}
/// <summary>名额只数通过敲门的会话:一堆还没敲门的连接占着,名册里的拨号方照样连得上(之前它们共用一个上限,节点多了之后后到的拨号方被永远拒绝)。</summary>
[Fact]
public async Task PeerListener_SessionLimitCountsOnlyAdmittedPeers_SoSilentConnectionsCannotLockOutTheRoster()
{
using NodeIdentity first = NodeIdentity.Generate();
using NodeIdentity second = NodeIdentity.Generate();
using NodeIdentity third = NodeIdentity.Generate();
await using var agent = new PeerFixture(id => id == "dlr_1" ? first.PublicKeySpki : id == "dlr_2" ? second.PublicKeySpki : third.PublicKeySpki, maxSessions: 2);
agent.KnockPeers.AddRange([first.PublicKeySpki, second.PublicKeySpki, third.PublicKeySpki]);
await agent.Listener.Started.Task;
// 20 个什么都不说的连接:占着待敲门的名额,但不占会话名额
var silent = new List<TcpClient>();
try
{
for (int i = 0; i < 20; i++)
{
var client = new TcpClient();
await client.ConnectAsync(IPAddress.Loopback, agent.Listener.Port);
silent.Add(client);
}
(_, TlsFrameTransport a) = await PeerConnector.ConnectAsync("127.0.0.1", agent.Listener.Port, agent.PublicKey, first, "dlr_1", null, 1 << 20, TimeProvider.System, default);
(_, TlsFrameTransport b) = await PeerConnector.ConnectAsync("127.0.0.1", agent.Listener.Port, agent.PublicKey, second, "dlr_2", null, 1 << 20, TimeProvider.System, default);
// 会话名额(2)满了:第三个在名册里的拨号方被拒绝,一个会话结束之后又能连上
await Assert.ThrowsAnyAsync<Exception>(async () =>
{
(_, TlsFrameTransport refused) = await PeerConnector.ConnectAsync("127.0.0.1", agent.Listener.Port, agent.PublicKey, third, "dlr_3", null, 1 << 20, TimeProvider.System, default);
await refused.DisposeAsync();
});
await a.DisposeAsync();
bool reconnected = false;
for (int attempt = 0; attempt < 50 && !reconnected; attempt++)
{
try
{
(_, TlsFrameTransport again) = await PeerConnector.ConnectAsync("127.0.0.1", agent.Listener.Port, agent.PublicKey, third, "dlr_3", null, 1 << 20, TimeProvider.System, default);
await again.DisposeAsync();
reconnected = true;
}
catch (Exception ex) when (ex is IOException or System.Security.Authentication.AuthenticationException or InvalidOperationException)
{
await Task.Delay(100);
}
}
Assert.True(reconnected);
await b.DisposeAsync();
}
finally
{
foreach (TcpClient client in silent)
{
client.Dispose();
}
}
}
[Fact]
public async Task PeerListener_WritesNothingToStrangers_NotEvenATlsServerHello()
{
@@ -85,6 +85,7 @@ public abstract class BlockBase : IBlock
}
double max = Math.Max(Spec.Int("rate_max_hz"), Spec.Int("rate_min_hz"));
ReportStart(max);
return new AdaptiveRate(new AdaptiveRateOptions
{
MinHz = Spec.Int("rate_min_hz"),
@@ -95,8 +96,38 @@ public abstract class BlockBase : IBlock
});
}
/// <summary>
/// 自适应开始:记一条「start → watch」(从最高频率起步)。审计据此知道这个任务从什么时候起在自适应,
/// 没有这条之前的时段是「没有记录」,不是某个状态。
/// </summary>
protected void ReportStart(double hz) =>
Env.OnRateChange?.Invoke(new RateChange
{
AtMs = Env.Time.GetUtcNow().ToUnixTimeMilliseconds(),
Task = Spec.Scope.Task?.Value,
Kind = Spec.KindId,
From = "start",
To = "watch",
Hz = Math.Round(hz, 3),
Reason = "start",
});
private void LogTransition(ProbeTransition change)
{
Env.OnRateChange?.Invoke(new RateChange
{
AtMs = Env.Time.GetUtcNow().ToUnixTimeMilliseconds(),
Task = Spec.Scope.Task?.Value,
Kind = Spec.KindId,
From = change.From.ToString().ToLowerInvariant(),
To = change.To.ToString().ToLowerInvariant(),
Hz = Math.Round(change.Hz, 3),
Reason = change.Reason,
LossPct = Math.Round(change.LossPct, 4),
Sent = change.Sent,
Lost = change.Lost,
});
var fields = new (string, object?)[]
{
("stream", Spec.StreamId), ("kind", Spec.KindId), ("from", change.From.ToString().ToLowerInvariant()), ("to", change.To.ToString().ToLowerInvariant()), ("hz", Math.Round(change.Hz, 1)),
@@ -143,6 +143,7 @@ public sealed class HttpDialer : BlockBase
}
double max = Math.Max(Spec.Int("rate_max_per_min"), Spec.Int("rate_min_per_min")) / 60.0;
ReportStart(max);
return new AdaptiveRate(new AdaptiveRateOptions
{
MinHz = Spec.Int("rate_min_per_min") / 60.0,
@@ -245,6 +245,9 @@ public sealed class BlockEnv
/// <summary>单个 Agent 任务:解析出来的地址能不能连(本地 <c>allowed_target_cidrs</c>);null 表示不限制。</summary>
public Func<IPAddress, bool>? AllowAddress { get; init; }
/// <summary>自适应频率的状态变化(用于审计);null 表示不记录。</summary>
public Action<RateChange>? OnRateChange { get; init; }
/// <summary>拨号方:对端 TLS 证书的 SPKI 指纹;null 表示不校验证书(目标任务,目标是任意地址)。</summary>
public byte[]? PeerSpkiSha256 { get; init; }
@@ -277,6 +280,7 @@ public sealed class BlockEnv
PeerAddress = PeerAddress,
Target = Target,
AllowAddress = AllowAddress,
OnRateChange = OnRateChange,
PeerSpkiSha256 = PeerSpkiSha256,
KnockKey = KnockKey,
Fragments = Fragments,
@@ -107,8 +107,8 @@ public sealed class BudgetGuard
}
/// <summary>
/// 目标在静态上可不可以接受:IP 字面量按 <c>allowed_target_cidrs</c> 检查;域名(包括 URL 里的)要解析之后才知道,
/// 这里只要求配置过白名单,运行时每次解析出的地址再用 <see cref="AddressAllowed"/> 检查。
/// 目标在静态上可不可以接受:IP 字面量直接用 <see cref="AddressAllowed"/> 判断;域名(包括 URL 里的)要解析之后才知道,
/// 这里先接受,运行时每次解析出的地址再逐个检查。
/// </summary>
public bool TargetAllowed(string? target)
{
@@ -117,9 +117,13 @@ public sealed class BudgetGuard
return false;
}
return IPAddress.TryParse(host, out IPAddress? address) ? AddressAllowed(address) : _cidrs.Count > 0;
return !IPAddress.TryParse(host, out IPAddress? address) || AddressAllowed(address);
}
/// <summary>
/// 这个地址能不能作为单个 Agent 任务的目标。<c>allowed_target_cidrs</c> 没配置时只允许公网地址(内网、回环、链路本地、组播等一律拒绝),
/// 防止被攻破的 Backend 借 Agent 探测它所在的内网;配置了就只允许列出的网段(包括内网段,由节点所有者显式决定)。
/// </summary>
public bool AddressAllowed(IPAddress address)
{
if (address.IsIPv4MappedToIPv6)
@@ -127,6 +131,11 @@ public sealed class BudgetGuard
address = address.MapToIPv4();
}
if (_limits.AllowedTargetCidrs.Count == 0)
{
return IsPublic(address);
}
foreach ((IPAddress network, int prefix) in _cidrs)
{
if (InCidr(address, network, prefix))
@@ -138,6 +147,28 @@ public sealed class BudgetGuard
return false;
}
private static readonly (IPAddress Network, int Prefix)[] s_nonPublic =
[
(IPAddress.Parse("0.0.0.0"), 8), (IPAddress.Parse("10.0.0.0"), 8), (IPAddress.Parse("100.64.0.0"), 10), (IPAddress.Parse("127.0.0.0"), 8),
(IPAddress.Parse("169.254.0.0"), 16), (IPAddress.Parse("172.16.0.0"), 12), (IPAddress.Parse("192.0.0.0"), 24), (IPAddress.Parse("192.168.0.0"), 16),
(IPAddress.Parse("198.18.0.0"), 15), (IPAddress.Parse("224.0.0.0"), 3),
(IPAddress.Parse("::"), 128), (IPAddress.Parse("::1"), 128), (IPAddress.Parse("fc00::"), 7), (IPAddress.Parse("fe80::"), 10), (IPAddress.Parse("ff00::"), 8),
];
/// <summary>不是公网:私有、回环、链路本地、运营商级 NAT、组播、保留地址。</summary>
private static bool IsPublic(IPAddress address)
{
foreach ((IPAddress network, int prefix) in s_nonPublic)
{
if (InCidr(address, network, prefix))
{
return false;
}
}
return true;
}
private bool PortAllowed(KindSpec kind, ParamValues values)
{
if (!kind.Params.Any(p => p.Name == "port"))
@@ -342,6 +342,10 @@ public sealed class NodeHeartbeatExtra
/// <summary>日志只报计数(不带文本):自上次心跳以来各级别的条数、最多几个出现最多的类型,以及最新一条的编号;文本由 Backend 用 <c>moshiro.logs.fetch</c> 按需拉取。</summary>
[JsonPropertyName("log_summary")]
public LogSummaryDto? LogSummary { get; set; }
/// <summary>自上次心跳以来自适应探测频率的状态变化(节点上限 512 条,超出丢最旧的)。</summary>
[JsonPropertyName("rate_changes")]
public List<RateChange>? RateChanges { get; set; }
}
/// <summary>某一类日志(消息文本的稳定哈希)在一个级别上出现了多少次。</summary>
@@ -357,6 +361,56 @@ public sealed class LogTypeCountDto
public long Count { get; set; }
}
/// <summary>
/// 自适应探测频率的一次状态变化(Quiet / Watch / Active / Dead):节点随心跳上报,Backend 存成事件(<c>rate_change</c>),
/// 事后可以回看每条线路、每个时段的探测频率为什么升降(例如晚高峰)。线路任务带 <c>listener</c>(对端)和 <c>ip_version</c>,目标任务带 <c>task</c>。
/// </summary>
public sealed class RateChange
{
[JsonPropertyName("at_ms")]
public long AtMs { get; set; }
/// <summary>这条线路的监听方节点 id(线路任务)。</summary>
[JsonPropertyName("listener")]
public string? Listener { get; set; }
/// <summary>拨号方节点 id:节点上报时不填,由 Backend 按上报者补上。</summary>
[JsonPropertyName("dialer")]
public string? Dialer { get; set; }
[JsonPropertyName("ip_version")]
public string? IpVersion { get; set; }
/// <summary>单个 Agent 任务的任务 id。</summary>
[JsonPropertyName("task")]
public string? Task { get; set; }
[JsonPropertyName("kind")]
public string Kind { get; set; } = "";
[JsonPropertyName("from")]
public string From { get; set; } = "";
[JsonPropertyName("to")]
public string To { get; set; } = "";
[JsonPropertyName("hz")]
public double Hz { get; set; }
[JsonPropertyName("reason")]
public string Reason { get; set; } = "";
/// <summary>变化时刻证明窗口内的丢包率(百分比)和发出、丢失的包数。</summary>
[JsonPropertyName("loss_pct")]
public double LossPct { get; set; }
[JsonPropertyName("sent")]
public int Sent { get; set; }
[JsonPropertyName("lost")]
public int Lost { get; set; }
}
public sealed class LogSummaryDto
{
[JsonPropertyName("last_id")]
@@ -826,6 +880,7 @@ public static class FragmentKeys
[JsonSerializable(typeof(LogsFetchParams))]
[JsonSerializable(typeof(LogsFetchResult))]
[JsonSerializable(typeof(LogSummaryDto))]
[JsonSerializable(typeof(RateChange))]
[JsonSerializable(typeof(UpdateStatus))]
[JsonSerializable(typeof(NodeHeartbeatExtra))]
[JsonSerializable(typeof(NodeHelloMeta))]
@@ -46,24 +46,37 @@ public sealed class MeasurementTests
}
[Fact]
public void BudgetGuard_TargetTasks_AreDeniedUnlessTheAddressIsInAnAllowedRange()
public void BudgetGuard_TargetTasks_DefaultToPublicAddressesOnly_AndAnExplicitListIsExactly_WhatItSays()
{
Assert.False(Guard().TargetAllowed("1.1.1.1"));
// 没配置白名单:公网地址允许,内网、回环、链路本地、运营商 NAT、组播一律拒绝;域名要解析后逐个地址检查
BudgetGuard open = Guard();
foreach (string allowed in new[] { "1.1.1.1", "1.1.1.1:443", "2606:4700::1111", "example.com:443", "https://example.com/health", "::ffff:8.8.8.8" })
{
Assert.True(open.TargetAllowed(allowed), allowed);
}
BudgetGuard some = Guard(new NodeLimits { AllowedTargetCidrs = ["192.0.2.0/24", "2001:db8::/32"] });
Assert.True(some.TargetAllowed("192.0.2.17"));
foreach (string denied in new[] { "10.1.2.3", "172.16.0.1", "172.31.255.255", "192.168.1.1:80", "127.0.0.1", "169.254.169.254", "100.64.0.1", "224.0.0.1", "0.0.0.0", "::1", "fe80::1", "fd00::1", "::ffff:10.0.0.1", "http://192.168.1.1/" })
{
Assert.False(open.TargetAllowed(denied), denied);
}
Assert.False(open.TargetAllowed(null));
Assert.True(open.AddressAllowed(IPAddress.Parse("172.32.0.1"))); // 172.16/12 的外面是公网
// 配置了白名单:只允许列出的网段,可以包含内网,其余(包括公网)都不允许
BudgetGuard some = Guard(new NodeLimits { AllowedTargetCidrs = ["192.0.2.0/24", "2001:db8::/32", "10.0.0.0/8"] });
Assert.True(some.TargetAllowed("192.0.2.17:443"));
Assert.True(some.TargetAllowed("2001:db8::1"));
Assert.True(some.TargetAllowed("10.9.9.9"));
Assert.False(some.TargetAllowed("1.1.1.1"));
Assert.False(some.TargetAllowed("192.0.3.1"));
Assert.False(some.TargetAllowed(null));
// 域名要解析之后才知道地址:配置过白名单就先接受,每个解析出的地址再由 AddressAllowed 检查
Assert.True(some.TargetAllowed("example.com:443"));
Assert.True(some.TargetAllowed("https://example.com/health"));
Assert.False(Guard().TargetAllowed("example.com"));
Assert.True(some.TargetAllowed("example.com"));
Assert.True(some.AddressAllowed(IPAddress.Parse("192.0.2.9")));
Assert.False(some.AddressAllowed(IPAddress.Parse("10.0.0.1")));
Assert.False(some.AddressAllowed(IPAddress.Parse("::ffff:10.0.0.1")));
Assert.False(some.AddressAllowed(IPAddress.Parse("8.8.8.8")));
Assert.False(some.AddressAllowed(IPAddress.Parse("::ffff:8.8.8.8")));
// 写了但一条也解析不出来:当成「什么都不允许」,不退回到默认
Assert.False(Guard(new NodeLimits { AllowedTargetCidrs = ["not-a-network"] }).AddressAllowed(IPAddress.Parse("1.1.1.1")));
}
// ---- RowStore ----
@@ -99,6 +99,7 @@ public static class ApiModule
RouteGroupBuilder publicGroup = app.MapGroup("/api/public").AddEndpointFilter(PublicGuard.InvokeAsync);
QueryEndpoints.Map(publicGroup);
TargetEndpoints.Map(publicGroup);
AdaptiveEndpoints.Map(publicGroup);
LiveEndpoints.Map(publicGroup);
HostEndpoints.Map(publicGroup);
NodeEndpoints.Map(publicGroup);
@@ -18,6 +18,7 @@ namespace MoShiro.Backend.Api;
[JsonSerializable(typeof(IdDto))]
[JsonSerializable(typeof(OverviewDto))]
[JsonSerializable(typeof(SeriesDto))]
[JsonSerializable(typeof(AdaptiveDto))]
[JsonSerializable(typeof(CompareDto))]
[JsonSerializable(typeof(DistributionDto))]
[JsonSerializable(typeof(HeatmapDto))]
@@ -240,3 +240,13 @@ public sealed record TargetDto(
IReadOnlyList<long> T,
IReadOnlyDictionary<string, double?[]> Values,
SummaryDto Summary);
/// <summary>一个时间桶里一种采集方法的自适应探测频率:各状态的秒数、时间加权的平均频率(Hz,不知道时为 null)、状态变化次数、变化时的最大丢包率(%)。</summary>
public sealed record AdaptiveBucketDto(DateTimeOffset Start, double QuietSeconds, double WatchSeconds, double ActiveSeconds, double DeadSeconds, double? AverageHz, int Changes, double MaxLossPct);
public sealed record AdaptiveKindDto(string Kind, IReadOnlyList<AdaptiveBucketDto> Buckets);
/// <summary>自适应探测频率的一次状态变化:从哪个状态到哪个状态、之后的频率、原因,以及当时的丢包率和包数。</summary>
public sealed record AdaptiveChangeDto(DateTimeOffset At, string Kind, string From, string To, double Hz, string Reason, double LossPct, int Sent, int Lost);
public sealed record AdaptiveDto(string LineId, DateTimeOffset From, DateTimeOffset To, string Step, IReadOnlyList<AdaptiveKindDto> Kinds, IReadOnlyList<AdaptiveChangeDto> Changes);
@@ -0,0 +1,68 @@
using Fleet.Core;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Routing;
using MoShiro.Backend.Application;
using MoShiro.Protocol;
namespace MoShiro.Backend.Api;
/// <summary>自适应探测频率的审计:一条线路在一段时间里每种采集方法各状态占多久、平均频率多少,以及每一次状态变化和原因。</summary>
public static class AdaptiveEndpoints
{
private const string Tag = "public";
private const long MaxRangeMs = 30 * Steps.Day;
private const int MaxBuckets = 1000;
public static void Map(RouteGroupBuilder group) =>
group.MapGet("/lines/{line_id}/adaptive", AdaptiveAsync)
.Api("lines_adaptive", "Adaptive probe-rate states of one line over time (audit)", Tag)
.Returns<AdaptiveDto>().ProducesProblems(403, 404, 422, 503);
private static Task<IResult> AdaptiveAsync(
HttpContext http, LineService lines, AdaptiveAuditService audit, TimeProvider time, string line_id, string? range, DateTimeOffset? from, DateTimeOffset? to, string? step) =>
QueryHelpers.GuardedAsync(http, async ct =>
{
Result<LineView> found = await QueryHelpers.FindVisibleLineAsync(http, lines, line_id, ct);
if (!found.TryGet(out LineView? view, out FleetError? error))
{
return Problems.From(http, error);
}
long now = time.GetUtcNow().ToUnixTimeMilliseconds();
long toMs = to?.ToUnixTimeMilliseconds() ?? now;
long spanMs;
if (from is { } start)
{
spanMs = toMs - start.ToUnixTimeMilliseconds();
}
else if (!RangeResolver.TryParseSpan(range ?? "48h", out spanMs))
{
return Problems.From(http, ValidationError.Single("range", "must look like 48h or 7d"));
}
if (spanMs <= 0 || spanMs > MaxRangeMs)
{
return Problems.From(http, ValidationError.Single("range", "must be between 1 minute and 30 days"));
}
long stepMs = Steps.Hour;
if (step is not null && (!RangeResolver.TryParseSpan(step, out stepMs) || stepMs < Steps.Minute))
{
return Problems.From(http, ValidationError.Single("step", "must look like 15m or 1h"));
}
if (spanMs / stepMs > MaxBuckets)
{
return Problems.From(http, ValidationError.Single("step", $"too fine for this range (at most {MaxBuckets} buckets)"));
}
long fromMs = toMs - spanMs;
AdaptiveResult result = await audit.GetAsync(view.Key, fromMs, toMs, stepMs, ct);
var dto = new AdaptiveDto(
view.Key.Id, Mappers.Time(result.FromMs), Mappers.Time(result.ToMs), Steps.Format(result.StepMs),
[.. result.Kinds.Select(k => new AdaptiveKindDto(k.Kind, [.. k.Buckets.Select(b => new AdaptiveBucketDto(Mappers.Time(b.StartMs), b.QuietSeconds, b.WatchSeconds, b.ActiveSeconds, b.DeadSeconds, b.AverageHz, b.Changes, b.MaxLossPct))]))],
[.. result.Changes.Select(c => new AdaptiveChangeDto(Mappers.Time(c.AtMs), c.Kind, c.From, c.To, c.Hz, c.Reason, c.LossPct, c.Sent, c.Lost))]);
return TypedResults.Json(dto, ApiJsonContext.Default.AdaptiveDto);
});
}
@@ -108,6 +108,22 @@ public sealed class MoShiroNodeEventSink : INodeEventSink
}
}
/// <summary>
/// 自适应探测频率的状态变化存成事件(<see cref="AdaptiveAudit.EventKind"/>):线路任务记在监听方名下(和线路的数据同一个归属),
/// 单个 Agent 任务记在上报的节点名下。时间以节点上报的为准,明显不对(未来、或太久以前)的用心跳的时间。
/// </summary>
private async ValueTask RecordRateChangesAsync(NodeId reporter, long heartbeatMs, List<RateChange> changes, CancellationToken ct)
{
foreach (RateChange change in changes.Take(256))
{
change.Dialer = reporter.Value;
long at = change.AtMs > heartbeatMs - 3_600_000 && change.AtMs <= heartbeatMs + 60_000 ? change.AtMs : heartbeatMs;
change.AtMs = at;
NodeId owner = change.Listener is { Length: > 0 } listener ? new NodeId(listener) : reporter;
await _events.AddAsync(owner, AdaptiveAudit.EventKind, at, System.Text.Json.JsonSerializer.Serialize(change, MoShiroJsonContext.Default.RateChange), ct);
}
}
private async ValueTask ApplyHeartbeatAsync(NodeId id, NodeEvent nodeEvent, System.Text.Json.JsonElement extra, CancellationToken ct)
{
await _nodes.SetVersionAsync(id.Value, nodeEvent.Node.NodeVersion, ct);
@@ -129,6 +145,11 @@ public sealed class MoShiroNodeEventSink : INodeEventSink
}
_nodeLogs.Observe(id, beat.LogSummary);
if (beat.RateChanges is { Count: > 0 } rateChanges)
{
await RecordRateChangesAsync(id, nodeEvent.AtMs, rateChanges, ct);
}
state.ReplaceDialers(beat.Dialers.Select(d => new ConnectedDialerInfo(d.Id.Value, d.Version, d.Remote, d.SinceUtcMs, d.IpVersion)));
foreach (ConnectedDialer dialer in beat.Dialers)
{
@@ -33,6 +33,7 @@ public static class ApplicationServiceCollectionExtensions
services.AddSingleton<LinkService>();
services.AddSingleton<LineService>();
services.AddSingleton<TargetService>();
services.AddSingleton<AdaptiveAuditService>();
services.AddSingleton<TestService>();
services.AddSingleton<IngestService>();
services.AddSingleton<WindowComputer>();
@@ -51,6 +52,8 @@ public static class ApplicationServiceCollectionExtensions
services.AddSingleton<ReleaseService>();
services.AddSingleton<ScheduleService>();
services.AddSingleton<BackendUpdates>();
services.AddSingleton<BackendAutoUpdater>();
services.AddHostedService(provider => provider.GetRequiredService<BackendAutoUpdater>());
services.AddSingleton<JoinTokenService>();
services.AddSingleton<RawBrowserService>();
services.AddSingleton<EventService>();
@@ -21,6 +21,15 @@ public sealed class BackendOptions
/// </summary>
public string? UpdateMode { get; set; } = Environment.GetEnvironmentVariable("MOSHIRO_UPDATE");
/// <summary>
/// 自动更新(环境变量 <c>MoShiro__AutoUpdate=true</c>):新版本登记之后过了 <see cref="AutoUpdateDelayMinutes"/> 分钟自动确认,和管理员点「更新并重启」是同一条路径
/// (先备份数据库,备份失败就不更新)。默认关:Backend 重启会让所有节点断一两分钟。同样需要 <see cref="UpdateMode"/> 是 <c>restart</c>。
/// </summary>
public bool AutoUpdate { get; set; }
/// <summary>新版本登记之后等多久再自动更新(分钟):连着推几次的时候只更新到最后一版,也给 CI 里同时发布的 Agent 版本留出时间。</summary>
public int AutoUpdateDelayMinutes { get; set; } = 10;
/// <summary>
/// 每个 Agent 都必须有属主(passkey):创建加入码时必须已经建立信任根,并且至少选一把操作员集合里的 passkey 作为属主。
/// 属主是节点敏感内容(extension 会话、加密数据)绑定的人,装上时固定在节点本地,面板改不了。默认开;没有信任根的老部署才需要关掉。
@@ -0,0 +1,156 @@
using System.Text.Json;
using Fleet.Core;
using MoShiro.Protocol;
namespace MoShiro.Backend.Application;
/// <summary>一个时间桶里一种采集方法的自适应探测频率:各状态各占多少秒、时间加权的平均频率、状态变化次数、变化时的最大丢包率。</summary>
public sealed record AdaptiveBucket(long StartMs, double QuietSeconds, double WatchSeconds, double ActiveSeconds, double DeadSeconds, double? AverageHz, int Changes, double MaxLossPct);
public sealed record AdaptiveKind(string Kind, IReadOnlyList<AdaptiveBucket> Buckets);
public sealed record AdaptiveResult(long FromMs, long ToMs, long StepMs, IReadOnlyList<AdaptiveKind> Kinds, IReadOnlyList<RateChange> Changes);
/// <summary>
/// 自适应探测频率的审计:节点把每一次状态变化(Quiet / Watch / Active / Dead、频率、原因、丢包)随心跳上报,Backend 存成事件(保留期和别的事件一样),
/// 这里把它们还原成「某个时段里各状态占多久、平均频率多少」,所以几天之后也能回答「晚高峰的时候自适应是什么状态」。
/// </summary>
public sealed class AdaptiveAuditService
{
private readonly IEventStore _events;
public AdaptiveAuditService(IEventStore events)
{
_events = events;
}
public async ValueTask<AdaptiveResult> GetAsync(LineKey line, long fromMs, long toMs, long stepMs, CancellationToken ct)
{
// 往前多取一段:区间开头的状态要由这之前最近的一次变化决定
IReadOnlyList<EventRecord> records = await _events.ListAsync(line.Listener, AdaptiveAudit.EventKind, fromMs - AdaptiveAudit.LookBackMs, toMs, null, 20_000, ct);
List<RateChange> changes = [];
foreach (EventRecord record in records.OrderBy(r => r.AtMs).ThenBy(r => r.Id))
{
RateChange? change = AdaptiveAudit.Parse(record);
if (change is not null && change.Dialer == line.Dialer.Value && (change.IpVersion ?? "ipv4") == line.IpVersion && change.Listener == line.Listener.Value)
{
changes.Add(change);
}
}
return AdaptiveAudit.Summarize(changes, fromMs, toMs, stepMs);
}
}
public static class AdaptiveAudit
{
public const string EventKind = "rate_change";
/// <summary>区间开头之前最多往回看多久,找那时的状态。</summary>
public const long LookBackMs = 24 * 3_600_000L;
public static RateChange? Parse(EventRecord record)
{
if (record.Detail is null)
{
return null;
}
try
{
RateChange? change = JsonSerializer.Deserialize(record.Detail, MoShiroJsonContext.Default.RateChange);
if (change is not null)
{
change.AtMs = record.AtMs;
}
return change;
}
catch (JsonException)
{
return null;
}
}
/// <summary>
/// 把按时间排好的状态变化还原成每个时间桶里各状态的秒数。一次变化之后到下一次变化之前,是变化后的状态和频率;
/// 第一次记录之前什么都不知道,不计入任何状态(节点启动时会记一条「start → watch」,所以每个自适应的任务从启动起就有记录;
/// 没有这条记录的时段是节点还没升级到会上报的版本)。一种采集方法一条时间线(同一条线路上每种方法各自变化)。
/// </summary>
public static AdaptiveResult Summarize(IReadOnlyList<RateChange> changes, long fromMs, long toMs, long stepMs)
{
int count = (int)Math.Max(1, Math.Ceiling((toMs - fromMs) / (double)stepMs));
var kinds = new List<AdaptiveKind>();
foreach (IGrouping<string, RateChange> group in changes.GroupBy(c => c.Kind).OrderBy(g => g.Key, StringComparer.Ordinal))
{
var quiet = new double[count];
var watch = new double[count];
var active = new double[count];
var dead = new double[count];
var hzWeighted = new double[count];
var hzSeconds = new double[count];
var changeCounts = new int[count];
var maxLoss = new double[count];
RateChange[] ordered = [.. group.OrderBy(c => c.AtMs)];
// 每一段:[这次变化, 下一次变化) 内是变化后的状态和频率;第一次记录之前不知道,不计
string state = ordered[0].To;
double? hz = ordered[0].Hz;
long cursor = ordered[0].AtMs;
foreach (RateChange change in ordered)
{
AddSegment(cursor, change.AtMs, state, hz);
int slot = (int)((change.AtMs - fromMs) / stepMs);
if (change.AtMs >= fromMs && change.AtMs < toMs && slot >= 0 && slot < count)
{
changeCounts[slot]++;
maxLoss[slot] = Math.Max(maxLoss[slot], change.LossPct);
}
cursor = change.AtMs;
state = change.To;
hz = change.Hz;
}
AddSegment(cursor, toMs, state, hz);
var buckets = new List<AdaptiveBucket>(count);
for (int i = 0; i < count; i++)
{
buckets.Add(new AdaptiveBucket(fromMs + (i * stepMs), Math.Round(quiet[i], 1), Math.Round(watch[i], 1), Math.Round(active[i], 1), Math.Round(dead[i], 1),
hzSeconds[i] > 0 ? Math.Round(hzWeighted[i] / hzSeconds[i], 3) : null, changeCounts[i], maxLoss[i]));
}
kinds.Add(new AdaptiveKind(group.Key, buckets));
void AddSegment(long start, long end, string segmentState, double? segmentHz)
{
start = Math.Max(start, fromMs);
end = Math.Min(end, toMs);
for (long t = start; t < end;)
{
int slot = (int)((t - fromMs) / stepMs);
long slotEnd = Math.Min(end, fromMs + ((slot + 1) * stepMs));
double seconds = (slotEnd - t) / 1000.0;
switch (segmentState)
{
case "quiet": quiet[slot] += seconds; break;
case "watch": watch[slot] += seconds; break;
case "active": active[slot] += seconds; break;
case "dead": dead[slot] += seconds; break;
}
if (segmentHz is { } known)
{
hzWeighted[slot] += known * seconds;
hzSeconds[slot] += seconds;
}
t = slotEnd;
}
}
}
RateChange[] inRange = [.. changes.Where(c => c.AtMs >= fromMs && c.AtMs < toMs).OrderByDescending(c => c.AtMs).Take(200)];
return new AdaptiveResult(fromMs, toMs, stepMs, kinds, inRange);
}
}
@@ -0,0 +1,100 @@
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using MoShiro.Protocol;
namespace MoShiro.Backend.Application;
/// <summary>
/// 自动更新(<see cref="BackendOptions.AutoUpdate"/>,默认关):每分钟看一眼面板上登记的 Backend 新版本,登记满 <see cref="BackendOptions.AutoUpdateDelayMinutes"/> 分钟之后
/// 替管理员确认。确认走 <see cref="BackendUpdates.ConfirmAsync"/>,所以数据库备份、清单签名、「只更新到登记时的那一版」都和手动更新一样;
/// 更新是有序退出,由 Quadlet 单元(<c>Pull=always</c>、<c>Restart=always</c>)拉起新镜像。
/// </summary>
public sealed partial class BackendAutoUpdater : BackgroundService
{
private static readonly TimeSpan s_startupDelay = TimeSpan.FromMinutes(2);
private static readonly TimeSpan s_interval = TimeSpan.FromMinutes(1);
private readonly BackendUpdates _updates;
private readonly BackendOptions _options;
private readonly TimeProvider _time;
private readonly ILogger<BackendAutoUpdater> _log;
public BackendAutoUpdater(BackendUpdates updates, BackendOptions options, TimeProvider time, ILogger<BackendAutoUpdater> log)
{
_updates = updates;
_options = options;
_time = time;
_log = log;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
if (!_options.AutoUpdate)
{
return;
}
try
{
await Task.Delay(s_startupDelay, _time, stoppingToken);
using var timer = new PeriodicTimer(s_interval, _time);
do
{
try
{
await RunOnceAsync(stoppingToken);
}
catch (Exception ex) when (ex is not OperationCanceledException)
{
Failed(ex);
}
}
while (await timer.WaitForNextTickAsync(stoppingToken));
}
catch (OperationCanceledException)
{
// 后端正在停止
}
}
/// <summary>检查一次;确认了更新(Backend 即将有序退出)返回 true。</summary>
public async ValueTask<bool> RunOnceAsync(CancellationToken ct)
{
if (!_options.AutoUpdate || !_updates.Enabled)
{
return false;
}
BackendUpdateView view = await _updates.GetAsync(ct);
if (view.State != "available" || view.Available is not { } offer)
{
return false;
}
// 清单的 seq 是发布时间(秒)
TimeSpan age = _time.GetUtcNow() - DateTimeOffset.FromUnixTimeSeconds(offer.Seq);
if (age < TimeSpan.FromMinutes(Math.Max(0, _options.AutoUpdateDelayMinutes)))
{
return false;
}
Fleet.Core.Result<bool> result = await _updates.ConfirmAsync(offer.Seq, ct);
if (!result.IsSuccess)
{
Refused(offer.Tag, result.Error.Message);
return false;
}
AutoConfirmed(offer.Tag, (int)age.TotalMinutes);
return true;
}
[LoggerMessage(Level = LogLevel.Information, Message = "auto update: {Tag} was offered {Minutes} minutes ago, confirming it")]
private partial void AutoConfirmed(string tag, int minutes);
[LoggerMessage(Level = LogLevel.Warning, Message = "auto update to {Tag} was not started: {Reason}")]
private partial void Refused(string tag, string reason);
[LoggerMessage(Level = LogLevel.Warning, Message = "the auto update check failed")]
private partial void Failed(Exception exception);
}
@@ -0,0 +1,108 @@
using System.Text.Json.Nodes;
using Fleet.Core;
using MoShiro.Backend.Application;
using MoShiro.Protocol;
using MoShiro.TestSupport;
namespace MoShiro.Backend.Tests;
public sealed class AdaptiveAuditTests : IClassFixture<BackendFixture>
{
private static readonly long s_t0 = new DateTimeOffset(2026, 10, 22, 10, 0, 0, TimeSpan.Zero).ToUnixTimeMilliseconds();
private const long Minute = 60_000;
private const long Hour = 3_600_000;
private readonly BackendFixture _backend;
public AdaptiveAuditTests(BackendFixture backend)
{
_backend = backend;
}
private static RateChange Change(long atMs, string kind, string from, string to, double hz, string reason = "quiet", double lossPct = 0) =>
new() { AtMs = atMs, Kind = kind, From = from, To = to, Hz = hz, Reason = reason, LossPct = lossPct, Sent = 1000, Lost = (int)(lossPct * 10) };
/// <summary>状态变化还原成每个时间桶里各状态的秒数:跨桶的一段要切开,区间开头的状态取第一次变化的「从」,不同方法各算各的。</summary>
[Fact]
public void Summarize_SplitsStateTimeAcrossBuckets_AndWeightsTheAverageRateByTime()
{
RateChange[] changes =
[
Change(s_t0, "quic_ping", "start", "watch", 10, "start"), // 10:00 起步:watch 10 Hz
Change(s_t0 + (30 * Minute), "quic_ping", "watch", "quiet", 2), // 10:30 起 quiet 2 Hz
Change(s_t0 + (90 * Minute), "quic_ping", "quiet", "active", 10, "sustained loss", 0.4), // 11:30 起 active 10 Hz
Change(s_t0 + (30 * Minute), "tcp_ping", "quiet", "watch", 20, "rtt above baseline"), // 这条方法 10:30 才有第一条记录
];
AdaptiveResult result = AdaptiveAudit.Summarize(changes, s_t0, s_t0 + (2 * Hour), Hour);
AdaptiveKind quic = Assert.Single(result.Kinds, k => k.Kind == "quic_ping");
Assert.Equal(2, quic.Buckets.Count);
AdaptiveBucket first = quic.Buckets[0];
Assert.Equal((1800, 1800, 0), ((int)first.WatchSeconds, (int)first.QuietSeconds, (int)first.ActiveSeconds)); // 10:00–10:30 watch,10:30–11:00 quiet
Assert.Equal(6, first.AverageHz); // (10 Hz × 30 min + 2 Hz × 30 min) / 60 min
Assert.Equal(2, first.Changes);
AdaptiveBucket second = quic.Buckets[1];
Assert.Equal((1800, 1800), ((int)second.QuietSeconds, (int)second.ActiveSeconds)); // 11:00–11:30 quiet,11:30–12:00 active
Assert.Equal(6, second.AverageHz); // (2 Hz × 30 min + 10 Hz × 30 min) / 60 min
Assert.Equal((1, 0.4), (second.Changes, second.MaxLossPct));
AdaptiveKind tcp = Assert.Single(result.Kinds, k => k.Kind == "tcp_ping");
Assert.Equal((0, 1800), ((int)tcp.Buckets[0].QuietSeconds, (int)tcp.Buckets[0].WatchSeconds)); // 10:00–10:30 没有记录,不算任何状态
Assert.Equal(3600, (int)tcp.Buckets[1].WatchSeconds);
Assert.Equal(4, result.Changes.Count);
Assert.Equal("quic_ping", result.Changes[0].Kind); // 最新的在前
}
/// <summary>变化在区间开头之前发生:那之后的状态延续到区间里。</summary>
[Fact]
public void Summarize_CarriesTheStateInFromBeforeTheRange()
{
RateChange[] changes = [Change(s_t0 - (2 * Hour), "quic_ping", "quiet", "active", 10, "sustained loss", 1)];
AdaptiveResult result = Summarize(changes);
AdaptiveBucket bucket = Assert.Single(Assert.Single(result.Kinds).Buckets);
Assert.Equal(3600, (int)bucket.ActiveSeconds);
Assert.Equal((10, 0), (bucket.AverageHz, bucket.Changes));
Assert.Empty(result.Changes); // 区间里没有变化
AdaptiveResult Summarize(RateChange[] c) => AdaptiveAudit.Summarize(c, s_t0, s_t0 + Hour, Hour);
}
/// <summary>端到端:拨号节点随心跳上报的状态变化存成事件,几天后还能通过接口按时段查到。</summary>
[Fact]
public async Task RateChangesReportedInHeartbeats_AreStored_AndServedByTheAdaptiveEndpoint()
{
await using AgentSimulator first = await _backend.ConnectAgentAsync();
await using AgentSimulator second = await _backend.ConnectAgentAsync();
(AgentSimulator dialer, AgentSimulator listener) = string.CompareOrdinal(first.NodeId.Value, second.NodeId.Value) < 0 ? (first, second) : (second, first); // 小 id 拨大 id
long now = TimeProvider.System.GetUtcNow().ToUnixTimeMilliseconds();
RateChange Of(long atMs, string from, string to, double hz, string reason, double loss)
{
RateChange change = Change(atMs, "quic_ping", from, to, hz, reason, loss);
change.Listener = listener.NodeId.Value;
change.IpVersion = "ipv4";
return change;
}
await dialer.SendRateChangesAsync(Of(now - (20 * Minute), "watch", "quiet", 2, "quiet", 0), Of(now - (10 * Minute), "quiet", "active", 10, "sustained loss", 0.5));
string lineId = listener.NodeId.Value + "~" + dialer.NodeId.Value;
JsonNode? audit = null;
await BackendFixture.EventuallyAsync(async () =>
{
audit = await _backend.GetJsonAsync(_backend.Admin, $"/api/public/lines/{lineId}/adaptive?range=1h&step=10m");
return audit["changes"]!.AsArray().Count == 2;
});
JsonNode quic = Assert.Single(audit!["kinds"]!.AsArray())!;
Assert.Equal("quic_ping", quic["kind"]!.GetValue<string>());
double active = quic["buckets"]!.AsArray().Sum(b => b!["active_seconds"]!.GetValue<double>());
Assert.InRange(active, 500, 700); // 最近 10 分钟 active
JsonNode newest = audit!["changes"]![0]!;
Assert.Equal(("quiet", "active", "sustained loss"), (newest["from"]!.GetValue<string>(), newest["to"]!.GetValue<string>(), newest["reason"]!.GetValue<string>()));
}
}
@@ -69,6 +69,33 @@ public sealed class BackendUpdateTests : IAsyncLifetime
Assert.Equal(HttpStatusCode.Conflict, replay.StatusCode);
}
/// <summary>自动更新默认关;打开之后只在新版本登记满设定的时间才确认,走的是和手动一样的路径(先备份)。</summary>
[Fact]
public async Task AutoUpdate_IsOffByDefault_AndConfirmsOnlyAfterTheOfferHasAged()
{
BackendOptions options = _backend.Services.GetRequiredService<BackendOptions>();
var updater = _backend.Services.GetRequiredService<BackendAutoUpdater>();
using HttpResponseMessage old = await PublishAsync(10); // seq 是发布时间(秒):1970 年,早就过了等待时间
Assert.Equal(HttpStatusCode.Created, old.StatusCode);
Assert.False(await updater.RunOnceAsync(default)); // 默认关
Assert.Equal(0, _backend.Restarts.Count);
options.AutoUpdate = true;
options.AutoUpdateDelayMinutes = 10;
using HttpResponseMessage fresh = await PublishAsync(TimeProvider.System.GetUtcNow().ToUnixTimeSeconds());
Assert.Equal(HttpStatusCode.Created, fresh.StatusCode);
Assert.False(await updater.RunOnceAsync(default)); // 刚登记的版本还要等
Assert.Equal(0, _backend.Restarts.Count);
options.AutoUpdateDelayMinutes = 0;
Assert.True(await updater.RunOnceAsync(default));
Assert.Equal(1, _backend.Restarts.Count);
Assert.Single(Directory.GetFiles(Path.Combine(_backend.DataDirectory, "backups"), "pre-update-*.db")); // 先备份
Assert.False(await updater.RunOnceAsync(default)); // 已经在更新,不会再确认第二次
Assert.Equal(1, _backend.Restarts.Count);
}
[Fact]
public async Task Confirm_BacksUpTheDatabase_ThenExits_AndRejectsASupersededSequence()
{
@@ -223,6 +223,13 @@ public sealed class AgentSimulator : IAsyncDisposable
new HeartbeatParams { Extra = JsonSerializer.SerializeToElement(new NodeHeartbeatExtra { Update = status }, MoShiroJsonContext.Default.NodeHeartbeatExtra) },
FleetRpcContext.Default.HeartbeatParams));
/// <summary>像拨号节点那样,随心跳上报自适应探测频率的状态变化。</summary>
public Task SendRateChangesAsync(params RateChange[] changes) =>
SendAsync(RpcMessage.Notification(
FleetMethods.Heartbeat,
new HeartbeatParams { Extra = JsonSerializer.SerializeToElement(new NodeHeartbeatExtra { RateChanges = [.. changes] }, MoShiroJsonContext.Default.NodeHeartbeatExtra) },
FleetRpcContext.Default.HeartbeatParams));
public async Task<RpcMessage> RequestRawAsync(string method, JsonElement? parameters = null) => await RequestAsync(method, parameters);
public async Task<TaskSet> WaitForTaskSetAsync(Func<TaskSet, bool> predicate, TimeSpan? timeout = null)
+186
View File
@@ -0,0 +1,186 @@
import { useState } from 'react'
import { api } from '@shared/api/client'
import type { AdaptiveBucketDto, AdaptiveDto } from '@shared/api/types'
import { fmtAgo } from '@shared/format'
import { useQuery } from '@shared/hooks'
import { Segmented } from '@shared/ui'
const WINDOWS = [
{ value: '24h', label: '24 小时', step: '30m' },
{ value: '48h', label: '48 小时', step: '1h' },
{ value: '7d', label: '7 天', step: '3h' },
{ value: '30d', label: '30 天', step: '12h' },
] as const
type Window = (typeof WINDOWS)[number]['value']
const STATES = [
{ key: 'quiet', label: '平稳', level: 'ok' },
{ key: 'watch', label: '观察', level: 'warn' },
{ key: 'active', label: '活跃', level: 'bad' },
{ key: 'dead', label: '不通', level: 'dead' },
] as const
type StateKey = (typeof STATES)[number]['key']
const KIND_NAMES: Record<string, string> = { quic_ping: 'QUIC 往返', tcp_ping: 'TCP 往返', tcp_connect: 'TCP 建连', http_request: 'HTTP 请求' }
const STATE_LABEL: Record<string, string> = { start: '启动', quiet: '平稳', watch: '观察', active: '活跃', dead: '不通' }
const secondsOf = (b: AdaptiveBucketDto, state: StateKey) => b[`${state}_seconds` as const]
/** 一个桶里占时最长的状态;没有任何记录(节点那段时间没有变化可报,也不知道状态)是 null。 */
function dominant(b: AdaptiveBucketDto): StateKey | null {
let best: StateKey | null = null
let seconds = 0
for (const s of STATES) {
const v = secondsOf(b, s.key)
if (v > seconds) {
seconds = v
best = s.key
}
}
return best
}
const hhmm = (iso: string, withDate: boolean) => {
const d = new Date(iso)
const time = d.toLocaleTimeString('zh-CN', { hour: '2-digit', minute: '2-digit', hour12: false })
return withDate ? `${d.getMonth() + 1}/${d.getDate()} ${time}` : time
}
function Row({ kind, buckets, step }: { kind: string; buckets: AdaptiveBucketDto[]; step: string }) {
const [hover, setHover] = useState<number | null>(null)
const stepMs = buckets.length > 1 ? Date.parse(buckets[1].start) - Date.parse(buckets[0].start) : 3_600_000
const at = hover === null ? null : buckets[hover]
// 刻度:每个本地零点、以及(短窗口时)每 6 小时一个
const ticks = buckets.flatMap((b, i) => {
const d = new Date(b.start)
const midnight = d.getHours() === 0 && d.getMinutes() === 0
const sixHour = stepMs <= 3_600_000 * 3 && d.getMinutes() === 0 && d.getHours() % 6 === 0
return midnight || sixHour ? [{ pos: (i / buckets.length) * 100, label: midnight ? `${d.getMonth() + 1}/${d.getDate()}` : `${String(d.getHours()).padStart(2, '0')}:00` }] : []
})
return (
<div className="adaptive-row">
<div className="adaptive-name">{KIND_NAMES[kind] ?? kind}</div>
<div className="adaptive-body" onMouseLeave={() => setHover(null)}>
<div className="bars" style={{ ['--cells' as string]: buckets.length }} role="img" aria-label={`${KIND_NAMES[kind] ?? kind} 自适应状态,每格 ${step}`}>
{buckets.map((b, i) => {
const state = dominant(b)
const level = state === null ? 'none' : STATES.find((s) => s.key === state)!.level
return <span key={b.start} className={`bar ${level}${hover === i ? ' on' : ''}`} onMouseEnter={() => setHover(i)} />
})}
</div>
<div className="line-mini-axis" aria-hidden="true">
{ticks.map((t) => (
<span key={t.pos} style={{ left: `${t.pos}%` }}>
{t.label}
</span>
))}
</div>
<div className={`pop num adaptive-pop${at ? ' open' : ''}`} role="tooltip" aria-hidden={!at}>
{at ? (
<>
<b className="pop-time">
{hhmm(at.start, true)}–{hhmm(new Date(Date.parse(at.start) + stepMs).toISOString(), false)}
</b>
<div className="pop-group">
{STATES.map((s) => (
<span key={s.key}>
{s.label} <em>{Math.round(secondsOf(at, s.key) / 60)} 分钟</em>
</span>
))}
<span>
平均频率 <em>{at.average_hz === null ? '—' : `${at.average_hz} Hz`}</em>
</span>
<span>
状态变化 <em>{at.changes} 次</em>
</span>
{at.max_loss_pct > 0 ? (
<span>
变化时丢包 <em className="warn">{at.max_loss_pct}%</em>
</span>
) : null}
</div>
</>
) : null}
</div>
</div>
</div>
)
}
/**
* 自适应探测频率的审计:每种方法每个时段(一格)里占时最长的状态,鼠标悬停看各状态的分钟数、平均频率和变化次数;
* 下面是最近的每一次状态变化和原因。数据来自节点上报的状态变化(存 30 天以上),所以几天后还能回看晚高峰时的情况。
*/
export function AdaptiveAudit({ lineId }: { lineId: string }) {
const [window, setWindow] = useState<Window>('48h')
const info = WINDOWS.find((w) => w.value === window) ?? WINDOWS[1]
const query = useQuery(() => api.get<AdaptiveDto>(`/api/public/lines/${encodeURIComponent(lineId)}/adaptive`, { range: info.value, step: info.step }), [lineId, info.value], 60_000)
const data = query.data
if (!data || data.kinds.length === 0) {
return (
<section className="card adaptive-card">
<div className="adaptive-head">
<b>自适应探测频率</b>
<span className="faint">{query.loading ? '加载中…' : '这段时间没有记录:线路一直平稳(没有状态变化),或者拨号节点还没升级到会上报的版本'}</span>
</div>
</section>
)
}
return (
<section className="card adaptive-card">
<div className="adaptive-head">
<b>自适应探测频率</b>
<span className="faint">每格 {info.step},颜色是这一格里占时最长的状态</span>
<span className="right">
<Segmented<Window> options={WINDOWS.map((w) => ({ value: w.value, label: w.label }))} value={window} onChange={setWindow} />
</span>
</div>
<div className="adaptive-legend">
{STATES.map((s) => (
<span key={s.key}>
<i className={`bar ${s.level}`} /> {s.label}
</span>
))}
<span>
<i className="bar none" /> 无记录
</span>
</div>
{data.kinds.map((k) => (
<Row key={k.kind} kind={k.kind} buckets={k.buckets} step={info.step} />
))}
{data.changes.length > 0 ? (
<details className="adaptive-changes">
<summary>最近 {data.changes.length} 次状态变化</summary>
<table className="table">
<thead>
<tr>
<th>时间</th>
<th>方法</th>
<th>状态</th>
<th>频率</th>
<th>原因</th>
<th>丢包</th>
</tr>
</thead>
<tbody>
{data.changes.map((c) => (
<tr key={`${c.at}-${c.kind}-${c.to}`}>
<td title={fmtAgo(c.at)}>{hhmm(c.at, true)}</td>
<td>{KIND_NAMES[c.kind] ?? c.kind}</td>
<td>
{STATE_LABEL[c.from] ?? c.from} → {STATE_LABEL[c.to] ?? c.to}
</td>
<td className="num">{c.hz} Hz</td>
<td className="muted">{c.reason}</td>
<td className="num">
{c.loss_pct}%({c.lost}/{c.sent})
</td>
</tr>
))}
</tbody>
</table>
</details>
) : null}
</section>
)
}
+17 -10
View File
@@ -100,12 +100,17 @@ export function LineStrips({ lineId, dialer, listener, range, m }: Props) {
<span className={`line-mini-badge ${levelOf(value)}`} style={{ gridRow }}>{text}</span>
</div>
)
const direction = (id: string, route: string, title: string, rows: ReturnType<typeof row>[]) => (
// 一个协议一组:QUIC 的去程和回程放在一起,TCP 的放在一起;每个方向上面一行写明是谁到谁,组底下一条时间轴
const group = (id: string, proto: string, label: (dir: string) => string, entries: { route: string; dir: 'up' | 'down'; value: number | null | undefined; text: string; values: (number | null)[] }[]) => (
<div className="line-mini-direction" key={id}>
<div className="line-mini-route" title={title}>{route}</div>
{rows}
{entries.flatMap((e, k) => [
<div className="line-mini-route" key={`${id}-${e.dir}-route`} style={{ gridRow: k * 2 + 1 }} title={`${e.dir === 'up' ? '去程' : '回程'}:${e.route}`}>
{e.route}
</div>,
row(`${id}-${e.dir}`, k * 2 + 2, proto, label(e.dir === 'up' ? '去程' : '回程'), e.value, e.text, e.values),
])}
{ticks.length > 0 ? (
<div className="line-mini-axis" aria-hidden="true">
<div className="line-mini-axis" aria-hidden="true" style={{ gridRow: entries.length * 2 + 1 }}>
{ticks.map((tick) => (
<span key={tick.pos} style={{ left: `${tick.pos}%` }}>{tick.label}</span>
))}
@@ -113,6 +118,8 @@ export function LineStrips({ lineId, dialer, listener, range, m }: Props) {
) : null}
</div>
)
const up = `${dialer} → ${listener}`
const down = `${listener} → ${dialer}`
return (
<div className="strips" ref={box} onMouseMove={onMove} onMouseLeave={() => setHover(null)}>
<div className="line-mini-stack">
@@ -123,13 +130,13 @@ export function LineStrips({ lineId, dialer, listener, range, m }: Props) {
<small>悬停看 p95 / p99</small>
{series.error ? <span className="line-mini-badge bad" title={`上一次取数失败:${series.error.message}`}>数据未更新</span> : null}
</div>
{direction('up', `${dialer} → ${listener}`, `去程:${dialer} → ${listener}`, [
row('up-quic', 2, 'QUIC', '去程丢包', m['loss.up'], lossText(m['loss.up']), lossUp),
row('up-tcp', 3, 'TCP', '去程重传', m['tcp.retrans.up'], retransText(m['tcp.retrans.up']), retransUp),
{group('quic', 'QUIC', (d) => `${d}丢包`, [
{ route: up, dir: 'up', value: m['loss.up'], text: lossText(m['loss.up']), values: lossUp },
{ route: down, dir: 'down', value: m['loss.down'], text: lossText(m['loss.down']), values: lossDown },
])}
{direction('down', `${listener} → ${dialer}`, `回程:${listener} → ${dialer}`, [
row('down-quic', 2, 'QUIC', '回程丢包', m['loss.down'], lossText(m['loss.down']), lossDown),
row('down-tcp', 3, 'TCP', '回程重传', m['tcp.retrans.down'], retransText(m['tcp.retrans.down']), retransDown),
{group('tcp', 'TCP', (d) => `${d}重传`, [
{ route: up, dir: 'up', value: m['tcp.retrans.up'], text: retransText(m['tcp.retrans.up']), values: retransUp },
{ route: down, dir: 'down', value: m['tcp.retrans.down'], text: retransText(m['tcp.retrans.down']), values: retransDown },
])}
</div>
<div className={`pop num${hover !== null && at !== null ? ' open' : ''}`} role="tooltip" aria-hidden={hover === null}>
+80
View File
@@ -2678,3 +2678,83 @@ input {
.pop .pop-note {
white-space: normal;
}
/* ── 线路详情:自适应探测频率审计 ── */
.adaptive-card {
display: grid;
gap: 12px;
margin-top: 16px;
padding: 16px 20px;
}
.adaptive-head {
display: flex;
flex-wrap: wrap;
align-items: center;
gap: 8px 12px;
}
.adaptive-legend {
display: flex;
flex-wrap: wrap;
gap: 4px 16px;
color: var(--app-ink-muted);
font-size: 12px;
}
.adaptive-legend .bar {
display: inline-block;
width: 12px;
height: 10px;
margin-right: 4px;
vertical-align: middle;
}
.adaptive-row {
display: grid;
grid-template-columns: 88px minmax(0, 1fr);
align-items: start;
gap: 8px;
}
.adaptive-name {
padding-top: 0;
font-size: 13px;
font-weight: 600;
}
.adaptive-body {
position: relative;
display: grid;
gap: 3px;
}
.adaptive-body .bars {
height: 14px;
}
.adaptive-body .line-mini-axis {
grid-column: auto;
grid-row: auto;
}
/* 提示框弹在这一行的上方,不盖住别的行 */
.adaptive-pop {
top: auto;
bottom: calc(100% + 6px);
left: 0;
width: 190px;
transform-origin: left bottom;
}
.adaptive-changes summary {
cursor: pointer;
color: var(--app-ink-muted);
font-size: 13px;
}
/* 自适应审计里「不通」:比「没有记录」(灰、半透明)深 */
.bar.dead {
background: var(--app-ink-muted);
opacity: 1;
}
+9 -4
View File
@@ -10,6 +10,7 @@ import { NodeTag, useNodeInfos } from '@shared/nodes'
import { useQuery } from '@shared/hooks'
import { Chip, Dot, LoadingBar, Segmented, Spinner, Stat } from '@shared/ui'
import { PageError, ScoreChips } from '../lib'
import { AdaptiveAudit } from '../AdaptiveAudit'
import { Minimap } from '../Minimap'
import { readSpec, usePeriod, usePeriodSeries, writeSpec } from '../periods'
import { dayBounds, dayKey, RangePicker, WINDOWS, type WindowValue } from '../ranges'
@@ -55,6 +56,8 @@ interface ChartDef {
group: MetricGroup
/** 占满一行。 */
wide?: boolean
/** 没有才是好的指标:整段为 0 时图上显示的说明 */
quiet?: string
}
const GROUPS = [
@@ -66,10 +69,10 @@ const GROUPS = [
const CHARTS: ChartDef[] = [
{ id: 'rtt', title: '延迟', group: 'quic', unit: 'ms', series: [{ key: 'rtt.p50', label: 'p50', color: '--chart-1' }, { key: 'rtt.p95', label: 'p95', color: '--chart-2' }, { key: 'rtt.p99', label: 'p99', color: '--chart-3' }] },
{ id: 'jitter', title: '抖动(p95)', group: 'quic', unit: 'ms', series: [{ key: 'jitter.up.p95', label: '去程', color: '--chart-2' }, { key: 'jitter.down.p95', label: '回程', color: '--chart-1' }] },
{ id: 'loss', title: '丢包', group: 'quic', unit: '%', series: [{ key: 'loss.up', label: '去程', color: '--chart-2' }, { key: 'loss.down', label: '回程', color: '--chart-3' }, { key: 'loss.total', label: '合计', color: '--chart-1', dash: [4, 3] }] },
{ id: 'loss-runs', title: '最长连续丢包', group: 'quic', unit: '包', series: [{ key: 'loss.run.up_max', label: '去程', color: '--chart-2' }, { key: 'loss.run.down_max', label: '回程', color: '--chart-3' }] },
{ id: 'loss', title: '丢包', group: 'quic', unit: '%', fill: true, quiet: '✓ 这段时间没有丢包', series: [{ key: 'loss.up', label: '去程', color: '--chart-2' }, { key: 'loss.down', label: '回程', color: '--chart-3' }, { key: 'loss.total', label: '合计', color: '--chart-1', dash: [4, 3] }] },
{ id: 'loss-runs', title: '最长连续丢包', group: 'quic', unit: '包', fill: true, quiet: '✓ 这段时间没有连续丢包', series: [{ key: 'loss.run.up_max', label: '去程', color: '--chart-2' }, { key: 'loss.run.down_max', label: '回程', color: '--chart-3' }] },
{ id: 'tcp', title: '往返时延(RTT)', group: 'tcp-connect', unit: 'ms', series: [{ key: 'tcp.rtt.p50', label: 'p50', color: '--chart-1' }, { key: 'tcp.rtt.p95', label: 'p95', color: '--chart-2' }, { key: 'tcp.rtt.max', label: '最大', color: '--chart-3' }] },
{ id: 'retrans', title: '重传率', group: 'tcp-connect', unit: '%', series: [{ key: 'tcp.retrans.up', label: '去程', color: '--chart-2' }, { key: 'tcp.retrans.down', label: '回程', color: '--chart-3' }, { key: 'tcp.retrans', label: '合计', color: '--chart-1', dash: [4, 3] }] },
{ id: 'retrans', title: '重传率', group: 'tcp-connect', unit: '%', fill: true, quiet: '✓ 这段时间没有重传', series: [{ key: 'tcp.retrans.up', label: '去程', color: '--chart-2' }, { key: 'tcp.retrans.down', label: '回程', color: '--chart-3' }, { key: 'tcp.retrans', label: '合计', color: '--chart-1', dash: [4, 3] }] },
{ id: 'connect-time', title: '建连耗时(p95)', group: 'tcp-connect', unit: 'ms', series: [{ key: 'connect.tcp_ms.p95', label: 'TCP 建连', color: '--chart-1' }, { key: 'connect.tls_ms.p95', label: 'TLS 握手', color: '--chart-2' }] },
{ id: 'connect-ok', title: '建连成功率', group: 'tcp-connect', unit: '%', zero: false, series: [{ key: 'connect.success', label: '成功率', color: '--chart-3' }] },
{ id: 'score', title: '综合评分', group: 'score', wide: true, unit: '', series: [{ key: 'score.reality', label: 'REALITY', color: '--chart-1' }, { key: 'score.hy2', label: 'Hy2', color: '--chart-2' }, { key: 'score.mc', label: 'MC', color: '--chart-4' }, { key: 'score.base', label: '综合', color: '--chart-5', dash: [4, 3] }] },
@@ -177,7 +180,7 @@ export function Line() {
]
: raw.map(({ x, values }) => ({ label: x.label, values, color: x.color, dash: x.dash, fill: def.fill }))
if (!list.some((x) => x.values.some((v) => v !== null))) return null
return { def, node: <TimeChart t={s.t} series={list} unit={def.unit} gaps={gaps} zero={def.zero !== false} height={metricGroup === 'quic' ? 220 : def.wide ? 230 : 190} /> }
return { def, node: <TimeChart t={s.t} series={list} unit={def.unit} quiet={def.quiet} gaps={gaps} zero={def.zero !== false} height={metricGroup === 'quic' ? 220 : def.wide ? 230 : 190} /> }
})
return { group, cards: cards.filter((c): c is NonNullable<typeof c> => c !== null) }
}).filter((g) => g.cards.length > 0)
@@ -355,6 +358,8 @@ export function Line() {
{chartSections}
</section>
<AdaptiveAudit lineId={id} />
{test.data && result ? (
<div className="card">
<div className="card-title">最近一次测试 · {test.data.plan}</div>
+32
View File
@@ -86,3 +86,35 @@ export interface TargetDto {
values: Record<string, (number | null)[]>
summary: { metrics: Record<string, number | null>; samples: Record<string, number>; coverage: number }
}
export interface AdaptiveBucketDto {
start: string
quiet_seconds: number
watch_seconds: number
active_seconds: number
dead_seconds: number
average_hz: number | null
changes: number
max_loss_pct: number
}
export interface AdaptiveChangeDto {
at: string
kind: string
from: string
to: string
hz: number
reason: string
loss_pct: number
sent: number
lost: number
}
export interface AdaptiveDto {
line_id: string
from: string
to: string
step: string
kinds: { kind: string; buckets: AdaptiveBucketDto[] }[]
changes: AdaptiveChangeDto[]
}
+41 -6
View File
@@ -26,6 +26,8 @@ interface Props {
gaps?: { from: number; to: number }[]
/** 为真时 `t` 是「从起点算起的毫秒数」而不是 Unix 毫秒:横轴显示 0:00、0:15…,用来把两段时间叠在一起比。 */
offset?: boolean
/** 丢包、重传这类「没有才是好」的图:整段都是 0(或没有值)时,在图上盖一句说明,而不是留一条贴着底边的平线。 */
quiet?: string
}
/** 偏移量(毫秒)→ h:mm:ss / m:ss。 */
@@ -38,6 +40,33 @@ export function fmtOffset(ms: number): string {
return h > 0 ? `${h}:${two(m)}${sec ? ':' + two(sec) : ''}` : `${m}:${two(sec)}`
}
/**
* 稀疏:有值的点里,前后都是空的(孤立的)占三成以上。每 10 秒才一次的建连放在 5 秒的格子里就是隔一格有一格没有,
* 每个点都是孤立的,连不成线,图上什么都看不见;这种序列画出点,并把点连起来。
*/
function isSparse(values: (number | null)[]): boolean {
let filled = 0
let isolated = 0
for (let i = 0; i < values.length; i++) {
if (values[i] === null) continue
filled++
if ((i === 0 || values[i - 1] === null) && (i === values.length - 1 || values[i + 1] === null)) isolated++
}
return filled > 0 && isolated / filled >= 0.3
}
/**
* Y 轴范围:默认从 0 开始,上面留 8%。百分比的轴不会超过 100%(成功率贴着 100% 时不再出现 100.75% 这种刻度),
* 一条平线(最大等于最小)也给出可读的范围。
*/
function yRange(min: number, max: number, zero: boolean, unit: string): [number, number] {
const percent = unit === '%'
const lo = zero ? Math.min(0, min) : percent ? Math.max(0, min - Math.max(1, (max - min) * 0.2)) : min
let hi = max <= min ? min + (percent ? Math.max(1, Math.abs(min) * 0.01) : 1) : max + (max - min) * 0.08
if (percent && max <= 100) hi = Math.min(100, hi)
return [lo, hi > lo ? hi : lo + 1]
}
function cssVar(name: string): string {
return getComputedStyle(document.documentElement).getPropertyValue(name).trim()
}
@@ -61,7 +90,7 @@ function formatValue(v: number | null | undefined, unit: string): string {
* - 在图上拖动选一段放大预览,双击或点「还原」恢复;数据刷新时保持放大的区间;
* - 鼠标悬停显示这个时刻每条线的值。
*/
export function TimeChart({ t, series, height = 220, unit = '', zero = true, gaps = [], offset = false }: Props) {
export function TimeChart({ t, series, height = 220, unit = '', zero = true, gaps = [], offset = false, quiet }: Props) {
const host = useRef<HTMLDivElement>(null)
const tip = useRef<HTMLDivElement>(null)
const plot = useRef<uPlot | null>(null)
@@ -70,7 +99,10 @@ export function TimeChart({ t, series, height = 220, unit = '', zero = true, gap
// 数据变了(轮询、实时)只更新数据,不重建图:重建会闪一下,还会丢掉鼠标位置
const latest = useRef({ t, series, gaps })
latest.current = { t, series, gaps }
const structure = JSON.stringify([series.map((x) => [x.label, x.color, x.dash, x.fill, x.faint]), unit, zero, offset, height])
// 稀疏的序列(例如每分钟只有几次的建连):大部分格子是空的,单独的一个点画成线就看不见,所以画出点并把点连起来
const sparse = series.map((x) => isSparse(x.values))
const allQuiet = quiet !== undefined && series.filter((x) => !x.faint).every((x) => x.values.every((v) => v === null || v <= 0))
const structure = JSON.stringify([series.map((x, i) => [x.label, x.color, x.dash, x.fill, x.faint, sparse[i]]), unit, zero, offset, height])
const dataOf = (): AlignedData => [latest.current.t.map((v) => v / 1000), ...latest.current.series.map((x) => x.values.map((v) => (v === null ? null : v)))]
const reset = useCallback(() => {
@@ -97,7 +129,7 @@ export function TimeChart({ t, series, height = 220, unit = '', zero = true, gap
select: { show: true, left: 0, top: 0, width: 0, height: 0 },
scales: {
x: { time: !offset, range: (_u, min, max) => (zoom.current ? [zoom.current.min, zoom.current.max] : [min, max]) },
y: { range: (_u, min, max) => [zero ? Math.min(0, min) : min, max <= min ? min + 1 : max + (max - min) * 0.08] },
y: { range: (_u, min, max) => yRange(min, max, zero, unit) },
},
axes: [
{ stroke: axis, grid: { stroke: grid, width: 1 }, ticks: { stroke: grid }, font, ...(offset ? { values: (_u: uPlot, ticks: number[]) => ticks.map((v) => fmtOffset(v * 1000)) } : {}) },
@@ -113,8 +145,8 @@ export function TimeChart({ t, series, height = 220, unit = '', zero = true, gap
width: s.faint ? 1 : 1.8,
dash: s.dash,
fill: s.fill ? color + '22' : undefined,
spanGaps: false,
points: { show: false },
spanGaps: sparse[series.indexOf(s)],
points: { show: sparse[series.indexOf(s)], size: 5 },
}
}),
],
@@ -218,7 +250,10 @@ export function TimeChart({ t, series, height = 220, unit = '', zero = true, gap
)}
</span>
</div>
<div ref={host} className="chart" />
<div className="chart-body">
<div ref={host} className="chart" />
{allQuiet ? <div className="chart-quiet">{quiet}</div> : null}
</div>
<div ref={tip} className="chart-tip" />
</div>
)
+1 -1
View File
@@ -11,7 +11,7 @@ export const TARGET_STATUS: Record<TargetDto['status'], { label: string; tone: '
/** Agent 拒绝任务的原因,翻成人话。 */
export const REJECT_REASONS: Record<string, string> = {
policy_denied: '本机不允许这个目标:把它所在的网段加进 agent.json 的 limits.allowed_target_cidrs',
policy_denied: '本机不允许这个目标:默认只允许公网地址(目标在内网、回环或链路本地);如果配置过 agent.json 的 limits.allowed_target_cidrs,则只允许其中列出的网段',
budget_exceeded: '超出本机的速率预算',
unsupported: '这个 Agent 不支持这种采集方法',
invalid_params: '参数不合法',
+17
View File
@@ -1088,3 +1088,20 @@ pre.code {
align-items: center;
gap: 4px;
}
/* 「没有才是好」的图整段为 0 时:盖一句说明,图的坐标轴留着,说明在绘图区中间 */
.chart-body {
position: relative;
}
.chart-quiet {
position: absolute;
inset: 0 0 28px 56px;
display: flex;
align-items: center;
justify-content: center;
color: var(--app-ok);
font-size: 14px;
font-weight: 600;
pointer-events: none;
}