Stefan/evcc-io
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Log: separate ring buffer budget for trace and higher levels (#32698)

Browse source
This commit is contained in:
andig 2026-08-10 17:59:23 +02:00 • committed by GitHub
parent 232d367461
commit 4caddeec5d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 125 additions and 60 deletions
Download patch file Download diff file Expand all files Collapse all files

149
util/logstash/log.go
View file

@ -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 {

36
util/logstash/log_test.go
View file

@ -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) {