From d63c3bff741227500c472f54903cbfc06c37f5ec Mon Sep 17 00:00:00 2001 From: justme Date: Thu, 22 Oct 2026 10:44:46 +0800 Subject: [PATCH] agent: keep adaptive rate changes apart from the logs and report them in the heartbeat --- backend/agent/src/MoShiro.Agent/AgentApp.cs | 6 +- .../src/MoShiro.Agent/Dial/DialConnection.cs | 8 ++- .../agent/src/MoShiro.Agent/RateChangeLog.cs | 40 ++++++++++++++ .../src/MoShiro.Agent/Tasks/TargetRunners.cs | 5 +- .../MoShiro.Measurement/Blocks/BlockBase.cs | 14 +++++ .../MoShiro.Measurement/Core/Contracts.cs | 4 ++ .../shared/MoShiro.Protocol/Rpc/MoShiroRpc.cs | 55 +++++++++++++++++++ 7 files changed, 128 insertions(+), 4 deletions(-) create mode 100644 backend/agent/src/MoShiro.Agent/RateChangeLog.cs diff --git a/backend/agent/src/MoShiro.Agent/AgentApp.cs b/backend/agent/src/MoShiro.Agent/AgentApp.cs index b80ffdf..c7f5659 100644 --- a/backend/agent/src/MoShiro.Agent/AgentApp.cs +++ b/backend/agent/src/MoShiro.Agent/AgentApp.cs @@ -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, diff --git a/backend/agent/src/MoShiro.Agent/Dial/DialConnection.cs b/backend/agent/src/MoShiro.Agent/Dial/DialConnection.cs index 5d6dd6a..6434e30 100644 --- a/backend/agent/src/MoShiro.Agent/Dial/DialConnection.cs +++ b/backend/agent/src/MoShiro.Agent/Dial/DialConnection.cs @@ -11,7 +11,7 @@ using MoShiro.Protocol; namespace MoShiro.Agent; /// 所有连接共用的东西。 -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); /// /// 与一个 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(transport, s_router, session, _env.Time); diff --git a/backend/agent/src/MoShiro.Agent/RateChangeLog.cs b/backend/agent/src/MoShiro.Agent/RateChangeLog.cs new file mode 100644 index 0000000..364d259 --- /dev/null +++ b/backend/agent/src/MoShiro.Agent/RateChangeLog.cs @@ -0,0 +1,40 @@ +using System.Collections.Concurrent; +using MoShiro.Protocol; + +namespace MoShiro.Agent; + +/// +/// 自适应探测频率的状态变化，等着随下一次心跳上报给 Backend（Backend 存成事件，用于事后审计）。 +/// 和普通日志分开存：日志环形缓冲几分钟到几小时就会被覆盖，而这里的变化要能在几天后回看。有界，满了丢最旧的。 +/// +public sealed class RateChangeLog +{ + public const int Capacity = 512; + + private readonly ConcurrentQueue _queue = new(); + + public void Add(RateChange change) + { + _queue.Enqueue(change); + while (_queue.Count > Capacity && _queue.TryDequeue(out _)) + { + } + } + + /// 取走最多 条（最旧的在前）；没有返回 null，心跳里就不带这个字段。 + public List? Take(int max) + { + List? taken = null; + while ((taken?.Count ?? 0) < max) + { + if (!_queue.TryDequeue(out RateChange? change)) + { + break; + } + + (taken ??= []).Add(change); + } + + return taken; + } +} diff --git a/backend/agent/src/MoShiro.Agent/Tasks/TargetRunners.cs b/backend/agent/src/MoShiro.Agent/Tasks/TargetRunners.cs index 6b14ded..c946c03 100644 --- a/backend/agent/src/MoShiro.Agent/Tasks/TargetRunners.cs +++ b/backend/agent/src/MoShiro.Agent/Tasks/TargetRunners.cs @@ -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))); diff --git a/backend/shared/MoShiro.Measurement/Blocks/BlockBase.cs b/backend/shared/MoShiro.Measurement/Blocks/BlockBase.cs index 478b75b..4c5ebd5 100644 --- a/backend/shared/MoShiro.Measurement/Blocks/BlockBase.cs +++ b/backend/shared/MoShiro.Measurement/Blocks/BlockBase.cs @@ -97,6 +97,20 @@ public abstract class BlockBase : IBlock 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)), diff --git a/backend/shared/MoShiro.Measurement/Core/Contracts.cs b/backend/shared/MoShiro.Measurement/Core/Contracts.cs index 44519c5..1c5dd3a 100644 --- a/backend/shared/MoShiro.Measurement/Core/Contracts.cs +++ b/backend/shared/MoShiro.Measurement/Core/Contracts.cs @@ -245,6 +245,9 @@ public sealed class BlockEnv /// 单个 Agent 任务：解析出来的地址能不能连（本地 allowed_target_cidrs）；null 表示不限制。 public Func? AllowAddress { get; init; } + /// 自适应频率的状态变化（用于审计）；null 表示不记录。 + public Action? OnRateChange { get; init; } + /// 拨号方：对端 TLS 证书的 SPKI 指纹；null 表示不校验证书（目标任务，目标是任意地址）。 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, diff --git a/backend/shared/MoShiro.Protocol/Rpc/MoShiroRpc.cs b/backend/shared/MoShiro.Protocol/Rpc/MoShiroRpc.cs index 129b238..8b02fd8 100644 --- a/backend/shared/MoShiro.Protocol/Rpc/MoShiroRpc.cs +++ b/backend/shared/MoShiro.Protocol/Rpc/MoShiroRpc.cs @@ -342,6 +342,10 @@ public sealed class NodeHeartbeatExtra /// 日志只报计数（不带文本）：自上次心跳以来各级别的条数、最多几个出现最多的类型，以及最新一条的编号；文本由 Backend 用 moshiro.logs.fetch 按需拉取。 [JsonPropertyName("log_summary")] public LogSummaryDto? LogSummary { get; set; } + + /// 自上次心跳以来自适应探测频率的状态变化（节点上限 512 条，超出丢最旧的）。 + [JsonPropertyName("rate_changes")] + public List? RateChanges { get; set; } } /// 某一类日志（消息文本的稳定哈希）在一个级别上出现了多少次。 @@ -357,6 +361,56 @@ public sealed class LogTypeCountDto public long Count { get; set; } } +/// +/// 自适应探测频率的一次状态变化（Quiet / Watch / Active / Dead）：节点随心跳上报，Backend 存成事件（rate_change）， +/// 事后可以回看每条线路、每个时段的探测频率为什么升降（例如晚高峰）。线路任务带 listener（对端）和 ip_version，目标任务带 task。 +/// +public sealed class RateChange +{ + [JsonPropertyName("at_ms")] + public long AtMs { get; set; } + + /// 这条线路的监听方节点 id（线路任务）。 + [JsonPropertyName("listener")] + public string? Listener { get; set; } + + /// 拨号方节点 id：节点上报时不填，由 Backend 按上报者补上。 + [JsonPropertyName("dialer")] + public string? Dialer { get; set; } + + [JsonPropertyName("ip_version")] + public string? IpVersion { get; set; } + + /// 单个 Agent 任务的任务 id。 + [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; } = ""; + + /// 变化时刻证明窗口内的丢包率（百分比）和发出、丢失的包数。 + [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))]