perf(filter): Implement valuer cache

internal/etw/consumer.go

@@ -131,9 +131,11 @@ func (c *Consumer) ProcessEvent(ev *etw.EventRecord) error {
 		eventsExcluded.Add(1)
 		return nil
 	}
-	if c.filter != nil && !evt.IsStackWalk() && !c.filter.Run(evt) {
+
+	if c.filter != nil && !evt.IsStackWalk() && !c.filter.Eval(evt) {
 		return nil
 	}
+
 	// Increment sequence
 	if !evt.IsState() {
 		c.sequencer.Increment()

pkg/cap/reader_windows.go

@@ -169,7 +169,7 @@ func (r *reader) read(evt *event.Event, eventsc chan *event.Event) {
 	if evt.Type.OnlyState() {
 		return
 	}
-	if r.filter != nil && !r.filter.Run(evt) {
+	if r.filter != nil && !r.filter.Eval(evt) {
 		capDroppedByFilter.Add(1)
 		return
 	}

pkg/filament/filament_test.go

@@ -24,15 +24,16 @@ package filament
 import (
 	"bufio"
 	"bytes"
+	"net"
+	"strings"
+	"testing"
+	"time"
+
 	"github.com/rabbitstack/fibratus/pkg/config"
 	"github.com/rabbitstack/fibratus/pkg/event"
 	"github.com/rabbitstack/fibratus/pkg/event/params"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"net"
-	"strings"
-	"testing"
-	"time"
 )
 
 func init() {
@@ -117,5 +118,5 @@ func TestFilamentFilter(t *testing.T) {
 		Name:   "CreateProcess",
 	}
 
-	require.True(t, filament.Filter().Run(evt))
+	require.True(t, filament.Filter().Eval(evt))
 }

pkg/filter/fields/id.go

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2016-present by Nedim Sabic Sabic
+ * https://www.fibratus.io
+ * 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 fields
+
+const MaxFieldID int16 = 23
+
+// ID returns the filter index used by the valuer cache.
+// Note that not all fields need to provide the identifier.
+func (f Field) ID() int16 {
+	switch f {
+	case EvtName:
+		return 0
+	case EvtPID:
+		return 1
+	case RegistryPath:
+		return 2
+	case FilePath:
+		return 3
+	case FileExtension:
+		return 4
+	case PsExe:
+		return 5
+	case PsName:
+		return 6
+	case PsPid:
+		return 7
+	case FileOperation:
+		return 8
+	case FileStatus:
+		return 9
+	case RegistryStatus:
+		return 10
+	case ModuleName:
+		return 11
+	case ModulePath:
+		return 12
+	case ModuleSignatureExists, DllSignatureExists:
+		return 13
+	case ModuleSignatureTrusted, DllSignatureTrusted:
+		return 14
+	case ThreadCallstackSummary:
+		return 15
+	case ThreadCallstackModules:
+		return 16
+	case ThreadCallstackSymbols:
+		return 17
+	case PsParentName:
+		return 18
+	case PsSID:
+		return 19
+	case PsCmdline:
+		return 20
+	case PsTokenIntegrityLevel:
+		return 21
+	case PsAccessMaskNames:
+		return MaxFieldID - 1
+	}
+	return -1
+}

pkg/filter/filter.go

@@ -26,7 +26,6 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
-	"sync"
 
 	errs "github.com/rabbitstack/fibratus/pkg/errors"
 	"github.com/rabbitstack/fibratus/pkg/event"
@@ -47,13 +46,21 @@ var (
 type Filter interface {
 	// Compile compiles the filter by parsing the sequence/expression.
 	Compile() error
-	// Run runs a filter with a single expression. The return value decides
-	// if the incoming event has successfully matched the filter expression.
-	Run(evt *event.Event) bool
-	// RunSequence runs a filter with sequence expressions. Sequence rules depend
-	// on the state machine transitions and partial matches to decide whether the
-	// rule is fired.
-	RunSequence(evt *event.Event, seqID int, partials map[int][]*event.Event, rawMatch bool) bool
+	// Eval evaluates the event against filter expression. Returns true if the filter
+	// has matched against the event, or false othwerise. Creates the valuer cache within
+	// the lifetime of the method call.
+	Eval(evt *event.Event) bool
+	// EvalWithValuer evaluates the event against filter. Returns true if the filter
+	// has matched against the event, or false othwerise.
+	// The valuer cache is acquired before the evaluation stage and provides a fast
+	// access to extracted field values.
+	EvalWithValuer(evt *event.Event, valuer *ValuerCache) bool
+	// EvalSequence evalutes the event against sequence expresions. Sequence rules
+	// depend on the state machine transitions and partial matches to decide whether
+	// the rule is fired.
+	// The valuer cache is acquired before the evaluation stage and provides a fast
+	// access to extracted field values.
+	EvalSequence(evt *event.Event, valuer *ValuerCache, seqID int, partials map[int][]*event.Event, rawMatch bool) bool
 	// GetStringFields returns field names mapped to their string values.
 	GetStringFields() map[fields.Field][]string
 	// GetFields returns all fields used in the filter expression.
@@ -243,11 +250,17 @@ func (f *filter) Compile() error {
 	return f.checkBoundRefs()
 }
 
-func (f *filter) Run(e *event.Event) bool {
+func (f *filter) Eval(e *event.Event) bool {
+	valuer := AcquireValuerCache()
+	defer valuer.Release()
+	return f.EvalWithValuer(e, valuer)
+}
+
+func (f *filter) EvalWithValuer(e *event.Event, cache *ValuerCache) bool {
 	if f.expr == nil {
 		return false
 	}
-	return ql.Eval(f.expr, f.mapValuer(e), f.hasFunctions)
+	return ql.Eval(f.expr, f.mapValuer(e, cache), f.hasFunctions)
 }
 
 // evalBoundSequence evaluates the sequence with bound fields
@@ -380,16 +393,15 @@ func (f *filter) evalSequence(
 	return match
 }
 
-func (f *filter) RunSequence(e *event.Event, seqID int, partials map[int][]*event.Event, rawMatch bool) bool {
+func (f *filter) EvalSequence(e *event.Event, valuerCache *ValuerCache, seqID int, partials map[int][]*event.Event, rawMatch bool) bool {
 	if f.seq == nil {
 		return false
 	}
 	nseqs := len(f.seq.Expressions)
 	if seqID > nseqs-1 {
 		return false
 	}
-	valuer := f.mapValuer(e)
-	defer valuerPool.Put(valuer)
+	valuer := f.mapValuer(e, valuerCache)
 	expr := f.seq.Expressions[seqID]
 
 	if rawMatch {
@@ -482,39 +494,38 @@ func InterpolateFields(s string, evts []*event.Event) string {
 	return r
 }
 
-var valuerPool = sync.Pool{
-	New: func() any {
-		return make(map[string]any)
-	},
-}
-
 // mapValuer for each field present in the AST, we run the
-// accessors and extract the field values that are
-// supplied to the valuer. The valuer feeds the
-// expression with correct values.
-func (f *filter) mapValuer(evt *event.Event) map[string]any {
-	valuer := valuerPool.Get().(map[string]any)
+// accessors and extract the field values that are supplied
+// to the valuer. The valuer feeds the expression with correct
+// values. If the field value is present in the valuer cache then
+// we directly populate the valuer for the field.
+func (f *filter) mapValuer(evt *event.Event, valuerCache *ValuerCache) map[string]any {
 	for _, field := range f.fields {
-		for _, accessor := range f.accessors {
-			if !accessor.IsFieldAccessible(evt) {
-				continue
-			}
-			v, err := accessor.Get(field, evt)
-			if v == nil || err != nil {
-				if v == nil {
-					valuer[field.String()] = defaultAccessorValue(field)
-				}
-				if err != nil && !errs.IsParamNotFound(err) {
-					valuer[field.String()] = defaultAccessorValue(field)
-					accessorErrors.Add(err.Error(), 1)
-				}
-				continue
+		valuerCache.populateValuer(field, func() any {
+			return f.extractField(field, evt)
+		})
+	}
+	return valuerCache.valuer
+}
+
+// extractField extracts the field value from the accessor.
+func (f *filter) extractField(field Field, evt *event.Event) any {
+	for _, accessor := range f.accessors {
+		if !accessor.IsFieldAccessible(evt) {
+			continue
+		}
+		v, err := accessor.Get(field, evt)
+		if err != nil {
+			if !errs.IsParamNotFound(err) {
+				accessorErrors.Add(err.Error(), 1)
 			}
-			valuer[field.String()] = v
-			break
+			return defaultAccessorValue(field)
+		}
+		if v != nil {
+			return v
 		}
 	}
-	return valuer
+	return defaultAccessorValue(field)
 }
 
 // addField appends a new field to the filter fields list.