From 4334390423b4c9f11b6dc5d8ef984de4dd1179e3 Mon Sep 17 00:00:00 2001 From: Veer Singh <8453348+digitalveer@users.noreply.github.com> Date: Mon, 27 Jul 2026 11:04:28 -0700 Subject: [PATCH] Cache the joined log buffer and skip it for single-line rules The log buffer joins every buffered line and scans all of them for each pattern on every push, which is wasted work when a pattern cannot even match across lines. The buffer now caches the joined string, and Push invalidates the cache. CompilePattern marks the patterns that cannot match across lines (no start anchor, no way to match a newline, and the appended \z anchor binds every branch) and Match checks them against the most recent line alone. --- pkg/logcounter/log_counter.go | 7 +- pkg/systemlogmonitor/config.go | 6 +- pkg/systemlogmonitor/log_buffer.go | 112 +++++++-- pkg/systemlogmonitor/log_buffer_bench_test.go | 74 ++++++ .../log_buffer_equivalence_test.go | 216 ++++++++++++++++++ pkg/systemlogmonitor/log_buffer_test.go | 18 -- pkg/systemlogmonitor/log_monitor.go | 3 +- 7 files changed, 394 insertions(+), 42 deletions(-) create mode 100644 pkg/systemlogmonitor/log_buffer_bench_test.go create mode 100644 pkg/systemlogmonitor/log_buffer_equivalence_test.go diff --git a/pkg/logcounter/log_counter.go b/pkg/logcounter/log_counter.go index dd1ef6e69..1662092bf 100644 --- a/pkg/logcounter/log_counter.go +++ b/pkg/logcounter/log_counter.go @@ -21,7 +21,6 @@ package logcounter import ( "fmt" - "regexp" "time" "k8s.io/utils/clock" @@ -43,8 +42,8 @@ const ( type logCounter struct { logCh <-chan *systemtypes.Log buffer systemlogmonitor.LogBuffer - pattern *regexp.Regexp - revertPattern *regexp.Regexp + pattern *systemlogmonitor.Pattern + revertPattern *systemlogmonitor.Pattern clock clock.Clock } @@ -53,7 +52,7 @@ func NewJournaldLogCounter(options *options.LogCounterOptions) (types.LogCounter if err != nil { return nil, fmt.Errorf("invalid pattern %q: %w", options.Pattern, err) } - var revertPattern *regexp.Regexp + var revertPattern *systemlogmonitor.Pattern if options.RevertPattern != "" { revertPattern, err = systemlogmonitor.CompilePattern(options.RevertPattern) if err != nil { diff --git a/pkg/systemlogmonitor/config.go b/pkg/systemlogmonitor/config.go index 4ba76b7b6..a2ad39f27 100644 --- a/pkg/systemlogmonitor/config.go +++ b/pkg/systemlogmonitor/config.go @@ -17,8 +17,6 @@ limitations under the License. package systemlogmonitor import ( - "regexp" - watchertypes "k8s.io/node-problem-detector/pkg/systemlogmonitor/logwatchers/types" systemlogtypes "k8s.io/node-problem-detector/pkg/systemlogmonitor/types" "k8s.io/node-problem-detector/pkg/types" @@ -59,8 +57,8 @@ func (mc *MonitorConfig) ApplyDefaultConfiguration() { } } -func (mc MonitorConfig) compileRules() ([]*regexp.Regexp, error) { - patterns := make([]*regexp.Regexp, len(mc.Rules)) +func (mc MonitorConfig) compileRules() ([]*Pattern, error) { + patterns := make([]*Pattern, len(mc.Rules)) for i, rule := range mc.Rules { pattern, err := CompilePattern(rule.Pattern) if err != nil { diff --git a/pkg/systemlogmonitor/log_buffer.go b/pkg/systemlogmonitor/log_buffer.go index 0645d31ed..001abcc02 100644 --- a/pkg/systemlogmonitor/log_buffer.go +++ b/pkg/systemlogmonitor/log_buffer.go @@ -18,27 +18,41 @@ package systemlogmonitor import ( "regexp" + "regexp/syntax" + "slices" "strings" "k8s.io/node-problem-detector/pkg/systemlogmonitor/types" ) -// LogBuffer buffers the logs and supports match in the log buffer with regular expression. +// LogBuffer buffers the logs and matches a compiled pattern. type LogBuffer interface { // Push pushes log into the log buffer. Push(*types.Log) - // Match with regular expression in the log buffer. - Match(*regexp.Regexp) []*types.Log - // String returns a concatenated string of the buffered logs. - String() string + // Match with a compiled pattern in the log buffer. + Match(*Pattern) []*types.Log } +// Pattern is a compiled rule plus the facts that let Match narrow its scan. +type Pattern struct { + // regexp is the rule anchored to the end of the buffered logs. + regexp *regexp.Regexp + // lastLineOnly reports that the rule can match only in the last pushed line. + // Match then skips building the joined buffer. + lastLineOnly bool +} + +// logBuffer is not safe for concurrent use. type logBuffer struct { // buffer is a simple ring buffer. buffer []*types.Log msg []string max int current int + // joined caches the result of String. Push clears it. + joined string + // joinedOK reports whether joined is current. + joinedOK bool } // NewLogBuffer creates log buffer with max line number limit. Because we only match logs @@ -55,22 +69,74 @@ func NewLogBuffer(maxLines int) *logBuffer { // CompilePattern compiles a log buffer pattern that must match to the end of // the buffered logs. -func CompilePattern(expr string) (*regexp.Regexp, error) { +func CompilePattern(expr string) (*Pattern, error) { + // Compile expr alone first so an error cites the pattern as written. if _, err := regexp.Compile(expr); err != nil { return nil, err } - return regexp.Compile(expr + `\z`) + anchored := expr + `\z` + reg, err := regexp.Compile(anchored) + if err != nil { + return nil, err + } + p := &Pattern{regexp: reg} + tree, err := syntax.Parse(anchored, syntax.Perl) + if err != nil { + return p, nil + } + // A top-level alternation binds the appended anchor to its last branch only. + // Equal trees prove that the anchor covers every branch. + grouped, err := syntax.Parse(`(?:`+expr+`)\z`, syntax.Perl) + if err != nil { + return p, nil + } + p.lastLineOnly = tree.Equal(grouped) && isLastLineOnly(tree) + return p, nil +} + +// isLastLineOnly reports whether the rule accepts no newline and has no start anchor. +func isLastLineOnly(re *syntax.Regexp) bool { + switch re.Op { + case syntax.OpAnyChar: + // `(?s).` accepts a newline. + return false + case syntax.OpBeginText, syntax.OpBeginLine, syntax.OpEndLine: + // A start anchor marks the start of the whole buffer. + return false + case syntax.OpLiteral: + if slices.Contains(re.Rune, '\n') { + return false + } + case syntax.OpCharClass: + // Rune stores the character class as inclusive lo, hi pairs. + for i := 0; i+1 < len(re.Rune); i += 2 { + if re.Rune[i] <= '\n' && '\n' <= re.Rune[i+1] { + return false + } + } + } + for _, sub := range re.Sub { + if !isLastLineOnly(sub) { + return false + } + } + return true } func (b *logBuffer) Push(log *types.Log) { b.buffer[b.current%b.max] = log b.msg[b.current%b.max] = log.Message b.current++ + b.joinedOK = false + b.joined = "" } -func (b *logBuffer) Match(reg *regexp.Regexp) []*types.Log { +func (b *logBuffer) Match(p *Pattern) []*types.Log { + if p.lastLineOnly { + return b.matchLastLine(p.regexp) + } log := b.String() - loc := reg.FindStringIndex(log) + loc := p.regexp.FindStringIndex(log) if loc == nil { // No match return nil @@ -86,15 +152,33 @@ func (b *logBuffer) Match(reg *regexp.Regexp) []*types.Log { break } } - for i := 0; i < len(matched)/2; i++ { - matched[i], matched[len(matched)-i-1] = matched[len(matched)-i-1], matched[i] - } + slices.Reverse(matched) return matched } +// matchLastLine matches a lastLineOnly rule against the most recently pushed line. +func (b *logBuffer) matchLastLine(reg *regexp.Regexp) []*types.Log { + if b.current == 0 { + return nil + } + last := (b.current - 1) % b.max + if !reg.MatchString(b.msg[last]) { + return nil + } + return []*types.Log{b.buffer[last]} +} + func (b *logBuffer) String() string { - logs := append(b.msg[b.current%b.max:], b.msg[:b.current%b.max]...) - return concatLogs(logs) + if b.joinedOK { + return b.joined + } + head := b.current % b.max + lines := make([]string, 0, b.max) + lines = append(lines, b.msg[head:]...) + lines = append(lines, b.msg[:head]...) + b.joined = concatLogs(lines) + b.joinedOK = true + return b.joined } // tail returns current tail index. diff --git a/pkg/systemlogmonitor/log_buffer_bench_test.go b/pkg/systemlogmonitor/log_buffer_bench_test.go new file mode 100644 index 000000000..a04fd30ca --- /dev/null +++ b/pkg/systemlogmonitor/log_buffer_bench_test.go @@ -0,0 +1,74 @@ +/* +Copyright 2026 The Kubernetes Authors All rights reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package systemlogmonitor + +import ( + "fmt" + "testing" + + "k8s.io/node-problem-detector/pkg/systemlogmonitor/types" +) + +// kernelMonitorPatterns is the rule set of config/kernel-monitor.json. +var kernelMonitorPatterns = []string{ + `Killed process \d+ (.+) total-vm:\d+kB, anon-rss:\d+kB, file-rss:\d+kB.*`, + `task [\S ]+:\w+ blocked for more than \w+ seconds\.`, + `unregister_netdevice: waiting for \w+ to become free. Usage count = \d+`, + `BUG: unable to handle kernel NULL pointer dereference at .*`, + `divide error: 0000 \[#\d+\] SMP`, + `EXT4-fs error .*`, + `EXT4-fs warning .*`, + `Buffer I/O error .*`, + `XFS .* Shutting down filesystem.?`, + `CE memory read error .*`, + `.*\[Hardware Error\]: event severity: corrected$`, + `.*\[Hardware Error\]: event severity: recoverable$`, + `.*\[Hardware Error\]: event severity: fatal$`, + `task docker:\w+ blocked for more than \w+ seconds\.`, +} + +// benchmarkLine matches none of the rules above, the common case on a healthy node. +const benchmarkLine = "systemd[1]: Started Session 4321 of user core." + +// BenchmarkPushAndMatchAll measures the per-line cost of the monitor hot path. +// Each iteration pushes one line and evaluates every rule against the buffer. +func BenchmarkPushAndMatchAll(b *testing.B) { + for _, bufferSize := range []int{10, 100} { + b.Run(fmt.Sprintf("buffer=%d", bufferSize), func(b *testing.B) { + buf := NewLogBuffer(bufferSize) + for range bufferSize { + buf.Push(&types.Log{Message: benchmarkLine}) + } + patterns := make([]*Pattern, 0, len(kernelMonitorPatterns)) + for _, expr := range kernelMonitorPatterns { + p, err := CompilePattern(expr) + if err != nil { + b.Fatalf("failed to compile %q: %v", expr, err) + } + patterns = append(patterns, p) + } + log := &types.Log{Message: benchmarkLine} + b.ReportAllocs() + for b.Loop() { + buf.Push(log) + for _, p := range patterns { + buf.Match(p) + } + } + }) + } +} diff --git a/pkg/systemlogmonitor/log_buffer_equivalence_test.go b/pkg/systemlogmonitor/log_buffer_equivalence_test.go new file mode 100644 index 000000000..e90698106 --- /dev/null +++ b/pkg/systemlogmonitor/log_buffer_equivalence_test.go @@ -0,0 +1,216 @@ +/* +Copyright 2026 The Kubernetes Authors All rights reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package systemlogmonitor + +import ( + "math/rand" + "reflect" + "regexp" + "slices" + "testing" + + "k8s.io/node-problem-detector/pkg/systemlogmonitor/types" +) + +// referenceMatch is the unoptimized matcher: build the whole buffer, scan all of it. +// Match must agree with it on every input. +func referenceMatch(b *logBuffer, reg *regexp.Regexp) []*types.Log { + log := concatLogs(append(append([]string{}, b.msg[b.current%b.max:]...), b.msg[:b.current%b.max]...)) + loc := reg.FindStringIndex(log) + if loc == nil { + return nil + } + s := len(log) - loc[0] - 1 + total := 0 + matched := []*types.Log{} + for i := b.tail(); i >= b.current && b.buffer[i%b.max] != nil; i-- { + matched = append(matched, b.buffer[i%b.max]) + total += len(b.msg[i%b.max]) + 1 + if total > s { + break + } + } + slices.Reverse(matched) + return matched +} + +// equivalencePatterns mixes shipped rules with rules that attack the last-line shortcut. +var equivalencePatterns = []string{ + // Shipped rules. + `Killed process \d+ (.+) total-vm:\d+kB, anon-rss:\d+kB, file-rss:\d+kB.*`, + `task [\S ]+:\w+ blocked for more than \w+ seconds\.`, + `unregister_netdevice: waiting for \w+ to become free. Usage count = \d+`, + `BUG: unable to handle kernel NULL pointer dereference at .*`, + `EXT4-fs error .*`, + `XFS .* Shutting down filesystem.?`, + `.*\[Hardware Error\]: event severity: fatal$`, + `Error syncing pod .*skipping.*failed to "StartContainer".*`, + // Rules that must fall back to the full buffer. + `(?s)first.*second`, + `^only line`, + `(?m)^line \w+$`, + `alpha\nbeta`, + `alpha[\s\S]*beta`, + `alpha[\x00-\x7f]+beta`, + `\Aalpha`, + // Top-level alternations that the appended anchor does not bind. + `alpha|beta`, + `abort|abandon`, + `alpha|`, + // Rules that stay on the last line but stress the edges. + `\balpha\b`, + `alpha$`, + `a*`, + ``, + `(alpha|beta) gamma`, + `[^x]+`, +} + +// equivalenceLines are pushed in random order so matches land at every ring offset. +var equivalenceLines = []string{ + "alpha gamma", + "beta gamma", + "abort now", + "only line", + "first", + "second", + "line one", + "line two", + "alpha", + "beta", + "task docker:1234 blocked for more than 120 seconds.", + "EXT4-fs error (device sda1): ext4_find_entry:1455: inode #2", + "mce: [Hardware Error]: event severity: fatal", + "Killed process 1234 (mysqld) total-vm:100kB, anon-rss:20kB, file-rss:3kB", + "", + "trailing\nembedded newline", +} + +func TestMatchEquivalence(t *testing.T) { + rng := rand.New(rand.NewSource(1)) + // Compile each rule once, before the trial loops. + patterns := make([]*Pattern, 0, len(equivalencePatterns)) + refRegexps := make([]*regexp.Regexp, 0, len(equivalencePatterns)) + for _, expr := range equivalencePatterns { + p, err := CompilePattern(expr) + if err != nil { + t.Fatalf("failed to compile %q: %v", expr, err) + } + patterns = append(patterns, p) + refRegexps = append(refRegexps, regexp.MustCompile(expr+`\z`)) + } + for _, maxLines := range []int{1, 2, 3, 5, 10} { + for trial := range 200 { + buf := NewLogBuffer(maxLines) + ref := NewLogBuffer(maxLines) + // Push a random number of lines, from none to more than the ring. + for range rng.Intn(maxLines*2 + 1) { + log := &types.Log{Message: equivalenceLines[rng.Intn(len(equivalenceLines))]} + buf.Push(log) + ref.Push(log) + for i, expr := range equivalencePatterns { + want := referenceMatch(ref, refRegexps[i]) + got := buf.Match(patterns[i]) + if len(want) == 0 && len(got) == 0 { + continue + } + if !reflect.DeepEqual(want, got) { + t.Fatalf("maxLines=%d trial=%d pattern=%q buffer=%q:\nwant %v\ngot %v", + maxLines, trial, expr, ref.String(), messages(want), messages(got)) + } + } + } + } + } +} + +func messages(logs []*types.Log) []string { + out := []string{} + for _, log := range logs { + out = append(out, log.Message) + } + return out +} + +// TestLastLineOnlyClassification pins the lastLineOnly verdict for each rule shape. +func TestLastLineOnlyClassification(t *testing.T) { + for expr, want := range map[string]bool{ + `EXT4-fs error .*`: true, + `task \S+ blocked`: true, + `alpha$`: true, + `\balpha\b`: true, + // A negated class holds the newline unless the rule excludes it. + `[^x]+`: false, + `[^x\n]+`: true, + `(alpha|beta)+ gamma`: true, + `(?s)alpha.*beta`: false, + `^alpha`: false, + `\Aalpha`: false, + `(?m)^alpha$`: false, + "alpha\nbeta": false, + `alpha[\s\S]*beta`: false, + `alpha[\x00-\x7f]beta`: false, + `alpha[\n]beta`: false, + `alpha(beta|\n)`: false, + `alpha{1,3}[\t-\r]beta`: false, + // The appended anchor reaches the last branch of a top-level alternation only. + `alpha|beta`: false, + // The parser factors the shared prefix out, so the root stays a concatenation. + `abort|abandon`: false, + `a|`: false, + `(alpha|beta) gamma`: true, + } { + p, err := CompilePattern(expr) + if err != nil { + t.Fatalf("failed to compile %q: %v", expr, err) + } + if got := p.lastLineOnly; got != want { + t.Errorf("pattern %q: lastLineOnly = %v, want %v", expr, got, want) + } + } + // Every shipped kernel rule must keep the last line shortcut. + for _, expr := range kernelMonitorPatterns { + p, err := CompilePattern(expr) + if err != nil { + t.Fatalf("failed to compile %q: %v", expr, err) + } + if !p.lastLineOnly { + t.Errorf("kernel rule %q: lastLineOnly = false, want true", expr) + } + } +} + +// TestMatchAlternationSpansBuffer pins the reported repro for a top-level alternation. +// The first branch matches an older line, so the last line shortcut must not apply. +func TestMatchAlternationSpansBuffer(t *testing.T) { + b := NewLogBuffer(2) + b.Push(&types.Log{Message: "kernel: oom-kill:constraint=CONSTRAINT_NONE"}) + b.Push(&types.Log{Message: "kubelet: node ready"}) + expr := `oom-kill|Out of memory` + p, err := CompilePattern(expr) + if err != nil { + t.Fatalf("failed to compile %q: %v", expr, err) + } + want := referenceMatch(b, regexp.MustCompile(expr+`\z`)) + if len(want) == 0 { + t.Fatalf("pattern %q: the reference matcher found nothing", expr) + } + got := b.Match(p) + if !reflect.DeepEqual(want, got) { + t.Errorf("pattern %q: want %v, got %v", expr, messages(want), messages(got)) + } +} diff --git a/pkg/systemlogmonitor/log_buffer_test.go b/pkg/systemlogmonitor/log_buffer_test.go index c37d9fb7a..bf5800d38 100644 --- a/pkg/systemlogmonitor/log_buffer_test.go +++ b/pkg/systemlogmonitor/log_buffer_test.go @@ -118,21 +118,3 @@ func TestMatch(t *testing.T) { } } } - -func BenchmarkMatch(b *testing.B) { - buf := NewLogBuffer(10) - for i := 0; i < 10; i++ { - buf.Push(&types.Log{Message: "Out of memory: Kill process 20744 (mysqld) score 318 or sacrifice child"}) - } - // A pattern from the default kernel monitor configuration which does not - // match the buffered logs. - expr := `task [\S ]+:\w+ blocked for more than \w+ seconds\.` - pattern, err := CompilePattern(expr) - if err != nil { - b.Fatalf("failed to compile pattern %q: %v", expr, err) - } - b.ResetTimer() - for i := 0; i < b.N; i++ { - buf.Match(pattern) - } -} diff --git a/pkg/systemlogmonitor/log_monitor.go b/pkg/systemlogmonitor/log_monitor.go index 365f0878d..15b0c40ec 100644 --- a/pkg/systemlogmonitor/log_monitor.go +++ b/pkg/systemlogmonitor/log_monitor.go @@ -20,7 +20,6 @@ import ( "encoding/json" "fmt" "os" - "regexp" "time" "k8s.io/klog/v2" @@ -51,7 +50,7 @@ type logMonitor struct { watcher watchertypes.LogWatcher buffer LogBuffer config MonitorConfig - patterns []*regexp.Regexp + patterns []*Pattern conditions []types.Condition logCh <-chan *systemlogtypes.Log output chan *types.Status