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