From a7ebf61551cdd482dfb5e68044bc76cd3f1a4b3c Mon Sep 17 00:00:00 2001 From: justme Date: Tue, 20 Oct 2026 19:12:11 +0800 Subject: [PATCH] [CI] [SKIP TESTS] Add detailed Probe-Agent connection diagnostics --- .../src/MoShiro.Agent/DataPlane/Registries.cs | 4 + .../agent/src/MoShiro.Agent/Peer/PeerHub.cs | 33 ++++- .../fleet/src/Fleet.Core/ContextNodeLog.cs | 8 ++ .../src/Fleet.Hosting/BufferedNodeLog.cs | 2 +- backend/fleet/src/Fleet.Link/Peer.cs | 47 ++++++- backend/fleet/tests/Fleet.Tests/LinkTests.cs | 13 +- .../src/MoShiro.Probe/Link/AgentConnection.cs | 60 +++++++-- .../MoShiro.Measurement/Core/Contracts.cs | 3 + .../Net/MeasurementDatagramRouter.cs | 53 +++++++- .../Net/MsQuicConnection.cs | 123 +++++++++++++++--- .../MoShiro.Measurement/Net/MsQuicListener.cs | 3 +- .../Net/QuicDatagramTransport.cs | 4 +- .../MsQuicConnectionTests.cs | 33 ++++- .../Agents/NodeRuntimeLogService.cs | 2 +- .../NodeRuntimeLogTests.cs | 27 ++++ docs/ops/deployment.md | 12 ++ 16 files changed, 376 insertions(+), 51 deletions(-) create mode 100644 backend/fleet/src/Fleet.Core/ContextNodeLog.cs create mode 100644 backend/tests/MoShiro.Backend.Tests/NodeRuntimeLogTests.cs diff --git a/backend/agent/src/MoShiro.Agent/DataPlane/Registries.cs b/backend/agent/src/MoShiro.Agent/DataPlane/Registries.cs index 239cc7b..91a38d3 100644 --- a/backend/agent/src/MoShiro.Agent/DataPlane/Registries.cs +++ b/backend/agent/src/MoShiro.Agent/DataPlane/Registries.cs @@ -56,6 +56,10 @@ public sealed class ProbeSession : DataSession public string ProbeId => Info.PeerId; + public override string DiagnosticPeerId => ProbeId; + + public string? CloseReason { get; set; } + public string IpVersion => Info.Remote is IPEndPoint endpoint && endpoint.Address.AddressFamily == AddressFamily.InterNetworkV6 && !endpoint.Address.IsIPv4MappedToIPv6 ? IpVersions.V6 : IpVersions.V4; public string PathKey => ProbeId + "~" + IpVersion; diff --git a/backend/agent/src/MoShiro.Agent/Peer/PeerHub.cs b/backend/agent/src/MoShiro.Agent/Peer/PeerHub.cs index 67769a9..8d73405 100644 --- a/backend/agent/src/MoShiro.Agent/Peer/PeerHub.cs +++ b/backend/agent/src/MoShiro.Agent/Peer/PeerHub.cs @@ -112,6 +112,7 @@ public sealed class PeerHub public async Task OnSessionAsync(PeerSessionInfo info, IRpcTransport transport, CancellationToken ct) { var probe = new ProbeSession(info) { ConnectedAtUtcMs = _time.GetUtcNow().ToUnixTimeMilliseconds() }; + var log = new ContextNodeLog(_log, ("probe", probe.ProbeId), ("ip_version", probe.IpVersion), ("control_id", Guid.NewGuid().ToString("N"))); var rpc = new RpcSession(transport, _router, new PeerContext(probe, _clock), _time); probe.Rpc = rpc; @@ -151,6 +152,8 @@ public sealed class PeerHub ProbeSession? previous = _sessions.Replace(probe); if (previous?.Rpc is { } old) { + previous.CloseReason = "replaced"; + log.Info("replacing previous probe control session"); await old.CloseAsync("replaced", CancellationToken.None); } @@ -159,8 +162,8 @@ public sealed class PeerHub using CancellationTokenSource linked = CancellationTokenSource.CreateLinkedTokenSource(ct); Task run = rpc.RunAsync(linked.Token); - Task watchdog = WatchIdleAsync(probe, rpc, linked.Token); - _log.Info("probe connected", ("probe", info.PeerId), ("version", info.Version ?? ""), ("remote", probe.RemoteAddress ?? "")); + Task watchdog = WatchIdleAsync(probe, rpc, log, linked.Token); + log.Info("probe connected", ("phase", "control_session"), ("version", info.Version ?? ""), ("remote", info.Remote)); await _uplink.EmitEventAsync(AgentEventKinds.ProbeConnected, info.PeerId, Describe(probe)); await PushAsync(probe, initial, linked.Token); try @@ -171,6 +174,12 @@ public sealed class PeerHub { // 停止 } + catch (Exception ex) + { + probe.CloseReason ??= "control_error"; + log.Warn("probe control session failed", ("exception", ex.GetType().Name), ("err", ex.Message)); + throw; + } finally { _sessions.Remove(probe); @@ -178,7 +187,10 @@ public sealed class PeerHub await watchdog; await rpc.DisposeAsync(); await _tasks.CloseSessionAsync(probe); - _log.Info("probe disconnected", ("probe", info.PeerId)); + log.Info("probe disconnected", ("phase", "control_session"), + ("reason", probe.CloseReason ?? (ct.IsCancellationRequested ? "service_stopping" : "peer_closed")), + ("duration_ms", _time.GetUtcNow().ToUnixTimeMilliseconds() - probe.ConnectedAtUtcMs), + ("datagram_bound", probe.DatagramConnection is not null), ("data_remote", probe.DatagramRemoteEndPoint)); OnProbeLost?.Invoke(probe); await _uplink.EmitEventAsync(AgentEventKinds.ProbeDisconnected, info.PeerId, Describe(probe)); } @@ -188,8 +200,9 @@ public sealed class PeerHub /// Probe 每 30 秒发一次时钟同步请求，所以会话里至少每 30 秒有一帧：超过 没有收到任何东西， /// 说明 Probe 已经静默死亡（断电、进程被冻结），关闭会话，让 Backend 看到 probe_disconnected，而不是一直显示在线。 /// - private async Task WatchIdleAsync(ProbeSession probe, RpcSession rpc, CancellationToken ct) + private async Task WatchIdleAsync(ProbeSession probe, RpcSession rpc, INodeLog log, CancellationToken ct) { + long lastSummary = _time.GetTimestamp(); using var timer = new PeriodicTimer(TimeSpan.FromSeconds(10), _time); try { @@ -197,10 +210,18 @@ public sealed class PeerHub { if (rpc.IdleFor > IdleLimit) { - _log.Warn("probe session is silent, closing it", ("probe", probe.ProbeId), ("idle_s", (long)rpc.IdleFor.TotalSeconds)); + probe.CloseReason = "idle_timeout"; + log.Warn("probe session is silent, closing it", ("phase", "control_session"), ("idle_s", (long)rpc.IdleFor.TotalSeconds), ("limit_s", IdleLimit.TotalSeconds)); await rpc.CloseAsync("idle", CancellationToken.None); return; } + if (_time.GetElapsedTime(lastSummary) >= TimeSpan.FromSeconds(60)) + { + lastSummary = _time.GetTimestamp(); + log.Info("probe session summary", ("phase", "control_session"), ("remote", probe.Info.Remote), + ("control_idle_s", (long)rpc.IdleFor.TotalSeconds), ("datagram_bound", probe.DatagramConnection is not null), + ("data_remote", probe.DatagramRemoteEndPoint)); + } } } catch (OperationCanceledException) @@ -225,6 +246,7 @@ public sealed class PeerHub { if (probe.ProbeId == probeId && probe.Rpc is { } rpc) { + probe.CloseReason = "disconnected"; await rpc.CloseAsync("disconnected", CancellationToken.None); disconnected = true; } @@ -240,6 +262,7 @@ public sealed class PeerHub { if (!allowed.Contains(probe.ProbeId) && probe.Rpc is { } rpc) { + probe.CloseReason = "revoked"; _log.Info("probe is no longer allowed, closing its session", ("probe", probe.ProbeId)); _ = rpc.CloseAsync("revoked", CancellationToken.None).AsTask(); } diff --git a/backend/fleet/src/Fleet.Core/ContextNodeLog.cs b/backend/fleet/src/Fleet.Core/ContextNodeLog.cs new file mode 100644 index 0000000..f78b8fc --- /dev/null +++ b/backend/fleet/src/Fleet.Core/ContextNodeLog.cs @@ -0,0 +1,8 @@ +namespace Fleet.Core; + +/// Adds stable diagnostic context to every event in one operation. +public sealed class ContextNodeLog(INodeLog inner, params (string Key, object? Value)[] context) : INodeLog +{ + public void Write(NodeLogLevel level, string message, params (string Key, object? Value)[] fields) => + inner.Write(level, message, [.. context, .. fields]); +} diff --git a/backend/fleet/src/Fleet.Hosting/BufferedNodeLog.cs b/backend/fleet/src/Fleet.Hosting/BufferedNodeLog.cs index 1d29ad5..81063a4 100644 --- a/backend/fleet/src/Fleet.Hosting/BufferedNodeLog.cs +++ b/backend/fleet/src/Fleet.Hosting/BufferedNodeLog.cs @@ -7,7 +7,7 @@ namespace Fleet.Hosting; public sealed class BufferedNodeLog : INodeLog, INodeLogBatchSource { private const int Capacity = 256; - private const int MaxFields = 12; + private const int MaxFields = 20; private const int MaxMessageLength = 512; private const int MaxFieldValueLength = 256; private const int MaxBatchBytes = 8 << 10; diff --git a/backend/fleet/src/Fleet.Link/Peer.cs b/backend/fleet/src/Fleet.Link/Peer.cs index 7a20f7a..a039fb1 100644 --- a/backend/fleet/src/Fleet.Link/Peer.cs +++ b/backend/fleet/src/Fleet.Link/Peer.cs @@ -96,6 +96,10 @@ public sealed class PeerListener private async Task HandleAsync(TcpClient client, CancellationToken ct) { + bool admitted = false; + string phase = "knock"; + long started = _time.GetTimestamp(); + EndPoint? remote = client.Client.RemoteEndPoint; try { client.NoDelay = true; @@ -108,19 +112,27 @@ public sealed class PeerListener return; } + admitted = true; + phase = "tls"; + _log.Info("peer knock accepted", ("remote", remote), ("phase", phase)); var tls = new SslStream(stream, leaveInnerStreamOpen: false); await tls.AuthenticateAsServerAsync( new SslServerAuthenticationOptions { ServerCertificate = _options.Certificate, ClientCertificateRequired = false, EnabledSslProtocols = System.Security.Authentication.SslProtocols.Tls12 | System.Security.Authentication.SslProtocols.Tls13 }, linked.Token); + phase = "peer_auth"; + _log.Info("peer TLS established", ("remote", remote), ("phase", phase), ("tls", tls.SslProtocol)); var transport = new TlsFrameTransport(tls, _options.MaxFrameBytes); PeerSessionInfo? info = await HandshakeAsync(transport, client.Client.RemoteEndPoint, linked.Token); if (info is null) { + _log.Warn("peer authentication rejected", ("remote", remote), ("phase", phase)); await transport.DisposeAsync(); return; } - _log.Info("peer session", ("probe", info.PeerId), ("result", "established")); + phase = "control_session"; + _log.Info("peer session", ("probe", info.PeerId), ("result", "established"), ("remote", remote), + ("elapsed_ms", (long)_time.GetElapsedTime(started).TotalMilliseconds)); await using (transport) { await _onSession(info, transport, ct); @@ -128,7 +140,14 @@ public sealed class PeerListener } catch (Exception ex) when (ex is IOException or System.Security.Authentication.AuthenticationException or OperationCanceledException or SocketException or ObjectDisposedException) { - // 握手失败或连接断开：这个会话结束，监听继续 + // Invalid knocks stay silent. Only peers that passed admission get diagnostics. + if (admitted && !ct.IsCancellationRequested) + { + _log.Warn("peer control connection failed", ("remote", remote), ("phase", phase), + ("elapsed_ms", (long)_time.GetElapsedTime(started).TotalMilliseconds), + ("exception", ex.GetType().Name), ("err", ex.Message), + ("socket_error", ex is SocketException socket ? socket.SocketErrorCode : null)); + } } finally { @@ -239,25 +258,37 @@ public static class PeerTimings public static class PeerConnector { + public static Task<(PeerSessionInfo Info, TlsFrameTransport Transport)> ConnectAsync( + string host, int port, byte[] agentPublicKey, NodeIdentity identity, string peerId, string? version, int maxFrameBytes, TimeProvider time, CancellationToken ct) => + ConnectAsync(host, port, agentPublicKey, identity, peerId, version, maxFrameBytes, time, NullNodeLog.Instance, ct); + public static async Task<(PeerSessionInfo Info, TlsFrameTransport Transport)> ConnectAsync( - string host, int port, byte[] agentPublicKey, NodeIdentity identity, string peerId, string? version, int maxFrameBytes, TimeProvider time, CancellationToken ct) + string host, int port, byte[] agentPublicKey, NodeIdentity identity, string peerId, string? version, int maxFrameBytes, TimeProvider time, INodeLog log, CancellationToken ct) { // 证书按公钥的指纹固定；敲门密钥由双方身份密钥派生 byte[] expectedSpkiSha256 = NodeIdentity.Fingerprint(agentPublicKey); TcpClient client = IPAddress.TryParse(host, out IPAddress? address) ? new TcpClient(address.AddressFamily) : new TcpClient(); client.NoDelay = true; PeerSockets.EnableKeepAlive(client.Client); + string phase = "tcp_connect"; + long started = time.GetTimestamp(); try { using var timeout = new CancellationTokenSource(PeerTimings.Handshake, time); using CancellationTokenSource linked = CancellationTokenSource.CreateLinkedTokenSource(ct, timeout.Token); await client.ConnectAsync(host, port, linked.Token); + log.Info("peer TCP connected", ("remote", client.Client.RemoteEndPoint), ("local", client.Client.LocalEndPoint), + ("phase", phase), ("elapsed_ms", (long)time.GetElapsedTime(started).TotalMilliseconds)); + phase = "knock_send"; Stream stream = client.GetStream(); await stream.WriteAsync(Knock.Build(Knock.DeriveKey(identity, agentPublicKey), time.GetUtcNow().ToUnixTimeMilliseconds()), linked.Token); + phase = "tls"; var tls = new SslStream( stream, leaveInnerStreamOpen: false, (_, certificate, _, _) => certificate is not null && SpkiMatches(certificate, expectedSpkiSha256)); await tls.AuthenticateAsClientAsync(new SslClientAuthenticationOptions { TargetHost = host }, linked.Token); + log.Info("peer TLS established", ("phase", phase), ("tls", tls.SslProtocol), ("elapsed_ms", (long)time.GetElapsedTime(started).TotalMilliseconds)); + phase = "peer_auth"; var transport = new TlsFrameTransport(tls, maxFrameBytes); RpcFrame? first = await transport.ReceiveAsync(linked.Token); @@ -288,10 +319,18 @@ public static class PeerConnector throw new InvalidOperationException("session id mismatch: the key agreement did not produce the same credentials"); } + log.Info("peer authentication established", ("phase", phase), ("elapsed_ms", (long)time.GetElapsedTime(started).TotalMilliseconds)); return (new PeerSessionInfo(peerId, credentials, client.Client.RemoteEndPoint), transport); } - catch + catch (Exception ex) { + if (!ct.IsCancellationRequested) + { + log.Warn("peer control handshake failed", ("phase", phase), ("remote", host), ("port", port), + ("elapsed_ms", (long)time.GetElapsedTime(started).TotalMilliseconds), ("timeout_ms", PeerTimings.Handshake.TotalMilliseconds), + ("exception", ex.GetType().Name), ("err", ex.Message), + ("socket_error", ex is SocketException socket ? socket.SocketErrorCode : null)); + } client.Dispose(); throw; } diff --git a/backend/fleet/tests/Fleet.Tests/LinkTests.cs b/backend/fleet/tests/Fleet.Tests/LinkTests.cs index e8d45f9..9d2e063 100644 --- a/backend/fleet/tests/Fleet.Tests/LinkTests.cs +++ b/backend/fleet/tests/Fleet.Tests/LinkTests.cs @@ -364,13 +364,18 @@ public sealed class LinkTests agent.KnockPeers.Add(probe.PublicKeySpki); await agent.Listener.Started.Task; - (PeerSessionInfo client, TlsFrameTransport transport) = await PeerConnector.ConnectAsync("127.0.0.1", agent.Listener.Port, agent.PublicKey, probe, "prb_1", null, 1 << 20, TimeProvider.System, default); + var logs = new Fleet.Hosting.BufferedNodeLog(NullNodeLog.Instance, TimeProvider.System); + var scoped = new ContextNodeLog(logs, ("attempt_id", "test-attempt")); + (PeerSessionInfo client, TlsFrameTransport transport) = await PeerConnector.ConnectAsync("127.0.0.1", agent.Listener.Port, agent.PublicKey, probe, "prb_1", null, 1 << 20, TimeProvider.System, scoped, default); await using (transport) { PeerSessionInfo server = await seen.Task.WaitAsync(TimeSpan.FromSeconds(5)); Assert.Equal("prb_1", server.PeerId); Assert.Equal(client.Credentials.Key, server.Credentials.Key); Assert.Equal(client.Credentials.SessionId, server.Credentials.SessionId); + NodeLogBatch batch = logs.TakeBatch(32); + Assert.Equal(new[] { "tcp_connect", "tls", "peer_auth" }, batch.Records.Select(x => x.Fields["phase"])); + Assert.All(batch.Records, entry => Assert.Equal("test-attempt", entry.Fields["attempt_id"])); } } @@ -383,8 +388,12 @@ public sealed class LinkTests agent.KnockPeers.AddRange([probe.PublicKeySpki, impostor.PublicKeySpki]); // 敲门通过之后，名册外的 ID 和伪造的签名仍然被拒绝 await agent.Listener.Started.Task; + var logs = new Fleet.Hosting.BufferedNodeLog(NullNodeLog.Instance, TimeProvider.System); InvalidOperationException unknown = await Assert.ThrowsAsync(async () => - await PeerConnector.ConnectAsync("127.0.0.1", agent.Listener.Port, agent.PublicKey, probe, "prb_unknown", null, 1 << 20, TimeProvider.System, default)); + await PeerConnector.ConnectAsync("127.0.0.1", agent.Listener.Port, agent.PublicKey, probe, "prb_unknown", null, 1 << 20, TimeProvider.System, logs, default)); + NodeLogRecord failure = Assert.Single(logs.TakeBatch(32).Records, entry => entry.Message == "peer control handshake failed"); + Assert.Equal("peer_auth", failure.Fields["phase"]); + Assert.Equal("InvalidOperationException", failure.Fields["exception"]); InvalidOperationException forged = await Assert.ThrowsAsync(async () => await PeerConnector.ConnectAsync("127.0.0.1", agent.Listener.Port, agent.PublicKey, impostor, "prb_1", null, 1 << 20, TimeProvider.System, default)); Assert.Equal(unknown.Message, forged.Message); diff --git a/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs b/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs index 1ae62e1..60867f9 100644 --- a/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs +++ b/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs @@ -35,6 +35,9 @@ public sealed class AgentConnection : IDisposable public static RpcRouter Router => s_router; private readonly ConnectionEnv _env; + private INodeLog _attemptLog; + private string _phase = "idle"; + private long _attemptStarted; private volatile ConnectionSession? _session; private int _previousQuicSourcePort; private QuicDatagramTransport? _activeDatagrams; @@ -53,6 +56,7 @@ public sealed class AgentConnection : IDisposable { Endpoint = endpoint; _env = env; + _attemptLog = env.Log; } public KnownAgent Endpoint { get; } @@ -141,12 +145,16 @@ public sealed class AgentConnection : IDisposable var backoff = new ProbeReconnectBackoff(); while (!ct.IsCancellationRequested) { + _attemptStarted = _env.Time.GetTimestamp(); + _attemptLog = new ContextNodeLog(_env.Log, ("agent", Endpoint.AgentId), ("ip_version", Endpoint.IpVersion), + ("attempt_id", Guid.NewGuid().ToString("N"))); SetAttempt(ProbeLinkAttemptState.Connecting, null, null, 0); try { string reason = await RunOnceAsync(backoff, ct); SetAttempt(ProbeLinkAttemptState.RetryWait, reason, null, backoff.Attempts); - _env.Log.Warn("agent session ended", ("agent", Endpoint.AgentId), ("err", reason)); + _attemptLog.Warn("agent session ended", ("phase", _phase), ("err", reason), + ("elapsed_ms", (long)_env.Time.GetElapsedTime(_attemptStarted).TotalMilliseconds)); } catch (OperationCanceledException) when (ct.IsCancellationRequested) { @@ -154,13 +162,18 @@ public sealed class AgentConnection : IDisposable } catch (Exception ex) { - SetAttempt(ProbeLinkAttemptState.RetryWait, $"{ex.GetType().Name}: {ex.Message}", null, backoff.Attempts + 1); - _env.Log.Error("agent connection attempt failed; retry will continue", ("agent", Endpoint.AgentId), ("exception", ex.GetType().FullName), ("err", ex.Message)); + SetAttempt(ProbeLinkAttemptState.RetryWait, $"{_phase}: {ex.GetType().Name}: {ex.Message}", null, backoff.Attempts + 1); + _attemptLog.Error("agent connection attempt failed; retry will continue", ("phase", _phase), + ("exception", ex.GetType().FullName), ("err", ex.Message), + ("elapsed_ms", (long)_env.Time.GetElapsedTime(_attemptStarted).TotalMilliseconds), + ("socket_error", ex is SocketException socket ? socket.SocketErrorCode : null)); } try { TimeSpan delay = backoff.Next(); + _attemptLog.Info("agent retry scheduled", ("retry_attempt", backoff.Attempts), ("retry_in_ms", (long)delay.TotalMilliseconds), + ("phase", _phase)); lock (_handshakeGate) { _attemptState = ProbeLinkAttemptState.RetryWait; @@ -175,7 +188,7 @@ public sealed class AgentConnection : IDisposable if (completed == retryTask) { await retryTask; - _env.Log.Info("manual link retry requested", ("agent", Endpoint.AgentId)); + _attemptLog.Info("manual link retry requested"); } else { @@ -193,6 +206,9 @@ public sealed class AgentConnection : IDisposable private async Task RunOnceAsync(ProbeReconnectBackoff backoff, CancellationToken ct) { + _phase = "dns"; + _attemptLog.Info("agent connection attempt started", ("phase", _phase), ("host", Endpoint.Host), + ("tcp_port", Endpoint.TcpPort), ("quic_ports", string.Join(',', Endpoint.EffectiveQuicPorts))); AddressFamily family = Endpoint.IpVersion == IpVersions.V6 ? AddressFamily.InterNetworkV6 : AddressFamily.InterNetwork; using var resolveTimeout = new CancellationTokenSource(TimeSpan.FromSeconds(10), _env.Time); using CancellationTokenSource resolveCt = CancellationTokenSource.CreateLinkedTokenSource(ct, resolveTimeout.Token); @@ -205,9 +221,12 @@ public sealed class AgentConnection : IDisposable throw new SocketException((int)SocketError.HostNotFound); } + _attemptLog.Info("agent address resolved", ("phase", _phase), ("remote", address), + ("elapsed_ms", (long)_env.Time.GetElapsedTime(_attemptStarted).TotalMilliseconds)); + _phase = "control_handshake"; byte[] agentKey = Convert.FromBase64String(Endpoint.PublicKey); (PeerSessionInfo info, TlsFrameTransport transport) = await PeerConnector.ConnectAsync( - address.ToString(), Endpoint.TcpPort, agentKey, _env.Identity, _env.ProbeId, ProbeApp.Version, 8 << 20, _env.Time, ct); + address.ToString(), Endpoint.TcpPort, agentKey, _env.Identity, _env.ProbeId, ProbeApp.Version, 8 << 20, _env.Time, _attemptLog, ct); var data = new DataSession(info.Credentials.SessionIdValue, info.Credentials.Key); var remote = (IPEndPoint)info.Remote!; IReadOnlyList quicPorts = Endpoint.EffectiveQuicPorts; @@ -216,7 +235,7 @@ public sealed class AgentConnection : IDisposable data, _env.Clock, new DataPlaneCounters(), - _env.Log, + _attemptLog, new IPEndPoint(remote.Address, quicPort), agentKey, Volatile.Read(ref _previousQuicSourcePort)); @@ -230,7 +249,7 @@ public sealed class AgentConnection : IDisposable var blockEnv = new BlockEnv { Clock = _env.Clock, - Log = _env.Log, + Log = _attemptLog, Session = () => data, Datagrams = datagrams, PeerAddress = remote.Address, @@ -250,7 +269,9 @@ public sealed class AgentConnection : IDisposable try { // 先同步时钟（块头里要带偏移）；Agent 随后推送线路任务集 + _phase = "clock_sync"; await SyncClockAsync(rpc, clock, rounds: 5, cts.Token); + _attemptLog.Info("agent clock synchronized", ("phase", _phase), ("offset_us", clock.OffsetUs)); lock (_handshakeGate) { _handshakeAttempts++; @@ -259,6 +280,7 @@ public sealed class AgentConnection : IDisposable try { SetAttempt(ProbeLinkAttemptState.Connecting, null, null, 0); + _phase = "quic_handshake"; await datagrams.ConnectAsync(cts.Token); lock (_handshakeGate) { @@ -294,8 +316,10 @@ public sealed class AgentConnection : IDisposable backoff.Reset(); _retrySignal.Wait(0, CancellationToken.None); // 若“立即重试”恰好赶在本次拨号成功前到达，本次成功已满足请求。 _session = session; + _phase = "measuring"; SetAttempt(ProbeLinkAttemptState.Connected, null, null, 0); - _env.Log.Info("agent session up", ("agent", Endpoint.AgentId), ("offset_us", clock.OffsetUs)); + _attemptLog.Info("agent session up", ("phase", _phase), ("remote", remote), ("quic_port", quicPort), + ("local_udp_ports", string.Join(',', datagrams.BoundPorts)), ("handshake_ms", datagrams.HandshakeDurationMs), ("offset_us", clock.OffsetUs)); Task maintenance = MaintainAsync(rpc, session, clock, datagrams, cts.Token); Task finished = await Task.WhenAny(run, receive, pump, maintenance); string reason = finished == run @@ -346,7 +370,7 @@ public sealed class AgentConnection : IDisposable { if (!await Bounded.WaitAsync(work(), Bounded.Teardown, _env.Time)) { - _env.Log.Warn("agent session teardown step timed out, moving on", ("agent", Endpoint.AgentId), ("step", step)); + _attemptLog.Warn("agent session teardown step timed out, moving on", ("phase", _phase), ("step", step), ("timeout_ms", Bounded.Teardown.TotalMilliseconds)); } } @@ -356,7 +380,7 @@ public sealed class AgentConnection : IDisposable { await task; } - catch (Exception ex) when (ex is OperationCanceledException or IOException or ObjectDisposedException or System.Net.Sockets.SocketException or InvalidOperationException) + catch (Exception ex) when (ex is OperationCanceledException or IOException or ObjectDisposedException or System.Net.Sockets.SocketException or InvalidOperationException or TimeoutException) { // 会话结束时的正常噪声 } @@ -392,12 +416,13 @@ public sealed class AgentConnection : IDisposable long seenIn = datagrams.Counters.PacketsIn; long seenOut = datagrams.Counters.PacketsOut; long inChangedUs = lastClockUs; + long lastReportUs = lastClockUs; while (await flush.WaitForNextTickAsync(ct)) { if (rpc.IdleFor > s_idleLimit) { // Agent 已经不认这个会话（它重启、重连、或者空闲超时关了会话，而 FIN 没有到达）：Probe 这边结束它，让外层按退避重拨 - _env.Log.Warn("agent session is silent, reconnecting", ("agent", Endpoint.AgentId), ("idle_s", (long)rpc.IdleFor.TotalSeconds)); + _attemptLog.Warn("agent session is silent, reconnecting", ("phase", "control_session"), ("idle_s", (long)rpc.IdleFor.TotalSeconds)); throw new IOException("the Agent stopped answering on this session"); } @@ -411,11 +436,22 @@ public sealed class AgentConnection : IDisposable } else if (packetsOut != seenOut && _env.Clock.NowUs() - inChangedUs > s_dataIdleLimit.TotalMicroseconds) { - _env.Log.Warn("agent datagram path is silent, reconnecting", ("agent", Endpoint.AgentId), ("idle_s", (_env.Clock.NowUs() - inChangedUs) / 1_000_000)); + _attemptLog.Warn("agent datagram path is silent, reconnecting", ("phase", "quic_data"), + ("idle_s", (_env.Clock.NowUs() - inChangedUs) / 1_000_000), ("packets_in", packetsIn), ("packets_out", packetsOut), + ("dropped_invalid", datagrams.Counters.DroppedInvalid), ("dropped_rate_limited", datagrams.Counters.DroppedRateLimited)); throw new IOException("the Agent has not answered any measurement datagram"); } seenOut = packetsOut; + if (_env.Clock.NowUs() - lastReportUs >= 60_000_000) + { + lastReportUs = _env.Clock.NowUs(); + QuicTransportSnapshot? snapshot = datagrams.GetTransportSnapshot(); + _attemptLog.Info("agent measurement summary", ("phase", "quic_data"), ("packets_in", packetsIn), ("packets_out", packetsOut), + ("dropped_invalid", datagrams.Counters.DroppedInvalid), ("dropped_rate_limited", datagrams.Counters.DroppedRateLimited), + ("control_idle_s", (long)rpc.IdleFor.TotalSeconds), ("roles", session.RoleCount), + ("transport_packets_sent", snapshot?.PacketsSent), ("path_mtu", snapshot?.PathMtu)); + } session.FlushLineBlocks(); if (_env.Clock.NowUs() - lastClockUs >= s_clockInterval.TotalMicroseconds) { diff --git a/backend/shared/MoShiro.Measurement/Core/Contracts.cs b/backend/shared/MoShiro.Measurement/Core/Contracts.cs index 301df12..b8df778 100644 --- a/backend/shared/MoShiro.Measurement/Core/Contracts.cs +++ b/backend/shared/MoShiro.Measurement/Core/Contracts.cs @@ -23,6 +23,9 @@ public class DataSession public byte[] Key { get; } + /// Public node identity for diagnostics; never the data-session identifier or key. + public virtual string? DiagnosticPeerId => null; + public IDatagramConnection? DatagramConnection => Volatile.Read(ref _datagramConnection); public void BindDatagramConnection(IDatagramConnection connection) => Volatile.Write(ref _datagramConnection, connection); diff --git a/backend/shared/MoShiro.Measurement/Net/MeasurementDatagramRouter.cs b/backend/shared/MoShiro.Measurement/Net/MeasurementDatagramRouter.cs index c38b55d..893c17b 100644 --- a/backend/shared/MoShiro.Measurement/Net/MeasurementDatagramRouter.cs +++ b/backend/shared/MoShiro.Measurement/Net/MeasurementDatagramRouter.cs @@ -1,6 +1,7 @@ using System.Collections.Concurrent; using System.Net; using System.Net.Sockets; +using Fleet.Core; using MoShiro.Protocol; namespace MoShiro.Measurement; @@ -46,13 +47,16 @@ internal sealed class MeasurementDatagramRouter private readonly IDataSessions _sessions; private readonly IAdmission? _admission; private readonly MonoClock _clock; + private readonly INodeLog _log; + private long _lastDropLog; private readonly ConcurrentDictionary _handlers = new(); - public MeasurementDatagramRouter(IDataSessions sessions, IAdmission? admission, MonoClock clock, DataPlaneCounters counters) + public MeasurementDatagramRouter(IDataSessions sessions, IAdmission? admission, MonoClock clock, DataPlaneCounters counters, INodeLog? log = null) { _sessions = sessions; _admission = admission; _clock = clock; + _log = log ?? NullNodeLog.Instance; Counters = counters; } @@ -69,17 +73,34 @@ internal sealed class MeasurementDatagramRouter if (datagram.Length > payload.Length) { Counters.DropInvalid(); + LogDrop(connection, "oversize", datagram.Length); return; } datagram.CopyTo(payload); payload = payload[..datagram.Length]; - if (!MeasurementDatagram.TryParseHeader(payload, out MeasurementDatagramHeader header) - || !_handlers.TryGetValue(header.StreamId, out QuicDatagramHandler? handler) - || _sessions.Find(header.SessionId) is not { } session - || !MeasurementDatagram.VerifyMac(payload, session.Key)) + if (!MeasurementDatagram.TryParseHeader(payload, out MeasurementDatagramHeader header)) { Counters.DropInvalid(); + LogDrop(connection, "invalid_header", datagram.Length); + return; + } + if (!_handlers.TryGetValue(header.StreamId, out QuicDatagramHandler? handler)) + { + Counters.DropInvalid(); + LogDrop(connection, "unknown_stream", datagram.Length, header.StreamId); + return; + } + if (_sessions.Find(header.SessionId) is not { } session) + { + Counters.DropInvalid(); + LogDrop(connection, "unknown_session", datagram.Length, header.StreamId); + return; + } + if (!MeasurementDatagram.VerifyMac(payload, session.Key)) + { + Counters.DropInvalid(); + LogDrop(connection, "invalid_mac", datagram.Length, header.StreamId); return; } @@ -87,6 +108,7 @@ internal sealed class MeasurementDatagramRouter if (remote is null) { Counters.DropInvalid(); + LogDrop(connection, "missing_remote", datagram.Length, header.StreamId); return; } @@ -97,14 +119,35 @@ internal sealed class MeasurementDatagramRouter { case Admission.Invalid: Counters.DropInvalid(); + LogDrop(connection, "admission_invalid", datagram.Length, header.StreamId); return; case Admission.RateLimited: Counters.DropRateLimited(); + LogDrop(connection, "rate_limited", datagram.Length, header.StreamId); return; } } session.BindDatagramConnection(connection); + connection.NoteAuthenticatedDatagram(session.DiagnosticPeerId, header.StreamId); handler(new QuicDatagramDelivery(header, payload, receiveUs, session, connection, Counters)); } + + private void LogDrop(MsQuicConnection connection, string reason, int bytes, uint? stream = null) + { + // One bounded throttle for this transport, regardless of attacker-controlled sources/streams. + long now = _clock.Time.GetTimestamp(); + long previous = Interlocked.Read(ref _lastDropLog); + if (previous != 0 && _clock.Time.GetElapsedTime(previous - 1, now) < TimeSpan.FromSeconds(30)) + { + return; + } + if (Interlocked.CompareExchange(ref _lastDropLog, now + 1, previous) != previous) + { + return; + } + _log.Warn("measurement datagram dropped", ("phase", "measurement"), ("reason", reason), + ("quic_id", connection.DiagnosticId), ("remote", connection.RemoteEndPoint), ("bytes", bytes), ("stream", stream), + ("invalid_total", Counters.DroppedInvalid), ("rate_limited_total", Counters.DroppedRateLimited)); + } } diff --git a/backend/shared/MoShiro.Measurement/Net/MsQuicConnection.cs b/backend/shared/MoShiro.Measurement/Net/MsQuicConnection.cs index 40a1d01..47a1218 100644 --- a/backend/shared/MoShiro.Measurement/Net/MsQuicConnection.cs +++ b/backend/shared/MoShiro.Measurement/Net/MsQuicConnection.cs @@ -40,7 +40,7 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection private readonly byte[]? _expectedPeerSpki; private readonly Action? _closed; private readonly Action? _peerAddressChanged; - private readonly INodeLog _log; + private readonly ContextNodeLog _log; private readonly Dictionary _sends = []; private nint _registration; private nint _configuration; @@ -51,6 +51,14 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection private double? _handshakeDurationMs; private QuicTransportSnapshot? _transportSnapshot; private long _statisticsAt; + private NativeCounters? _nativeCounters; + private readonly long _createdAt = Stopwatch.GetTimestamp(); + private int _localPort; + private int _receivedDatagram; + private long _sendRejected; + private long _sendRejectedAt; + + private sealed record NativeCounters(ulong Sent, ulong Received, long At); private MsQuicConnection( nint handle, @@ -67,7 +75,7 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection _configuration = configuration; _expectedPeerSpki = expectedPeerSpki; DatagramHandler = datagramHandler; - _log = log; + _log = new ContextNodeLog(log, ("quic_id", DiagnosticId), ("side", expectedPeerSpki is null ? "agent" : "probe")); _closed = closed; _peerAddressChanged = peerAddressChanged; _self = GCHandle.Alloc(this); @@ -75,6 +83,8 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection public MsQuicDatagramCallback? DatagramHandler { get; set; } + internal string DiagnosticId { get; } = Guid.NewGuid().ToString("N"); + public IPEndPoint? RemoteEndPoint { get; private set; } public IPEndPoint? LocalEndPoint => Volatile.Read(ref _handle) is nint handle && handle != 0 @@ -131,6 +141,7 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection double? loss = stats.SendTotalPackets > 0 ? lost * 100.0 / stats.SendTotalPackets : null; Volatile.Write(ref _transportSnapshot, new QuicTransportSnapshot(loss, stats.SendTotalPackets, stats.SendCongestionCount, stats.SendEcnCongestionCount, stats.SendPathMtu)); + Volatile.Write(ref _nativeCounters, new NativeCounters(stats.SendTotalPackets, stats.RecvTotalPackets, now)); } catch (InvalidOperationException) { @@ -153,11 +164,8 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection long handshakeStarted = 0; try { - log.Info("opening MsQuic client registration"); registration = MsQuicApi.RegistrationOpen("moshiro-probe"); - log.Info("opening MsQuic client configuration"); configuration = MsQuicApi.ConfigurationOpen(registration, QuicAlpn.Value); - log.Info("loading MsQuic client credentials"); LoadClientCredentials(configuration); var endpoint = new IPEndPoint(address.AddressFamily == AddressFamily.InterNetworkV6 ? IPAddress.IPv6Any : IPAddress.Any, localPort); SocketAddress serialized = endpoint.Serialize(); @@ -171,19 +179,18 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection try { nint handle = MsQuicApi.ConnectionOpen(registration, s_callback, 0); - log.Info("opened MsQuic client connection"); connection = new MsQuicConnection(handle, registration, configuration, expectedPeerSpki, datagramHandler, log, null, null); + connection.RemoteEndPoint = new IPEndPoint(address, port); + connection._localPort = localPort; registration = 0; configuration = 0; MsQuicApi.SetContext(handle, GCHandle.ToIntPtr(connection._self)); - log.Info("enabling MsQuic datagram receive"); MsQuicApi.SetParam(handle, DatagramReceiveEnabled, 1); - log.Info("binding MsQuic local address", ("local_port", localPort)); MsQuicApi.ConnectionBind(handle, localAddress); - log.Info("starting MsQuic QUIC handshake", ("remote", address), ("port", port), ("local_port", localPort)); + connection._log.Info("starting MsQuic QUIC handshake", ("phase", "quic_handshake"), ("remote", connection.RemoteEndPoint), + ("local_port", localPort), ("timeout_ms", 10_000)); handshakeStarted = Stopwatch.GetTimestamp(); MsQuicApi.ConnectionStart(handle, connection._configuration, nativeFamily, name, checked((ushort)port)); - log.Info("MsQuic connection start returned"); } finally { @@ -194,10 +201,14 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection connection._handshakeDurationMs = Stopwatch.GetElapsedTime(handshakeStarted).TotalMilliseconds; return connection; } - catch + catch (Exception ex) { if (connection is not null) { + if (!ct.IsCancellationRequested) + { + connection.LogState(NodeLogLevel.Warn, "QUIC handshake failed", ("exception", ex.GetType().Name)); + } await connection.DisposeAsync(); } @@ -211,6 +222,10 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection MsQuicApi.RegistrationClose(registration); } + if (ex is TimeoutException) + { + throw new TimeoutException($"QUIC handshake timed out after 10000 ms: remote={address}:{port}, local_udp_port={localPort}", ex); + } throw; } } @@ -226,12 +241,25 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection var connection = new MsQuicConnection(handle, 0, 0, null, datagramHandler, log, closed, peerAddressChanged); MsQuicApi.SetContext(handle, GCHandle.ToIntPtr(connection._self)); MsQuicApi.SetCallbackHandler(handle, s_callback, GCHandle.ToIntPtr(connection._self)); + connection.RefreshRemoteEndPoint(); + connection._log.Info("accepted MsQuic QUIC connection", ("phase", "quic_handshake"), ("remote", connection.RemoteEndPoint)); MsQuicApi.SetParam(handle, DatagramReceiveEnabled, 1); // Transport parameters must be set before SetConfiguration starts TLS. MsQuicApi.ConnectionSetConfiguration(handle, configuration); return connection; } + private int _authenticatedDatagram; + + internal void NoteAuthenticatedDatagram(string? peerId, uint streamId) + { + if (Interlocked.Exchange(ref _authenticatedDatagram, 1) == 0) + { + _log.Info("first authenticated measurement datagram", ("phase", "measurement"), + ("peer", peerId), ("remote", RemoteEndPoint), ("stream", streamId)); + } + } + public bool Send(ReadOnlySpan datagram) { // ConnectionClose must not free the handle or a pending send while @@ -247,6 +275,7 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection nint handle = Volatile.Read(ref _handle); if (handle == 0 || !IsReady || datagram.Length > MaxDatagramBytes) { + LogSendRejected(handle == 0 ? "connection_closed" : !IsReady ? "datagram_not_ready" : "datagram_too_large", datagram.Length, null); return false; } @@ -282,9 +311,50 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection } ReleaseSend(context); + LogSendRejected("native_send_failed", datagram.Length, status); return false; } + private void LogSendRejected(string reason, int bytes, uint? status) + { + long rejected = Interlocked.Increment(ref _sendRejected); + long now = Stopwatch.GetTimestamp(); + long previous = Interlocked.Read(ref _sendRejectedAt); + if (previous != 0 && Stopwatch.GetElapsedTime(previous, now) < TimeSpan.FromSeconds(30)) + { + return; + } + if (Interlocked.CompareExchange(ref _sendRejectedAt, now, previous) != previous) + { + return; + } + + _log.Warn("QUIC datagram submission rejected", ("phase", "quic_data"), ("reason", reason), ("remote", RemoteEndPoint), + ("bytes", bytes), ("max_datagram", MaxDatagramBytes), ("rejected_total", rejected), ("status", status is { } s ? $"0x{s:X8}" : null)); + } + + // Diagnostics only read managed snapshots. Never query the native worker from + // timeout, heartbeat, or teardown paths: the worker may itself be blocked. + private void LogState(NodeLogLevel level, string message, params (string Key, object? Value)[] details) + { + NativeCounters? counters = Volatile.Read(ref _nativeCounters); + _log.Write(level, message, [ + ("remote", RemoteEndPoint), ("local_port", _localPort), + ("phase", _connected.Task.IsCompletedSuccessfully ? "quic_data" : "quic_handshake"), + ("elapsed_ms", (long)Stopwatch.GetElapsedTime(_createdAt).TotalMilliseconds), + ("handshake_completed", _connected.Task.IsCompletedSuccessfully), + ("packets_sent", counters?.Sent), ("packets_received", counters?.Received), + ("stats_age_ms", counters is null ? null : (long?)Stopwatch.GetElapsedTime(counters.At).TotalMilliseconds), + .. details, + ]); + } + + internal static string StatusName(uint status) => status == 0 ? "success" : !OperatingSystem.IsLinux() ? "native_error" : status switch + { + 1 => "invalid_state", 22 => "invalid_parameter", 62 => "connection_idle", 103 => "handshake_failure", + 110 => "connection_timeout", 111 => "connection_refused", 113 => "unreachable", 125 => "aborted", _ => "native_error", + }; + public async ValueTask DisposeAsync() { lock (_nativeGate) @@ -375,22 +445,31 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection switch (eventType) { case ConnectionEventConnected: - connection._log.Info("MsQuic QUIC handshake completed"); connection.RefreshRemoteEndPoint(); + try + { + connection._localPort = connection.LocalEndPoint?.Port ?? connection._localPort; + } + catch (InvalidOperationException) + { + // An unavailable diagnostic address must not fail a successful handshake. + } connection._connected.TrySetResult(); + connection.LogState(NodeLogLevel.Info, "MsQuic QUIC handshake completed"); break; case ConnectionEventShutdownByTransport: { uint status = unchecked((uint)Marshal.ReadInt32(eventPointer, IntPtr.Size)); ulong errorCode = unchecked((ulong)Marshal.ReadInt64(eventPointer, IntPtr.Size * 2)); - connection._log.Error("MsQuic QUIC transport shutdown", ("status", $"0x{status:X8}"), ("error_code", errorCode)); + connection.LogState(NodeLogLevel.Error, "MsQuic QUIC transport shutdown", ("status", $"0x{status:X8}"), + ("status_name", StatusName(status)), ("error_code", errorCode)); connection._connected.TrySetException(new IOException($"QUIC connection closed during handshake (transport status 0x{status:X8}, error code {errorCode}).")); break; } case ConnectionEventShutdownByPeer: { ulong errorCode = unchecked((ulong)Marshal.ReadInt64(eventPointer, IntPtr.Size)); - connection._log.Warn("MsQuic QUIC peer shutdown", ("error_code", errorCode)); + connection.LogState(NodeLogLevel.Warn, "MsQuic QUIC peer shutdown", ("error_code", errorCode)); connection._connected.TrySetException(new IOException($"QUIC connection closed during handshake by peer (error code {errorCode}).")); break; } @@ -403,6 +482,7 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection IPEndPoint? endpoint = ReadEndpoint(address); if (endpoint is not null) { + connection._log.Info("QUIC peer address changed", ("previous", connection.RemoteEndPoint), ("remote", endpoint)); connection.RemoteEndPoint = endpoint; connection._peerAddressChanged?.Invoke(endpoint); } @@ -446,8 +526,13 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection private void OnDatagramStateChanged(nint eventPointer) { (bool sendEnabled, int maxSendLength) = ReadDatagramState(eventPointer); - Volatile.Write(ref _maxDatagramBytes, maxSendLength); - Volatile.Write(ref _sendEnabled, sendEnabled ? 1 : 0); + int previousMax = Interlocked.Exchange(ref _maxDatagramBytes, maxSendLength); + int previousEnabled = Interlocked.Exchange(ref _sendEnabled, sendEnabled ? 1 : 0); + if (previousMax != maxSendLength || previousEnabled != (sendEnabled ? 1 : 0)) + { + _log.Info("QUIC datagram negotiation changed", ("phase", "datagram_negotiation"), ("remote", RemoteEndPoint), + ("send_enabled", sendEnabled), ("max_datagram", maxSendLength)); + } } internal static (bool SendEnabled, int MaxSendLength) ReadDatagramState(nint eventPointer) @@ -475,6 +560,10 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection var buffer = Marshal.PtrToStructure(bufferPointer); if (buffer.Buffer != 0 && buffer.Length > 0 && buffer.Length <= 65535) { + if (Interlocked.Exchange(ref _receivedDatagram, 1) == 0) + { + _log.Info("first QUIC datagram received", ("phase", "quic_data"), ("remote", RemoteEndPoint), ("bytes", buffer.Length)); + } unsafe { DatagramHandler?.Invoke(this, new ReadOnlySpan((void*)buffer.Buffer, checked((int)buffer.Length))); @@ -581,6 +670,8 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection private void OnShutdownComplete(nint handle, nint eventPointer) { + UpdateTransportSnapshot(); + LogState(NodeLogLevel.Info, "QUIC connection shutdown completed", ("rejected_sends", Interlocked.Read(ref _sendRejected))); if (!_connected.Task.IsCompletedSuccessfully) { byte flags = Marshal.ReadByte(eventPointer, IntPtr.Size); diff --git a/backend/shared/MoShiro.Measurement/Net/MsQuicListener.cs b/backend/shared/MoShiro.Measurement/Net/MsQuicListener.cs index dcf2dac..6201f29 100644 --- a/backend/shared/MoShiro.Measurement/Net/MsQuicListener.cs +++ b/backend/shared/MoShiro.Measurement/Net/MsQuicListener.cs @@ -76,6 +76,8 @@ internal sealed class MsQuicListener : IAsyncDisposable MsQuicApi.ListenerStart(listener, QuicAlpn.Value, address); effectivePort = ReadPort(MsQuicApi.GetAddress(listener, ListenerLocalAddress)); bound.Add(effectivePort); + _log.Info("QUIC listener ready", ("bind", bindAddress), ("port", effectivePort), + ("ip_version", bindAddress.AddressFamily == AddressFamily.InterNetworkV6 ? "ipv6" : "ipv4")); } } @@ -131,7 +133,6 @@ internal sealed class MsQuicListener : IAsyncDisposable } nint connectionHandle = Marshal.ReadIntPtr(@event, IntPtr.Size * 2); - owner._log.Info("accepted MsQuic QUIC connection"); MsQuicConnection? connection = null; connection = MsQuicConnection.Accept( connectionHandle, diff --git a/backend/shared/MoShiro.Measurement/Net/QuicDatagramTransport.cs b/backend/shared/MoShiro.Measurement/Net/QuicDatagramTransport.cs index 59ea32c..94e391b 100644 --- a/backend/shared/MoShiro.Measurement/Net/QuicDatagramTransport.cs +++ b/backend/shared/MoShiro.Measurement/Net/QuicDatagramTransport.cs @@ -30,7 +30,7 @@ public sealed class QuicDatagramTransport : IDisposable, IAsyncDisposable /// Constructs an Agent-side QUIC listener. public QuicDatagramTransport(IDataSessions sessions, IAdmission? admission, MonoClock clock, DataPlaneCounters counters, INodeLog log, IPAddress bind, IReadOnlyList ports, X509Certificate2 serverCertificate) { - _router = new MeasurementDatagramRouter(sessions, admission, clock, counters); + _router = new MeasurementDatagramRouter(sessions, admission, clock, counters, log); _log = log; _onDatagram = _router.Dispatch; _bind = bind; @@ -41,7 +41,7 @@ public sealed class QuicDatagramTransport : IDisposable, IAsyncDisposable /// Constructs a Probe-side QUIC dialer for one Agent session. public QuicDatagramTransport(DataSession session, MonoClock clock, DataPlaneCounters counters, INodeLog log, IPEndPoint remote, byte[] peerSpki, int previousSourcePort) { - _router = new MeasurementDatagramRouter(new SingleSession(session), null, clock, counters); + _router = new MeasurementDatagramRouter(new SingleSession(session), null, clock, counters, log); _log = log; _onDatagram = _router.Dispatch; _bind = remote.AddressFamily == AddressFamily.InterNetworkV6 ? IPAddress.IPv6Any : IPAddress.Any; diff --git a/backend/shared/tests/MoShiro.Measurement.Tests/MsQuicConnectionTests.cs b/backend/shared/tests/MoShiro.Measurement.Tests/MsQuicConnectionTests.cs index 8c22e5d..81ffc8e 100644 --- a/backend/shared/tests/MoShiro.Measurement.Tests/MsQuicConnectionTests.cs +++ b/backend/shared/tests/MoShiro.Measurement.Tests/MsQuicConnectionTests.cs @@ -1,4 +1,5 @@ using System.Net; +using System.Collections.Concurrent; using System.Runtime.InteropServices; using System.Security.Cryptography; using System.Security.Cryptography.X509Certificates; @@ -31,11 +32,23 @@ public sealed class MsQuicConnectionTests [Fact] public async Task DatagramConnection_CanSendCloseAndReconnect_WithoutBlockingNativeWorkers() { + var log = new CaptureLog(); + var counters = new DataPlaneCounters(); + var router = new MeasurementDatagramRouter(new SingleSession(new DataSession(1, new byte[32])), null, + new MonoClock(TimeProvider.System), counters, log); using ECDsa key = ECDsa.Create(ECCurve.NamedCurves.nistP256); var request = new CertificateRequest("CN=localhost", key, HashAlgorithmName.SHA256); using X509Certificate2 certificate = request.CreateSelfSigned(DateTimeOffset.UnixEpoch, DateTimeOffset.UnixEpoch.AddYears(100)); await using var listener = new MsQuicListener(IPAddress.Loopback, [0], certificate, - (connection, data) => connection.Send(data), NullNodeLog.Instance, null); + (connection, data) => + { + // Bad measurement payloads must be counted without flooding logs. + for (int i = 0; i < 100; i++) + { + router.Dispatch(connection, data); + } + connection.Send(data); + }, log, null); listener.Start(); for (int attempt = 0; attempt < 3; attempt++) @@ -43,7 +56,7 @@ public sealed class MsQuicConnectionTests var reply = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10)); MsQuicConnection client = await MsQuicConnection.ConnectAsync(IPAddress.Loopback, listener.BoundPorts[0], 0, - key.ExportSubjectPublicKeyInfo(), (_, data) => reply.TrySetResult(data.ToArray()), NullNodeLog.Instance, timeout.Token); + key.ExportSubjectPublicKeyInfo(), (_, data) => reply.TrySetResult(data.ToArray()), log, timeout.Token); try { while (!client.IsReady) @@ -72,6 +85,22 @@ public sealed class MsQuicConnectionTests await client.WaitForShutdownAsync().WaitAsync(TimeSpan.FromSeconds(3)); } } + Assert.True(counters.DroppedInvalid >= 300); + var dropped = Assert.Single(log.Items, x => x.Message == "measurement datagram dropped"); + Assert.Equal("invalid_header", dropped.Fields["reason"]); + Assert.NotNull(dropped.Fields["remote"]); + Assert.NotNull(dropped.Fields["quic_id"]); + var connected = log.Items.Where(x => x.Message == "MsQuic QUIC handshake completed").ToArray(); + Assert.Equal(6, connected.Select(x => x.Fields["quic_id"]).Distinct().Count()); + Assert.All(connected, entry => Assert.NotNull(entry.Fields["remote"])); + Assert.DoesNotContain(log.Items.SelectMany(x => x.Fields.Keys), key => key is "key" or "session_id" or "mac" or "signature"); } + private sealed class CaptureLog : INodeLog + { + public ConcurrentQueue<(string Message, Dictionary Fields)> Items { get; } = new(); + + public void Write(NodeLogLevel level, string message, params (string Key, object? Value)[] fields) => + Items.Enqueue((message, fields.ToDictionary(x => x.Key, x => x.Value))); + } } diff --git a/backend/src/MoShiro.Backend.Application/Agents/NodeRuntimeLogService.cs b/backend/src/MoShiro.Backend.Application/Agents/NodeRuntimeLogService.cs index e537ed3..5cd32f9 100644 --- a/backend/src/MoShiro.Backend.Application/Agents/NodeRuntimeLogService.cs +++ b/backend/src/MoShiro.Backend.Application/Agents/NodeRuntimeLogService.cs @@ -78,7 +78,7 @@ public sealed class NodeRuntimeLogService(TimeProvider time) return safe; } - foreach ((string key, string value) in fields.Take(8)) + foreach ((string key, string value) in fields.Take(20)) { string safeKey = Truncate(key, 48); string lower = key.ToLowerInvariant(); diff --git a/backend/tests/MoShiro.Backend.Tests/NodeRuntimeLogTests.cs b/backend/tests/MoShiro.Backend.Tests/NodeRuntimeLogTests.cs new file mode 100644 index 0000000..82a32fb --- /dev/null +++ b/backend/tests/MoShiro.Backend.Tests/NodeRuntimeLogTests.cs @@ -0,0 +1,27 @@ +using Fleet.Core; +using MoShiro.Backend.Application; +using MoShiro.Protocol; + +namespace MoShiro.Backend.Tests; + +public sealed class NodeRuntimeLogTests +{ + [Fact] + public void DiagnosticFieldsSurviveUpload_WhileSecretsAndOversizedFieldsRemainBounded() + { + var service = new NodeRuntimeLogService(TimeProvider.System); + var node = new NodeId("prb_diagnostics"); + Dictionary fields = Enumerable.Range(0, 18).ToDictionary(i => "field_" + i, _ => "value"); + fields.Add("api_token", "must-not-survive"); + fields.Add("detail", new string('x', 500)); + fields.Add("overflow", "discard"); + service.Append(node, [new NodeLogRecord(TimeProvider.System.GetUtcNow().ToUnixTimeMilliseconds(), "warn", "handshake failed", fields)], 0); + + NodeRuntimeLog entry = Assert.Single(service.List(node, 10).Items); + Assert.Equal(20, entry.Fields.Count); + Assert.Equal("value", entry.Fields["field_17"]); + Assert.Equal("[redacted]", entry.Fields["api_token"]); + Assert.Equal(128, entry.Fields["detail"].Length); + Assert.False(entry.Fields.ContainsKey("overflow")); + } +} diff --git a/docs/ops/deployment.md b/docs/ops/deployment.md index 40118ef..69d6fe4 100644 --- a/docs/ops/deployment.md +++ b/docs/ops/deployment.md @@ -4,6 +4,18 @@ ## 1. 构建与打包 +### Probe ↔ Agent 连接日志 + +管理端节点日志收集节点通过心跳上传的近期日志；原始日志可从 systemd journal、容器 logs 或 OpenWrt `logread` 查看。排障时同时查看 Probe 和 Agent，并按时间、`remote` 端点对应两端记录。 + +- Probe 每次重连生成独立 `attempt_id`，附带 `agent`、`ip_version`。阶段依次为 DNS、TCP 连接、敲门、TLS、对等认证、时钟同步、QUIC 握手和测量。失败记录包括阶段、耗时、异常及重试间隔。 +- Agent 控制会话带 `probe`、`ip_version`、`control_id`，记录替换、主动断开、吊销、空闲超时、控制连接异常和会话时长。通过敲门验证前的连接保持静默。 +- 每条 QUIC 连接带独立 `quic_id`，记录远端、本地 UDP 端口、握手、DATAGRAM 协商状态及最大长度、首个数据报、首个通过认证的测量数据报和关闭状态。首个认证数据报在 Agent 端关联公开 Probe ID；这些关联 ID 均为本地诊断标记，两端的 ID 不相同。 +- QUIC 原生关闭码同时保留十六进制和可识别的原因，例如 Linux `0x0000006E` 为 `connection_timeout`。原生收发统计来自回调缓存，`stats_age_ms` 表示缓存年龄；空值表示尚无快照，不能据此断言没有发包，确认网络丢包仍需抓包。 +- 测量数据报拒绝区分格式、未知流、未知会话、MAC 校验、准入策略和限速。拒绝日志每个传输实例最多每 30 秒一条，计数始终累计；发送失败也按连接限频。正常会话每 60 秒输出摘要，避免逐包日志干扰测量。 + +日志不记录会话密钥、会话凭据、MAC、签名或包内容。近期日志缓冲仍有容量和心跳批次预算；大量重连时可能覆盖旧记录，管理端的 dropped 数量表示丢弃的日志条数。 + ### IPv6 状态诊断 Agent 和 Probe 在 hello 初始化及心跳中上报 IPv6 检测结果，包括监听状态、全局地址、默认路由和网络命名空间。容器可提供只读宿主 `/proc` 挂载，附带宿主网络状态。只传布尔状态，不传 IP 或网卡名称。Docker 和 Podman 推荐宿主网络模式；自定义桥接网络时，需要启用 IPv6 并提供 `/proc` 宿主挂载，前端会提示宿主和容器网络状态是否不同。