diff --git a/backend/agent/src/MoShiro.Agent/AgentApp.cs b/backend/agent/src/MoShiro.Agent/AgentApp.cs index a3880a8..614de02 100644 --- a/backend/agent/src/MoShiro.Agent/AgentApp.cs +++ b/backend/agent/src/MoShiro.Agent/AgentApp.cs @@ -28,7 +28,7 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable private readonly LoadedNodeIdentity _identity; private readonly TimeProvider _time; private readonly INodeLog _log; - private readonly INodeLogBatchSource? _logBatchSource; + private readonly INodeLogSource? _logSource; private readonly MonoClock _clock; private readonly AgentCounters _counters = new(); private readonly SessionRegistry _sessions = new(); @@ -62,7 +62,7 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable _identity = identity; _time = time; _log = log; - _logBatchSource = log as INodeLogBatchSource; + _logSource = log as INodeLogSource; _clock = new MonoClock(time); _uplink = new BackendUplink(time); @@ -320,6 +320,23 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable async (_, request, _) => Result.Ok(new TestRunResult { Accepted = await _hub.DisconnectAsync(request.DialerId.Value) })); } + if (Allow(MoShiroMethods.LogsFetch)) + { + router.On( + MoShiroMethods.LogsFetch, + MoShiroJsonContext.Default.LogsFetchParams, + MoShiroJsonContext.Default.LogsFetchResult, + (_, request, _) => + { + NodeLogPage page = _logSource?.Read(request.AfterId, request.Limit, request.MinLevel) ?? new NodeLogPage([], 0); + return ValueTask.FromResult(Result.Ok(new LogsFetchResult + { + LastId = page.LastId, + Items = [.. page.Items.Select(e => new LogEntryDto { Id = e.Id, Type = e.Type, AtUtcMs = e.Record.AtUtcMs, Level = e.Record.Level, Message = e.Record.Message, Fields = e.Record.Fields })], + })); + }); + } + if (Allow(MoShiroMethods.LinkRetry)) { router.OnNotify(MoShiroMethods.LinkRetry, MoShiroJsonContext.Default.LinkRetryParams, (_, request, _) => @@ -407,7 +424,7 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable long seconds = Math.Max(1, (nowMs - _lastAtMs) / 1000); long packetsIn = _counters.PacketsIn; long packetsOut = _counters.PacketsOut; - NodeLogBatch logBatch = _logBatchSource?.TakeBatch(16) ?? new NodeLogBatch([], 0); + NodeLogSummary? logSummary = _logSource?.TakeSummary(); UpdateStatus? updateStatus = _nativeUpdate?.Status() ?? _update?.Status(); var extra = new NodeHeartbeatExtra { @@ -421,8 +438,14 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable DroppedRateLimited = _counters.DroppedRateLimited, QueueBytes = _outbox.PendingBytes, Update = updateStatus, - RuntimeLogs = logBatch.Records.Count == 0 ? null : [.. logBatch.Records], - DroppedRuntimeLogs = logBatch.Dropped == 0 ? null : logBatch.Dropped, + LogSummary = logSummary is null ? null : new LogSummaryDto + { + LastId = logSummary.LastId, + Info = logSummary.Info, + Warn = logSummary.Warn, + Error = logSummary.Error, + Types = [.. logSummary.Types.Select(t => new LogTypeCountDto { Type = t.Type, Level = t.Level, Count = t.Count })], + }, }; _lastIn = packetsIn; _lastOut = packetsOut; diff --git a/backend/fleet/src/Fleet.Core/NodeSupport.cs b/backend/fleet/src/Fleet.Core/NodeSupport.cs index 641322e..68ecc62 100644 --- a/backend/fleet/src/Fleet.Core/NodeSupport.cs +++ b/backend/fleet/src/Fleet.Core/NodeSupport.cs @@ -15,15 +15,27 @@ public interface INodeLog void Write(NodeLogLevel level, string message, params (string Key, object? Value)[] fields); } -/// One structured node log line queued for delivery to the control plane. +/// One structured node log line. public sealed record NodeLogRecord(long AtUtcMs, string Level, string Message, Dictionary Fields); -public sealed record NodeLogBatch(IReadOnlyList Records, long Dropped); +/// A retained log line: grows with every line, is a stable hash of the message text (the template), so counters can be sent without the text. +public sealed record NodeLogEntry(long Id, uint Type, NodeLogRecord Record); -/// Optional bounded log source used by node heartbeats to upload recent local logs. -public interface INodeLogBatchSource +/// How many lines of one type and level were written since the last summary. +public sealed record NodeLogTypeCount(uint Type, string Level, long Count); + +/// What a node tells the control plane about its logs without sending any text: counters since the last summary and the newest line id. +public sealed record NodeLogSummary(long LastId, long Info, long Warn, long Error, IReadOnlyList Types); + +public sealed record NodeLogPage(IReadOnlyList Items, long LastId); + +/// Bounded local log retention. The control plane gets counters in heartbeats and pulls the text when it needs it. +public interface INodeLogSource { - NodeLogBatch TakeBatch(int maxRecords); + NodeLogSummary TakeSummary(); + + /// Lines with an id above , oldest first, at most , at or above (info, warn or error). + NodeLogPage Read(long afterId, int limit, string? minLevel); } public sealed class NullNodeLog : INodeLog diff --git a/backend/fleet/src/Fleet.Hosting/BufferedNodeLog.cs b/backend/fleet/src/Fleet.Hosting/BufferedNodeLog.cs index 81063a4..22df52e 100644 --- a/backend/fleet/src/Fleet.Hosting/BufferedNodeLog.cs +++ b/backend/fleet/src/Fleet.Hosting/BufferedNodeLog.cs @@ -3,20 +3,28 @@ using Fleet.Core; namespace Fleet.Hosting; -/// Writes logs locally and keeps a bounded copy for the next authenticated heartbeat. -public sealed class BufferedNodeLog : INodeLog, INodeLogBatchSource +/// +/// Writes logs locally and keeps a bounded ring of recent lines. Heartbeats carry only counters (see ); +/// the text leaves the node when the control plane asks for it (see ). +/// +public sealed class BufferedNodeLog : INodeLog, INodeLogSource { - private const int Capacity = 256; + private const int Capacity = 2048; private const int MaxFields = 20; private const int MaxMessageLength = 512; private const int MaxFieldValueLength = 256; - private const int MaxBatchBytes = 8 << 10; + private const int MaxPageBytes = 96 << 10; + private const int MaxSummaryTypes = 8; private readonly object _gate = new(); - private readonly Queue _pending = []; + private readonly LinkedList _ring = []; + private readonly Dictionary<(uint Type, string Level), long> _counts = []; private readonly INodeLog _local; private readonly TimeProvider _time; - private long _dropped; + private long _lastId; + private long _info; + private long _warn; + private long _error; public BufferedNodeLog(INodeLog local, TimeProvider time) { @@ -35,50 +43,93 @@ public sealed class BufferedNodeLog : INodeLog, INodeLogBatchSource safeFields[key.Length <= 64 ? key : key[..64]] = Truncate(safeValue, MaxFieldValueLength); } - var record = new NodeLogRecord( - _time.GetUtcNow().ToUnixTimeMilliseconds(), - level.ToString().ToLowerInvariant(), - Truncate(message, MaxMessageLength), - safeFields); - + string levelName = level.ToString().ToLowerInvariant(); + var record = new NodeLogRecord(_time.GetUtcNow().ToUnixTimeMilliseconds(), levelName, Truncate(message, MaxMessageLength), safeFields); + uint type = TypeOf(record.Message); lock (_gate) { - if (_pending.Count == Capacity) + _ring.AddLast(new NodeLogEntry(++_lastId, type, record)); + if (_ring.Count > Capacity) { - _pending.Dequeue(); - _dropped++; + _ring.RemoveFirst(); } - _pending.Enqueue(record); + _counts[(type, levelName)] = _counts.GetValueOrDefault((type, levelName)) + 1; + switch (level) + { + case NodeLogLevel.Error: + _error++; + break; + case NodeLogLevel.Warn: + _warn++; + break; + default: + _info++; + break; + } } } - public NodeLogBatch TakeBatch(int maxRecords) + public NodeLogSummary TakeSummary() { - int count = Math.Clamp(maxRecords, 0, 32); lock (_gate) { - var records = new List(Math.Min(count, _pending.Count)); + var summary = new NodeLogSummary( + _lastId, _info, _warn, _error, + [.. _counts.OrderByDescending(pair => pair.Value).Take(MaxSummaryTypes).Select(pair => new NodeLogTypeCount(pair.Key.Type, pair.Key.Level, pair.Value))]); + _counts.Clear(); + _info = _warn = _error = 0; + return summary; + } + } + + public NodeLogPage Read(long afterId, int limit, string? minLevel) + { + int rank = LevelRank(minLevel); + lock (_gate) + { + var items = new List(Math.Min(Math.Max(limit, 0), 200)); int bytes = 0; - while (records.Count < count && _pending.TryPeek(out NodeLogRecord? record)) + foreach (NodeLogEntry entry in _ring) { - int recordBytes = 64 + record.Message.Length + record.Fields.Sum(pair => pair.Key.Length + pair.Value.Length + 8); - if (records.Count > 0 && bytes + recordBytes > MaxBatchBytes) + if (entry.Id <= afterId || LevelRank(entry.Record.Level) < rank) + { + continue; + } + + int entryBytes = 64 + entry.Record.Message.Length + entry.Record.Fields.Sum(pair => pair.Key.Length + pair.Value.Length + 8); + if (items.Count >= Math.Clamp(limit, 0, 200) || (items.Count > 0 && bytes + entryBytes > MaxPageBytes)) { break; } - _pending.Dequeue(); - records.Add(record); - bytes += recordBytes; + items.Add(entry); + bytes += entryBytes; } - long dropped = _dropped; - _dropped = 0; - return new NodeLogBatch(records, dropped); + return new NodeLogPage(items, _lastId); } } + /// The line's type: a stable 32-bit FNV-1a hash of its message text, so the same template always maps to the same number. + public static uint TypeOf(string message) + { + uint hash = 2166136261; + foreach (char c in message) + { + hash = (hash ^ c) * 16777619; + } + + return hash; + } + + private static int LevelRank(string? level) => level switch + { + "error" => 2, + "warn" => 1, + _ => 0, + }; + private static bool IsSensitiveKey(string key) { string lower = key.ToLowerInvariant(); diff --git a/backend/fleet/tests/Fleet.Tests/BufferedNodeLogTests.cs b/backend/fleet/tests/Fleet.Tests/BufferedNodeLogTests.cs new file mode 100644 index 0000000..16e78b3 --- /dev/null +++ b/backend/fleet/tests/Fleet.Tests/BufferedNodeLogTests.cs @@ -0,0 +1,65 @@ +using Fleet.Core; +using Fleet.Hosting; + +namespace Fleet.Tests; + +public sealed class BufferedNodeLogTests +{ + [Fact] + public void Summary_CountsByLevelAndTypeWithoutText_AndResetsAfterEachTake() + { + var log = new BufferedNodeLog(NullNodeLog.Instance, TimeProvider.System); + for (int i = 0; i < 5; i++) + { + log.Warn("peer control handshake failed", ("attempt", i)); + } + + log.Info("link up"); + log.Error("MsQuic QUIC transport shutdown"); + + NodeLogSummary first = log.TakeSummary(); + Assert.Equal((1L, 5L, 1L), (first.Info, first.Warn, first.Error)); + Assert.Equal(7, first.LastId); + NodeLogTypeCount top = first.Types[0]; + Assert.Equal((BufferedNodeLog.TypeOf("peer control handshake failed"), "warn", 5L), (top.Type, top.Level, top.Count)); + + NodeLogSummary second = log.TakeSummary(); + Assert.Equal((0L, 0L, 0L), (second.Info, second.Warn, second.Error)); + Assert.Empty(second.Types); + Assert.Equal(7, second.LastId); + } + + [Fact] + public void Read_ReturnsLinesAfterAnIdAtOrAboveALevel_OldestFirst_AndTheRingKeepsTheNewest() + { + var log = new BufferedNodeLog(NullNodeLog.Instance, TimeProvider.System); + log.Info("one"); + log.Warn("two"); + log.Error("three"); + log.Info("four"); + + Assert.Equal(["two", "three"], log.Read(1, 2, null).Items.Select(e => e.Record.Message)); + Assert.Equal(["three"], log.Read(0, 10, "error").Items.Select(e => e.Record.Message)); + Assert.Equal(["two", "three"], log.Read(0, 10, "warn").Items.Select(e => e.Record.Message)); + + for (int i = 0; i < 3000; i++) + { + log.Info("filler", ("n", i)); + } + + NodeLogPage page = log.Read(0, 200, null); + Assert.Equal(200, page.Items.Count); + Assert.True(page.Items[0].Id > 4); // 最早的几条已经被环挤掉 + Assert.Equal(3004, page.LastId); + } + + [Fact] + public void SensitiveFieldsAreRedacted() + { + var log = new BufferedNodeLog(NullNodeLog.Instance, TimeProvider.System); + log.Info("auth", ("token", "abc"), ("agent", "agt_x")); + NodeLogEntry entry = Assert.Single(log.Read(0, 10, null).Items); + Assert.Equal("[redacted]", entry.Record.Fields["token"]); + Assert.Equal("agt_x", entry.Record.Fields["agent"]); + } +} diff --git a/backend/fleet/tests/Fleet.Tests/LinkTests.cs b/backend/fleet/tests/Fleet.Tests/LinkTests.cs index 430e1ee..395530e 100644 --- a/backend/fleet/tests/Fleet.Tests/LinkTests.cs +++ b/backend/fleet/tests/Fleet.Tests/LinkTests.cs @@ -373,9 +373,9 @@ public sealed class LinkTests Assert.Equal("dlr_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"])); + NodeLogPage page = logs.Read(0, 32, null); + Assert.Equal(new[] { "tcp_connect", "tls", "peer_auth" }, page.Items.Select(x => x.Record.Fields["phase"])); + Assert.All(page.Items, entry => Assert.Equal("test-attempt", entry.Record.Fields["attempt_id"])); } } @@ -391,7 +391,7 @@ public sealed class LinkTests 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, dialer, "dlr_unknown", null, 1 << 20, TimeProvider.System, logs, default)); - NodeLogRecord failure = Assert.Single(logs.TakeBatch(32).Records, entry => entry.Message == "peer control handshake failed"); + NodeLogRecord failure = Assert.Single(logs.Read(0, 32, null).Items, entry => entry.Record.Message == "peer control handshake failed").Record; Assert.Equal("peer_auth", failure.Fields["phase"]); Assert.Equal("InvalidOperationException", failure.Fields["exception"]); InvalidOperationException forged = await Assert.ThrowsAsync(async () => diff --git a/backend/openapi/v1.json b/backend/openapi/v1.json index d1780d5..841d177 100644 --- a/backend/openapi/v1.json +++ b/backend/openapi/v1.json @@ -1897,7 +1897,7 @@ "tags": [ "admin-nodes" ], - "summary": "Recent runtime logs of a node", + "summary": "Recent runtime logs of a node: pulled from the node when it is online, otherwise what the backend kept", "operationId": "node_runtime_logs", "parameters": [ { @@ -9859,4 +9859,4 @@ "name": "agent" } ] -} \ No newline at end of file +} diff --git a/backend/shared/MoShiro.Protocol/Rpc/MoShiroMethodTable.cs b/backend/shared/MoShiro.Protocol/Rpc/MoShiroMethodTable.cs index 4d5d76f..c30e526 100644 --- a/backend/shared/MoShiro.Protocol/Rpc/MoShiroMethodTable.cs +++ b/backend/shared/MoShiro.Protocol/Rpc/MoShiroMethodTable.cs @@ -19,6 +19,7 @@ public static class MoShiroMethodTable MethodSpec.Request(MoShiroMethods.TestRun, RpcSide.Server), MethodSpec.Request(MoShiroMethods.TestCancel, RpcSide.Server), MethodSpec.Request(MoShiroMethods.DialerDisconnect, RpcSide.Server), + MethodSpec.Request(MoShiroMethods.LogsFetch, RpcSide.Server), MethodSpec.Notification(MoShiroMethods.LinkRetry, RpcSide.Server), MethodSpec.Notification(MoShiroMethods.UpdateOffer, RpcSide.Server), MethodSpec.Notification(MoShiroMethods.UpdateApply, RpcSide.Server), diff --git a/backend/shared/MoShiro.Protocol/Rpc/MoShiroRpc.cs b/backend/shared/MoShiro.Protocol/Rpc/MoShiroRpc.cs index df35297..ee50931 100644 --- a/backend/shared/MoShiro.Protocol/Rpc/MoShiroRpc.cs +++ b/backend/shared/MoShiro.Protocol/Rpc/MoShiroRpc.cs @@ -17,6 +17,7 @@ public static class MoShiroMethods public const string UpdateOffer = "moshiro.update.offer"; public const string UpdateApply = "moshiro.update.apply"; public const string Schedule = "moshiro.schedule"; + public const string LogsFetch = "moshiro.logs.fetch"; // 拨号方 ↔ 监听方(点对点会话) public const string Clock = "moshiro.clock"; @@ -338,11 +339,84 @@ public sealed class NodeHeartbeatExtra [JsonPropertyName("update")] public UpdateStatus? Update { get; set; } - [JsonPropertyName("runtime_logs")] - public List? RuntimeLogs { get; set; } + /// 日志只报计数(不带文本):自上次心跳以来各级别的条数、最多几个出现最多的类型,以及最新一条的编号;文本由 Backend 用 moshiro.logs.fetch 按需拉取。 + [JsonPropertyName("log_summary")] + public LogSummaryDto? LogSummary { get; set; } +} - [JsonPropertyName("dropped_runtime_logs")] - public long? DroppedRuntimeLogs { get; set; } +/// 某一类日志(消息文本的稳定哈希)在一个级别上出现了多少次。 +public sealed class LogTypeCountDto +{ + [JsonPropertyName("type")] + public uint Type { get; set; } + + [JsonPropertyName("level")] + public string Level { get; set; } = "info"; + + [JsonPropertyName("count")] + public long Count { get; set; } +} + +public sealed class LogSummaryDto +{ + [JsonPropertyName("last_id")] + public long LastId { get; set; } + + [JsonPropertyName("info")] + public long Info { get; set; } + + [JsonPropertyName("warn")] + public long Warn { get; set; } + + [JsonPropertyName("error")] + public long Error { get; set; } + + [JsonPropertyName("types")] + public List Types { get; set; } = []; +} + +/// Backend → 节点:拉取编号大于 after_id 的日志(最旧的在前),级别不低于 min_level。 +public sealed class LogsFetchParams +{ + [JsonPropertyName("after_id")] + public long AfterId { get; set; } + + [JsonPropertyName("limit")] + public int Limit { get; set; } = 100; + + [JsonPropertyName("min_level")] + public string? MinLevel { get; set; } +} + +public sealed class LogEntryDto +{ + [JsonPropertyName("id")] + public long Id { get; set; } + + [JsonPropertyName("type")] + public uint Type { get; set; } + + [JsonPropertyName("at_utc_ms")] + public long AtUtcMs { get; set; } + + [JsonPropertyName("level")] + public string Level { get; set; } = "info"; + + [JsonPropertyName("message")] + public string Message { get; set; } = ""; + + [JsonPropertyName("fields")] + public Dictionary Fields { get; set; } = []; +} + +public sealed class LogsFetchResult +{ + [JsonPropertyName("items")] + public List Items { get; set; } = []; + + /// 节点上最新一条的编号:比 after_id 小说明节点重启过,编号从头开始。 + [JsonPropertyName("last_id")] + public long LastId { get; set; } } /// @@ -749,6 +823,9 @@ public static class FragmentKeys [JsonSerializable(typeof(DialTargets))] [JsonSerializable(typeof(UpdateOfferParams))] [JsonSerializable(typeof(ScheduleParams))] +[JsonSerializable(typeof(LogsFetchParams))] +[JsonSerializable(typeof(LogsFetchResult))] +[JsonSerializable(typeof(LogSummaryDto))] [JsonSerializable(typeof(UpdateStatus))] [JsonSerializable(typeof(NodeHeartbeatExtra))] [JsonSerializable(typeof(NodeHelloMeta))] diff --git a/backend/src/MoShiro.Backend.Api/Admin/AgentEndpoints.cs b/backend/src/MoShiro.Backend.Api/Admin/AgentEndpoints.cs index 43823dc..74e9ca5 100644 --- a/backend/src/MoShiro.Backend.Api/Admin/AgentEndpoints.cs +++ b/backend/src/MoShiro.Backend.Api/Admin/AgentEndpoints.cs @@ -23,7 +23,7 @@ public static class AgentEndpoints .Api("agents_get", "One agent", AgentsTag).Returns().ProducesProblems(401, 403, 404) .Requires(Capability.FleetRead); group.MapGet("/nodes/{id}/logs", ListNodeLogsAsync) - .Api("node_runtime_logs", "Recent runtime logs of a node", "admin-nodes").Returns().ProducesProblems(401, 403) + .Api("node_runtime_logs", "Recent runtime logs of a node: pulled from the node when it is online, otherwise what the backend kept", "admin-nodes").Returns().ProducesProblems(401, 403) .Requires(Capability.FleetRead); group.MapPatch("/agents/{id}", PatchAsync) .Accepts("application/merge-patch+json", "application/json") @@ -69,10 +69,13 @@ public static class AgentEndpoints private static async Task ListAsync(AgentService agents, CancellationToken ct) => TypedResults.Json(new PageDto([.. (await agents.ListAsync(ct)).Select(Mappers.ToDto)], null), ApiJsonContext.Default.PageDtoAgentDto); - private static Microsoft.AspNetCore.Http.HttpResults.JsonHttpResult ListNodeLogsAsync(NodeRuntimeLogService logs, string id, int? limit) => - TypedResults.Json( - ToDto(logs.List(new NodeId(id), Math.Clamp(limit ?? 100, 1, 200))), - ApiJsonContext.Default.NodeRuntimeLogsDto); + /// 先向节点拉一次最新的日志(离线或超时就算了),再返回 Backend 里保留的;页面开着时前端每几秒请求一次,就是实时拉取。 + private static async Task ListNodeLogsAsync(NodeRuntimeLogService logs, string id, int? limit, CancellationToken ct) + { + var node = new NodeId(id); + await logs.PullAsync(node, 100, null, ct); + return TypedResults.Json(ToDto(logs.List(node, Math.Clamp(limit ?? 100, 1, 200))), ApiJsonContext.Default.NodeRuntimeLogsDto); + } private static NodeRuntimeLogsDto ToDto(NodeRuntimeLogSnapshot snapshot) => new([.. snapshot.Items.Select(item => new NodeRuntimeLogDto(item.Id, Mappers.Time(item.AtUtcMs), item.Level, item.Message, item.Fields))], snapshot.Dropped); diff --git a/backend/src/MoShiro.Backend.Application/Agents/MoShiroFleetHooks.cs b/backend/src/MoShiro.Backend.Application/Agents/MoShiroFleetHooks.cs index 9499dc2..fb526f9 100644 --- a/backend/src/MoShiro.Backend.Application/Agents/MoShiroFleetHooks.cs +++ b/backend/src/MoShiro.Backend.Application/Agents/MoShiroFleetHooks.cs @@ -128,7 +128,7 @@ public sealed class MoShiroNodeEventSink : INodeEventSink return; } - _nodeLogs.Append(id, beat.RuntimeLogs, beat.DroppedRuntimeLogs ?? 0); + _nodeLogs.Observe(id, beat.LogSummary); 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/Agents/NodeRuntimeLogService.cs b/backend/src/MoShiro.Backend.Application/Agents/NodeRuntimeLogService.cs index 15cc6f1..0c63c8a 100644 --- a/backend/src/MoShiro.Backend.Application/Agents/NodeRuntimeLogService.cs +++ b/backend/src/MoShiro.Backend.Application/Agents/NodeRuntimeLogService.cs @@ -8,47 +8,89 @@ public sealed record NodeRuntimeLog(long Id, long AtUtcMs, string Level, string public sealed record NodeRuntimeLogSnapshot(IReadOnlyList Items, long Dropped); -/// Bounded in-memory recent logs uploaded by authenticated Agent and 拨号方 heartbeats. -public sealed class NodeRuntimeLogService(TimeProvider time) +/// +/// 节点日志:心跳只带各级别的计数(不带文本),文本由这里按需向节点拉取。管理端打开日志页时拉最新的; +/// 节点报告了错误,也会自动拉一次警告以上的日志留着,供事后查看(Backend 内存里每个节点保留最近 条)。 +/// +public sealed class NodeRuntimeLogService(TimeProvider time, IAgentGateway gateway) { - private const int CapacityPerNode = 256; - private const int MaxBatch = 32; + private const int CapacityPerNode = 512; + private const int PullLimit = 200; + private static readonly TimeSpan s_autoPullEvery = TimeSpan.FromMinutes(1); + private readonly ConcurrentDictionary _nodes = new(); - public void Append(NodeId node, IReadOnlyList? records, long droppedByNode) + /// 心跳里的日志计数:有新的错误就在后台自动拉一次警告以上的日志。 + public void Observe(NodeId node, LogSummaryDto? summary) { - if ((records is null || records.Count == 0) && droppedByNode <= 0) + if (summary is null || summary.Error <= 0) { return; } LogBuffer buffer = _nodes.GetOrAdd(node, static _ => new LogBuffer()); + long now = time.GetUtcNow().ToUnixTimeMilliseconds(); lock (buffer.Gate) { - buffer.DroppedByNode += Math.Max(0, droppedByNode); - if (records is null) + if (buffer.Pulling || now - buffer.LastAutoPullMs < s_autoPullEvery.TotalMilliseconds) { return; } - foreach (NodeLogRecord record in records.Take(MaxBatch)) + buffer.LastAutoPullMs = now; + } + + _ = Task.Run(async () => + { + try { - long now = time.GetUtcNow().ToUnixTimeMilliseconds(); - long at = record.AtUtcMs < now - 86_400_000 || record.AtUtcMs > now + 86_400_000 ? now : record.AtUtcMs; - var entry = new NodeRuntimeLog( - ++buffer.LastId, - at, - record.Level is "info" or "warn" or "error" ? record.Level : "info", - Truncate(record.Message ?? "", 384), - SafeFields(record.Fields)); + await PullAsync(node, PullLimit, "warn", CancellationToken.None); + } + catch (Exception ex) when (ex is not OutOfMemoryException) + { + // 拉取是尽力而为:节点刚好断开或超时,下次有新错误再拉 + } + }); + } - if (buffer.Items.Count == CapacityPerNode) - { - buffer.Items.Dequeue(); - buffer.DroppedByBackend++; - } + /// 向节点拉取上次之后的新日志;节点离线或超时返回 false,已有的日志照常可看。 + public async ValueTask PullAsync(NodeId node, int limit, string? minLevel, CancellationToken ct) + { + LogBuffer buffer = _nodes.GetOrAdd(node, static _ => new LogBuffer()); + long after; + lock (buffer.Gate) + { + if (buffer.Pulling) + { + return true; // 已经有一次拉取在进行,它的结果同样会进缓冲区 + } - buffer.Items.Enqueue(entry); + buffer.Pulling = true; + after = buffer.LastFetchedId; + } + + try + { + using var timeout = CancellationTokenSource.CreateLinkedTokenSource(ct); + timeout.CancelAfter(TimeSpan.FromSeconds(5)); + Result result = await gateway.FetchLogsAsync(node, new LogsFetchParams { AfterId = after, Limit = Math.Clamp(limit, 1, PullLimit), MinLevel = minLevel }, timeout.Token); + if (!result.TryGet(out LogsFetchResult? page, out _)) + { + return false; + } + + Append(buffer, after, page); + return true; + } + catch (OperationCanceledException) when (!ct.IsCancellationRequested) + { + return false; + } + finally + { + lock (buffer.Gate) + { + buffer.Pulling = false; } } } @@ -62,9 +104,56 @@ public sealed class NodeRuntimeLogService(TimeProvider time) lock (buffer.Gate) { - return new NodeRuntimeLogSnapshot( - [.. buffer.Items.Reverse().Take(Math.Clamp(limit, 1, 200))], - buffer.DroppedByNode + buffer.DroppedByBackend); + return new NodeRuntimeLogSnapshot([.. buffer.Items.Reverse().Take(Math.Clamp(limit, 1, 200))], buffer.Dropped); + } + } + + private void Append(LogBuffer buffer, long after, LogsFetchResult page) + { + long now = time.GetUtcNow().ToUnixTimeMilliseconds(); + lock (buffer.Gate) + { + if (page.LastId < after) + { + // 节点重启过,编号从头开始:旧的编号不再有意义 + buffer.LastFetchedId = 0; + after = 0; + } + + long expected = after + 1; + foreach (LogEntryDto entry in page.Items) + { + if (entry.Id <= buffer.LastFetchedId) + { + continue; + } + + if (entry.Id > expected && buffer.LastFetchedId > 0) + { + buffer.Dropped += entry.Id - expected; // 中间的几条在节点上已经被环挤掉了 + } + + expected = entry.Id + 1; + long at = entry.AtUtcMs < now - 86_400_000 || entry.AtUtcMs > now + 86_400_000 ? now : entry.AtUtcMs; + if (buffer.Items.Count == CapacityPerNode) + { + buffer.Items.Dequeue(); + buffer.Dropped++; + } + + buffer.Items.Enqueue(new NodeRuntimeLog( + entry.Id, + at, + entry.Level is "info" or "warn" or "error" ? entry.Level : "info", + Truncate(entry.Message, 384), + SafeFields(entry.Fields))); + buffer.LastFetchedId = entry.Id; + } + + if (page.Items.Count == 0 && page.LastId > buffer.LastFetchedId) + { + buffer.LastFetchedId = page.LastId; // 没有符合级别的新日志:下次从这里接着拉,不用重复扫 + } } } @@ -80,17 +169,11 @@ public sealed class NodeRuntimeLogService(TimeProvider time) foreach ((string key, string value) in fields.Take(20)) { - string safeKey = Truncate(key, 48); string lower = key.ToLowerInvariant(); - bool sensitive = lower.Contains("token", StringComparison.Ordinal) - || lower.Contains("secret", StringComparison.Ordinal) - || lower.Contains("password", StringComparison.Ordinal) - || lower.Contains("private", StringComparison.Ordinal) - || lower.Contains("signature", StringComparison.Ordinal) - || lower.Contains("authorization", StringComparison.Ordinal) - || lower.Contains("cookie", StringComparison.Ordinal) - || lower == "mac"; - safe[safeKey] = sensitive ? "[redacted]" : Truncate(value ?? "", 128); + bool sensitive = lower.Contains("token", StringComparison.Ordinal) || lower.Contains("secret", StringComparison.Ordinal) || lower.Contains("password", StringComparison.Ordinal) + || lower.Contains("private", StringComparison.Ordinal) || lower.Contains("signature", StringComparison.Ordinal) || lower.Contains("authorization", StringComparison.Ordinal) + || lower.Contains("cookie", StringComparison.Ordinal) || lower == "mac"; + safe[Truncate(key, 64)] = sensitive ? "[redacted]" : Truncate(value, 128); } return safe; @@ -99,9 +182,15 @@ public sealed class NodeRuntimeLogService(TimeProvider time) private sealed class LogBuffer { public object Gate { get; } = new(); - public Queue Items { get; } = []; - public long LastId; - public long DroppedByNode; - public long DroppedByBackend; + + public Queue Items { get; } = new(); + + public long LastFetchedId { get; set; } + + public long Dropped { get; set; } + + public long LastAutoPullMs { get; set; } + + public bool Pulling { get; set; } } } diff --git a/backend/src/MoShiro.Backend.Application/Ports/IAgentGateway.cs b/backend/src/MoShiro.Backend.Application/Ports/IAgentGateway.cs index 92b5a93..4496b33 100644 --- a/backend/src/MoShiro.Backend.Application/Ports/IAgentGateway.cs +++ b/backend/src/MoShiro.Backend.Application/Ports/IAgentGateway.cs @@ -25,6 +25,9 @@ public interface IAgentGateway ValueTask CancelTestAsync(NodeId agent, TestId test, CancellationToken ct); + /// 向节点拉取日志(编号大于 AfterId 的、不低于 MinLevel 的,最旧的在前)。节点离线时失败。 + ValueTask> FetchLogsAsync(NodeId node, LogsFetchParams request, CancellationToken ct); + /// 让监听方断开某个拨号方的会话。返回监听方是否真的有这个会话。 ValueTask> DisconnectDialerAsync(NodeId listener, NodeId dialer, CancellationToken ct); diff --git a/backend/src/MoShiro.Backend.Gateway/FleetGateway.cs b/backend/src/MoShiro.Backend.Gateway/FleetGateway.cs index 3b5574a..8b8564d 100644 --- a/backend/src/MoShiro.Backend.Gateway/FleetGateway.cs +++ b/backend/src/MoShiro.Backend.Gateway/FleetGateway.cs @@ -50,6 +50,15 @@ public sealed class FleetGateway : IAgentGateway await Connections.RequestAsync( agent, MoShiroMethods.TestCancel, new TestCancelParams { TestId = test }, MoShiroJsonContext.Default.TestCancelParams, MoShiroJsonContext.Default.TestRunResult, ct); + public async ValueTask> FetchLogsAsync(NodeId node, LogsFetchParams request, CancellationToken ct) + { + Result result = await Connections.RequestAsync( + node, MoShiroMethods.LogsFetch, request, MoShiroJsonContext.Default.LogsFetchParams, MoShiroJsonContext.Default.LogsFetchResult, ct); + return result.TryGet(out LogsFetchResult? page, out FleetError? error) + ? Result.Ok(page) + : Result.Fail(error.Code == FleetErrors.NodeOffline ? MoShiroErrors.AgentOffline : error.Code, error.Message); + } + public async ValueTask> DisconnectDialerAsync(NodeId listener, NodeId dialer, CancellationToken ct) { Result result = await Connections.RequestAsync( diff --git a/backend/tests/MoShiro.Backend.Tests/NodeRuntimeLogTests.cs b/backend/tests/MoShiro.Backend.Tests/NodeRuntimeLogTests.cs index 5389e32..36a4038 100644 --- a/backend/tests/MoShiro.Backend.Tests/NodeRuntimeLogTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/NodeRuntimeLogTests.cs @@ -1,27 +1,53 @@ -using Fleet.Core; -using MoShiro.Backend.Application; +using System.Text.Json.Nodes; using MoShiro.Protocol; +using MoShiro.TestSupport; namespace MoShiro.Backend.Tests; -public sealed class NodeRuntimeLogTests +public sealed class NodeRuntimeLogTests : IClassFixture { - [Fact] - public void DiagnosticFieldsSurviveUpload_WhileSecretsAndOversizedFieldsRemainBounded() + private readonly BackendFixture _backend; + + public NodeRuntimeLogTests(BackendFixture backend) { - var service = new NodeRuntimeLogService(TimeProvider.System); - var node = new NodeId("agt_diagnostics"); + _backend = backend; + } + + private static LogEntryDto Entry(long id, string level, string message, Dictionary? fields = null) => + new() { Id = id, Type = (uint)id, AtUtcMs = TimeProvider.System.GetUtcNow().ToUnixTimeMilliseconds(), Level = level, Message = message, Fields = fields ?? [] }; + + private async Task LogsAsync(AgentSimulator agent) => + await _backend.GetJsonAsync(_backend.Admin, $"/api/admin/nodes/{agent.NodeId.Value}/logs?limit=50"); + + [Fact] + public async Task Logs_ArePulledFromTheNodeOnDemand_OnlyTheNewLinesEachTime_WithSecretsAndOversizedFieldsBounded() + { + await using AgentSimulator agent = await _backend.ConnectAgentAsync(); 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); + agent.Logs.Add(Entry(1, "info", "link up")); + agent.Logs.Add(Entry(2, "warn", "handshake failed", fields)); - 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")); + JsonNode first = await LogsAsync(agent); + Assert.Equal(["handshake failed", "link up"], first["items"]!.AsArray().Select(i => i!["message"]!.GetValue())); // 新的在前 + JsonNode warn = first["items"]![0]!; + Assert.Equal(20, warn["fields"]!.AsObject().Count); + Assert.Equal("[redacted]", warn["fields"]!["api_token"]!.GetValue()); + Assert.Equal(128, warn["fields"]!["detail"]!.GetValue().Length); + Assert.Null(warn["fields"]!["overflow"]); + + agent.Logs.Add(Entry(3, "error", "transport shutdown")); + JsonNode second = await LogsAsync(agent); + Assert.Equal(3, second["items"]!.AsArray().Count); + Assert.Equal(2, agent.LogFetches.Last().AfterId); // 第二次只拉编号 2 之后的 + } + + [Fact] + public async Task Logs_OfAnOfflineNode_AreWhatTheBackendKept() + { + JsonNode logs = await _backend.GetJsonAsync(_backend.Admin, "/api/admin/nodes/agt_nobody/logs"); + Assert.Empty(logs["items"]!.AsArray()); } } diff --git a/backend/tests/MoShiro.TestSupport/AgentSimulator.cs b/backend/tests/MoShiro.TestSupport/AgentSimulator.cs index c41bfe4..3fbc8c8 100644 --- a/backend/tests/MoShiro.TestSupport/AgentSimulator.cs +++ b/backend/tests/MoShiro.TestSupport/AgentSimulator.cs @@ -17,6 +17,11 @@ namespace MoShiro.TestSupport; /// public sealed class AgentSimulator : IAsyncDisposable { + /// 节点上保留的日志(moshiro.logs.fetch 按编号返回它们)和收到的拉取请求。 + public System.Collections.Concurrent.ConcurrentBag Logs { get; } = []; + + public System.Collections.Concurrent.ConcurrentQueue LogFetches { get; } = new(); + private readonly NodeIdentity _identity; private readonly NodeIdentity _tlsKey; private readonly ClientWebSocket _socket = new(); @@ -446,6 +451,13 @@ public sealed class AgentSimulator : IAsyncDisposable await SendAsync(RpcMessage.Failure(message.Id, new FleetError(new ErrorCode(errorCode ?? "test_busy"), "refused"))); } + break; + case MoShiroMethods.LogsFetch: + LogsFetchParams fetch = message.Params!.Value.Deserialize(MoShiroJsonContext.Default.LogsFetchParams)!; + LogFetches.Enqueue(fetch); + var page = new LogsFetchResult { LastId = Logs.IsEmpty ? 0 : Logs.Max(l => l.Id) }; + page.Items.AddRange(Logs.Where(l => l.Id > fetch.AfterId).OrderBy(l => l.Id).Take(Math.Max(1, fetch.Limit))); + await SendAsync(RpcMessage.Success(message.Id!.Value, page, MoShiroJsonContext.Default.LogsFetchResult)); break; case MoShiroMethods.TestCancel: TestCancelParams cancel = message.Params!.Value.Deserialize(MoShiroJsonContext.Default.TestCancelParams)!; diff --git a/docs/design/protocol.md b/docs/design/protocol.md index 712f264..78ca075 100644 --- a/docs/design/protocol.md +++ b/docs/design/protocol.md @@ -17,7 +17,7 @@ MoShiro 的两条控制链路都建立在 Fleet 之上: | 更新 | 通知 `moshiro.update.offer` / `moshiro.update.apply`(B→A、B→P),心跳附加字段 `update` | `offer` 带 CI 签名的发布清单原文和签名,节点只写进数据目录;`apply` 请求更新;`update` 是宿主机更新器的状态(`state`、`current`、`available`、`error`、`at`),见 [deployment.md](../ops/deployment.md) | | 上报原始块 | 发件箱,二进制通道 16 | `RawChunk`(自描述,见 [measurement.md](measurement.md) 第 4 节),`msg_id = chunk_id` | | 拨号方 交付列块 | 点对点会话的发件箱,二进制通道 17 | 列块:`u32 stream_id \| u32 first_seq \| u32 count \| 列数据(Brotli)` | -| 节点自身状态 | `fleet.heartbeat` 的附加字段 | Agent:`sessions`、`dialers[]`、`pps_in/out`、`dropped_invalid`、`dropped_rate_limited`、`queue_bytes`、`update`;拨号方:只有 `update` | +| 节点自身状态 | `fleet.heartbeat` 的附加字段 | Agent:`sessions`、`dialers[]`、`pps_in/out`、`dropped_invalid`、`dropped_rate_limited`、`queue_bytes`、`update`、`log_summary`(自上次心跳以来各级别日志的条数、出现最多的几个类型 `type`(消息文本的稳定哈希)和最新一条的编号,不带文本);拨号方:只有 `update` | | 版本与能力 | `fleet.hello` | capabilities 包括 `moshiro.kinds:<列表>`、`moshiro.blocks:<列表>`;Agent 附加 `platform`(系统、架构、版本、tag、提交、构建时间)、`tls_pubkey`(面向 拨号方 的证书公钥,面板自己算指纹)、`addresses`(本机公网地址)、`listen`、`local_limits` 摘要;拨号方 附加 `platform` | ## 2. `moshiro.*` 方法 @@ -32,6 +32,7 @@ MoShiro 的两条控制链路都建立在 Fleet 之上: | `moshiro.test.run` | B→A | 请求 | `test_id`、`dialer_id`(在哪个 拨号方 上跑)、`plan`、`phases[]` → `accepted`,或错误 `test_busy`、`dialer_offline` | | `moshiro.dialer.disconnect` | B→A | 请求 | `dialer_id` → `accepted`(Agent 是否真的有这个会话);Agent 关闭该 拨号方 的会话 | | `moshiro.test.cancel` | B→A | 请求 | `test_id` | +| `moshiro.logs.fetch` | B→A | 请求 | `after_id`、`limit`(≤ 200)、`min_level` → `items[]`(`id`、`type`、`at_utc_ms`、`level`、`message`、`fields`)、`last_id`:按需拉取节点本地保留的日志(最近 2048 条),最旧的在前;`last_id` 比 `after_id` 小说明节点重启过。管理端打开日志页时拉,节点报告了错误时 Backend 自动拉一次警告以上的 | | `moshiro.test.status` | A→B | 通知 | `test_id`、`status`(`running`、`phase_done`、`succeeded`、`failed`、`aborted`)、`phase?`、`error_code?` | | `moshiro.event` | A→B | 通知 | `kind`(`dialer_connected`(`detail` 是 `dialer_id`,`dialer` 里带版本、远端地址和连接时间)、`dialer_disconnected`、`endpoint_changed`、`session_rejected`、`clock_step`、`gap`、`cpu_saturated`(节点进程 CPU 饱和,可能影响测量,由 拨号方 或 Agent 上报))、`at_utc`、`detail` | | `moshiro.host` | A、P→B | 通知 | `HostSnapshot`:主机资源占用的汇总数字(CPU、内存、根分区、网卡累计流量与网速、运行时长;没有型号、主机名、IP),每 `host_interval_s` 秒一次;`host_stats` 为 false 时不发,见 [enrollment.md](../security/enrollment.md) | diff --git a/web/shared/api/schema.ts b/web/shared/api/schema.ts index 681b289..9d15583 100644 --- a/web/shared/api/schema.ts +++ b/web/shared/api/schema.ts @@ -508,7 +508,7 @@ export interface paths { path?: never; cookie?: never; }; - /** Recent runtime logs of a node */ + /** Recent runtime logs of a node: pulled from the node when it is online, otherwise what the backend kept */ get: operations["node_runtime_logs"]; put?: never; post?: never;