diff --git a/backend/agent/src/MoShiro.Agent/AgentApp.cs b/backend/agent/src/MoShiro.Agent/AgentApp.cs index 3f531c3..f368007 100644 --- a/backend/agent/src/MoShiro.Agent/AgentApp.cs +++ b/backend/agent/src/MoShiro.Agent/AgentApp.cs @@ -24,6 +24,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 MonoClock _clock; private readonly AgentCounters _counters = new(); private readonly SessionRegistry _sessions = new(); @@ -37,6 +38,7 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable private readonly TcpAcceptor _tcp; private readonly PeerListener _peers; private readonly PeerHub _hub; + private readonly ChunkAssembler _chunks; private readonly TaskManager _tasks; private readonly TestRunner _tests; private readonly FleetLink _link; @@ -52,6 +54,7 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable _identity = identity; _time = time; _log = log; + _logBatchSource = log as INodeLogBatchSource; _clock = new MonoClock(time); _uplink = new BackendUplink(time); @@ -82,10 +85,10 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable MaxPacketsPerSecond = config.Limits.MaxTotalPps, MaxBulkMbps = config.Limits.MaxTotalMbps, }; - var chunks = new ChunkAssembler(_streams, _clock, _counters, _outbox, identity.NodeId, time, log); - var targets = new TargetRunners(chunks, identity.NodeId, _clock, time, log); + _chunks = new ChunkAssembler(_streams, _clock, _counters, _outbox, identity.NodeId, time, log); + var targets = new TargetRunners(_chunks, identity.NodeId, _clock, time, log); _tasks = new TaskManager(CreateGuard, _streams, env, targets, () => _datagrams.BoundPorts, () => _tcp.BoundPorts, _trust); - _hub = new PeerHub(_sessions, _clock, chunks, _tasks, _uplink, time, log); + _hub = new PeerHub(_sessions, _clock, _chunks, _tasks, _uplink, time, log); _tests = new TestRunner(_tasks, _sessions, _streams, env, _uplink, config.Limits.MaxConcurrentTests, transport => transport == "quic_datagram" ? _datagrams.BoundPorts : _tcp.BoundPorts, time, log); _tasks.OnLineTasksChanged = (probe, set) => _ = _hub.PushToAsync(probe, set).AsTask(); @@ -179,11 +182,12 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable _log.Info("agent listening", ("control", ControlPort), ("quic", string.Join(',', QuicPorts)), ("tcp", string.Join(',', TcpPorts))); try { - await Task.WhenAll(quic, tcp, peers, _link.RunAsync(linked.Token), ReportHostAsync(linked.Token)); + await Task.WhenAll(quic, tcp, peers, _link.RunAsync(linked.Token), ReportHostAsync(linked.Token), _chunks.RunTrafficReportingAsync(linked.Token)); } finally { await _tasks.StopAsync(); + _chunks.FlushTraffic(); await linked.CancelAsync(); } } @@ -351,6 +355,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); var extra = new AgentHeartbeatExtra { RuntimeStatus = _update?.Status().State == "applying" ? NodeRuntimeStatus.Updating : NodeRuntimeStatus.Running, @@ -363,6 +368,8 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable DroppedRateLimited = _counters.DroppedRateLimited, QueueBytes = _outbox.PendingBytes, Update = _update?.Status(), + RuntimeLogs = logBatch.Records.Count == 0 ? null : [.. logBatch.Records], + DroppedRuntimeLogs = logBatch.Dropped == 0 ? null : logBatch.Dropped, }; _lastIn = packetsIn; _lastOut = packetsOut; diff --git a/backend/agent/src/MoShiro.Agent/Chunks/ChunkAssembler.cs b/backend/agent/src/MoShiro.Agent/Chunks/ChunkAssembler.cs index 5753a42..a4eb2c6 100644 --- a/backend/agent/src/MoShiro.Agent/Chunks/ChunkAssembler.cs +++ b/backend/agent/src/MoShiro.Agent/Chunks/ChunkAssembler.cs @@ -18,6 +18,7 @@ public sealed class ChunkAssembler private readonly NodeId _agent; private readonly TimeProvider _time; private readonly INodeLog _log; + private readonly TaskTrafficReporter _traffic; public ChunkAssembler(StreamRegistry streams, MonoClock clock, AgentCounters counters, Outbox outbox, NodeId agent, TimeProvider time, INodeLog log) { @@ -28,6 +29,7 @@ public sealed class ChunkAssembler _agent = agent; _time = time; _log = log; + _traffic = new TaskTrafficReporter(log, time); } /// @@ -68,9 +70,20 @@ public sealed class ChunkAssembler return true; } + public Task RunTrafficReportingAsync(CancellationToken ct) => _traffic.RunAsync(ct); + + public void FlushTraffic() => _traffic.Flush(); + /// Agent 自己采集的块(目标任务)直接发出。 public void Emit(RawChunk chunk) { - _outbox.Enqueue(MoShiroChannels.RawChunk, chunk.Header.ChunkId.Value, RawChunkCodec.Encode(chunk)); + byte[] payload = RawChunkCodec.Encode(chunk, out int uncompressedBytes); + _outbox.Enqueue(MoShiroChannels.RawChunk, chunk.Header.ChunkId.Value, payload); + _traffic.Record( + chunk.Header.TaskId?.Value ?? chunk.Header.TestId?.Value ?? "unknown", + chunk.Header.Kind, + chunk.Streams.Sum(stream => (long)stream.Count), + uncompressedBytes, + payload.Length); } } diff --git a/backend/agent/src/MoShiro.Agent/Chunks/TaskTrafficReporter.cs b/backend/agent/src/MoShiro.Agent/Chunks/TaskTrafficReporter.cs new file mode 100644 index 0000000..042ae3a --- /dev/null +++ b/backend/agent/src/MoShiro.Agent/Chunks/TaskTrafficReporter.cs @@ -0,0 +1,107 @@ +using Fleet.Core; +using Fleet.Rpc; + +namespace MoShiro.Agent; + +/// Aggregates per-task Agent → Backend chunk sizes into low-volume diagnostic logs. +internal sealed class TaskTrafficReporter +{ + private static readonly TimeSpan s_window = TimeSpan.FromMinutes(1); + + private readonly object _gate = new(); + private readonly INodeLog _log; + private readonly TimeProvider _time; + private readonly Dictionary<(string Task, string Kind), Totals> _totals = []; + private long _windowStarted; + + public TaskTrafficReporter(INodeLog log, TimeProvider time) + { + _log = log; + _time = time; + _windowStarted = time.GetTimestamp(); + } + + public void Record(string task, string kind, long records, int uncompressedBytes, int compressedBytes) + { + lock (_gate) + { + var key = (task, kind); + if (!_totals.TryGetValue(key, out Totals? totals)) + { + totals = new Totals(); + _totals.Add(key, totals); + } + + totals.Chunks++; + totals.Records += records; + totals.UncompressedBytes += uncompressedBytes; + totals.CompressedBytes += compressedBytes; + } + } + + public async Task RunAsync(CancellationToken ct) + { + using var timer = new PeriodicTimer(s_window, _time); + try + { + while (await timer.WaitForNextTickAsync(ct)) + { + Flush(); + } + } + catch (OperationCanceledException) + { + // The final snapshot is flushed by the Agent shutdown path. + } + } + + public void Flush() + { + List completed; + long windowSeconds; + lock (_gate) + { + if (_totals.Count == 0) + { + _windowStarted = _time.GetTimestamp(); + return; + } + + TimeSpan elapsed = _time.GetElapsedTime(_windowStarted); + windowSeconds = Math.Max(1, (long)elapsed.TotalSeconds); + completed = [.. _totals.Select(pair => new Snapshot(pair.Key.Task, pair.Key.Kind, pair.Value.Chunks, pair.Value.Records, pair.Value.UncompressedBytes, pair.Value.CompressedBytes))]; + _totals.Clear(); + _windowStarted = _time.GetTimestamp(); + } + + foreach (Snapshot item in completed) + { + double savedPercent = item.UncompressedBytes == 0 + ? 0 + : 100d * (item.UncompressedBytes - item.CompressedBytes) / item.UncompressedBytes; + long queuedBytes = item.CompressedBytes + (long)item.Chunks * OutboxFrame.HeaderBytes; + _log.Info( + "task upload traffic", + ("link", "agent-backend"), + ("task", item.Task), + ("kind", item.Kind), + ("chunks", item.Chunks), + ("records", item.Records), + ("raw_bytes", item.UncompressedBytes), + ("compressed_bytes", item.CompressedBytes), + ("queued_bytes", queuedBytes), + ("saved_pct", Math.Round(savedPercent, 1)), + ("window_s", windowSeconds)); + } + } + + private sealed class Totals + { + public long Chunks; + public long Records; + public long UncompressedBytes; + public long CompressedBytes; + } + + private sealed record Snapshot(string Task, string Kind, long Chunks, long Records, long UncompressedBytes, long CompressedBytes); +} diff --git a/backend/agent/src/MoShiro.Agent/Program.cs b/backend/agent/src/MoShiro.Agent/Program.cs index b2072ec..bb3ac4b 100644 --- a/backend/agent/src/MoShiro.Agent/Program.cs +++ b/backend/agent/src/MoShiro.Agent/Program.cs @@ -3,7 +3,7 @@ using Fleet.Hosting; using MoShiro.Agent; TimeProvider time = TimeProvider.System; -INodeLog log = new ConsoleNodeLog(time); +INodeLog log = new BufferedNodeLog(new ConsoleNodeLog(time), time); if (PlatformCheck.VerifyCrypto() is { } problem) { diff --git a/backend/agent/src/MoShiro.Agent/Tasks/TaskManager.cs b/backend/agent/src/MoShiro.Agent/Tasks/TaskManager.cs index 8253825..308d576 100644 --- a/backend/agent/src/MoShiro.Agent/Tasks/TaskManager.cs +++ b/backend/agent/src/MoShiro.Agent/Tasks/TaskManager.cs @@ -161,6 +161,7 @@ public sealed class TaskManager foreach ((ProbeSession probe, Dictionary current) in _sessionLines) { var next = new Dictionary(); + bool lineTasksChanged = current.Count != line.Count; foreach ((string id, (TaskSpec task, TaskDecision decision)) in line) { LineEntry candidate = BuildLineEntry(task, decision, probe); @@ -170,6 +171,7 @@ public sealed class TaskManager } else { + lineTasksChanged = true; Start(candidate); next[id] = candidate; } @@ -189,7 +191,13 @@ public sealed class TaskManager current[id] = entry; } - changes.Add((probe, SetFor(current))); + // The Agent receives a full task snapshot for peer/trust changes and target-only tasks too. + // Those changes do not affect a Probe's line blocks, so avoid retransmitting the same full + // line_task_set to every Probe. New sessions still get their initial snapshot in OpenSession. + if (lineTasksChanged) + { + changes.Add((probe, SetFor(current))); + } } } diff --git a/backend/fleet/src/Fleet.Core/NodeSupport.cs b/backend/fleet/src/Fleet.Core/NodeSupport.cs index e5a1ab5..273494c 100644 --- a/backend/fleet/src/Fleet.Core/NodeSupport.cs +++ b/backend/fleet/src/Fleet.Core/NodeSupport.cs @@ -15,6 +15,17 @@ 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. +public sealed record NodeLogRecord(long AtUtcMs, string Level, string Message, Dictionary Fields); + +public sealed record NodeLogBatch(IReadOnlyList Records, long Dropped); + +/// Optional bounded log source used by node heartbeats to upload recent local logs. +public interface INodeLogBatchSource +{ + NodeLogBatch TakeBatch(int maxRecords); +} + public sealed class NullNodeLog : INodeLog { public static NullNodeLog Instance { get; } = new(); diff --git a/backend/fleet/src/Fleet.Hosting/BufferedNodeLog.cs b/backend/fleet/src/Fleet.Hosting/BufferedNodeLog.cs new file mode 100644 index 0000000..1d29ad5 --- /dev/null +++ b/backend/fleet/src/Fleet.Hosting/BufferedNodeLog.cs @@ -0,0 +1,96 @@ +using System.Globalization; +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 +{ + private const int Capacity = 256; + private const int MaxFields = 12; + private const int MaxMessageLength = 512; + private const int MaxFieldValueLength = 256; + private const int MaxBatchBytes = 8 << 10; + + private readonly object _gate = new(); + private readonly Queue _pending = []; + private readonly INodeLog _local; + private readonly TimeProvider _time; + private long _dropped; + + public BufferedNodeLog(INodeLog local, TimeProvider time) + { + _local = local; + _time = time; + } + + public void Write(NodeLogLevel level, string message, params (string Key, object? Value)[] fields) + { + _local.Write(level, message, fields); + + var safeFields = new Dictionary(Math.Min(fields.Length, MaxFields), StringComparer.Ordinal); + foreach ((string key, object? value) in fields.Take(MaxFields)) + { + string safeValue = IsSensitiveKey(key) ? "[redacted]" : Convert.ToString(value, CultureInfo.InvariantCulture) ?? ""; + safeFields[key.Length <= 64 ? key : key[..64]] = Truncate(safeValue, MaxFieldValueLength); + } + + var record = new NodeLogRecord( + _time.GetUtcNow().ToUnixTimeMilliseconds(), + level.ToString().ToLowerInvariant(), + Truncate(message, MaxMessageLength), + safeFields); + + lock (_gate) + { + if (_pending.Count == Capacity) + { + _pending.Dequeue(); + _dropped++; + } + + _pending.Enqueue(record); + } + } + + public NodeLogBatch TakeBatch(int maxRecords) + { + int count = Math.Clamp(maxRecords, 0, 32); + lock (_gate) + { + var records = new List(Math.Min(count, _pending.Count)); + int bytes = 0; + while (records.Count < count && _pending.TryPeek(out NodeLogRecord? record)) + { + int recordBytes = 64 + record.Message.Length + record.Fields.Sum(pair => pair.Key.Length + pair.Value.Length + 8); + if (records.Count > 0 && bytes + recordBytes > MaxBatchBytes) + { + break; + } + + _pending.Dequeue(); + records.Add(record); + bytes += recordBytes; + } + + long dropped = _dropped; + _dropped = 0; + return new NodeLogBatch(records, dropped); + } + } + + private static bool IsSensitiveKey(string key) + { + string lower = key.ToLowerInvariant(); + return 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"; + } + + private static string Truncate(string value, int maxLength) => value.Length <= maxLength ? value : value[..maxLength]; +} diff --git a/backend/openapi/v1.json b/backend/openapi/v1.json index 1de1968..e4ea1db 100644 --- a/backend/openapi/v1.json +++ b/backend/openapi/v1.json @@ -1892,6 +1892,70 @@ "x-moshiro-capability": "fleet.revoke" } }, + "/api/admin/nodes/{id}/logs": { + "get": { + "tags": [ + "admin-nodes" + ], + "summary": "Recent Agent or Probe runtime logs", + "operationId": "node_runtime_logs", + "parameters": [ + { + "name": "id", + "in": "path", + "required": true, + "schema": { + "type": "string" + } + }, + { + "name": "limit", + "in": "query", + "schema": { + "pattern": "^-?(?:0|[1-9]\\d*)$", + "type": [ + "integer", + "string" + ], + "format": "int32" + } + } + ], + "responses": { + "200": { + "description": "OK", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/NodeRuntimeLogsDto" + } + } + } + }, + "401": { + "description": "Unauthorized", + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/ProblemDto" + } + } + } + }, + "403": { + "description": "Forbidden", + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/ProblemDto" + } + } + } + } + }, + "x-moshiro-capability": "fleet.read" + } + }, "/api/admin/agents/{id}/approve": { "post": { "tags": [ @@ -7606,6 +7670,64 @@ } } }, + "NodeRuntimeLogDto": { + "required": [ + "id", + "at", + "level", + "message", + "fields" + ], + "type": "object", + "properties": { + "id": { + "pattern": "^-?(?:0|[1-9]\\d*)$", + "type": [ + "integer", + "string" + ], + "format": "int64" + }, + "at": { + "format": "date-time" + }, + "level": { + "type": "string" + }, + "message": { + "type": "string" + }, + "fields": { + "type": "object", + "additionalProperties": { + "type": "string" + } + } + } + }, + "NodeRuntimeLogsDto": { + "required": [ + "items", + "dropped" + ], + "type": "object", + "properties": { + "items": { + "type": "array", + "items": { + "$ref": "#/components/schemas/NodeRuntimeLogDto" + } + }, + "dropped": { + "pattern": "^-?(?:0|[1-9]\\d*)$", + "type": [ + "integer", + "string" + ], + "format": "int64" + } + } + }, "NodeRuntimeStatus": { "enum": [ "initializing", @@ -9963,6 +10085,9 @@ { "name": "admin-agents" }, + { + "name": "admin-nodes" + }, { "name": "admin-join-tokens" }, diff --git a/backend/probe/src/MoShiro.Probe/ProbeApp.cs b/backend/probe/src/MoShiro.Probe/ProbeApp.cs index 19a4a5b..72728fb 100644 --- a/backend/probe/src/MoShiro.Probe/ProbeApp.cs +++ b/backend/probe/src/MoShiro.Probe/ProbeApp.cs @@ -23,6 +23,7 @@ public sealed class ProbeApp : IDisposable private readonly ConcurrentDictionary _pendingRetries = new(StringComparer.Ordinal); private readonly SemaphoreSlim _changed = new(0); private readonly INodeLog _log; + private readonly INodeLogBatchSource? _logBatchSource; private readonly FleetLink _link; private readonly ReportingSection _reporting; private readonly TimeProvider _time; @@ -31,6 +32,7 @@ public sealed class ProbeApp : IDisposable public ProbeApp(ProbeFile file, LoadedNodeIdentity identity, string configDirectory, TimeProvider time, INodeLog log, RestartController? restart = null) { _log = log; + _logBatchSource = log as INodeLogBatchSource; _time = time; _reporting = file.Reporting; var trust = TrustAnchor.Load(file.Fleet.IdentityDir, log); @@ -52,7 +54,16 @@ public sealed class ProbeApp : IDisposable { UpdateStatus? updateStatus = update?.Status(); NodeRuntimeStatus runtimeStatus = updateStatus?.State == "applying" ? NodeRuntimeStatus.Updating : _env.Directory.Agents.Count == 0 ? NodeRuntimeStatus.ProbePaused : NodeRuntimeStatus.Running; - return JsonSerializer.SerializeToElement(new ProbeHeartbeatExtra { RuntimeStatus = runtimeStatus, Update = updateStatus, Ipv6 = Ipv6Diagnostics.Sample(null, Environment.GetEnvironmentVariable("MOSHIRO_NETWORK_SCOPE"), Environment.GetEnvironmentVariable("HOST_PROC")), QuicLinks = SnapshotQuicLinks() }, MoShiroJsonContext.Default.ProbeHeartbeatExtra); + NodeLogBatch logBatch = _logBatchSource?.TakeBatch(16) ?? new NodeLogBatch([], 0); + return JsonSerializer.SerializeToElement(new ProbeHeartbeatExtra + { + RuntimeStatus = runtimeStatus, + Update = updateStatus, + Ipv6 = Ipv6Diagnostics.Sample(null, Environment.GetEnvironmentVariable("MOSHIRO_NETWORK_SCOPE"), Environment.GetEnvironmentVariable("HOST_PROC")), + QuicLinks = SnapshotQuicLinks(), + RuntimeLogs = logBatch.Records.Count == 0 ? null : [.. logBatch.Records], + DroppedRuntimeLogs = logBatch.Dropped == 0 ? null : logBatch.Dropped, + }, MoShiroJsonContext.Default.ProbeHeartbeatExtra); }, HelloMeta = () => JsonSerializer.SerializeToElement(new ProbeHelloMeta { Owners = [.. trust.Controllers], Platform = PlatformDetector.Detect(), Ipv6 = Ipv6Diagnostics.Sample(null, Environment.GetEnvironmentVariable("MOSHIRO_NETWORK_SCOPE"), Environment.GetEnvironmentVariable("HOST_PROC")) }, MoShiroJsonContext.Default.ProbeHelloMeta), }; diff --git a/backend/probe/src/MoShiro.Probe/Program.cs b/backend/probe/src/MoShiro.Probe/Program.cs index 85a0464..ac255a2 100644 --- a/backend/probe/src/MoShiro.Probe/Program.cs +++ b/backend/probe/src/MoShiro.Probe/Program.cs @@ -3,7 +3,7 @@ using Fleet.Hosting; using MoShiro.Probe; TimeProvider time = TimeProvider.System; -INodeLog log = new ConsoleNodeLog(time); +INodeLog log = new BufferedNodeLog(new ConsoleNodeLog(time), time); if (PlatformCheck.VerifyCrypto() is { } problem) { diff --git a/backend/shared/MoShiro.Protocol/Raw/RawChunkCodec.cs b/backend/shared/MoShiro.Protocol/Raw/RawChunkCodec.cs index aaaca96..f869dc7 100644 --- a/backend/shared/MoShiro.Protocol/Raw/RawChunkCodec.cs +++ b/backend/shared/MoShiro.Protocol/Raw/RawChunkCodec.cs @@ -59,8 +59,13 @@ public static class RawChunkCodec } public static byte[] Encode(RawChunk chunk) + => Encode(chunk, out _); + + /// Encode a chunk and return the pre-compression payload size for traffic diagnostics. + public static byte[] Encode(RawChunk chunk, out int uncompressedBytes) { byte[] plain = EncodeBody(chunk); + uncompressedBytes = plain.Length; using var compressed = new MemoryStream(); using (var brotli = new BrotliStream(compressed, CompressionLevel.Optimal, leaveOpen: true)) { diff --git a/backend/shared/MoShiro.Protocol/Rpc/MoShiroRpc.cs b/backend/shared/MoShiro.Protocol/Rpc/MoShiroRpc.cs index b13964b..f9daf60 100644 --- a/backend/shared/MoShiro.Protocol/Rpc/MoShiroRpc.cs +++ b/backend/shared/MoShiro.Protocol/Rpc/MoShiroRpc.cs @@ -322,6 +322,12 @@ public sealed class ProbeHeartbeatExtra [JsonPropertyName("update")] public UpdateStatus? Update { get; set; } + + [JsonPropertyName("runtime_logs")] + public List? RuntimeLogs { get; set; } + + [JsonPropertyName("dropped_runtime_logs")] + public long? DroppedRuntimeLogs { get; set; } } /// Probe's rolling QUIC connection health snapshot for one Agent/IP path. @@ -546,6 +552,12 @@ public sealed class AgentHeartbeatExtra [JsonPropertyName("update")] public UpdateStatus? Update { get; set; } + + [JsonPropertyName("runtime_logs")] + public List? RuntimeLogs { get; set; } + + [JsonPropertyName("dropped_runtime_logs")] + public long? DroppedRuntimeLogs { get; set; } } /// Agent 下发给 Probe 的一条线路任务(期望状态类型 line_task_set 里的一项)。 diff --git a/backend/src/MoShiro.Backend.Api/Admin/AgentEndpoints.cs b/backend/src/MoShiro.Backend.Api/Admin/AgentEndpoints.cs index f618c85..2cefeac 100644 --- a/backend/src/MoShiro.Backend.Api/Admin/AgentEndpoints.cs +++ b/backend/src/MoShiro.Backend.Api/Admin/AgentEndpoints.cs @@ -23,6 +23,9 @@ public static class AgentEndpoints group.MapGet("/agents/{id}", GetAsync) .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 Agent or Probe runtime logs", "admin-nodes").Returns().ProducesProblems(401, 403) + .Requires(Capability.FleetRead); group.MapPatch("/agents/{id}", PatchAsync) .Accepts("application/merge-patch+json", "application/json") .Api("agents_update", "Update name, labels, public host or visibility (merge-patch, If-Match supported)", AgentsTag) @@ -84,6 +87,14 @@ 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); + + 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); + private static async Task GetAsync(HttpContext http, AgentService agents, string id, CancellationToken ct) { AgentView? view = await agents.GetAsync(new NodeId(id), ct); diff --git a/backend/src/MoShiro.Backend.Api/Contracts/AdminDtos.cs b/backend/src/MoShiro.Backend.Api/Contracts/AdminDtos.cs index 2c136be..6b7556b 100644 --- a/backend/src/MoShiro.Backend.Api/Contracts/AdminDtos.cs +++ b/backend/src/MoShiro.Backend.Api/Contracts/AdminDtos.cs @@ -32,6 +32,10 @@ public sealed record RejectedDto(string Id, string Reason); public sealed record HeartbeatDto(int Sessions, long PpsIn, long PpsOut, long DroppedInvalid, long DroppedRateLimited, long QueueBytes, DateTimeOffset At); +public sealed record NodeRuntimeLogDto(long Id, DateTimeOffset At, string Level, string Message, IReadOnlyDictionary Fields); + +public sealed record NodeRuntimeLogsDto(IReadOnlyList Items, long Dropped); + public sealed record AgentDto( string Id, string Name, diff --git a/backend/src/MoShiro.Backend.Api/Contracts/ApiJsonContext.cs b/backend/src/MoShiro.Backend.Api/Contracts/ApiJsonContext.cs index ee4b20f..1bdd0ed 100644 --- a/backend/src/MoShiro.Backend.Api/Contracts/ApiJsonContext.cs +++ b/backend/src/MoShiro.Backend.Api/Contracts/ApiJsonContext.cs @@ -36,6 +36,7 @@ namespace MoShiro.Backend.Api; [JsonSerializable(typeof(LoginResponse))] [JsonSerializable(typeof(MeDto))] [JsonSerializable(typeof(AgentDto))] +[JsonSerializable(typeof(NodeRuntimeLogsDto))] [JsonSerializable(typeof(PageDto))] [JsonSerializable(typeof(PageDto))] [JsonSerializable(typeof(AgentPatchRequest))] diff --git a/backend/src/MoShiro.Backend.Application/Agents/MoShiroFleetHooks.cs b/backend/src/MoShiro.Backend.Application/Agents/MoShiroFleetHooks.cs index d4db58c..8a7f80d 100644 --- a/backend/src/MoShiro.Backend.Application/Agents/MoShiroFleetHooks.cs +++ b/backend/src/MoShiro.Backend.Application/Agents/MoShiroFleetHooks.cs @@ -72,9 +72,11 @@ public sealed class MoShiroNodeEventSink : INodeEventSink private readonly NodeOwnerService _owners; private readonly ProbeQuicRuntime _quicRuntime; private readonly ProbeRuntime _probeRuntime; + private readonly NodeRuntimeLogService _nodeLogs; - public MoShiroNodeEventSink(IAgentStore agents, AgentRuntime runtime, IEventStore events, LiveHub live, AgentService service, NodeInfoService nodes, ReleaseService releases, ScheduleService schedule, IProbeStore probes, NodeOwnerService owners, ProbeQuicRuntime quicRuntime, ProbeRuntime probeRuntime) + public MoShiroNodeEventSink(IAgentStore agents, AgentRuntime runtime, IEventStore events, LiveHub live, AgentService service, NodeInfoService nodes, ReleaseService releases, ScheduleService schedule, IProbeStore probes, NodeOwnerService owners, ProbeQuicRuntime quicRuntime, ProbeRuntime probeRuntime, NodeRuntimeLogService nodeLogs) { + _nodeLogs = nodeLogs; _owners = owners; _quicRuntime = quicRuntime; _probeRuntime = probeRuntime; @@ -125,6 +127,7 @@ public sealed class MoShiroNodeEventSink : INodeEventSink _releases.RecordStatus(id, beat?.Update); if (beat is not null) { + _nodeLogs.Append(id, beat.RuntimeLogs, beat.DroppedRuntimeLogs ?? 0); state.ReplaceProbes(beat.Probes.Select(p => new ConnectedProbeInfo(p.Id.Value, p.Version, p.Remote, p.SinceUtcMs, p.IpVersion))); foreach (ConnectedProbe p in beat.Probes) { @@ -185,6 +188,7 @@ public sealed class MoShiroNodeEventSink : INodeEventSink _releases.RecordStatus(id, heartbeat?.Update); if (heartbeat is not null) { + _nodeLogs.Append(id, heartbeat.RuntimeLogs, heartbeat.DroppedRuntimeLogs ?? 0); _probeRuntime.RecordHeartbeat(id.Value, nodeEvent.AtMs, heartbeat); } if (heartbeat?.QuicLinks is { } quicLinks) diff --git a/backend/src/MoShiro.Backend.Application/Agents/NodeRuntimeLogService.cs b/backend/src/MoShiro.Backend.Application/Agents/NodeRuntimeLogService.cs new file mode 100644 index 0000000..e537ed3 --- /dev/null +++ b/backend/src/MoShiro.Backend.Application/Agents/NodeRuntimeLogService.cs @@ -0,0 +1,107 @@ +using System.Collections.Concurrent; +using Fleet.Core; +using MoShiro.Protocol; + +namespace MoShiro.Backend.Application; + +public sealed record NodeRuntimeLog(long Id, long AtUtcMs, string Level, string Message, IReadOnlyDictionary Fields); + +public sealed record NodeRuntimeLogSnapshot(IReadOnlyList Items, long Dropped); + +/// Bounded in-memory recent logs uploaded by authenticated Agent and Probe heartbeats. +public sealed class NodeRuntimeLogService(TimeProvider time) +{ + private const int CapacityPerNode = 256; + private const int MaxBatch = 32; + private readonly ConcurrentDictionary _nodes = new(); + + public void Append(NodeId node, IReadOnlyList? records, long droppedByNode) + { + if ((records is null || records.Count == 0) && droppedByNode <= 0) + { + return; + } + + LogBuffer buffer = _nodes.GetOrAdd(node, static _ => new LogBuffer()); + lock (buffer.Gate) + { + buffer.DroppedByNode += Math.Max(0, droppedByNode); + if (records is null) + { + return; + } + + foreach (NodeLogRecord record in records.Take(MaxBatch)) + { + 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)); + + if (buffer.Items.Count == CapacityPerNode) + { + buffer.Items.Dequeue(); + buffer.DroppedByBackend++; + } + + buffer.Items.Enqueue(entry); + } + } + } + + public NodeRuntimeLogSnapshot List(NodeId node, int limit) + { + if (!_nodes.TryGetValue(node, out LogBuffer? buffer)) + { + return new NodeRuntimeLogSnapshot([], 0); + } + + lock (buffer.Gate) + { + return new NodeRuntimeLogSnapshot( + [.. buffer.Items.Reverse().Take(Math.Clamp(limit, 1, 200))], + buffer.DroppedByNode + buffer.DroppedByBackend); + } + } + + private static string Truncate(string value, int maxLength) => value.Length <= maxLength ? value : value[..maxLength]; + + private static Dictionary SafeFields(IReadOnlyDictionary? fields) + { + var safe = new Dictionary(StringComparer.Ordinal); + if (fields is null) + { + return safe; + } + + foreach ((string key, string value) in fields.Take(8)) + { + 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); + } + + return safe; + } + + private sealed class LogBuffer + { + public object Gate { get; } = new(); + public Queue Items { get; } = []; + public long LastId; + public long DroppedByNode; + public long DroppedByBackend; + } +} diff --git a/backend/src/MoShiro.Backend.Application/ApplicationServiceCollectionExtensions.cs b/backend/src/MoShiro.Backend.Application/ApplicationServiceCollectionExtensions.cs index dc5f090..c42f29a 100644 --- a/backend/src/MoShiro.Backend.Application/ApplicationServiceCollectionExtensions.cs +++ b/backend/src/MoShiro.Backend.Application/ApplicationServiceCollectionExtensions.cs @@ -17,6 +17,7 @@ public static class ApplicationServiceCollectionExtensions services.AddSingleton(); services.AddSingleton(); + services.AddSingleton(); services.AddSingleton(); services.AddSingleton(); services.AddSingleton(); diff --git a/web/admin/src/RuntimeLogs.tsx b/web/admin/src/RuntimeLogs.tsx new file mode 100644 index 0000000..053bf38 --- /dev/null +++ b/web/admin/src/RuntimeLogs.tsx @@ -0,0 +1,46 @@ +import type { NodeRuntimeLogsDto } from '@shared/api/types' +import { fmtTime } from '@shared/format' +import { useQuery } from '@shared/hooks' +import { Chip, Empty, ErrorNotice, Spinner } from '@shared/ui' +import { api } from '@shared/api/client' + +export function RuntimeLogs({ nodeId }: { nodeId: string }) { + const logs = useQuery( + () => api.get(`/api/admin/nodes/${encodeURIComponent(nodeId)}/logs`, { limit: 100 }), + [nodeId], + 5000, + ) + + return ( +
+
+ 运行日志 + 最近 {logs.data?.items.length ?? 0} 条 +
+ + {logs.data?.dropped ?

节点或后端缓冲区累计丢弃 {logs.data.dropped} 条较早日志。

: null} + {logs.loading && !logs.data ? : null} + {!logs.loading && (logs.data?.items.length ?? 0) === 0 ? 还没有收到运行日志 : null} + {(logs.data?.items.length ?? 0) > 0 ? ( +
+ {logs.data?.items.map((entry) => ( +
+ + {entry.level} +
+
{entry.message}
+ {Object.keys(entry.fields).length ? ( +
+ {Object.entries(entry.fields).map(([key, value]) => ( + {key}={value} + ))} +
+ ) : null} +
+
+ ))} +
+ ) : null} +
+ ) +} diff --git a/web/admin/src/admin.css b/web/admin/src/admin.css index c910dcb..a7282b7 100644 --- a/web/admin/src/admin.css +++ b/web/admin/src/admin.css @@ -17,6 +17,54 @@ overflow: auto; } +.runtime-logs { + display: grid; + gap: 1px; + max-height: 520px; + overflow: auto; + border-radius: var(--r-card); + background: var(--app-outline-soft); +} + +.runtime-log { + display: grid; + grid-template-columns: 150px auto minmax(0, 1fr); + align-items: start; + gap: 10px; + padding: 9px 10px; + background: var(--app-card); + font-size: 13px; +} + +.runtime-log-message { + min-width: 0; + overflow-wrap: anywhere; +} + +.runtime-log-fields { + display: flex; + flex-wrap: wrap; + gap: 4px 10px; + margin-top: 4px; + color: var(--app-ink-muted); + font-size: 11px; +} + +.runtime-log-fields b { + color: var(--app-ink-faint); + font-weight: 500; +} + +@media (max-width: 700px) { + .runtime-log { + grid-template-columns: 1fr auto; + } + + .runtime-log-message { + grid-column: 1 / -1; + } +} + .side .brand { padding: 8px 14px 16px; font-size: 18px; diff --git a/web/admin/src/pages/AgentDetail.tsx b/web/admin/src/pages/AgentDetail.tsx index b5ca975..a1e539b 100644 --- a/web/admin/src/pages/AgentDetail.tsx +++ b/web/admin/src/pages/AgentDetail.tsx @@ -8,6 +8,7 @@ import { useQuery } from '@shared/hooks' import { Chip, Confirm, Dialog, Empty, ErrorNotice, Field, Spinner, useAction } from '@shared/ui' import { AgentStatus } from './Agents' import { IPV6_STATES } from '@shared/ipVersions' +import { RuntimeLogs } from '../RuntimeLogs' function EditDialog({ agent, onClose, onSaved }: { agent: AgentDto; onClose: () => void; onSaved: () => void }) { const { busy, run } = useAction() @@ -358,6 +359,7 @@ export function AgentDetail() { )} + ) : null} diff --git a/web/admin/src/pages/ProbeDetail.tsx b/web/admin/src/pages/ProbeDetail.tsx index 44761df..fe09e72 100644 --- a/web/admin/src/pages/ProbeDetail.tsx +++ b/web/admin/src/pages/ProbeDetail.tsx @@ -7,6 +7,7 @@ import { useQuery } from '@shared/hooks' import { NodeTag, RuntimeStatusChip, useNodeInfos } from '@shared/nodes' import { IPV6_STATES } from '@shared/ipVersions' import { Chip, Dot, Empty, ErrorNotice, Spinner } from '@shared/ui' +import { RuntimeLogs } from '../RuntimeLogs' export function ProbeDetail() { const { id = '' } = useParams() @@ -111,6 +112,7 @@ export function ProbeDetail() { )} + ) : null} diff --git a/web/display/src/pages/Host.tsx b/web/display/src/pages/Host.tsx index 74a8d95..a5311f3 100644 --- a/web/display/src/pages/Host.tsx +++ b/web/display/src/pages/Host.tsx @@ -50,9 +50,13 @@ export function Host() {

{node?.name ?? id}

- {info?.country ? : null} - {info?.isp_key ? : null} - {info?.os_name ? {[info.os_name, info.arch].filter(Boolean).join(' / ')} : null} + {info?.country || info?.isp_key || info?.os_name ? ( + + {info.country ? : null} + {info.isp_key ? : null} + {info.os_name ? {[info.os_name, info.arch].filter(Boolean).join(' / ')} : null} + + ) : null} {node ? <>{node.kind === 'agent' ? 'Agent' : 'Probe'}{node.online ? (node.stale ? '状态数据暂未更新' : '在线') : '离线'} : 节点信息加载中…} {node?.last_seen_at && !node.online ? 最后在线 {fmtAgo(node.last_seen_at)} : null} diff --git a/web/shared/api/schema.ts b/web/shared/api/schema.ts index a560345..d35f243 100644 --- a/web/shared/api/schema.ts +++ b/web/shared/api/schema.ts @@ -501,6 +501,23 @@ export interface paths { patch: operations["agents_update"]; trace?: never; }; + "/api/admin/nodes/{id}/logs": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + /** Recent Agent or Probe runtime logs */ + get: operations["node_runtime_logs"]; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/api/admin/agents/{id}/approve": { parameters: { query?: never; @@ -1830,6 +1847,22 @@ export interface components { passkey: null | string; account: null | string; }; + NodeRuntimeLogDto: { + /** Format: int64 */ + id: number | string; + /** Format: date-time */ + at: unknown; + level: string; + message: string; + fields: { + [key: string]: string; + }; + }; + NodeRuntimeLogsDto: { + items: components["schemas"]["NodeRuntimeLogDto"][]; + /** Format: int64 */ + dropped: number | string; + }; /** @enum {unknown} */ NodeRuntimeStatus: "initializing" | "running" | "probe_paused" | "maintenance" | "stopping" | "offline" | "updating"; NodeUpdateDto: { @@ -3843,6 +3876,48 @@ export interface operations { }; }; }; + node_runtime_logs: { + parameters: { + query?: { + limit?: number | string; + }; + header?: never; + path: { + id: string; + }; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description OK */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["NodeRuntimeLogsDto"]; + }; + }; + /** @description Unauthorized */ + 401: { + headers: { + [name: string]: unknown; + }; + content: { + "application/problem+json": components["schemas"]["ProblemDto"]; + }; + }; + /** @description Forbidden */ + 403: { + headers: { + [name: string]: unknown; + }; + content: { + "application/problem+json": components["schemas"]["ProblemDto"]; + }; + }; + }; + }; agents_approve: { parameters: { query?: never; diff --git a/web/shared/api/types.ts b/web/shared/api/types.ts index 3804051..67b1445 100644 --- a/web/shared/api/types.ts +++ b/web/shared/api/types.ts @@ -7,6 +7,7 @@ type NarrowValue = unknown extends V ? string : number extends V ? Exclude = { [K in keyof T]: NarrowValue } export type AgentDto = Narrow +export type NodeRuntimeLogsDto = Narrow export type NodeRuntimeStatus = S['NodeRuntimeStatus'] export type AgentIpv6Dto = Narrow export type ConnectedProbeDto = Narrow