Merge pull request #1323 from hakman/precompile-log-buffer-regexps

Precompile log buffer regular expressions

pkg/logcounter/log_counter.go

@@ -21,6 +21,7 @@ package logcounter
 
 import (
 	"fmt"
+	"regexp"
 	"time"
 
 	"k8s.io/utils/clock"
@@ -42,12 +43,24 @@ const (
 type logCounter struct {
 	logCh         <-chan *systemtypes.Log
 	buffer        systemlogmonitor.LogBuffer
-	pattern       string
-	revertPattern string
+	pattern       *regexp.Regexp
+	revertPattern *regexp.Regexp
 	clock         clock.Clock
 }
 
 func NewJournaldLogCounter(options *options.LogCounterOptions) (types.LogCounter, error) {
+	pattern, err := systemlogmonitor.CompilePattern(options.Pattern)
+	if err != nil {
+		return nil, fmt.Errorf("invalid pattern %q: %w", options.Pattern, err)
+	}
+	var revertPattern *regexp.Regexp
+	if options.RevertPattern != "" {
+		revertPattern, err = systemlogmonitor.CompilePattern(options.RevertPattern)
+		if err != nil {
+			return nil, fmt.Errorf("invalid revert pattern %q: %w", options.RevertPattern, err)
+		}
+	}
+
 	watcher := journald.NewJournaldWatcher(watchertypes.WatcherConfig{
 		Plugin:       "journald",
 		PluginConfig: map[string]string{journaldSourceKey: options.JournaldSource},
@@ -62,8 +75,8 @@ func NewJournaldLogCounter(options *options.LogCounterOptions) (types.LogCounter
 	return &logCounter{
 		logCh:         logCh,
 		buffer:        systemlogmonitor.NewLogBuffer(bufferSize),
-		pattern:       options.Pattern,
-		revertPattern: options.RevertPattern,
+		pattern:       pattern,
+		revertPattern: revertPattern,
 		clock:         clock.RealClock{},
 	}, nil
 }
@@ -86,7 +99,7 @@ func (e *logCounter) Count() (count int, err error) {
 			if len(e.buffer.Match(e.pattern)) != 0 {
 				count++
 			}
-			if e.revertPattern != "" && len(e.buffer.Match(e.revertPattern)) != 0 {
+			if e.revertPattern != nil && len(e.buffer.Match(e.revertPattern)) != 0 {
 				count--
 			}
 		case <-e.clock.After(timeout):

pkg/logcounter/log_counter_test.go

@@ -20,27 +20,67 @@ limitations under the License.
 package logcounter
 
 import (
+	"strings"
 	"testing"
 	"time"
 
 	testclock "k8s.io/utils/clock/testing"
 
+	"k8s.io/node-problem-detector/cmd/logcounter/options"
 	"k8s.io/node-problem-detector/pkg/logcounter/types"
 	"k8s.io/node-problem-detector/pkg/systemlogmonitor"
 	systemtypes "k8s.io/node-problem-detector/pkg/systemlogmonitor/types"
 )
 
-func NewTestLogCounter(pattern string, startTime time.Time) (types.LogCounter, *testclock.FakeClock, chan *systemtypes.Log) {
+func newTestLogCounter(t *testing.T, pattern string, startTime time.Time) (types.LogCounter, *testclock.FakeClock, chan *systemtypes.Log) {
+	t.Helper()
+	compiledPattern, err := systemlogmonitor.CompilePattern(pattern)
+	if err != nil {
+		t.Fatalf("failed to compile pattern %q: %v", pattern, err)
+	}
 	logCh := make(chan *systemtypes.Log)
 	clock := testclock.NewFakeClock(startTime)
 	return &logCounter{
 		logCh:   logCh,
 		buffer:  systemlogmonitor.NewLogBuffer(bufferSize),
-		pattern: pattern,
+		pattern: compiledPattern,
 		clock:   clock,
 	}, clock, logCh
 }
 
+func TestNewJournaldLogCounterRejectsInvalidPatterns(t *testing.T) {
+	for _, tc := range []struct {
+		name          string
+		pattern       string
+		revertPattern string
+		errorContains string
+	}{
+		{
+			name:          "pattern",
+			pattern:       "[",
+			errorContains: `invalid pattern "["`,
+		},
+		{
+			name:          "revert pattern",
+			revertPattern: "[",
+			errorContains: `invalid revert pattern "["`,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			_, err := NewJournaldLogCounter(&options.LogCounterOptions{
+				Pattern:       tc.pattern,
+				RevertPattern: tc.revertPattern,
+			})
+			if err == nil {
+				t.Fatal("expected invalid pattern error")
+			}
+			if !strings.Contains(err.Error(), tc.errorContains) {
+				t.Errorf("expected error containing %q, got %q", tc.errorContains, err)
+			}
+		})
+	}
+}
+
 func TestCount(t *testing.T) {
 	startTime := time.Now()
 	for _, tc := range []struct {
@@ -113,7 +153,7 @@ func TestCount(t *testing.T) {
 		},
 	} {
 		t.Run(tc.description, func(t *testing.T) {
-			counter, fakeClock, logCh := NewTestLogCounter(tc.pattern, startTime)
+			counter, fakeClock, logCh := newTestLogCounter(t, tc.pattern, startTime)
 			go func(logs []*systemtypes.Log, ch chan<- *systemtypes.Log) {
 				for _, log := range logs {
 					ch <- log

pkg/systemlogmonitor/config.go

@@ -59,13 +59,14 @@ func (mc *MonitorConfig) ApplyDefaultConfiguration() {
 	}
 }
 
-// ValidateRules verifies whether the regular expressions in the rules are valid.
-func (mc MonitorConfig) ValidateRules() error {
-	for _, rule := range mc.Rules {
-		_, err := regexp.Compile(rule.Pattern)
+func (mc MonitorConfig) compileRules() ([]*regexp.Regexp, error) {
+	patterns := make([]*regexp.Regexp, len(mc.Rules))
+	for i, rule := range mc.Rules {
+		pattern, err := CompilePattern(rule.Pattern)
 		if err != nil {
-			return err
+			return nil, err
 		}
+		patterns[i] = pattern
 	}
-	return nil
+	return patterns, nil
 }

pkg/systemlogmonitor/log_buffer.go

@@ -28,7 +28,7 @@ type LogBuffer interface {
 	// Push pushes log into the log buffer.
 	Push(*types.Log)
 	// Match with regular expression in the log buffer.
-	Match(string) []*types.Log
+	Match(*regexp.Regexp) []*types.Log
 	// String returns a concatenated string of the buffered logs.
 	String() string
 }
@@ -39,8 +39,6 @@ type logBuffer struct {
 	msg     []string
 	max     int
 	current int
-	// regexps caches compiled regular expressions.
-	regexps map[string]*regexp.Regexp
 }
 
 // NewLogBuffer creates log buffer with max line number limit. Because we only match logs
@@ -49,26 +47,28 @@ type logBuffer struct {
 // lines of patterns we support.
 func NewLogBuffer(maxLines int) *logBuffer {
 	return &logBuffer{
-		buffer:  make([]*types.Log, maxLines),
-		msg:     make([]string, maxLines),
-		max:     maxLines,
-		regexps: make(map[string]*regexp.Regexp),
+		buffer: make([]*types.Log, maxLines),
+		msg:    make([]string, maxLines),
+		max:    maxLines,
 	}
 }
 
+// CompilePattern compiles a log buffer pattern that must match to the end of
+// the buffered logs.
+func CompilePattern(expr string) (*regexp.Regexp, error) {
+	if _, err := regexp.Compile(expr); err != nil {
+		return nil, err
+	}
+	return regexp.Compile(expr + `\z`)
+}
+
 func (b *logBuffer) Push(log *types.Log) {
 	b.buffer[b.current%b.max] = log
 	b.msg[b.current%b.max] = log.Message
 	b.current++
 }
 
-func (b *logBuffer) Match(expr string) []*types.Log {
-	// The expression should be checked outside, and it must match to the end.
-	reg, ok := b.regexps[expr]
-	if !ok {
-		reg = regexp.MustCompile(expr + `\z`)
-		b.regexps[expr] = reg
-	}
+func (b *logBuffer) Match(reg *regexp.Regexp) []*types.Log {
 	log := b.String()
 	loc := reg.FindStringIndex(log)
 	if loc == nil {

pkg/systemlogmonitor/log_buffer_test.go

@@ -61,6 +61,12 @@ func TestPush(t *testing.T) {
 	}
 }
 
+func TestCompilePatternRejectsInvalidExpression(t *testing.T) {
+	if _, err := CompilePattern(`foo\`); err == nil {
+		t.Fatal("expected invalid expression error")
+	}
+}
+
 func TestMatch(t *testing.T) {
 	max := 4
 	for c, test := range []struct {
@@ -97,7 +103,11 @@ func TestMatch(t *testing.T) {
 			b.Push(&types.Log{Message: log})
 		}
 		for i, expr := range test.exprs {
-			logs := b.Match(expr)
+			pattern, err := CompilePattern(expr)
+			if err != nil {
+				t.Fatalf("case %d.%d: failed to compile pattern %q: %v", c+1, i+1, expr, err)
+			}
+			logs := b.Match(pattern)
 			got := []string{}
 			for _, log := range logs {
 				got = append(got, log.Message)
@@ -117,8 +127,12 @@ func BenchmarkMatch(b *testing.B) {
 	// 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(expr)
+		buf.Match(pattern)
 	}
 }

pkg/systemlogmonitor/log_monitor.go

@@ -20,6 +20,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"os"
+	"regexp"
 	"time"
 
 	"k8s.io/klog/v2"
@@ -50,6 +51,7 @@ type logMonitor struct {
 	watcher    watchertypes.LogWatcher
 	buffer     LogBuffer
 	config     MonitorConfig
+	patterns   []*regexp.Regexp
 	conditions []types.Condition
 	logCh      <-chan *systemlogtypes.Log
 	output     chan *types.Status
@@ -73,7 +75,7 @@ func NewLogMonitorOrDie(configPath string) types.Monitor {
 	}
 	// Apply default configurations
 	(&l.config).ApplyDefaultConfiguration()
-	err = l.config.ValidateRules()
+	l.patterns, err = l.config.compileRules()
 	if err != nil {
 		klog.Fatalf("Failed to validate %s matching rules %+v: %v", l.configPath, l.config.Rules, err)
 	}
@@ -152,8 +154,8 @@ func (l *logMonitor) parseLog(log *systemlogtypes.Log) {
 	// Once there is new log, log monitor will push it into the log buffer and try
 	// to match each rule. If any rule is matched, log monitor will report a status.
 	l.buffer.Push(log)
-	for _, rule := range l.config.Rules {
-		matched := l.buffer.Match(rule.Pattern)
+	for i, rule := range l.config.Rules {
+		matched := l.buffer.Match(l.patterns[i])
 		if len(matched) == 0 {
 			continue
 		}