Add the ability to set the insertion filter into CLI (#843)

* Add the ability to set the insertion filter into CLI

* Resolve the conflict

* Upgrade Basenine to `v0.5.0` and check the insertion filter against syntax errors inside CLI

* Upgrade Basenine to `v0.5.1`

* Update the sums

.gitignore

@@ -30,7 +30,7 @@ build
 pprof/*
 
 # Database Files
-*.bin
+*.db
 *.gob
 
 # Nohup Files - https://man7.org/linux/man-pages/man1/nohup.1p.html

Dockerfile

@@ -78,8 +78,8 @@ RUN go build -ldflags="-extldflags=-static -s -w \
     -X 'github.com/up9inc/mizu/agent/pkg/version.Ver=${VER}'" -o mizuagent .
 
 # Download Basenine executable, verify the sha1sum
-ADD https://github.com/up9inc/basenine/releases/download/v0.4.17/basenine_linux_${GOARCH} ./basenine_linux_${GOARCH}
-ADD https://github.com/up9inc/basenine/releases/download/v0.4.17/basenine_linux_${GOARCH}.sha256 ./basenine_linux_${GOARCH}.sha256
+ADD https://github.com/up9inc/basenine/releases/download/v0.5.1/basenine_linux_${GOARCH} ./basenine_linux_${GOARCH}
+ADD https://github.com/up9inc/basenine/releases/download/v0.5.1/basenine_linux_${GOARCH}.sha256 ./basenine_linux_${GOARCH}.sha256
 RUN shasum -a 256 -c basenine_linux_${GOARCH}.sha256
 RUN chmod +x ./basenine_linux_${GOARCH}
 RUN mv ./basenine_linux_${GOARCH} ./basenine

acceptanceTests/extensions_test.go

@@ -40,7 +40,7 @@ func TestRedis(t *testing.T) {
 		return
 	}
 
-	apiServerUrl := GetApiServerUrl(defaultApiServerPort)
+	apiServerUrl := GetApiServerUrl(DefaultApiServerPort)
 
 	if err := WaitTapPodsReady(apiServerUrl); err != nil {
 		t.Errorf("failed to start tap pods on time, err: %v", err)
@@ -55,7 +55,7 @@ func TestRedis(t *testing.T) {
 		return
 	}
 
-	redisExternalIp, err := kubernetesProvider.GetServiceExternalIp(ctx, defaultNamespaceName, "redis")
+	redisExternalIp, err := kubernetesProvider.GetServiceExternalIp(ctx, DefaultNamespaceName, "redis")
 	if err != nil {
 		t.Errorf("failed to get redis external ip, err: %v", err)
 		return
@@ -65,7 +65,7 @@ func TestRedis(t *testing.T) {
 		Addr: fmt.Sprintf("%v:6379", redisExternalIp),
 	})
 
-	for i := 0; i < defaultEntriesCount/5; i++ {
+	for i := 0; i < DefaultEntriesCount/5; i++ {
 		requestErr := rdb.Ping(ctx).Err()
 		if requestErr != nil {
 			t.Errorf("failed to send redis request, err: %v", requestErr)
@@ -73,7 +73,7 @@ func TestRedis(t *testing.T) {
 		}
 	}
 
-	for i := 0; i < defaultEntriesCount/5; i++ {
+	for i := 0; i < DefaultEntriesCount/5; i++ {
 		requestErr := rdb.Set(ctx, "key", "value", -1).Err()
 		if requestErr != nil {
 			t.Errorf("failed to send redis request, err: %v", requestErr)
@@ -81,7 +81,7 @@ func TestRedis(t *testing.T) {
 		}
 	}
 
-	for i := 0; i < defaultEntriesCount/5; i++ {
+	for i := 0; i < DefaultEntriesCount/5; i++ {
 		requestErr := rdb.Exists(ctx, "key").Err()
 		if requestErr != nil {
 			t.Errorf("failed to send redis request, err: %v", requestErr)
@@ -89,7 +89,7 @@ func TestRedis(t *testing.T) {
 		}
 	}
 
-	for i := 0; i < defaultEntriesCount/5; i++ {
+	for i := 0; i < DefaultEntriesCount/5; i++ {
 		requestErr := rdb.Get(ctx, "key").Err()
 		if requestErr != nil {
 			t.Errorf("failed to send redis request, err: %v", requestErr)
@@ -97,7 +97,7 @@ func TestRedis(t *testing.T) {
 		}
 	}
 
-	for i := 0; i < defaultEntriesCount/5; i++ {
+	for i := 0; i < DefaultEntriesCount/5; i++ {
 		requestErr := rdb.Del(ctx, "key").Err()
 		if requestErr != nil {
 			t.Errorf("failed to send redis request, err: %v", requestErr)
@@ -138,7 +138,7 @@ func TestAmqp(t *testing.T) {
 		return
 	}
 
-	apiServerUrl := GetApiServerUrl(defaultApiServerPort)
+	apiServerUrl := GetApiServerUrl(DefaultApiServerPort)
 
 	if err := WaitTapPodsReady(apiServerUrl); err != nil {
 		t.Errorf("failed to start tap pods on time, err: %v", err)
@@ -153,7 +153,7 @@ func TestAmqp(t *testing.T) {
 		return
 	}
 
-	rabbitmqExternalIp, err := kubernetesProvider.GetServiceExternalIp(ctx, defaultNamespaceName, "rabbitmq")
+	rabbitmqExternalIp, err := kubernetesProvider.GetServiceExternalIp(ctx, DefaultNamespaceName, "rabbitmq")
 	if err != nil {
 		t.Errorf("failed to get RabbitMQ external ip, err: %v", err)
 		return
@@ -169,7 +169,7 @@ func TestAmqp(t *testing.T) {
 	// Temporary fix for missing amqp entries
 	time.Sleep(10 * time.Second)
 
-	for i := 0; i < defaultEntriesCount/5; i++ {
+	for i := 0; i < DefaultEntriesCount/5; i++ {
 		ch, err := conn.Channel()
 		if err != nil {
 			t.Errorf("failed to open a channel, err: %v", err)

acceptanceTests/logs_test.go

@@ -36,7 +36,7 @@ func TestLogs(t *testing.T) {
 		return
 	}
 
-	apiServerUrl := GetApiServerUrl(defaultApiServerPort)
+	apiServerUrl := GetApiServerUrl(DefaultApiServerPort)
 
 	if err := WaitTapPodsReady(apiServerUrl); err != nil {
 		t.Errorf("failed to start tap pods on time, err: %v", err)
@@ -137,7 +137,7 @@ func TestLogsPath(t *testing.T) {
 		return
 	}
 
-	apiServerUrl := GetApiServerUrl(defaultApiServerPort)
+	apiServerUrl := GetApiServerUrl(DefaultApiServerPort)
 
 	if err := WaitTapPodsReady(apiServerUrl); err != nil {
 		t.Errorf("failed to start tap pods on time, err: %v", err)

acceptanceTests/tap_test.go

@@ -53,14 +53,14 @@ func basicTapTest(t *testing.T, shouldCheckSrcAndDest bool, extraArgs... string)
 				return
 			}
 
-			apiServerUrl := GetApiServerUrl(defaultApiServerPort)
+			apiServerUrl := GetApiServerUrl(DefaultApiServerPort)
 
 			if err := WaitTapPodsReady(apiServerUrl); err != nil {
 				t.Errorf("failed to start tap pods on time, err: %v", err)
 				return
 			}
 
-			proxyUrl := GetProxyUrl(defaultNamespaceName, defaultServiceName)
+			proxyUrl := GetProxyUrl(DefaultNamespaceName, DefaultServiceName)
 			for i := 0; i < entriesCount; i++ {
 				if _, requestErr := ExecuteHttpGetRequest(fmt.Sprintf("%v/get", proxyUrl)); requestErr != nil {
 					t.Errorf("failed to send proxy request, err: %v", requestErr)
@@ -127,8 +127,8 @@ func TestTapGuiPort(t *testing.T) {
 				return
 			}
 
-			proxyUrl := GetProxyUrl(defaultNamespaceName, defaultServiceName)
-			for i := 0; i < defaultEntriesCount; i++ {
+			proxyUrl := GetProxyUrl(DefaultNamespaceName, DefaultServiceName)
+			for i := 0; i < DefaultEntriesCount; i++ {
 				if _, requestErr := ExecuteHttpGetRequest(fmt.Sprintf("%v/get", proxyUrl)); requestErr != nil {
 					t.Errorf("failed to send proxy request, err: %v", requestErr)
 					return
@@ -175,7 +175,7 @@ func TestTapAllNamespaces(t *testing.T) {
 		return
 	}
 
-	apiServerUrl := GetApiServerUrl(defaultApiServerPort)
+	apiServerUrl := GetApiServerUrl(DefaultApiServerPort)
 
 	if err := WaitTapPodsReady(apiServerUrl); err != nil {
 		t.Errorf("failed to start tap pods on time, err: %v", err)
@@ -224,7 +224,7 @@ func TestTapMultipleNamespaces(t *testing.T) {
 		return
 	}
 
-	apiServerUrl := GetApiServerUrl(defaultApiServerPort)
+	apiServerUrl := GetApiServerUrl(DefaultApiServerPort)
 
 	if err := WaitTapPodsReady(apiServerUrl); err != nil {
 		t.Errorf("failed to start tap pods on time, err: %v", err)
@@ -270,7 +270,7 @@ func TestTapRegex(t *testing.T) {
 		return
 	}
 
-	apiServerUrl := GetApiServerUrl(defaultApiServerPort)
+	apiServerUrl := GetApiServerUrl(DefaultApiServerPort)
 
 	if err := WaitTapPodsReady(apiServerUrl); err != nil {
 		t.Errorf("failed to start tap pods on time, err: %v", err)
@@ -318,7 +318,7 @@ func TestTapDryRun(t *testing.T) {
 	}()
 
 	go func() {
-		time.Sleep(shortRetriesCount * time.Second)
+		time.Sleep(ShortRetriesCount * time.Second)
 		resultChannel <- "fail"
 	}()
 
@@ -358,17 +358,17 @@ func TestTapRedact(t *testing.T) {
 		return
 	}
 
-	apiServerUrl := GetApiServerUrl(defaultApiServerPort)
+	apiServerUrl := GetApiServerUrl(DefaultApiServerPort)
 
 	if err := WaitTapPodsReady(apiServerUrl); err != nil {
 		t.Errorf("failed to start tap pods on time, err: %v", err)
 		return
 	}
 
-	proxyUrl := GetProxyUrl(defaultNamespaceName, defaultServiceName)
+	proxyUrl := GetProxyUrl(DefaultNamespaceName, DefaultServiceName)
 	requestHeaders := map[string]string{"User-Header": "Mizu"}
 	requestBody := map[string]string{"User": "Mizu"}
-	for i := 0; i < defaultEntriesCount; i++ {
+	for i := 0; i < DefaultEntriesCount; i++ {
 		if _, requestErr := ExecuteHttpPostRequestWithHeaders(fmt.Sprintf("%v/post", proxyUrl), requestHeaders, requestBody); requestErr != nil {
 			t.Errorf("failed to send proxy request, err: %v", requestErr)
 			return
@@ -410,17 +410,17 @@ func TestTapNoRedact(t *testing.T) {
 		return
 	}
 
-	apiServerUrl := GetApiServerUrl(defaultApiServerPort)
+	apiServerUrl := GetApiServerUrl(DefaultApiServerPort)
 
 	if err := WaitTapPodsReady(apiServerUrl); err != nil {
 		t.Errorf("failed to start tap pods on time, err: %v", err)
 		return
 	}
 
-	proxyUrl := GetProxyUrl(defaultNamespaceName, defaultServiceName)
+	proxyUrl := GetProxyUrl(DefaultNamespaceName, DefaultServiceName)
 	requestHeaders := map[string]string{"User-Header": "Mizu"}
 	requestBody := map[string]string{"User": "Mizu"}
-	for i := 0; i < defaultEntriesCount; i++ {
+	for i := 0; i < DefaultEntriesCount; i++ {
 		if _, requestErr := ExecuteHttpPostRequestWithHeaders(fmt.Sprintf("%v/post", proxyUrl), requestHeaders, requestBody); requestErr != nil {
 			t.Errorf("failed to send proxy request, err: %v", requestErr)
 			return
@@ -462,15 +462,15 @@ func TestTapRegexMasking(t *testing.T) {
 		return
 	}
 
-	apiServerUrl := GetApiServerUrl(defaultApiServerPort)
+	apiServerUrl := GetApiServerUrl(DefaultApiServerPort)
 
 	if err := WaitTapPodsReady(apiServerUrl); err != nil {
 		t.Errorf("failed to start tap pods on time, err: %v", err)
 		return
 	}
 
-	proxyUrl := GetProxyUrl(defaultNamespaceName, defaultServiceName)
-	for i := 0; i < defaultEntriesCount; i++ {
+	proxyUrl := GetProxyUrl(DefaultNamespaceName, DefaultServiceName)
+	for i := 0; i < DefaultEntriesCount; i++ {
 		response, requestErr := http.Post(fmt.Sprintf("%v/post", proxyUrl), "text/plain", bytes.NewBufferString("Mizu"))
 		if _, requestErr = ExecuteHttpRequest(response, requestErr); requestErr != nil {
 			t.Errorf("failed to send proxy request, err: %v", requestErr)
@@ -515,25 +515,25 @@ func TestTapIgnoredUserAgents(t *testing.T) {
 		return
 	}
 
-	apiServerUrl := GetApiServerUrl(defaultApiServerPort)
+	apiServerUrl := GetApiServerUrl(DefaultApiServerPort)
 
 	if err := WaitTapPodsReady(apiServerUrl); err != nil {
 		t.Errorf("failed to start tap pods on time, err: %v", err)
 		return
 	}
 
-	proxyUrl := GetProxyUrl(defaultNamespaceName, defaultServiceName)
+	proxyUrl := GetProxyUrl(DefaultNamespaceName, DefaultServiceName)
 
 	ignoredUserAgentCustomHeader := "Ignored-User-Agent"
 	headers := map[string]string{"User-Agent": ignoredUserAgentValue, ignoredUserAgentCustomHeader: ""}
-	for i := 0; i < defaultEntriesCount; i++ {
+	for i := 0; i < DefaultEntriesCount; i++ {
 		if _, requestErr := ExecuteHttpGetRequestWithHeaders(fmt.Sprintf("%v/get", proxyUrl), headers); requestErr != nil {
 			t.Errorf("failed to send proxy request, err: %v", requestErr)
 			return
 		}
 	}
 
-	for i := 0; i < defaultEntriesCount; i++ {
+	for i := 0; i < DefaultEntriesCount; i++ {
 		if _, requestErr := ExecuteHttpGetRequest(fmt.Sprintf("%v/get", proxyUrl)); requestErr != nil {
 			t.Errorf("failed to send proxy request, err: %v", requestErr)
 			return
@@ -569,7 +569,7 @@ func TestTapDumpLogs(t *testing.T) {
 		return
 	}
 
-	apiServerUrl := GetApiServerUrl(defaultApiServerPort)
+	apiServerUrl := GetApiServerUrl(DefaultApiServerPort)
 
 	if err := WaitTapPodsReady(apiServerUrl); err != nil {
 		t.Errorf("failed to start tap pods on time, err: %v", err)
@@ -651,7 +651,7 @@ func TestTapDumpLogs(t *testing.T) {
 }
 
 func TestIpResolving(t *testing.T) {
-	namespace := allNamespaces
+	namespace := AllNamespaces
 
 	t.Log("add permissions for ip-resolution for current user")
 	if err := ApplyKubeFilesForTest(

acceptanceTests/testsUtils.go

@@ -24,14 +24,14 @@ import (
 )
 
 const (
-	longRetriesCount      = 100
-	shortRetriesCount     = 10
-	defaultApiServerPort  = shared.DefaultApiServerPort
-	defaultNamespaceName  = "mizu-tests"
-	defaultServiceName    = "httpbin"
-	defaultEntriesCount   = 50
-	waitAfterTapPodsReady = 3 * time.Second
-	allNamespaces         = ""
+	LongRetriesCount      = 100
+	ShortRetriesCount     = 10
+	DefaultApiServerPort  = shared.DefaultApiServerPort
+	DefaultNamespaceName  = "mizu-tests"
+	DefaultServiceName    = "httpbin"
+	DefaultEntriesCount   = 50
+	WaitAfterTapPodsReady = 3 * time.Second
+	AllNamespaces         = ""
 )
 
 type PodDescriptor struct {
@@ -181,7 +181,7 @@ func ApplyKubeFile(kubeContext string, namespace string, filename string) (error
 		"--context", kubeContext,
 		"-f", filename,
 	}
-	if namespace != allNamespaces {
+	if namespace != AllNamespaces {
 		cmdArgs = append(cmdArgs, "-n", namespace)
 	}
 	cmd := exec.Command("kubectl", cmdArgs...)
@@ -199,7 +199,7 @@ func DeleteKubeFile(kubeContext string, namespace string, filename string) error
 		"--context", kubeContext,
 		"-f", filename,
 	}
-	if namespace != allNamespaces {
+	if namespace != AllNamespaces {
 		cmdArgs = append(cmdArgs, "-n", namespace)
 	}
 	cmd := exec.Command("kubectl", cmdArgs...)
@@ -304,11 +304,11 @@ func WaitTapPodsReady(apiServerUrl string) error {
 		if connectedTappersCount == 0 {
 			return fmt.Errorf("no connected tappers running")
 		}
-		time.Sleep(waitAfterTapPodsReady)
+		time.Sleep(WaitAfterTapPodsReady)
 		return nil
 	}
 
-	return retriesExecute(longRetriesCount, tapPodsReadyFunc)
+	return retriesExecute(LongRetriesCount, tapPodsReadyFunc)
 }
 
 func jsonBytesToInterface(jsonBytes []byte) (interface{}, error) {

agent/go.mod

@@ -22,7 +22,7 @@ require (
 	github.com/ory/kratos-client-go v0.8.2-alpha.1
 	github.com/patrickmn/go-cache v2.1.0+incompatible
 	github.com/stretchr/testify v1.7.0
-	github.com/up9inc/basenine/client/go v0.0.0-20220220204122-0ef8cb24fab1
+	github.com/up9inc/basenine/client/go v0.0.0-20220302073458-c32e0adf1500
 	github.com/up9inc/mizu/shared v0.0.0
 	github.com/up9inc/mizu/tap v0.0.0
 	github.com/up9inc/mizu/tap/api v0.0.0

agent/go.sum

@@ -855,6 +855,10 @@ github.com/ugorji/go/codec v1.2.6 h1:7kbGefxLoDBuYXOms4yD7223OpNMMPNPZxXk5TvFcyQ
 github.com/ugorji/go/codec v1.2.6/go.mod h1:V6TCNZ4PHqoHGFZuSG1W8nrCzzdgA2DozYxWFFpvxTw=
 github.com/up9inc/basenine/client/go v0.0.0-20220220204122-0ef8cb24fab1 h1:0XN8s3HtwUBr9hbWRAFulFMsu1f2cabfJbwpz/sOoLA=
 github.com/up9inc/basenine/client/go v0.0.0-20220220204122-0ef8cb24fab1/go.mod h1:SvJGPoa/6erhUQV7kvHBwM/0x5LyO6XaG2lUaCaKiUI=
+github.com/up9inc/basenine/client/go v0.0.0-20220301135911-d2111357b14e h1:nv/A/AeF8PcU91aHAj6o2cU8fl/46v0ZLj7wgIKjv+o=
+github.com/up9inc/basenine/client/go v0.0.0-20220301135911-d2111357b14e/go.mod h1:SvJGPoa/6erhUQV7kvHBwM/0x5LyO6XaG2lUaCaKiUI=
+github.com/up9inc/basenine/client/go v0.0.0-20220302073458-c32e0adf1500 h1:T1QHxt65NMete/GobVSvcHnwZAQibvahhrMTCgtnSS4=
+github.com/up9inc/basenine/client/go v0.0.0-20220302073458-c32e0adf1500/go.mod h1:SvJGPoa/6erhUQV7kvHBwM/0x5LyO6XaG2lUaCaKiUI=
 github.com/vektah/gqlparser v1.1.2/go.mod h1:1ycwN7Ij5njmMkPPAOaRFY4rET2Enx7IkVv3vaXspKw=
 github.com/vishvananda/netns v0.0.0-20211101163701-50045581ed74 h1:gga7acRE695APm9hlsSMoOoE65U4/TcqNj90mc69Rlg=
 github.com/vishvananda/netns v0.0.0-20211101163701-50045581ed74/go.mod h1:DD4vA1DwXk04H54A1oHXtwZmA0grkVMdPxx/VGLCah0=

agent/main.go

@@ -140,7 +140,7 @@ func runInApiServerMode(namespace string) *gin.Engine {
 	if err := config.LoadConfig(); err != nil {
 		logger.Log.Fatalf("Error loading config file %v", err)
 	}
-	app.ConfigureBasenineServer(shared.BasenineHost, shared.BaseninePort, config.Config.MaxDBSizeBytes, config.Config.LogLevel)
+	app.ConfigureBasenineServer(shared.BasenineHost, shared.BaseninePort, config.Config.MaxDBSizeBytes, config.Config.LogLevel, config.Config.InsertionFilter)
 	startTime = time.Now().UnixNano() / int64(time.Millisecond)
 	api.StartResolving(namespace)
 

agent/pkg/app/main.go

@@ -62,7 +62,7 @@ func LoadExtensions() {
 	controllers.InitExtensionsMap(ExtensionsMap)
 }
 
-func ConfigureBasenineServer(host string, port string, dbSize int64, logLevel logging.Level) {
+func ConfigureBasenineServer(host string, port string, dbSize int64, logLevel logging.Level, insertionFilter string) {
 	if !wait.New(
 		wait.WithProto("tcp"),
 		wait.WithWait(200*time.Millisecond),
@@ -86,6 +86,11 @@ func ConfigureBasenineServer(host string, port string, dbSize int64, logLevel lo
 			}
 		}
 	}
+
+	// Set the insertion filter that comes from the config
+	if err := basenine.InsertionFilter(host, port, insertionFilter); err != nil {
+		logger.Log.Errorf("Error while setting the insertion filter: %v", err)
+	}
 }
 
 func GetEntryInputChannel() chan *tapApi.OutputChannelItem {

agent/pkg/elastic/esClient.go

@@ -4,13 +4,14 @@ import (
 	"bytes"
 	"crypto/tls"
 	"encoding/json"
+	"net/http"
+	"sync"
+	"time"
+
 	"github.com/elastic/go-elasticsearch/v7"
 	"github.com/up9inc/mizu/shared"
 	"github.com/up9inc/mizu/shared/logger"
 	"github.com/up9inc/mizu/tap/api"
-	"net/http"
-	"sync"
-	"time"
 )
 
 type client struct {
@@ -31,6 +32,9 @@ func GetInstance() *client {
 
 func (client *client) Configure(config shared.ElasticConfig) {
 	if config.Url == "" || config.User == "" || config.Password == "" {
+		if client.es != nil {
+			client.es = nil
+		}
 		logger.Log.Infof("No elastic configuration was supplied, elastic exporter disabled")
 		return
 	}
@@ -46,13 +50,13 @@ func (client *client) Configure(config shared.ElasticConfig) {
 
 	es, err := elasticsearch.NewClient(cfg)
 	if err != nil {
-		logger.Log.Fatalf("Failed to initialize elastic client %v", err)
+		logger.Log.Errorf("Failed to initialize elastic client %v", err)
 	}
 
 	// Have the client instance return a response
 	res, err := es.Info()
 	if err != nil {
-		logger.Log.Fatalf("Elastic client.Info() ERROR: %v", err)
+		logger.Log.Errorf("Elastic client.Info() ERROR: %v", err)
 	} else {
 		client.es = es
 		client.index = "mizu_traffic_http_" + time.Now().Format("2006_01_02_15_04")

agent/pkg/oas/oas_generator.go

@@ -33,10 +33,23 @@ func (g *oasGenerator) Start() {
 	g.entriesChan = make(chan EntryWithSource, 100) // buffer up to 100 entries for OAS processing
 	g.ServiceSpecs = &sync.Map{}
 	g.started = true
-	go instance.runGeneretor()
+	go instance.runGenerator()
 }
 
-func (g *oasGenerator) runGeneretor() {
+func (g *oasGenerator) Stop() {
+	if !g.started {
+		return
+	}
+	g.cancel()
+	g.Reset()
+	g.started = false
+}
+
+func (g *oasGenerator) IsStarted() bool {
+	return g.started
+}
+
+func (g *oasGenerator) runGenerator() {
 	for {
 		select {
 		case <-g.ctx.Done():

agent/pkg/servicemap/servicemap.go

@@ -32,6 +32,7 @@ type serviceMap struct {
 
 type ServiceMap interface {
 	Enable()
+	Disable()
 	IsEnabled() bool
 	NewTCPEntry(source *tapApi.TCP, destination *tapApi.TCP, protocol *tapApi.Protocol)
 	GetStatus() ServiceMapStatus
@@ -159,6 +160,11 @@ func (s *serviceMap) Enable() {
 	s.enabled = true
 }
 
+func (s *serviceMap) Disable() {
+	s.Reset()
+	s.enabled = false
+}
+
 func (s *serviceMap) IsEnabled() bool {
 	return s.enabled
 }

cli/cmd/tap.go

@@ -3,9 +3,10 @@ package cmd
 import (
 	"errors"
 	"fmt"
-	"github.com/up9inc/mizu/cli/up9"
 	"os"
 
+	"github.com/up9inc/mizu/cli/up9"
+
 	"github.com/creasty/defaults"
 	"github.com/spf13/cobra"
 	"github.com/up9inc/mizu/cli/auth"
@@ -115,6 +116,7 @@ func init() {
 	tapCmd.Flags().StringSliceP(configStructs.PlainTextFilterRegexesTapName, "r", defaultTapConfig.PlainTextFilterRegexes, "List of regex expressions that are used to filter matching values from text/plain http bodies")
 	tapCmd.Flags().Bool(configStructs.DisableRedactionTapName, defaultTapConfig.DisableRedaction, "Disables redaction of potentially sensitive request/response headers and body values")
 	tapCmd.Flags().String(configStructs.HumanMaxEntriesDBSizeTapName, defaultTapConfig.HumanMaxEntriesDBSize, "Override the default max entries db size")
+	tapCmd.Flags().String(configStructs.InsertionFilterName, defaultTapConfig.InsertionFilter, "Set the insertion filter. Accepts string or a file path.")
 	tapCmd.Flags().Bool(configStructs.DryRunTapName, defaultTapConfig.DryRun, "Preview of all pods matching the regex, without tapping them")
 	tapCmd.Flags().StringP(configStructs.WorkspaceTapName, "w", defaultTapConfig.Workspace, "Uploads traffic to your UP9 workspace for further analysis (requires auth)")
 	tapCmd.Flags().String(configStructs.EnforcePolicyFile, defaultTapConfig.EnforcePolicyFile, "Yaml file path with policy rules")

cli/cmd/tapRunner.go

@@ -154,6 +154,7 @@ func finishTapExecution(kubernetesProvider *kubernetes.Provider) {
 func getTapMizuAgentConfig() *shared.MizuAgentConfig {
 	mizuAgentConfig := shared.MizuAgentConfig{
 		MaxDBSizeBytes:         config.Config.Tap.MaxEntriesDBSizeBytes(),
+		InsertionFilter:        config.Config.Tap.GetInsertionFilter(),
 		AgentImage:             config.Config.AgentImage,
 		PullPolicy:             config.Config.ImagePullPolicyStr,
 		LogLevel:               config.Config.LogLevel(),

cli/config/configStructs/tapConfig.go

@@ -3,10 +3,16 @@ package configStructs
 import (
 	"errors"
 	"fmt"
+	"io/fs"
+	"io/ioutil"
+	"os"
 	"regexp"
 
+	"github.com/up9inc/mizu/cli/uiUtils"
 	"github.com/up9inc/mizu/shared"
 
+	basenine "github.com/up9inc/basenine/server/lib"
+	"github.com/up9inc/mizu/shared/logger"
 	"github.com/up9inc/mizu/shared/units"
 )
 
@@ -18,6 +24,7 @@ const (
 	PlainTextFilterRegexesTapName = "regex-masking"
 	DisableRedactionTapName       = "no-redact"
 	HumanMaxEntriesDBSizeTapName  = "max-entries-db-size"
+	InsertionFilterName           = "insertion-filter"
 	DryRunTapName                 = "dry-run"
 	WorkspaceTapName              = "workspace"
 	EnforcePolicyFile             = "traffic-validation-file"
@@ -27,26 +34,27 @@ const (
 )
 
 type TapConfig struct {
-	UploadIntervalSec       int              `yaml:"upload-interval" default:"10"`
-	PodRegexStr             string           `yaml:"regex" default:".*"`
-	GuiPort                 uint16           `yaml:"gui-port" default:"8899"`
-	ProxyHost               string           `yaml:"proxy-host" default:"127.0.0.1"`
-	Namespaces              []string         `yaml:"namespaces"`
-	Analysis                bool             `yaml:"analysis" default:"false"`
-	AllNamespaces           bool             `yaml:"all-namespaces" default:"false"`
-	PlainTextFilterRegexes  []string         `yaml:"regex-masking"`
-	IgnoredUserAgents       []string         `yaml:"ignored-user-agents"`
-	DisableRedaction        bool             `yaml:"no-redact" default:"false"`
-	HumanMaxEntriesDBSize   string           `yaml:"max-entries-db-size" default:"200MB"`
-	DryRun                  bool             `yaml:"dry-run" default:"false"`
-	Workspace               string           `yaml:"workspace"`
-	EnforcePolicyFile       string           `yaml:"traffic-validation-file"`
-	ContractFile            string           `yaml:"contract"`
-	AskUploadConfirmation   bool             `yaml:"ask-upload-confirmation" default:"true"`
-	ApiServerResources      shared.Resources `yaml:"api-server-resources"`
-	TapperResources         shared.Resources `yaml:"tapper-resources"`
-	ServiceMesh             bool             `yaml:"service-mesh" default:"false"`
-	Tls                     bool             `yaml:"tls" default:"false"`
+	UploadIntervalSec      int              `yaml:"upload-interval" default:"10"`
+	PodRegexStr            string           `yaml:"regex" default:".*"`
+	GuiPort                uint16           `yaml:"gui-port" default:"8899"`
+	ProxyHost              string           `yaml:"proxy-host" default:"127.0.0.1"`
+	Namespaces             []string         `yaml:"namespaces"`
+	Analysis               bool             `yaml:"analysis" default:"false"`
+	AllNamespaces          bool             `yaml:"all-namespaces" default:"false"`
+	PlainTextFilterRegexes []string         `yaml:"regex-masking"`
+	IgnoredUserAgents      []string         `yaml:"ignored-user-agents"`
+	DisableRedaction       bool             `yaml:"no-redact" default:"false"`
+	HumanMaxEntriesDBSize  string           `yaml:"max-entries-db-size" default:"200MB"`
+	InsertionFilter        string           `yaml:"insertion-filter" default:""`
+	DryRun                 bool             `yaml:"dry-run" default:"false"`
+	Workspace              string           `yaml:"workspace"`
+	EnforcePolicyFile      string           `yaml:"traffic-validation-file"`
+	ContractFile           string           `yaml:"contract"`
+	AskUploadConfirmation  bool             `yaml:"ask-upload-confirmation" default:"true"`
+	ApiServerResources     shared.Resources `yaml:"api-server-resources"`
+	TapperResources        shared.Resources `yaml:"tapper-resources"`
+	ServiceMesh            bool             `yaml:"service-mesh" default:"false"`
+	Tls                    bool             `yaml:"tls" default:"false"`
 }
 
 func (config *TapConfig) PodRegex() *regexp.Regexp {
@@ -59,6 +67,25 @@ func (config *TapConfig) MaxEntriesDBSizeBytes() int64 {
 	return maxEntriesDBSizeBytes
 }
 
+func (config *TapConfig) GetInsertionFilter() string {
+	insertionFilter := config.InsertionFilter
+	if fs.ValidPath(insertionFilter) {
+		if _, err := os.Stat(insertionFilter); err == nil {
+			b, err := ioutil.ReadFile(insertionFilter)
+			if err != nil {
+				logger.Log.Warningf(uiUtils.Warning, fmt.Sprintf("Couldn't read the file on path: %s, err: %v", insertionFilter, err))
+			} else {
+				insertionFilter = string(b)
+			}
+		}
+	}
+	_, err := basenine.Parse(insertionFilter)
+	if err != nil {
+		logger.Log.Warningf(uiUtils.Warning, fmt.Sprintf("Insertion filter syntax error: %v", err))
+	}
+	return insertionFilter
+}
+
 func (config *TapConfig) Validate() error {
 	_, compileErr := regexp.Compile(config.PodRegexStr)
 	if compileErr != nil {

cli/go.mod

@@ -11,6 +11,7 @@ require (
 	github.com/op/go-logging v0.0.0-20160315200505-970db520ece7
 	github.com/spf13/cobra v1.3.0
 	github.com/spf13/pflag v1.0.5
+	github.com/up9inc/basenine/server/lib v0.0.0-20220302073458-c32e0adf1500
 	github.com/up9inc/mizu/shared v0.0.0
 	github.com/up9inc/mizu/tap/api v0.0.0
 	golang.org/x/oauth2 v0.0.0-20211104180415-d3ed0bb246c8
@@ -32,8 +33,10 @@ require (
 	github.com/MakeNowJust/heredoc v1.0.0 // indirect
 	github.com/PuerkitoBio/purell v1.1.1 // indirect
 	github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578 // indirect
+	github.com/alecthomas/participle/v2 v2.0.0-alpha7 // indirect
 	github.com/chai2010/gettext-go v0.0.0-20160711120539-c6fed771bfd5 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/dlclark/regexp2 v1.4.0 // indirect
 	github.com/docker/go-units v0.4.0 // indirect
 	github.com/evanphx/json-patch v5.6.0+incompatible // indirect
 	github.com/exponent-io/jsonpath v0.0.0-20210407135951-1de76d718b3f // indirect
@@ -68,6 +71,7 @@ require (
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/monochromegane/go-gitignore v0.0.0-20200626010858-205db1a8cc00 // indirect
 	github.com/mxk/go-flowrate v0.0.0-20140419014527-cca7078d478f // indirect
+	github.com/ohler55/ojg v1.12.13 // indirect
 	github.com/peterbourgon/diskv v2.0.1+incompatible // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect

cli/go.sum

@@ -83,6 +83,10 @@ github.com/PuerkitoBio/purell v1.1.1 h1:WEQqlqaGbrPkxLJWfBwQmfEAE1Z7ONdDLqrN38tN
 github.com/PuerkitoBio/purell v1.1.1/go.mod h1:c11w/QuzBsJSee3cPx9rAFu61PvFxuPbtSwDGJws/X0=
 github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578 h1:d+Bc7a5rLufV/sSk/8dngufqelfh6jnri85riMAaF/M=
 github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578/go.mod h1:uGdkoq3SwY9Y+13GIhn11/XLaGBb4BfwItxLd5jeuXE=
+github.com/alecthomas/participle/v2 v2.0.0-alpha7 h1:cK4vjj0VSgb3lN1nuKA5F7dw+1s1pWBe5bx7nNCnN+c=
+github.com/alecthomas/participle/v2 v2.0.0-alpha7/go.mod h1:NumScqsC42o9x+dGj8/YqsIfhrIQjFEOFovxotbBirA=
+github.com/alecthomas/repr v0.0.0-20181024024818-d37bc2a10ba1 h1:GDQdwm/gAcJcLAKQQZGOJ4knlw+7rfEQQcmwTbt4p5E=
+github.com/alecthomas/repr v0.0.0-20181024024818-d37bc2a10ba1/go.mod h1:xTS7Pm1pD1mvyM075QCDSRqH6qRLXylzS24ZTpRiSzQ=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
@@ -149,6 +153,8 @@ github.com/denisbrodbeck/machineid v1.0.1 h1:geKr9qtkB876mXguW2X6TU4ZynleN6ezuMS
 github.com/denisbrodbeck/machineid v1.0.1/go.mod h1:dJUwb7PTidGDeYyUBmXZ2GphQBbjJCrnectwCyxcUSI=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
 github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954/go.mod h1:vAd38F8PWV+bWy6jNmig1y/TA+kYO4g3RSRF0IAv0no=
+github.com/dlclark/regexp2 v1.4.0 h1:F1rxgk7p4uKjwIQxBs9oAXe5CqrXlCduYEJvrF4u93E=
+github.com/dlclark/regexp2 v1.4.0/go.mod h1:2pZnwuY/m+8K6iRw6wQdMtk+rH5tNGR1i55kozfMjCc=
 github.com/docker/distribution v2.7.1+incompatible/go.mod h1:J2gT2udsDAN96Uj4KfcMRqY0/ypR+oyYUYmja8H+y+w=
 github.com/docker/go-units v0.4.0 h1:3uh0PgVws3nIA0Q+MwDC8yjEPf9zjRfZZWXZYDct3Tw=
 github.com/docker/go-units v0.4.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
@@ -479,6 +485,8 @@ github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e h1:fD57ERR4JtEqsWb
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
 github.com/nxadm/tail v1.4.4 h1:DQuhQpB1tVlglWS2hLQ5OV6B5r8aGxSrPc5Qo6uTN78=
 github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A=
+github.com/ohler55/ojg v1.12.13 h1:FvfVpYzLgMraLcg3rrXiRXaihOP6fnzQNEU9YyZ/AmM=
+github.com/ohler55/ojg v1.12.13/go.mod h1:LBbIVRAgoFbYBXQhRhuEpaJIqq+goSO63/FQ+nyJU88=
 github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U=
 github.com/olekukonko/tablewriter v0.0.4/go.mod h1:zq6QwlOf5SlnkVbMSr5EoBv3636FWnp+qbPhuoO21uA=
 github.com/onsi/ginkgo v0.0.0-20170829012221-11459a886d9c/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
@@ -590,6 +598,10 @@ github.com/subosito/gotenv v1.2.0/go.mod h1:N0PQaV/YGNqwC0u51sEeR/aUtSLEXKX9iv69
 github.com/tmc/grpc-websocket-proxy v0.0.0-20190109142713-0ad062ec5ee5/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
 github.com/tv42/httpunix v0.0.0-20150427012821-b75d8614f926/go.mod h1:9ESjWnEqriFuLhtthL60Sar/7RFoluCcXsuvEwTV5KM=
 github.com/ugorji/go v1.1.4/go.mod h1:uQMGLiO92mf5W77hV/PUCpI3pbzQx3CRekS0kk+RGrc=
+github.com/up9inc/basenine/server/lib v0.0.0-20220301135911-d2111357b14e h1:kp+03sWT7n46jzm7g8YebjRxRrjZPXzMAeUce7vftTs=
+github.com/up9inc/basenine/server/lib v0.0.0-20220301135911-d2111357b14e/go.mod h1:R9bG4y/iq89jNC0xZ25uKDqenyKFTR3X9acGDOkKWSE=
+github.com/up9inc/basenine/server/lib v0.0.0-20220302073458-c32e0adf1500 h1:PT9v4JUsx1m4X0vj0E8bHOz6hQSzhbYtIq65eDRgHq4=
+github.com/up9inc/basenine/server/lib v0.0.0-20220302073458-c32e0adf1500/go.mod h1:R9bG4y/iq89jNC0xZ25uKDqenyKFTR3X9acGDOkKWSE=
 github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2/go.mod h1:UETIi67q53MR2AWcXfiuqkDkRtnGDLqkBTpCHuJHxtU=
 github.com/xlab/treeprint v0.0.0-20181112141820-a009c3971eca/go.mod h1:ce1O1j6UtZfjr22oyGxGLbauSBp2YVXpARAosm7dHBg=
 github.com/xlab/treeprint v1.1.0 h1:G/1DjNkPpfZCFt9CSh6b5/nY4VimlbHF3Rh4obvtzDk=

shared/models.go

@@ -35,6 +35,7 @@ type Resources struct {
 
 type MizuAgentConfig struct {
 	MaxDBSizeBytes         int64         `json:"maxDBSizeBytes"`
+	InsertionFilter        string        `json:"insertionFilter"`
 	AgentImage             string        `json:"agentImage"`
 	PullPolicy             string        `json:"pullPolicy"`
 	LogLevel               logging.Level `json:"logLevel"`