Merge pull request #1329 from DigitalVeer/lastline-match

Cache the joined log buffer and skip it for single-line rules

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 {

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 {

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.

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)
+				}
+			}
+		})
+	}
+}

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))
+	}
+}

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)
-	}
-}

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