fleet: send log counters in heartbeats and pull the text on demand

Heartbeats carry per-level counts, the most frequent log types (a stable hash of the message) and the newest id, no text. The node keeps its last 2048 lines; moshiro.logs.fetch lets the backend pull them. The log page pulls when it is open, and the backend pulls warnings and errors once a minute after a node reports an error.

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);
 }
 
-/// <summary>One structured node log line queued for delivery to the control plane.</summary>
+/// <summary>One structured node log line.</summary>
 public sealed record NodeLogRecord(long AtUtcMs, string Level, string Message, Dictionary<string, string> Fields);
 
-public sealed record NodeLogBatch(IReadOnlyList<NodeLogRecord> Records, long Dropped);
+/// <summary>A retained log line: <paramref name="Id"/> grows with every line, <paramref name="Type"/> is a stable hash of the message text (the template), so counters can be sent without the text.</summary>
+public sealed record NodeLogEntry(long Id, uint Type, NodeLogRecord Record);
 
-/// <summary>Optional bounded log source used by node heartbeats to upload recent local logs.</summary>
-public interface INodeLogBatchSource
+/// <summary>How many lines of one type and level were written since the last summary.</summary>
+public sealed record NodeLogTypeCount(uint Type, string Level, long Count);
+
+/// <summary>What a node tells the control plane about its logs without sending any text: counters since the last summary and the newest line id.</summary>
+public sealed record NodeLogSummary(long LastId, long Info, long Warn, long Error, IReadOnlyList<NodeLogTypeCount> Types);
+
+public sealed record NodeLogPage(IReadOnlyList<NodeLogEntry> Items, long LastId);
+
+/// <summary>Bounded local log retention. The control plane gets counters in heartbeats and pulls the text when it needs it.</summary>
+public interface INodeLogSource
 {
-    NodeLogBatch TakeBatch(int maxRecords);
+    NodeLogSummary TakeSummary();
+
+    /// <summary>Lines with an id above <paramref name="afterId"/>, oldest first, at most <paramref name="limit"/>, at or above <paramref name="minLevel"/> (info, warn or error).</summary>
+    NodeLogPage Read(long afterId, int limit, string? minLevel);
 }
 
 public sealed class NullNodeLog : INodeLog

backend/fleet/src/Fleet.Hosting/BufferedNodeLog.cs

@@ -3,20 +3,28 @@ using Fleet.Core;
 
 namespace Fleet.Hosting;
 
-/// <summary>Writes logs locally and keeps a bounded copy for the next authenticated heartbeat.</summary>
-public sealed class BufferedNodeLog : INodeLog, INodeLogBatchSource
+/// <summary>
+/// Writes logs locally and keeps a bounded ring of recent lines. Heartbeats carry only counters (see <see cref="TakeSummary"/>);
+/// the text leaves the node when the control plane asks for it (see <see cref="Read"/>).
+/// </summary>
+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<NodeLogRecord> _pending = [];
+    private readonly LinkedList<NodeLogEntry> _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<NodeLogRecord>(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<NodeLogEntry>(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);
         }
     }
 
+    /// <summary>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.</summary>
+    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();

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"]);
+    }
+}

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<InvalidOperationException>(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<InvalidOperationException>(async () =>