Compare commits
3
Commits
557b169dae
...
af385cf688
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
af385cf688 | ||
|
|
9ad34acbac | ||
|
|
d63c3bff74 |
No files matched your search
@@ -45,6 +45,7 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable
|
|||||||
private readonly PeerListener _peers;
|
private readonly PeerListener _peers;
|
||||||
private readonly PeerHub _hub;
|
private readonly PeerHub _hub;
|
||||||
private readonly ChunkAssembler _chunks;
|
private readonly ChunkAssembler _chunks;
|
||||||
|
private readonly RateChangeLog _rateChanges = new();
|
||||||
private readonly TaskManager _tasks;
|
private readonly TaskManager _tasks;
|
||||||
private readonly TestRunner _tests;
|
private readonly TestRunner _tests;
|
||||||
private readonly FleetLink _link;
|
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);
|
_datagrams = new QuicDatagramTransport(_sessions, _admission, _clock, _counters.Data, log, bind, config.Listen.QuicPorts, identity.Certificate, quic);
|
||||||
// 预握手敲门:控制端口和数据端口共用一个关卡(限速和去重一起算);名册就是期望状态里的 peers
|
// 预握手敲门:控制端口和数据端口共用一个关卡(限速和去重一起算);名册就是期望状态里的 peers
|
||||||
_trust = TrustAnchor.Load(config.Fleet.IdentityDir, log);
|
_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);
|
_knock = new KnockGate(identity.PeerIdentity, RosterKeys, time);
|
||||||
_tcp = new TcpAcceptor(bind, config.Listen.TcpPorts, identity.Certificate, _sessions, _counters.Data, _knock);
|
_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,
|
MaxBulkMbps = config.Limits.MaxTotalMbps,
|
||||||
};
|
};
|
||||||
_chunks = new ChunkAssembler(_streams, _clock, _counters, _outbox, identity.NodeId, time, log);
|
_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);
|
_tasks = new TaskManager(CreateGuard, _streams, env, targets, () => _datagrams.BoundPorts, () => _tcp.BoundPorts, _trust, acceptLineTasks: _listening);
|
||||||
_hub = new PeerHub(_sessions, _clock, _chunks, _tasks, _uplink, time, log);
|
_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);
|
_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,
|
DroppedRateLimited = _counters.DroppedRateLimited,
|
||||||
QueueBytes = _outbox.PendingBytes,
|
QueueBytes = _outbox.PendingBytes,
|
||||||
Update = updateStatus,
|
Update = updateStatus,
|
||||||
|
RateChanges = _rateChanges.Take(64),
|
||||||
LogSummary = logSummary is null ? null : new LogSummaryDto
|
LogSummary = logSummary is null ? null : new LogSummaryDto
|
||||||
{
|
{
|
||||||
LastId = logSummary.LastId,
|
LastId = logSummary.LastId,
|
||||||
|
|||||||
@@ -11,7 +11,7 @@ using MoShiro.Protocol;
|
|||||||
namespace MoShiro.Agent;
|
namespace MoShiro.Agent;
|
||||||
|
|
||||||
/// <summary>所有连接共用的东西。</summary>
|
/// <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>
|
/// <summary>
|
||||||
/// 与一个 Agent 的点对点连接,分两层:
|
/// 与一个 Agent 的点对点连接,分两层:
|
||||||
@@ -190,6 +190,12 @@ public sealed class DialConnection : IDisposable
|
|||||||
Fragments = shipper,
|
Fragments = shipper,
|
||||||
MaxPacketsPerSecond = _env.Limits.MaxTotalPps,
|
MaxPacketsPerSecond = _env.Limits.MaxTotalPps,
|
||||||
MaxBulkMbps = _env.Limits.MaxTotalMbps,
|
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 session = new DialSession(blockEnv, clock, _outbox, _env);
|
||||||
var rpc = new RpcSession<DialSession>(transport, s_router, session, _env.Time);
|
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 MonoClock _clock;
|
||||||
private readonly TimeProvider _time;
|
private readonly TimeProvider _time;
|
||||||
private readonly INodeLog _log;
|
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;
|
_chunks = chunks;
|
||||||
_agent = agent;
|
_agent = agent;
|
||||||
_clock = clock;
|
_clock = clock;
|
||||||
@@ -92,6 +94,7 @@ public sealed class TargetRunners
|
|||||||
PeerAddress = address,
|
PeerAddress = address,
|
||||||
Target = spec.Target,
|
Target = spec.Target,
|
||||||
AllowAddress = allow,
|
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),
|
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)));
|
IBlock block = BlockCatalog.Find(spec.Kind)!.Dialer(env, new BlockSpec(decision.Kind!, decision.Resolved!, 0, port, new FragmentScope(spec.Id, null, null)));
|
||||||
|
|||||||
@@ -97,6 +97,20 @@ public abstract class BlockBase : IBlock
|
|||||||
|
|
||||||
private void LogTransition(ProbeTransition change)
|
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?)[]
|
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)),
|
("stream", Spec.StreamId), ("kind", Spec.KindId), ("from", change.From.ToString().ToLowerInvariant()), ("to", change.To.ToString().ToLowerInvariant()), ("hz", Math.Round(change.Hz, 1)),
|
||||||
|
|||||||
@@ -245,6 +245,9 @@ public sealed class BlockEnv
|
|||||||
/// <summary>单个 Agent 任务:解析出来的地址能不能连(本地 <c>allowed_target_cidrs</c>);null 表示不限制。</summary>
|
/// <summary>单个 Agent 任务:解析出来的地址能不能连(本地 <c>allowed_target_cidrs</c>);null 表示不限制。</summary>
|
||||||
public Func<IPAddress, bool>? AllowAddress { get; init; }
|
public Func<IPAddress, bool>? AllowAddress { get; init; }
|
||||||
|
|
||||||
|
/// <summary>自适应频率的状态变化(用于审计);null 表示不记录。</summary>
|
||||||
|
public Action<RateChange>? OnRateChange { get; init; }
|
||||||
|
|
||||||
/// <summary>拨号方:对端 TLS 证书的 SPKI 指纹;null 表示不校验证书(目标任务,目标是任意地址)。</summary>
|
/// <summary>拨号方:对端 TLS 证书的 SPKI 指纹;null 表示不校验证书(目标任务,目标是任意地址)。</summary>
|
||||||
public byte[]? PeerSpkiSha256 { get; init; }
|
public byte[]? PeerSpkiSha256 { get; init; }
|
||||||
|
|
||||||
@@ -277,6 +280,7 @@ public sealed class BlockEnv
|
|||||||
PeerAddress = PeerAddress,
|
PeerAddress = PeerAddress,
|
||||||
Target = Target,
|
Target = Target,
|
||||||
AllowAddress = AllowAddress,
|
AllowAddress = AllowAddress,
|
||||||
|
OnRateChange = OnRateChange,
|
||||||
PeerSpkiSha256 = PeerSpkiSha256,
|
PeerSpkiSha256 = PeerSpkiSha256,
|
||||||
KnockKey = KnockKey,
|
KnockKey = KnockKey,
|
||||||
Fragments = Fragments,
|
Fragments = Fragments,
|
||||||
|
|||||||
@@ -342,6 +342,10 @@ public sealed class NodeHeartbeatExtra
|
|||||||
/// <summary>日志只报计数(不带文本):自上次心跳以来各级别的条数、最多几个出现最多的类型,以及最新一条的编号;文本由 Backend 用 <c>moshiro.logs.fetch</c> 按需拉取。</summary>
|
/// <summary>日志只报计数(不带文本):自上次心跳以来各级别的条数、最多几个出现最多的类型,以及最新一条的编号;文本由 Backend 用 <c>moshiro.logs.fetch</c> 按需拉取。</summary>
|
||||||
[JsonPropertyName("log_summary")]
|
[JsonPropertyName("log_summary")]
|
||||||
public LogSummaryDto? LogSummary { get; set; }
|
public LogSummaryDto? LogSummary { get; set; }
|
||||||
|
|
||||||
|
/// <summary>自上次心跳以来自适应探测频率的状态变化(节点上限 512 条,超出丢最旧的)。</summary>
|
||||||
|
[JsonPropertyName("rate_changes")]
|
||||||
|
public List<RateChange>? RateChanges { get; set; }
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>某一类日志(消息文本的稳定哈希)在一个级别上出现了多少次。</summary>
|
/// <summary>某一类日志(消息文本的稳定哈希)在一个级别上出现了多少次。</summary>
|
||||||
@@ -357,6 +361,56 @@ public sealed class LogTypeCountDto
|
|||||||
public long Count { get; set; }
|
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
|
public sealed class LogSummaryDto
|
||||||
{
|
{
|
||||||
[JsonPropertyName("last_id")]
|
[JsonPropertyName("last_id")]
|
||||||
@@ -826,6 +880,7 @@ public static class FragmentKeys
|
|||||||
[JsonSerializable(typeof(LogsFetchParams))]
|
[JsonSerializable(typeof(LogsFetchParams))]
|
||||||
[JsonSerializable(typeof(LogsFetchResult))]
|
[JsonSerializable(typeof(LogsFetchResult))]
|
||||||
[JsonSerializable(typeof(LogSummaryDto))]
|
[JsonSerializable(typeof(LogSummaryDto))]
|
||||||
|
[JsonSerializable(typeof(RateChange))]
|
||||||
[JsonSerializable(typeof(UpdateStatus))]
|
[JsonSerializable(typeof(UpdateStatus))]
|
||||||
[JsonSerializable(typeof(NodeHeartbeatExtra))]
|
[JsonSerializable(typeof(NodeHeartbeatExtra))]
|
||||||
[JsonSerializable(typeof(NodeHelloMeta))]
|
[JsonSerializable(typeof(NodeHelloMeta))]
|
||||||
|
|||||||
@@ -99,6 +99,7 @@ public static class ApiModule
|
|||||||
RouteGroupBuilder publicGroup = app.MapGroup("/api/public").AddEndpointFilter(PublicGuard.InvokeAsync);
|
RouteGroupBuilder publicGroup = app.MapGroup("/api/public").AddEndpointFilter(PublicGuard.InvokeAsync);
|
||||||
QueryEndpoints.Map(publicGroup);
|
QueryEndpoints.Map(publicGroup);
|
||||||
TargetEndpoints.Map(publicGroup);
|
TargetEndpoints.Map(publicGroup);
|
||||||
|
AdaptiveEndpoints.Map(publicGroup);
|
||||||
LiveEndpoints.Map(publicGroup);
|
LiveEndpoints.Map(publicGroup);
|
||||||
HostEndpoints.Map(publicGroup);
|
HostEndpoints.Map(publicGroup);
|
||||||
NodeEndpoints.Map(publicGroup);
|
NodeEndpoints.Map(publicGroup);
|
||||||
|
|||||||
@@ -18,6 +18,7 @@ namespace MoShiro.Backend.Api;
|
|||||||
[JsonSerializable(typeof(IdDto))]
|
[JsonSerializable(typeof(IdDto))]
|
||||||
[JsonSerializable(typeof(OverviewDto))]
|
[JsonSerializable(typeof(OverviewDto))]
|
||||||
[JsonSerializable(typeof(SeriesDto))]
|
[JsonSerializable(typeof(SeriesDto))]
|
||||||
|
[JsonSerializable(typeof(AdaptiveDto))]
|
||||||
[JsonSerializable(typeof(CompareDto))]
|
[JsonSerializable(typeof(CompareDto))]
|
||||||
[JsonSerializable(typeof(DistributionDto))]
|
[JsonSerializable(typeof(DistributionDto))]
|
||||||
[JsonSerializable(typeof(HeatmapDto))]
|
[JsonSerializable(typeof(HeatmapDto))]
|
||||||
|
|||||||
@@ -240,3 +240,13 @@ public sealed record TargetDto(
|
|||||||
IReadOnlyList<long> T,
|
IReadOnlyList<long> T,
|
||||||
IReadOnlyDictionary<string, double?[]> Values,
|
IReadOnlyDictionary<string, double?[]> Values,
|
||||||
SummaryDto Summary);
|
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)
|
private async ValueTask ApplyHeartbeatAsync(NodeId id, NodeEvent nodeEvent, System.Text.Json.JsonElement extra, CancellationToken ct)
|
||||||
{
|
{
|
||||||
await _nodes.SetVersionAsync(id.Value, nodeEvent.Node.NodeVersion, ct);
|
await _nodes.SetVersionAsync(id.Value, nodeEvent.Node.NodeVersion, ct);
|
||||||
@@ -129,6 +145,11 @@ public sealed class MoShiroNodeEventSink : INodeEventSink
|
|||||||
}
|
}
|
||||||
|
|
||||||
_nodeLogs.Observe(id, beat.LogSummary);
|
_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)));
|
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)
|
foreach (ConnectedDialer dialer in beat.Dialers)
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -33,6 +33,7 @@ public static class ApplicationServiceCollectionExtensions
|
|||||||
services.AddSingleton<LinkService>();
|
services.AddSingleton<LinkService>();
|
||||||
services.AddSingleton<LineService>();
|
services.AddSingleton<LineService>();
|
||||||
services.AddSingleton<TargetService>();
|
services.AddSingleton<TargetService>();
|
||||||
|
services.AddSingleton<AdaptiveAuditService>();
|
||||||
services.AddSingleton<TestService>();
|
services.AddSingleton<TestService>();
|
||||||
services.AddSingleton<IngestService>();
|
services.AddSingleton<IngestService>();
|
||||||
services.AddSingleton<WindowComputer>();
|
services.AddSingleton<WindowComputer>();
|
||||||
|
|||||||
@@ -0,0 +1,155 @@
|
|||||||
|
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>
|
||||||
|
/// 把按时间排好的状态变化还原成每个时间桶里各状态的秒数。一次变化之后到下一次变化之前,是变化后的状态和频率;
|
||||||
|
/// 区间里第一次变化之前,状态是这次变化的「从」(频率不知道,不计入平均)。一种采集方法一条时间线(同一条线路上每种方法各自变化)。
|
||||||
|
/// </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].From;
|
||||||
|
double? hz = null;
|
||||||
|
long cursor = Math.Min(ordered[0].AtMs, fromMs);
|
||||||
|
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,107 @@
|
|||||||
|
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 + (30 * Minute), "quic_ping", "watch", "quiet", 2), // 10:00–10:30 watch(频率未知),之后 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"),
|
||||||
|
];
|
||||||
|
|
||||||
|
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(2, first.AverageHz); // 只有 quiet 那半小时知道频率
|
||||||
|
Assert.Equal(1, 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((1800, 1800), ((int)tcp.Buckets[0].QuietSeconds, (int)tcp.Buckets[0].WatchSeconds));
|
||||||
|
Assert.Equal(3600, (int)tcp.Buckets[1].WatchSeconds);
|
||||||
|
|
||||||
|
Assert.Equal(3, 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>()));
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -223,6 +223,13 @@ public sealed class AgentSimulator : IAsyncDisposable
|
|||||||
new HeartbeatParams { Extra = JsonSerializer.SerializeToElement(new NodeHeartbeatExtra { Update = status }, MoShiroJsonContext.Default.NodeHeartbeatExtra) },
|
new HeartbeatParams { Extra = JsonSerializer.SerializeToElement(new NodeHeartbeatExtra { Update = status }, MoShiroJsonContext.Default.NodeHeartbeatExtra) },
|
||||||
FleetRpcContext.Default.HeartbeatParams));
|
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<RpcMessage> RequestRawAsync(string method, JsonElement? parameters = null) => await RequestAsync(method, parameters);
|
||||||
|
|
||||||
public async Task<TaskSet> WaitForTaskSetAsync(Func<TaskSet, bool> predicate, TimeSpan? timeout = null)
|
public async Task<TaskSet> WaitForTaskSetAsync(Func<TaskSet, bool> predicate, TimeSpan? timeout = null)
|
||||||
|
|||||||
@@ -0,0 +1,183 @@
|
|||||||
|
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: 'none' },
|
||||||
|
] 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> = { 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>
|
||||||
|
))}
|
||||||
|
</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>
|
||||||
|
)
|
||||||
|
}
|
||||||
@@ -2678,3 +2678,77 @@ input {
|
|||||||
.pop .pop-note {
|
.pop .pop-note {
|
||||||
white-space: normal;
|
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;
|
||||||
|
}
|
||||||
@@ -10,6 +10,7 @@ import { NodeTag, useNodeInfos } from '@shared/nodes'
|
|||||||
import { useQuery } from '@shared/hooks'
|
import { useQuery } from '@shared/hooks'
|
||||||
import { Chip, Dot, LoadingBar, Segmented, Spinner, Stat } from '@shared/ui'
|
import { Chip, Dot, LoadingBar, Segmented, Spinner, Stat } from '@shared/ui'
|
||||||
import { PageError, ScoreChips } from '../lib'
|
import { PageError, ScoreChips } from '../lib'
|
||||||
|
import { AdaptiveAudit } from '../AdaptiveAudit'
|
||||||
import { Minimap } from '../Minimap'
|
import { Minimap } from '../Minimap'
|
||||||
import { readSpec, usePeriod, usePeriodSeries, writeSpec } from '../periods'
|
import { readSpec, usePeriod, usePeriodSeries, writeSpec } from '../periods'
|
||||||
import { dayBounds, dayKey, RangePicker, WINDOWS, type WindowValue } from '../ranges'
|
import { dayBounds, dayKey, RangePicker, WINDOWS, type WindowValue } from '../ranges'
|
||||||
@@ -357,6 +358,8 @@ export function Line() {
|
|||||||
{chartSections}
|
{chartSections}
|
||||||
</section>
|
</section>
|
||||||
|
|
||||||
|
<AdaptiveAudit lineId={id} />
|
||||||
|
|
||||||
{test.data && result ? (
|
{test.data && result ? (
|
||||||
<div className="card">
|
<div className="card">
|
||||||
<div className="card-title">最近一次测试 · {test.data.plan}</div>
|
<div className="card-title">最近一次测试 · {test.data.plan}</div>
|
||||||
|
|||||||
@@ -86,3 +86,35 @@ export interface TargetDto {
|
|||||||
values: Record<string, (number | null)[]>
|
values: Record<string, (number | null)[]>
|
||||||
summary: { metrics: Record<string, number | null>; samples: Record<string, number>; coverage: number }
|
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[]
|
||||||
|
}
|
||||||
Reference in new issue
Block a user