Add exclusion rules to filter unwanted events

Author: acbox

README.md

@@ -165,6 +165,64 @@ Open your browser to http://localhost:9090.
 
 An example of a useful query is [rate(kubewatch_event_count[5m])](<http://localhost:9090/graph?g0.range_input=1h&g0.expr=rate(kubewatch_event_count%5B1m%5D)&g0.tab=0>)
 
+## Event filtering
+
+Events can be excluded from Sloop by adding `exclusionRules` to the config file:
+
+```
+{
+  "defaultNamespace": "default",
+  "defaultKind": "Pod",
+  "defaultLookback": "1h",
+  [...]
+  "exclusionRules": {
+    "_all": [
+      {"==": [ { "var": "metadata.namespace" }, "kube-system" ]}
+    ],
+    "Pod": [
+      {"==": [ { "var": "metadata.name" }, "sloop-0" ]}
+    ],
+    "Job": [
+      {"in": [ { "var": "metadata.name" }, [ "cron1", "cron3" ] ]}
+    ]
+  }
+}`
+
+```
+
+Adding rules can help to reduce resources consumed by Sloop and remove unwanted noise from the UI for events that are of no interest.
+
+### Limiting rules to specific kinds
+
+ * Rules under the special key `_all` are evaluated against events for objects of any kind
+ * Rules under any other key are evaluated only against objects whose kind matches the key, e.g. `Pod` only applies to pods, `Job` only applies to jobs etc.
+
+### Rule format and supported operations
+
+Rules should follow the [JsonLogic](https://jsonlogic.com) format and are evaluated against the json representation of the Kubernetes API object related to the event (see below).
+
+Available operators, such as `==` and `in` shown above, are documented [here](https://jsonlogic.com/operations.html).
+
+### Data available to rule logic
+
+Kubernetes API conventions for [objects](https://github.com/kubernetes/community/blob/master/contributors/devel/sig-architecture/api-conventions.md#objects) require the following keys to exist in the json data for all resources, all of which can be referenced in rules:
+
+ * `metadata`
+ * `spec`
+ * `status`
+
+Some commonly useful fields under the `metadata` [object](https://pkg.go.dev/k8s.io/apimachinery/pkg/apis/meta/v1#ObjectMeta) are:
+
+ * `name`
+ * `namespace`
+ * `labels`
+
+#### Type specific data
+
+Some resources contain additional type-specific fields, for example `PersistentVolumeClaimSpec` objects have fields named `selector` and `storageClassName`.
+
+Type specific fields for each object and their corresponding keys in the object json representation are documented in the [core API](https://pkg.go.dev/k8s.io/[email protected]/core/v1), e.g. for `PersistentVolumeClaimSpec` objects the documentation is [here](https://pkg.go.dev/k8s.io/[email protected]/core/v1#PersistentVolumeClaimSpec).
+
 ## Contributing
 
 Refer to [CONTRIBUTING.md](CONTRIBUTING.md)<br>

go.mod

@@ -5,6 +5,7 @@ go 1.19
 require (
 	github.com/Jeffail/gabs/v2 v2.2.0
 	github.com/dgraph-io/badger/v2 v2.0.3
+	github.com/diegoholiveira/jsonlogic/v3 v3.2.7
 	github.com/ghodss/yaml v1.0.0
 	github.com/golang/glog v1.0.0
 	github.com/golang/protobuf v1.5.2
@@ -46,6 +47,8 @@ require (
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/mailru/easyjson v0.7.6 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.2 // indirect
+	github.com/mitchellh/copystructure v1.0.0 // indirect
+	github.com/mitchellh/reflectwalk v1.0.0 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect

go.sum

@@ -77,6 +77,8 @@ github.com/dgraph-io/ristretto v0.0.2/go.mod h1:KPxhHT9ZxKefz+PCeOGsrHpl1qZ7i70d
 github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2/go.mod h1:SqUrOPUnsFjfmXRMNPybcSiG0BgUW2AuFH8PAnS2iTw=
 github.com/dgryski/go-farm v0.0.0-20200201041132-a6ae2369ad13 h1:fAjc9m62+UWV/WAFKLNi6ZS0675eEUC9y3AlwSbQu1Y=
 github.com/dgryski/go-farm v0.0.0-20200201041132-a6ae2369ad13/go.mod h1:SqUrOPUnsFjfmXRMNPybcSiG0BgUW2AuFH8PAnS2iTw=
+github.com/diegoholiveira/jsonlogic/v3 v3.2.7 h1:awX07pFPnlntZzRNBcO4a2Ivxa77NMt+narq/6xcS0E=
+github.com/diegoholiveira/jsonlogic/v3 v3.2.7/go.mod h1:9oE8z9G+0OMxOoLHF3fhek3KuqD5CBqM0B6XFL08MSg=
 github.com/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815/go.mod h1:WwZ+bS3ebgob9U8Nd0kOddGdZWjyMGR8Wziv+TBNwSE=
 github.com/dustin/go-humanize v1.0.0 h1:VSnTsYCnlFHaM2/igO1h6X3HA71jcobQuxemgkq4zYo=
 github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
@@ -220,8 +222,12 @@ github.com/mailru/easyjson v0.7.6/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJ
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/matttproud/golang_protobuf_extensions v1.0.2 h1:hAHbPm5IJGijwng3PWk09JkG9WeqChjprR5s9bBZ+OM=
 github.com/matttproud/golang_protobuf_extensions v1.0.2/go.mod h1:BSXmuO+STAnVfrANrmjBb36TMTDstsz7MSK+HVaYKv4=
+github.com/mitchellh/copystructure v1.0.0 h1:Laisrj+bAB6b/yJwB5Bt3ITZhGJdqmxquMKeZ+mmkFQ=
+github.com/mitchellh/copystructure v1.0.0/go.mod h1:SNtv71yrdKgLRyLFxmLdkAbkKEFWgYaq1OVrnRcwhnw=
 github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
 github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
+github.com/mitchellh/reflectwalk v1.0.0 h1:9D+8oIskB4VJBN5SFlmc27fSlIBZaov1Wpk/IfikLNY=
+github.com/mitchellh/reflectwalk v1.0.0/go.mod h1:mSTlrgnPZtwu0c4WaC2kGObEpuNDbx0jmZXqmk4esnw=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=

pkg/sloop/common/utilities.go

@@ -37,3 +37,27 @@ func Contains(stringList []string, elem string) bool {
 func GetFilePath(filePath string, fileName string) string {
 	return path.Join(filePath, fileName)
 }
+
+func Max(x int, y int) int {
+	if x < y {
+		return y
+	}
+	return x
+}
+
+func Truncate(text string, width int, delimiter ...string) (string, error) {
+	d := "..."
+	if len(delimiter) > 0 {
+		d = delimiter[0]
+	}
+	d_len := len(d)
+	if width < 0 {
+		return "", fmt.Errorf("invalid width")
+	}
+	if len(text) <= width {
+		return text, nil
+	}
+	r := []rune(text)
+	truncated := r[:(Max(width, d_len)-d_len)]
+	return string(truncated) + d, nil
+}

pkg/sloop/common/utilities_test.go

@@ -60,3 +60,31 @@ func Test_GetFilePath(t *testing.T) {
 	actualOutput := GetFilePath(filePrefix, fileName)
 	assert.Equal(t, expectedOutput, actualOutput)
 }
+
+func Test_Truncate_StringLongerThanWidth(t *testing.T) {
+	stringLong := "Lorem ipsum dolor sit amet, consectetur adipiscing elit. Donec eget odio quis felis laoreet dictum."
+	expectedOutput := "Lorem ipsum dolor sit amet, consectetur adipiscing elit. Donec eget odio quis..."
+	actualOutput, _ := Truncate(stringLong, 80)
+	assert.Equal(t, expectedOutput, actualOutput)
+}
+
+func Test_Truncate_StringShorterThanWidth(t *testing.T) {
+	stringMedium := "Lorem ipsum dolor"
+	expectedOutput := "Lorem ipsum dolor"
+	actualOutput, _ := Truncate(stringMedium, 80)
+	assert.Equal(t, expectedOutput, actualOutput)
+}
+
+func Test_Truncate_WidthShorterThanDelimiter(t *testing.T) {
+	stringShort := "Lorem"
+	expectedOutput := "..."
+	actualOutput, _ := Truncate(stringShort, 1)
+	assert.Equal(t, expectedOutput, actualOutput)
+}
+
+func Test_Truncate_StringEmpty(t *testing.T) {
+	stringEmpty := ""
+	expectedOutput := ""
+	actualOutput, _ := Truncate(stringEmpty, 1)
+	assert.Equal(t, expectedOutput, actualOutput)
+}

pkg/sloop/ingress/kubewatcher.go

@@ -14,6 +14,9 @@ import (
 	"sync"
 	"sync/atomic"
 	"time"
+	"bytes"
+	"strings"
+	"reflect"
 
 	"github.com/golang/glog"
 	"github.com/golang/protobuf/ptypes"
@@ -32,6 +35,7 @@ import (
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
+	"github.com/diegoholiveira/jsonlogic/v3"
 )
 
 /*
@@ -67,6 +71,7 @@ type kubeWatcherImpl struct {
 	stopped        bool
 	refreshCrd     *time.Ticker
 	currentContext string
+	exclusionRules map[string][]any
 }
 
 var (
@@ -79,11 +84,12 @@ var (
 )
 
 // Todo: Add additional parameters for filtering
-func NewKubeWatcherSource(kubeClient kubernetes.Interface, outChan chan typed.KubeWatchResult, resync time.Duration, includeCrds bool, crdRefreshInterval time.Duration, masterURL string, kubeContext string, enableGranularMetrics bool) (KubeWatcher, error) {
+func NewKubeWatcherSource(kubeClient kubernetes.Interface, outChan chan typed.KubeWatchResult, resync time.Duration, includeCrds bool, crdRefreshInterval time.Duration, masterURL string, kubeContext string, enableGranularMetrics bool, exclusionRules map[string][]any) (KubeWatcher, error) {
 	kw := &kubeWatcherImpl{resync: resync, protection: &sync.Mutex{}}
 	kw.stopChan = make(chan struct{})
 	kw.crdInformers = make(map[crdGroupVersionResourceKind]*crdInformerInfo)
 	kw.outchan = outChan
+	kw.exclusionRules = exclusionRules
 
 	kw.startWellKnownInformers(kubeClient, enableGranularMetrics)
 	if includeCrds {
@@ -304,6 +310,13 @@ func (i *kubeWatcherImpl) processUpdate(kind string, obj interface{}, watchResul
 	}
 	glog.V(99).Infof("processUpdate: obj json: %v", resourceJson)
 
+	eventExcluded := i.eventExcluded(kind, resourceJson)
+	if eventExcluded {
+		objName := reflect.ValueOf(obj).Elem().FieldByName("ObjectMeta").FieldByName("Name")
+		glog.V(2).Infof("Event for object excluded: %s/%s", kind, objName)
+		return
+	}
+
 	kubeMetadata, err := kubeextractor.ExtractMetadata(resourceJson)
 	if err != nil || kubeMetadata.Namespace == "" {
 		// We are only grabbing namespace here for a prometheus metric, so if metadata extract fails we just log and continue
@@ -360,6 +373,41 @@ func (i *kubeWatcherImpl) refreshCrdInformers(masterURL string, kubeContext stri
 	}
 }
 
+func (i *kubeWatcherImpl) getExclusionRules(kind string) ([]any) {
+	kindRules, _ := i.exclusionRules[kind]
+	globalRules, _ := i.exclusionRules["_all"]
+	combinedRules := append(
+		kindRules,
+		globalRules...
+	)
+	glog.V(common.GlogVerbose).Infof("Fetched rules: %s", combinedRules)
+	return combinedRules
+}
+
+func (i *kubeWatcherImpl) eventExcluded(kind string, resourceJson string) (bool) {
+	filters := i.getExclusionRules(kind)
+	for _, logic := range filters {
+		logicJson, err := json.Marshal(logic)
+		if err != nil {
+			glog.Errorf(`Failed to parse event filtering rule "%s": %s`, logic, err)
+			return false
+		}
+		var result bytes.Buffer
+		err = jsonlogic.Apply(
+			strings.NewReader(string(logicJson)),
+			strings.NewReader(resourceJson),
+			&result,
+		)
+		resultBool := strings.Contains(result.String(), "true")
+		if resultBool {
+			truncated, _ := common.Truncate(resourceJson, 40)
+			glog.V(2).Infof(`Event matched logic: logic="%s" resource="%s"`, string(logicJson), truncated)
+			return true
+		}
+	}
+	return false
+}
+
 func (i *kubeWatcherImpl) Stop() {
 	glog.Infof("Stopping kubeWatcher")
 

pkg/sloop/ingress/kubewatcher_test.go

@@ -81,21 +81,107 @@ func Test_bigPicture(t *testing.T) {
 	masterURL := "url"
 	kubeContext := "" // empty string makes things work
 	enableGranularMetrics := true
-	kw, err := NewKubeWatcherSource(kubeClient, outChan, resync, includeCrds, time.Duration(10*time.Second), masterURL, kubeContext, enableGranularMetrics)
+	exclusionRules := map[string][]any{}
+
+	kw, err := NewKubeWatcherSource(kubeClient, outChan, resync, includeCrds, time.Duration(10*time.Second), masterURL, kubeContext, enableGranularMetrics, exclusionRules)
+	assert.NoError(t, err)
+
+	// create namespace
+	ns := "ns"
+	_, err = kubeClient.CoreV1().Namespaces().Create(context.TODO(), &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: ns}}, metav1.CreateOptions{})
+	if err != nil {
+		t.FailNow()
+	}
+
+	// create first service
+	svc := &corev1.Service{ObjectMeta: metav1.ObjectMeta{Name: "s1"}}
+	_, err = kubeClient.CoreV1().Services(ns).Create(context.TODO(), svc, metav1.CreateOptions{})
+	if err != nil {
+		t.Fatalf("Error creating service: %v\n", err)
+	}
+
+	// create second service
+	svc = &corev1.Service{ObjectMeta: metav1.ObjectMeta{Name: "s2"}}
+	_, err = kubeClient.CoreV1().Services(ns).Create(context.TODO(), svc, metav1.CreateOptions{})
+	if err != nil {
+		t.Fatalf("Error creating service: %v\n", err)
+	}
+
+	// await events
+	result1 := <-outChan
+	result2 := <-outChan
+	result3 := <-outChan
+
+	assert.Contains(t, result1.Payload, `"name":"ns"`)
+	assert.Contains(t, result2.Payload, `"name":"s1"`)
+	assert.Contains(t, result3.Payload, `"name":"s2"`)
+
+	kw.Stop()
+}
+
+// As above but specify non-default exclusion rules to exclude events for service named s2
+func Test_bigPictureWithExclusionRules(t *testing.T) {
+	newCrdClient = newTestCrdClient(reactionListOfOne) // force startCustomInformers() to use a fake clientset
+
+	kubeClient := kubernetesFake.NewSimpleClientset()
+	outChan := make(chan typed.KubeWatchResult, 5)
+	resync := 30 * time.Minute
+	includeCrds := true
+	masterURL := "url"
+	kubeContext := "" // empty string makes things work
+	enableGranularMetrics := true
+	exclusionRules := map[string][]any{
+		"_all": []any{
+			map[string]any{
+				"==": []any{
+					map[string]any{
+						"var": "metadata.name",
+					},
+					"s2",
+				},
+			},
+		},
+	}
+
+	kw, err := NewKubeWatcherSource(kubeClient, outChan, resync, includeCrds, time.Duration(10*time.Second), masterURL, kubeContext, enableGranularMetrics, exclusionRules)
 	assert.NoError(t, err)
 
-	// create service and await corresponding event
+	// create namespace
 	ns := "ns"
 	_, err = kubeClient.CoreV1().Namespaces().Create(context.TODO(), &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: ns}}, metav1.CreateOptions{})
 	if err != nil {
 		t.FailNow()
 	}
-	svc := &corev1.Service{ObjectMeta: metav1.ObjectMeta{Name: "s"}}
+
+	// create first service
+	svc := &corev1.Service{ObjectMeta: metav1.ObjectMeta{Name: "s1"}}
+	_, err = kubeClient.CoreV1().Services(ns).Create(context.TODO(), svc, metav1.CreateOptions{})
+	if err != nil {
+		t.Fatalf("Error creating service: %v\n", err)
+	}
+
+	// create second service, corresponding event should be excluded by exclusion rule
+	svc = &corev1.Service{ObjectMeta: metav1.ObjectMeta{Name: "s2"}}
+	_, err = kubeClient.CoreV1().Services(ns).Create(context.TODO(), svc, metav1.CreateOptions{})
+	if err != nil {
+		t.Fatalf("Error creating service: %v\n", err)
+	}
+
+	// create third service
+	svc = &corev1.Service{ObjectMeta: metav1.ObjectMeta{Name: "s3"}}
 	_, err = kubeClient.CoreV1().Services(ns).Create(context.TODO(), svc, metav1.CreateOptions{})
 	if err != nil {
 		t.Fatalf("Error creating service: %v\n", err)
 	}
-	_ = <-outChan
+
+	// await events
+	result1 := <-outChan
+	result2 := <-outChan
+	result3 := <-outChan
+
+	assert.Contains(t, result1.Payload, `"name":"ns"`)
+	assert.Contains(t, result2.Payload, `"name":"s1"`)
+	assert.Contains(t, result3.Payload, `"name":"s3"`) // s2 should've been excluded so expect s3
 
 	kw.Stop()
 }

pkg/sloop/server/internal/config/config.go

@@ -29,8 +29,9 @@ type SloopConfig struct {
 	// These fields can only come from command line
 	ConfigFile string
 	// These fields can only come from file because they use complex types
-	LeftBarLinks  []webserver.LinkTemplate         `json:"leftBarLinks"`
-	ResourceLinks []webserver.ResourceLinkTemplate `json:"resourceLinks"`
+	LeftBarLinks   []webserver.LinkTemplate         `json:"leftBarLinks"`
+	ResourceLinks  []webserver.ResourceLinkTemplate `json:"resourceLinks"`
+	ExclusionRules map[string][]any                 `json:"exclusionRules"`
 	// Normal fields that can come from file or cmd line
 	DisableKubeWatcher       bool          `json:"disableKubeWatch"`
 	KubeWatchResyncInterval  time.Duration `json:"kubeWatchResyncInterval"`
@@ -177,6 +178,7 @@ func getDefaultConfig() *SloopConfig {
 		EnableGranularMetrics:    false,
 		PrivilegedAccess:         true,
 		BadgerDetailLogEnabled:   false,
+		ExclusionRules:           map[string][]any{},
 	}
 	return &defaultConfig
 }