diff --git a/backend/src/MoShiro.Backend.Api/ApiModule.cs b/backend/src/MoShiro.Backend.Api/ApiModule.cs index 6e6ddf0..f6af390 100644 --- a/backend/src/MoShiro.Backend.Api/ApiModule.cs +++ b/backend/src/MoShiro.Backend.Api/ApiModule.cs @@ -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); diff --git a/backend/src/MoShiro.Backend.Api/Contracts/ApiJsonContext.cs b/backend/src/MoShiro.Backend.Api/Contracts/ApiJsonContext.cs index d56d60f..2b3a2c6 100644 --- a/backend/src/MoShiro.Backend.Api/Contracts/ApiJsonContext.cs +++ b/backend/src/MoShiro.Backend.Api/Contracts/ApiJsonContext.cs @@ -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))] diff --git a/backend/src/MoShiro.Backend.Api/Contracts/PublicDtos.cs b/backend/src/MoShiro.Backend.Api/Contracts/PublicDtos.cs index fe9226a..91f4189 100644 --- a/backend/src/MoShiro.Backend.Api/Contracts/PublicDtos.cs +++ b/backend/src/MoShiro.Backend.Api/Contracts/PublicDtos.cs @@ -240,3 +240,13 @@ public sealed record TargetDto( IReadOnlyList T, IReadOnlyDictionary Values, SummaryDto Summary); + +/// 一个时间桶里一种采集方法的自适应探测频率：各状态的秒数、时间加权的平均频率（Hz，不知道时为 null）、状态变化次数、变化时的最大丢包率（%）。 +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 Buckets); + +/// 自适应探测频率的一次状态变化：从哪个状态到哪个状态、之后的频率、原因，以及当时的丢包率和包数。 +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 Kinds, IReadOnlyList Changes); diff --git a/backend/src/MoShiro.Backend.Api/Public/AdaptiveEndpoints.cs b/backend/src/MoShiro.Backend.Api/Public/AdaptiveEndpoints.cs new file mode 100644 index 0000000..c240764 --- /dev/null +++ b/backend/src/MoShiro.Backend.Api/Public/AdaptiveEndpoints.cs @@ -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; + +/// 自适应探测频率的审计：一条线路在一段时间里每种采集方法各状态占多久、平均频率多少，以及每一次状态变化和原因。 +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().ProducesProblems(403, 404, 422, 503); + + private static Task 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 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); + }); +} diff --git a/backend/src/MoShiro.Backend.Application/Agents/MoShiroFleetHooks.cs b/backend/src/MoShiro.Backend.Application/Agents/MoShiroFleetHooks.cs index fb526f9..9398a48 100644 --- a/backend/src/MoShiro.Backend.Application/Agents/MoShiroFleetHooks.cs +++ b/backend/src/MoShiro.Backend.Application/Agents/MoShiroFleetHooks.cs @@ -108,6 +108,22 @@ public sealed class MoShiroNodeEventSink : INodeEventSink } } + /// + /// 自适应探测频率的状态变化存成事件（）：线路任务记在监听方名下（和线路的数据同一个归属）， + /// 单个 Agent 任务记在上报的节点名下。时间以节点上报的为准，明显不对（未来、或太久以前）的用心跳的时间。 + /// + private async ValueTask RecordRateChangesAsync(NodeId reporter, long heartbeatMs, List 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) { diff --git a/backend/src/MoShiro.Backend.Application/ApplicationServiceCollectionExtensions.cs b/backend/src/MoShiro.Backend.Application/ApplicationServiceCollectionExtensions.cs index bc32064..8471ccb 100644 --- a/backend/src/MoShiro.Backend.Application/ApplicationServiceCollectionExtensions.cs +++ b/backend/src/MoShiro.Backend.Application/ApplicationServiceCollectionExtensions.cs @@ -33,6 +33,7 @@ public static class ApplicationServiceCollectionExtensions services.AddSingleton(); services.AddSingleton(); services.AddSingleton(); + services.AddSingleton(); services.AddSingleton(); services.AddSingleton(); services.AddSingleton(); diff --git a/backend/src/MoShiro.Backend.Application/Query/AdaptiveAudit.cs b/backend/src/MoShiro.Backend.Application/Query/AdaptiveAudit.cs new file mode 100644 index 0000000..f680623 --- /dev/null +++ b/backend/src/MoShiro.Backend.Application/Query/AdaptiveAudit.cs @@ -0,0 +1,155 @@ +using System.Text.Json; +using Fleet.Core; +using MoShiro.Protocol; + +namespace MoShiro.Backend.Application; + +/// 一个时间桶里一种采集方法的自适应探测频率：各状态各占多少秒、时间加权的平均频率、状态变化次数、变化时的最大丢包率。 +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 Buckets); + +public sealed record AdaptiveResult(long FromMs, long ToMs, long StepMs, IReadOnlyList Kinds, IReadOnlyList Changes); + +/// +/// 自适应探测频率的审计：节点把每一次状态变化（Quiet / Watch / Active / Dead、频率、原因、丢包）随心跳上报，Backend 存成事件（保留期和别的事件一样）， +/// 这里把它们还原成「某个时段里各状态占多久、平均频率多少」，所以几天之后也能回答「晚高峰的时候自适应是什么状态」。 +/// +public sealed class AdaptiveAuditService +{ + private readonly IEventStore _events; + + public AdaptiveAuditService(IEventStore events) + { + _events = events; + } + + public async ValueTask GetAsync(LineKey line, long fromMs, long toMs, long stepMs, CancellationToken ct) + { + // 往前多取一段：区间开头的状态要由这之前最近的一次变化决定 + IReadOnlyList records = await _events.ListAsync(line.Listener, AdaptiveAudit.EventKind, fromMs - AdaptiveAudit.LookBackMs, toMs, null, 20_000, ct); + List 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"; + + /// 区间开头之前最多往回看多久，找那时的状态。 + 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; + } + } + + /// + /// 把按时间排好的状态变化还原成每个时间桶里各状态的秒数。一次变化之后到下一次变化之前，是变化后的状态和频率； + /// 区间里第一次变化之前，状态是这次变化的「从」（频率不知道，不计入平均）。一种采集方法一条时间线（同一条线路上每种方法各自变化）。 + /// + public static AdaptiveResult Summarize(IReadOnlyList changes, long fromMs, long toMs, long stepMs) + { + int count = (int)Math.Max(1, Math.Ceiling((toMs - fromMs) / (double)stepMs)); + var kinds = new List(); + foreach (IGrouping 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(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); + } +} diff --git a/backend/tests/MoShiro.Backend.Tests/AdaptiveAuditTests.cs b/backend/tests/MoShiro.Backend.Tests/AdaptiveAuditTests.cs new file mode 100644 index 0000000..a69f289 --- /dev/null +++ b/backend/tests/MoShiro.Backend.Tests/AdaptiveAuditTests.cs @@ -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 +{ + 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) }; + + /// 状态变化还原成每个时间桶里各状态的秒数：跨桶的一段要切开，区间开头的状态取第一次变化的「从」，不同方法各算各的。 + [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); // 最新的在前 + } + + /// 变化在区间开头之前发生：那之后的状态延续到区间里。 + [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); + } + + /// 端到端：拨号节点随心跳上报的状态变化存成事件，几天后还能通过接口按时段查到。 + [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()); + double active = quic["buckets"]!.AsArray().Sum(b => b!["active_seconds"]!.GetValue()); + Assert.InRange(active, 500, 700); // 最近 10 分钟 active + JsonNode newest = audit!["changes"]![0]!; + Assert.Equal(("quiet", "active", "sustained loss"), (newest["from"]!.GetValue(), newest["to"]!.GetValue(), newest["reason"]!.GetValue())); + } +} diff --git a/backend/tests/MoShiro.TestSupport/AgentSimulator.cs b/backend/tests/MoShiro.TestSupport/AgentSimulator.cs index 5d4d831..cc3350a 100644 --- a/backend/tests/MoShiro.TestSupport/AgentSimulator.cs +++ b/backend/tests/MoShiro.TestSupport/AgentSimulator.cs @@ -223,6 +223,13 @@ public sealed class AgentSimulator : IAsyncDisposable new HeartbeatParams { Extra = JsonSerializer.SerializeToElement(new NodeHeartbeatExtra { Update = status }, MoShiroJsonContext.Default.NodeHeartbeatExtra) }, FleetRpcContext.Default.HeartbeatParams)); + /// 像拨号节点那样，随心跳上报自适应探测频率的状态变化。 + 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 RequestRawAsync(string method, JsonElement? parameters = null) => await RequestAsync(method, parameters); public async Task WaitForTaskSetAsync(Func predicate, TimeSpan? timeout = null)