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 `_any` 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>

Tiltfile

@@ -0,0 +1,57 @@
+#
+# To make this Tiltfile work with the helm chart comment out
+# these two lines in helm/sloop/templates/statefulset.yaml:
+#
+#   command:
+#   - /sloop
+#
+
+load('ext://helm_resource', 'helm_resource')
+
+# Allow the cluster to avoid problems while having kubectl configured to talk to a remote cluster.
+allow_k8s_contexts('kind-sloop')
+
+# Load the restart_process extension with the docker_build_with_restart func for live reloading.
+load('ext://restart_process', 'docker_build_with_restart')
+
+# Building binary locally.
+local_resource(
+    'sloop-binary',
+    # gcflags disable all optimisations
+    'GOOS=linux go build -gcflags "all=-N -l" -o sloop pkg/sloop/main.go',
+    deps=[
+        './pkg/sloop/',
+    ],
+)
+
+# Use custom Dockerfile for Tilt builds, which only takes locally built binary for live reloading.
+dockerfile = '''
+    FROM golang:1.19-alpine
+    RUN go install github.com/go-delve/delve/cmd/dlv@latest
+    COPY sloop /sloop
+    '''
+
+# Wrap a docker_build to restart the given entrypoint after a Live Update.
+docker_build_with_restart(
+    'sloop-image',
+    '.',
+    dockerfile_contents=dockerfile,
+    # Uncomment to run development image without delve
+    #entrypoint='/sloop --v=2 --config=/sloopconfig/sloop.json',
+    # Uncomment to run development image with delve (connect using "dlv connect 127.0.0.1:50100")
+    entrypoint='/go/bin/dlv --listen=0.0.0.0:50100 --api-version=2 --headless=true --only-same-user=false --accept-multiclient --check-go-version=false exec /sloop -- --v=2 --config=/sloopconfig/sloop.json',
+    live_update=[
+        # Copy the binary so it gets restarted.
+        sync('sloop', '/sloop'),
+    ],
+)
+
+# Deploy resources via Helm chart, replacing image with local build under development
+helm_resource(
+  'sloop-helm',
+  './helm/sloop',
+  deps=['./sloop'],
+  image_keys=[('image.repository', 'image.tag')],
+  image_deps=['sloop-image'],
+  port_forwards=["50100:50100","8080:8080"],
+)

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,13 @@ func Contains(stringList []string, elem string) bool {
 func GetFilePath(filePath string, fileName string) string {
 	return path.Join(filePath, fileName)
 }
+
+func Truncate(text string, width int) (string, error) {
+	if width < 0 {
+		return "", fmt.Errorf("invalid width size")
+	}
+
+	r := []rune(text)
+	trunc := r[:width]
+	return string(trunc)+ "...", nil
+}

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

@@ -15,6 +15,7 @@ import (
 	"sync/atomic"
 	"testing"
 	"time"
+	"strings"
 
 	"github.com/salesforce/sloop/pkg/sloop/store/typed"
 	"github.com/stretchr/testify/assert"
@@ -81,21 +82,58 @@ 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{
+		"_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, should be filtered out 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)
 	}
-	_ = <-outChan
+
+	// 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)
+	}
+
+	// await events
+	result1 := <-outChan
+	result2 := <-outChan
+	result3 := <-outChan
+
+	if !strings.Contains(result1.Payload, `"name":"ns"`) { t.Error("Expected event not found") }
+	if !strings.Contains(result2.Payload, `"name":"s1"`) { t.Error("Expected event not found") }
+	if !strings.Contains(result3.Payload, `"name":"s2"`) { t.Error("Expected event not found") }
 
 	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
 }

pkg/sloop/server/server.go

@@ -103,7 +103,7 @@ func RealMain() error {
 			return errors.Wrap(err, "failed to create kubernetes client")
 		}
 
-		kubeWatcherSource, err = ingress.NewKubeWatcherSource(kubeClient, kubeWatchChan, conf.KubeWatchResyncInterval, conf.WatchCrds, conf.CrdRefreshInterval, conf.ApiServerHost, kubeContext, conf.EnableGranularMetrics)
+		kubeWatcherSource, err = ingress.NewKubeWatcherSource(kubeClient, kubeWatchChan, conf.KubeWatchResyncInterval, conf.WatchCrds, conf.CrdRefreshInterval, conf.ApiServerHost, kubeContext, conf.EnableGranularMetrics, conf.ExclusionRules)
 		if err != nil {
 			return errors.Wrap(err, "failed to initialize kubeWatcher")
 		}