diff --git a/util/logstash/log.go b/util/logstash/log.go index f20c05b18..5fb27f333 100644 --- a/util/logstash/log.go +++ b/util/logstash/log.go @@ -1,11 +1,11 @@ package logstash import ( + "bytes" "container/ring" "io" "maps" "slices" - "strings" "sync" jww "github.com/spf13/jwalterweatherman" @@ -25,45 +25,84 @@ func Size() int64 { return DefaultHandler.Size() } -type logger struct { - mu sync.RWMutex +// entry is a stored log line together with its global write sequence, used to +// restore chronological order when merging the trace and non-trace buffers. +type entry struct { + seq uint64 + text element +} + +// buffer is a ring of entries that grows lazily until it reaches size. +type buffer struct { data *ring.Ring size int - // length mirrors data.Len() so Write avoids an O(n) ring.Len() call on every - // log line (see Write). Invariant: any code that changes the number of nodes - // in data must keep length in sync. + // length mirrors data.Len() to keep add O(1); any change to the number of + // ring nodes must keep it in sync length int } -func New(size int) *logger { - l := &logger{ - data: ring.New(1), - size: size, +func newBuffer(size int) *buffer { + return &buffer{data: ring.New(1), size: size, length: 1} +} + +func (b *buffer) add(e entry) { + b.data.Value = e + + // grow the ring until it reaches the configured size; ring.Len() is avoided + // as it walks the whole ring, dominating CPU on weak hardware under trace + if b.length < b.size { + b.data.Link(ring.New(1)) + b.length++ + } + + b.data = b.data.Next() +} + +// visit calls fn for every stored entry, oldest first +func (b *buffer) visit(fn func(entry)) { + r := b.data + for range r.Len() { + if e, ok := r.Value.(entry); ok && e.text != "" { + fn(e) + } + r = r.Next() + } +} + +type logger struct { + mu sync.RWMutex + seq uint64 + // trace lines get their own budget so that chatty areas (mqtt, httpd) cannot + // evict the far rarer debug/info/error lines from the visible log + trace *buffer + other *buffer +} + +func New(size int) *logger { + return &logger{ + trace: newBuffer(size), + other: newBuffer(size), } - l.length = l.data.Len() // keep length in sync with the initial ring - return l } var _ io.Writer = (*logger)(nil) -func (l *logger) Write(p []byte) (n int, err error) { +func (l *logger) Write(p []byte) (int, error) { + if bytes.HasPrefix(p, []byte("[cache ]")) { + return len(p), nil + } + + e := element(p) + _, level := e.areaLevel() + l.mu.Lock() defer l.mu.Unlock() - if !strings.HasPrefix(string(p), "[cache ]") { - l.data.Value = element(string(p)) - - // dynamically grow the ring until it reaches the configured size. - // Track the length in O(1) instead of calling ring.Len(), which walks - // the whole ring on every write — once the ring is full (size 10000) - // that is 10000 pointer chases per log line and dominates CPU on weak - // hardware (e.g. Victron Venus OS / ARMv7) under verbose trace logging. - if l.length < l.size { - l.data.Link(ring.New(1)) - l.length++ - } - - l.data = l.data.Next() + l.seq++ + if level == jww.LevelTrace { + l.trace.add(entry{l.seq, e}) + } else { + l.other.add(entry{l.seq, e}) } return len(p), nil @@ -73,15 +112,10 @@ func (l *logger) Size() int64 { l.mu.RLock() defer l.mu.RUnlock() - r := l.data var size int64 - - for range r.Len() { - if e, ok := r.Value.(element); ok { - size += int64(len(e)) - } - r = r.Next() - } + sum := func(e entry) { size += int64(len(e.text)) } + l.trace.visit(sum) + l.other.visit(sum) return size } @@ -90,17 +124,14 @@ func (l *logger) Areas() []string { l.mu.RLock() defer l.mu.RUnlock() - r := l.data - areas := make(map[string]struct{}) - for range r.Len() { - r = r.Prev() - if e, ok := r.Value.(element); ok && e != "" { - if a, _ := e.areaLevel(); a != "" { - areas[a] = struct{}{} - } + collect := func(e entry) { + if a, _ := e.text.areaLevel(); a != "" { + areas[a] = struct{}{} } } + l.trace.visit(collect) + l.other.visit(collect) return slices.Sorted(maps.Keys(areas)) } @@ -109,15 +140,35 @@ func (l *logger) All(areas []string, level jww.Threshold, count int) []string { l.mu.RLock() defer l.mu.RUnlock() - r := l.data all := len(areas) == 0 && level == jww.LevelTrace - res := make([]string, 0, r.Len()) - for range r.Len() { - if e, ok := r.Value.(element); ok && e != "" && (all || e.match(areas, level)) { - res = append(res, string(e)) + matching := func(b *buffer) []entry { + var res []entry + b.visit(func(e entry) { + if all || e.text.match(areas, level) { + res = append(res, e) + } + }) + return res + } + + // trace entries can only match when trace is requested + var trace []entry + if level == jww.LevelTrace { + trace = matching(l.trace) + } + other := matching(l.other) + + // both buffers are chronologically ordered, merge them by sequence + res := make([]string, 0, len(trace)+len(other)) + for len(trace) > 0 || len(other) > 0 { + if len(other) == 0 || (len(trace) > 0 && trace[0].seq < other[0].seq) { + res = append(res, string(trace[0].text)) + trace = trace[1:] + } else { + res = append(res, string(other[0].text)) + other = other[1:] } - r = r.Next() } if count > 0 && len(res) > count { diff --git a/util/logstash/log_test.go b/util/logstash/log_test.go index 6915d6910..af0c7a144 100644 --- a/util/logstash/log_test.go +++ b/util/logstash/log_test.go @@ -1,6 +1,7 @@ package logstash import ( + "fmt" "testing" jww "github.com/spf13/jwalterweatherman" @@ -21,7 +22,7 @@ func TestLog(t *testing.T) { log.Write([]byte(s2)) log.Write([]byte(s3)) - idx := log.data + trace, other := log.trace.data, log.other.data assert.Equal(t, []string{s1, s2, s3}, log.All(nil, jww.LevelTrace, 0)) assert.Equal(t, []string{s1, s2, s3}, log.All([]string{}, jww.LevelTrace, 0)) @@ -32,13 +33,27 @@ func TestLog(t *testing.T) { assert.Equal(t, []string{}, log.All(nil, jww.LevelFatal, 0)) - assert.Equal(t, idx, log.data, "data should not be changed after All() call") + assert.Same(t, trace, log.trace.data, "data should not be changed after All() call") + assert.Same(t, other, log.other.data, "data should not be changed after All() call") assert.Equal(t, []string{"test1", "test2"}, log.Areas()) } +// TestTraceBudget verifies that a flood of trace lines cannot evict non-trace +// entries, which is what made the log view show mqtt/httpd only (#32673). +func TestTraceBudget(t *testing.T) { + log := New(3) + + log.Write([]byte(s2)) + for i := range 100 { + log.Write([]byte(fmt.Sprintf("[mqtt ] TRACE send %d", i))) + } + + assert.Equal(t, []string{s2}, log.All(nil, jww.LevelDebug, 0)) + assert.Equal(t, []string{"mqtt", "test2"}, log.Areas()) +} + // TestRingGrowsThenCaps writes more lines than the configured size and verifies // the ring grows up to size and then keeps only the most recent size entries. -// This guards the grow/cap accounting in Write (length tracked in O(1)). func TestRingGrowsThenCaps(t *testing.T) { const size = 3 log := New(size) @@ -61,28 +76,27 @@ func TestRingGrowsThenCaps(t *testing.T) { } // TestRingSkipCacheLinesDoesNotGrow verifies that "[cache ]" lines take the -// early-return path: they are not stored and must neither grow the ring nor the -// length counter, which has to stay in sync with data.Len(). +// early-return path and neither grow the ring nor the length counter. func TestRingSkipCacheLinesDoesNotGrow(t *testing.T) { log := New(3) // grow the ring past a single node first, so a stray cursor advance on the - // cache path would actually be observable (Next() on a 1-node ring is a no-op) + // cache path would actually be observable log.Write([]byte(s1)) log.Write([]byte(s2)) stored := log.All(nil, jww.LevelTrace, 0) - lenBefore := log.length - cursorBefore := log.data + lenBefore := log.trace.length + cursorBefore := log.trace.data for range 10 { log.Write([]byte("[cache ] cache line")) } assert.Equal(t, stored, log.All(nil, jww.LevelTrace, 0), "cache lines must not change stored content") - assert.Equal(t, log.data.Len(), log.length, "length counter must stay in sync with the ring") - assert.Equal(t, lenBefore, log.length, "cache lines must not grow the ring") - assert.Same(t, cursorBefore, log.data, "cache lines must not advance the write cursor") + assert.Equal(t, log.trace.data.Len(), log.trace.length, "length counter must stay in sync with the ring") + assert.Equal(t, lenBefore, log.trace.length, "cache lines must not grow the ring") + assert.Same(t, cursorBefore, log.trace.data, "cache lines must not advance the write cursor") } func BenchmarkLog(b *testing.B) {