tests: filter unit tests by PID + fix pidSet bugs

.gitignore

@@ -15,7 +15,6 @@ kbuild
 latest.json
 strange.txt
 compile_commands.json
-contrib/tester-progs/sigkill-unprivileged-user-ns-tester
 /release
 
 # project tool artifacts

api/v1/README.md

@@ -922,7 +922,7 @@ AggregationOptions defines configuration options for aggregating events.
 | namespace | [string](#string) | repeated |  |
 | health_check | [google.protobuf.BoolValue](#google-protobuf-BoolValue) |  |  |
 | pid | [uint32](#uint32) | repeated |  |
-| pid_set | [uint32](#uint32) | repeated |  |
+| pid_set | [uint32](#uint32) | repeated | Filter by the PID of a process and any of its descendants. Note that this filter is intended for testing and development purposes only and should not be used in production. In particular, PID cycling in the OS over longer periods of time may cause unexpected events to pass this filter. |
 | event_set | [EventType](#tetragon-EventType) | repeated |  |
 | pod_regex | [string](#string) | repeated | Filter by process.pod.name field using RE2 regular expression syntax: https://github.com/google/re2/wiki/Syntax |
 | arguments_regex | [string](#string) | repeated | Filter by process.arguments field using RE2 regular expression syntax: https://github.com/google/re2/wiki/Syntax |

api/v1/tetragon/events.proto

@@ -44,6 +44,10 @@ message Filter {
     repeated string namespace = 2;
     google.protobuf.BoolValue health_check = 3;
     repeated uint32 pid = 4;
+    // Filter by the PID of a process and any of its descendants. Note that this filter is
+    // intended for testing and development purposes only and should not be used in
+    // production. In particular, PID cycling in the OS over longer periods of time may
+    // cause unexpected events to pass this filter.
     repeated uint32 pid_set = 5;
     repeated EventType event_set = 6;
     // Filter by process.pod.name field using RE2 regular expression syntax:

cmd/tetragon/flags.go

@@ -74,6 +74,8 @@ const (
 	keyEnablePolicyFilter      = "enable-policy-filter"
 	keyEnablePolicyFilterDebug = "enable-policy-filter-debug"
 
+	keyEnablePidSetFilter = "enable-pid-set-filter"
+
 	keyEnableMsgHandlingLatency = "enable-msg-handling-latency"
 )
 
@@ -133,6 +135,8 @@ func readAndSetFlags() {
 	option.Config.EnablePolicyFilterDebug = viper.GetBool(keyEnablePolicyFilterDebug)
 	option.Config.EnableMsgHandlingLatency = viper.GetBool(keyEnableMsgHandlingLatency)
 
+	option.Config.EnablePidSetFilter = viper.GetBool(keyEnablePidSetFilter)
+
 	// deprecation timeline: deprecated -> 0.10.0, removed -> 0.11.0
 	// manually handle the deprecation of --config-file
 	if viper.IsSet(keyConfigFile) {

cmd/tetragon/main.go

@@ -62,11 +62,11 @@ var (
 )
 
 func getExportFilters() ([]*tetragon.Filter, []*tetragon.Filter, error) {
-	allowList, err := filters.ParseFilterList(viper.GetString(keyExportAllowlist))
+	allowList, err := filters.ParseFilterList(viper.GetString(keyExportAllowlist), viper.GetBool(keyEnablePidSetFilter))
 	if err != nil {
 		return nil, nil, err
 	}
-	denyList, err := filters.ParseFilterList(viper.GetString(keyExportDenylist))
+	denyList, err := filters.ParseFilterList(viper.GetString(keyExportDenylist), viper.GetBool(keyEnablePidSetFilter))
 	if err != nil {
 		return nil, nil, err
 	}
@@ -649,6 +649,9 @@ func execute() error {
 	flags.Bool(keyEnablePolicyFilter, false, "Enable policy filter code (beta)")
 	flags.Bool(keyEnablePolicyFilterDebug, false, "Enable policy filter debug messages")
 
+	// Provide option to enable the pidSet export filters.
+	flags.Bool(keyEnablePidSetFilter, false, "Enable pidSet export filters. Not recommended for production use")
+
 	flags.Bool(keyEnableMsgHandlingLatency, false, "Enable metrics for message handling latency")
 
 	viper.BindPFlags(flags)

contrib/tester-progs/.gitignore

@@ -13,3 +13,4 @@ libuprobe.so
 uprobe-test-1
 uprobe-test-2
 lseek-pipe
+threads-tester

docs/content/en/docs/reference/grpc-api.md

@@ -541,7 +541,7 @@ AggregationOptions defines configuration options for aggregating events.
 | namespace | [string](#string) | repeated |  |
 | health_check | [google.protobuf.BoolValue](#google-protobuf-BoolValue) |  |  |
 | pid | [uint32](#uint32) | repeated |  |
-| pid_set | [uint32](#uint32) | repeated |  |
+| pid_set | [uint32](#uint32) | repeated | Filter by the PID of a process and any of its descendants. Note that this filter is intended for testing and development purposes only and should not be used in production. In particular, PID cycling in the OS over longer periods of time may cause unexpected events to pass this filter. |
 | event_set | [EventType](#tetragon-EventType) | repeated |  |
 | pod_regex | [string](#string) | repeated | Filter by process.pod.name field using RE2 regular expression syntax: https://github.com/google/re2/wiki/Syntax |
 | arguments_regex | [string](#string) | repeated | Filter by process.arguments field using RE2 regular expression syntax: https://github.com/google/re2/wiki/Syntax |

pkg/filters/filters.go

@@ -17,6 +17,7 @@ package filters
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"io"
 	"strings"
 
@@ -27,7 +28,7 @@ import (
 )
 
 // ParseFilterList parses a list of process filters in JSON format into protobuf messages.
-func ParseFilterList(filters string) ([]*tetragon.Filter, error) {
+func ParseFilterList(filters string, enablePidSetFilters bool) ([]*tetragon.Filter, error) {
 	if filters == "" {
 		return nil, nil
 	}
@@ -41,6 +42,9 @@ func ParseFilterList(filters string) ([]*tetragon.Filter, error) {
 			}
 			return nil, err
 		}
+		if len(result.PidSet) != 0 && !enablePidSetFilters {
+			return nil, fmt.Errorf("pidSet filters use a best-effort approach for tracking PIDs and are intended for testing/development, not for production (pass the --enable-pid-set-filter to ignore)")
+		}
 		results = append(results, &result)
 	}
 	return results, nil

pkg/filters/filters_test.go

@@ -34,7 +34,7 @@ func TestParseFilterList(t *testing.T) {
 {"pid_set":[1]}
 {"event_set":["PROCESS_EXEC", "PROCESS_EXIT", "PROCESS_KPROBE", "PROCESS_TRACEPOINT"]}
 {"arguments_regex":["^--version$","^-a -b -c$"]}`
-	filterProto, err := ParseFilterList(f)
+	filterProto, err := ParseFilterList(f, true)
 	assert.NoError(t, err)
 	if diff := cmp.Diff(
 		[]*tetragon.Filter{
@@ -52,11 +52,14 @@ func TestParseFilterList(t *testing.T) {
 	); diff != "" {
 		t.Errorf("filter mismatch (-want +got):\n%s", diff)
 	}
-	_, err = ParseFilterList("invalid filter json")
+	_, err = ParseFilterList("invalid filter json", true)
 	assert.Error(t, err)
-	filterProto, err = ParseFilterList("")
+	filterProto, err = ParseFilterList("", true)
 	assert.NoError(t, err)
 	assert.Empty(t, filterProto)
+	filterProto, err = ParseFilterList(`{"pid_set":[1]}`, false)
+	assert.Error(t, err)
+	assert.Empty(t, filterProto)
 }
 
 func TestEventTypeFilterMatch(t *testing.T) {

pkg/filters/pidSet.go

@@ -18,52 +18,85 @@ import (
 	"context"
 
 	"github.com/cilium/tetragon/api/v1/tetragon"
+	"github.com/cilium/tetragon/pkg/logger"
 	v1 "github.com/cilium/tetragon/pkg/oldhubble/api/v1"
 	hubbleFilters "github.com/cilium/tetragon/pkg/oldhubble/filters"
 )
 
-/* pidSet is the set of pids we are interested in receiving events
- * for. This includes the pids listed in filter and any of their
- * children.
- */
-var pidSet map[uint32]bool
+// We could use an LRU here but we really don't want to evict old entries and risk failing
+// a test that uses this filter. Instead, we take the safer approach from the perspective
+// of testing and opt to grow the map indefinitely and log a warning if the size exceeeds
+// a pre-determined threshold. Since we have protections in place to prevent this filter
+// being used in production, this should be acceptable.
+type ChildCache = map[uint32]struct{}
 
-func filterByPidSet(pids []uint32) hubbleFilters.FilterFunc {
-	return func(ev *v1.Event) bool {
-		process := GetProcess(ev)
-		if process == nil {
-			return false
-		}
-		for _, pid := range pids {
-			if pid == process.Pid.GetValue() {
-				pidSet[pid] = true
-				return true
-			}
-		}
-		parent := GetParent(ev)
-		if parent == nil {
-			return false
-		}
-		if pidSet[parent.Pid.GetValue()] == true {
+func checkPidSetMembership(pid uint32, pidSet []uint32, childCache ChildCache) bool {
+	// Check the original pidSet. The reason for doing this separately is that we never
+	// want to drop the original pidSet from the cache. Keeping this separately in a slice
+	// is an easy way to achieve this.
+	for _, p := range pidSet {
+		if pid == p {
 			return true
 		}
-		for _, pid := range pids {
-			if pid == parent.Pid.GetValue() {
-				pidSet[pid] = true
-				return true
-			}
-		}
+	}
+	// Fall back to childCache to check children.
+	_, ok := childCache[pid]
+	return ok
+}
+
+func doFilterByPidSet(ev *v1.Event, pidSet []uint32, childCache ChildCache, childCacheWarning *int) bool {
+	process := GetProcess(ev)
+	if process == nil {
+		return false
+	}
+
+	// Check the process against our cache
+	pid := process.Pid.GetValue()
+	if checkPidSetMembership(pid, pidSet, childCache) {
+		return true
+	}
+
+	parent := GetParent(ev)
+	if parent == nil {
 		return false
 	}
+
+	// Check the parent against our cache
+	ppid := parent.Pid.GetValue()
+	if checkPidSetMembership(ppid, pidSet, childCache) {
+		// Add our own PID to the children cache so that we can match our future children.
+		childCache[pid] = struct{}{}
+		// If we exceeded the pre-determined warning limit, log a warning message and
+		// double it.
+		if len(childCache) == *childCacheWarning {
+			logger.GetLogger().Warnf("pidSet filter cache has exceeded %d entries. To prevent excess memory usage, consider disabling it.", childCacheWarning)
+			*childCacheWarning *= 2
+		}
+		return true
+	}
+
+	// No matches, return false
+	return false
 }
 
+func filterByPidSet(pidSet []uint32, childCache ChildCache, childCacheWarning int) hubbleFilters.FilterFunc {
+	return func(ev *v1.Event) bool {
+		return doFilterByPidSet(ev, pidSet, childCache, &childCacheWarning)
+	}
+}
+
+// PidSetFilter is a filter that matches on a process and all of its children by their
+// PID, up to maxChildCacheSize number of children.
 type PidSetFilter struct{}
 
 func (f *PidSetFilter) OnBuildFilter(_ context.Context, ff *tetragon.Filter) ([]hubbleFilters.FilterFunc, error) {
-	pidSet = make(map[uint32]bool)
 	var fs []hubbleFilters.FilterFunc
 	if ff.PidSet != nil {
-		fs = append(fs, filterByPidSet(ff.PidSet))
+		childCache := make(ChildCache)
+		childCacheWarning := 8192
+
+		pidSet := ff.PidSet
+		fs = append(fs, filterByPidSet(pidSet, childCache, childCacheWarning))
 	}
 	return fs, nil
 }