Fixing readme ignored-user-agents documentation (#288)

* Run acceptance tests on pull request

* Take `options.DisableRedaction` into account

* Log `defaultTapConfig`

* Pass the `SENSITIVE_DATA_FILTERING_OPTIONS` to tapper daemon set too

* Revert "Run acceptance tests on pull request"

This reverts commit ad79f1418f.

README.md

@@ -143,7 +143,7 @@ Setting `mizu-resources-namespace=mizu` resets Mizu to its default behavior
 User-agent filtering (like health checks) - can be configured using command-line options:
 
 ```shell
-$ mizu tap "^ca.*" --set ignored-user-agents=kube-probe --set ignored-user-agents=prometheus
+$ mizu tap "^ca.*" --set tap.ignored-user-agents=kube-probe --set tap.ignored-user-agents=prometheus
 +carts-66c77f5fbb-fq65r
 +catalogue-5f4cb7cf5-7zrmn
 Web interface is now available at http://localhost:8899

acceptanceTests/logs_test.go

@@ -0,0 +1,196 @@
+package acceptanceTests
+
+import (
+	"archive/zip"
+	"os/exec"
+	"testing"
+)
+
+func TestLogs(t *testing.T) {
+	if testing.Short() {
+		t.Skip("ignored acceptance test")
+	}
+
+	cliPath, cliPathErr := getCliPath()
+	if cliPathErr != nil {
+		t.Errorf("failed to get cli path, err: %v", cliPathErr)
+		return
+	}
+
+	tapCmdArgs := getDefaultTapCommandArgs()
+
+	tapNamespace := getDefaultTapNamespace()
+	tapCmdArgs = append(tapCmdArgs, tapNamespace...)
+
+	tapCmd := exec.Command(cliPath, tapCmdArgs...)
+	t.Logf("running command: %v", tapCmd.String())
+
+	t.Cleanup(func() {
+		if err := cleanupCommand(tapCmd); err != nil {
+			t.Logf("failed to cleanup tap command, err: %v", err)
+		}
+	})
+
+	if err := tapCmd.Start(); err != nil {
+		t.Errorf("failed to start tap command, err: %v", err)
+		return
+	}
+
+	apiServerUrl := getApiServerUrl(defaultApiServerPort)
+
+	if err := waitTapPodsReady(apiServerUrl); err != nil {
+		t.Errorf("failed to start tap pods on time, err: %v", err)
+		return
+	}
+
+	logsCmdArgs := getDefaultLogsCommandArgs()
+
+	logsCmd := exec.Command(cliPath, logsCmdArgs...)
+	t.Logf("running command: %v", logsCmd.String())
+
+	if err := logsCmd.Start(); err != nil {
+		t.Errorf("failed to start logs command, err: %v", err)
+		return
+	}
+
+	if err := logsCmd.Wait(); err != nil {
+		t.Errorf("failed to wait logs command, err: %v", err)
+		return
+	}
+
+	logsPath, logsPathErr := getLogsPath()
+	if logsPathErr != nil {
+		t.Errorf("failed to get logs path, err: %v", logsPathErr)
+		return
+	}
+
+	zipReader, zipError := zip.OpenReader(logsPath)
+	if zipError != nil {
+		t.Errorf("failed to get zip reader, err: %v", zipError)
+		return
+	}
+
+	t.Cleanup(func() {
+		if err := zipReader.Close(); err != nil {
+			t.Logf("failed to close zip reader, err: %v", err)
+		}
+	})
+
+	var logsFileNames []string
+	for _, file := range zipReader.File {
+		logsFileNames = append(logsFileNames, file.Name)
+	}
+
+	if !Contains(logsFileNames, "mizu.mizu-api-server.log") {
+		t.Errorf("api server logs not found")
+		return
+	}
+
+	if !Contains(logsFileNames, "mizu_cli.log") {
+		t.Errorf("cli logs not found")
+		return
+	}
+
+	if !Contains(logsFileNames, "mizu_events.log") {
+		t.Errorf("events logs not found")
+		return
+	}
+
+	if !ContainsPartOfValue(logsFileNames, "mizu.mizu-tapper-daemon-set") {
+		t.Errorf("tapper logs not found")
+		return
+	}
+}
+
+func TestLogsPath(t *testing.T) {
+	if testing.Short() {
+		t.Skip("ignored acceptance test")
+	}
+
+	cliPath, cliPathErr := getCliPath()
+	if cliPathErr != nil {
+		t.Errorf("failed to get cli path, err: %v", cliPathErr)
+		return
+	}
+
+	tapCmdArgs := getDefaultTapCommandArgs()
+
+	tapNamespace := getDefaultTapNamespace()
+	tapCmdArgs = append(tapCmdArgs, tapNamespace...)
+
+	tapCmd := exec.Command(cliPath, tapCmdArgs...)
+	t.Logf("running command: %v", tapCmd.String())
+
+	t.Cleanup(func() {
+		if err := cleanupCommand(tapCmd); err != nil {
+			t.Logf("failed to cleanup tap command, err: %v", err)
+		}
+	})
+
+	if err := tapCmd.Start(); err != nil {
+		t.Errorf("failed to start tap command, err: %v", err)
+		return
+	}
+
+	apiServerUrl := getApiServerUrl(defaultApiServerPort)
+
+	if err := waitTapPodsReady(apiServerUrl); err != nil {
+		t.Errorf("failed to start tap pods on time, err: %v", err)
+		return
+	}
+
+	logsCmdArgs := getDefaultLogsCommandArgs()
+
+	logsPath := "../logs.zip"
+	logsCmdArgs = append(logsCmdArgs, "-f", logsPath)
+
+	logsCmd := exec.Command(cliPath, logsCmdArgs...)
+	t.Logf("running command: %v", logsCmd.String())
+
+	if err := logsCmd.Start(); err != nil {
+		t.Errorf("failed to start logs command, err: %v", err)
+		return
+	}
+
+	if err := logsCmd.Wait(); err != nil {
+		t.Errorf("failed to wait logs command, err: %v", err)
+		return
+	}
+
+	zipReader, zipError := zip.OpenReader(logsPath)
+	if zipError != nil {
+		t.Errorf("failed to get zip reader, err: %v", zipError)
+		return
+	}
+
+	t.Cleanup(func() {
+		if err := zipReader.Close(); err != nil {
+			t.Logf("failed to close zip reader, err: %v", err)
+		}
+	})
+
+	var logsFileNames []string
+	for _, file := range zipReader.File {
+		logsFileNames = append(logsFileNames, file.Name)
+	}
+
+	if !Contains(logsFileNames, "mizu.mizu-api-server.log") {
+		t.Errorf("api server logs not found")
+		return
+	}
+
+	if !Contains(logsFileNames, "mizu_cli.log") {
+		t.Errorf("cli logs not found")
+		return
+	}
+
+	if !Contains(logsFileNames, "mizu_events.log") {
+		t.Errorf("events logs not found")
+		return
+	}
+
+	if !ContainsPartOfValue(logsFileNames, "mizu.mizu-tapper-daemon-set") {
+		t.Errorf("tapper logs not found")
+		return
+	}
+}

acceptanceTests/tap_test.go

@@ -1,11 +1,14 @@
 package acceptanceTests
 
 import (
+	"archive/zip"
 	"bytes"
 	"encoding/json"
 	"fmt"
+	"io/ioutil"
 	"net/http"
 	"os/exec"
+	"path"
 	"strings"
 	"testing"
 	"time"
@@ -502,10 +505,19 @@ func TestTapRedact(t *testing.T) {
 			return fmt.Errorf("failed to get entry, err: %v", requestErr)
 		}
 
-		entry := requestResult.(map[string]interface{})["entry"].(map[string]interface{})
-		entryRequest := entry["request"].(map[string]interface{})
+		data := requestResult.(map[string]interface{})["data"].(map[string]interface{})
+		entryJson := data["entry"].(string)
 
-		headers :=  entryRequest["headers"].([]interface{})
+		var entry map[string]interface{}
+		if parseErr := json.Unmarshal([]byte(entryJson), &entry); parseErr != nil {
+			return fmt.Errorf("failed to parse entry, err: %v", parseErr)
+		}
+
+		entryRequest := entry["request"].(map[string]interface{})
+		entryPayload := entryRequest["payload"].(map[string]interface{})
+		entryDetails := entryPayload["details"].(map[string]interface{})
+
+		headers :=  entryDetails["headers"].([]interface{})
 		for _, headerInterface := range headers {
 			header := headerInterface.(map[string]interface{})
 			if header["name"].(string) != "User-Agent" {
@@ -518,8 +530,8 @@ func TestTapRedact(t *testing.T) {
 			}
 		}
 
-		data := entryRequest["postData"].(map[string]interface{})
-		textDataStr := data["text"].(string)
+		postData := entryDetails["postData"].(map[string]interface{})
+		textDataStr := postData["text"].(string)
 
 		var textData map[string]string
 		if parseErr := json.Unmarshal([]byte(textDataStr), &textData); parseErr != nil {
@@ -608,10 +620,19 @@ func TestTapNoRedact(t *testing.T) {
 			return fmt.Errorf("failed to get entry, err: %v", requestErr)
 		}
 
-		entry := requestResult.(map[string]interface{})["entry"].(map[string]interface{})
-		entryRequest := entry["request"].(map[string]interface{})
+		data := requestResult.(map[string]interface{})["data"].(map[string]interface{})
+		entryJson := data["entry"].(string)
 
-		headers :=  entryRequest["headers"].([]interface{})
+		var entry map[string]interface{}
+		if parseErr := json.Unmarshal([]byte(entryJson), &entry); parseErr != nil {
+			return fmt.Errorf("failed to parse entry, err: %v", parseErr)
+		}
+
+		entryRequest := entry["request"].(map[string]interface{})
+		entryPayload := entryRequest["payload"].(map[string]interface{})
+		entryDetails := entryPayload["details"].(map[string]interface{})
+
+		headers :=  entryDetails["headers"].([]interface{})
 		for _, headerInterface := range headers {
 			header := headerInterface.(map[string]interface{})
 			if header["name"].(string) != "User-Agent" {
@@ -624,8 +645,8 @@ func TestTapNoRedact(t *testing.T) {
 			}
 		}
 
-		data := entryRequest["postData"].(map[string]interface{})
-		textDataStr := data["text"].(string)
+		postData := entryDetails["postData"].(map[string]interface{})
+		textDataStr := postData["text"].(string)
 
 		var textData map[string]string
 		if parseErr := json.Unmarshal([]byte(textDataStr), &textData); parseErr != nil {
@@ -714,11 +735,20 @@ func TestTapRegexMasking(t *testing.T) {
 			return fmt.Errorf("failed to get entry, err: %v", requestErr)
 		}
 
-		entry := requestResult.(map[string]interface{})["entry"].(map[string]interface{})
-		entryRequest := entry["request"].(map[string]interface{})
+		data := requestResult.(map[string]interface{})["data"].(map[string]interface{})
+		entryJson := data["entry"].(string)
 
-		data := entryRequest["postData"].(map[string]interface{})
-		textData := data["text"].(string)
+		var entry map[string]interface{}
+		if parseErr := json.Unmarshal([]byte(entryJson), &entry); parseErr != nil {
+			return fmt.Errorf("failed to parse entry, err: %v", parseErr)
+		}
+
+		entryRequest := entry["request"].(map[string]interface{})
+		entryPayload := entryRequest["payload"].(map[string]interface{})
+		entryDetails := entryPayload["details"].(map[string]interface{})
+
+		postData := entryDetails["postData"].(map[string]interface{})
+		textData := postData["text"].(string)
 
 		if textData != "[REDACTED]" {
 			return fmt.Errorf("unexpected result - body is not redacted")
@@ -731,3 +761,105 @@ func TestTapRegexMasking(t *testing.T) {
 		return
 	}
 }
+
+func TestTapDumpLogs(t *testing.T) {
+	if testing.Short() {
+		t.Skip("ignored acceptance test")
+	}
+
+	cliPath, cliPathErr := getCliPath()
+	if cliPathErr != nil {
+		t.Errorf("failed to get cli path, err: %v", cliPathErr)
+		return
+	}
+
+	tapCmdArgs := getDefaultTapCommandArgs()
+
+	tapNamespace := getDefaultTapNamespace()
+	tapCmdArgs = append(tapCmdArgs, tapNamespace...)
+
+	tapCmdArgs = append(tapCmdArgs, "--set", "dump-logs=true")
+
+	tapCmd := exec.Command(cliPath, tapCmdArgs...)
+	t.Logf("running command: %v", tapCmd.String())
+
+	if err := tapCmd.Start(); err != nil {
+		t.Errorf("failed to start tap command, err: %v", err)
+		return
+	}
+
+	apiServerUrl := getApiServerUrl(defaultApiServerPort)
+
+	if err := waitTapPodsReady(apiServerUrl); err != nil {
+		t.Errorf("failed to start tap pods on time, err: %v", err)
+		return
+	}
+
+	if err := cleanupCommand(tapCmd); err != nil {
+		t.Errorf("failed to cleanup tap command, err: %v", err)
+		return
+	}
+
+	mizuFolderPath, mizuPathErr := getMizuFolderPath()
+	if mizuPathErr != nil {
+		t.Errorf("failed to get mizu folder path, err: %v", mizuPathErr)
+		return
+	}
+
+	files, readErr := ioutil.ReadDir(mizuFolderPath)
+	if readErr != nil {
+		t.Errorf("failed to read mizu folder files, err: %v", readErr)
+		return
+	}
+
+	var dumpsLogsPath string
+	for _, file := range files {
+		fileName := file.Name()
+		if strings.Contains(fileName, "mizu_logs") {
+			dumpsLogsPath = path.Join(mizuFolderPath, fileName)
+			break
+		}
+	}
+
+	if dumpsLogsPath == "" {
+		t.Errorf("dump logs file not found")
+		return
+	}
+
+	zipReader, zipError := zip.OpenReader(dumpsLogsPath)
+	if zipError != nil {
+		t.Errorf("failed to get zip reader, err: %v", zipError)
+		return
+	}
+
+	t.Cleanup(func() {
+		if err := zipReader.Close(); err != nil {
+			t.Logf("failed to close zip reader, err: %v", err)
+		}
+	})
+
+	var logsFileNames []string
+	for _, file := range zipReader.File {
+		logsFileNames = append(logsFileNames, file.Name)
+	}
+
+	if !Contains(logsFileNames, "mizu.mizu-api-server.log") {
+		t.Errorf("api server logs not found")
+		return
+	}
+
+	if !Contains(logsFileNames, "mizu_cli.log") {
+		t.Errorf("cli logs not found")
+		return
+	}
+
+	if !Contains(logsFileNames, "mizu_events.log") {
+		t.Errorf("events logs not found")
+		return
+	}
+
+	if !ContainsPartOfValue(logsFileNames, "mizu.mizu-tapper-daemon-set") {
+		t.Errorf("tapper logs not found")
+		return
+	}
+}

acceptanceTests/testsUtils.go

@@ -9,6 +9,7 @@ import (
 	"os"
 	"os/exec"
 	"path"
+	"strings"
 	"syscall"
 	"time"
 )
@@ -32,13 +33,22 @@ func getCliPath() (string, error) {
 	return cliPath, nil
 }
 
-func getConfigPath() (string, error) {
+func getMizuFolderPath() (string, error) {
 	home, homeDirErr := os.UserHomeDir()
 	if homeDirErr != nil {
 		return "", homeDirErr
 	}
 
-	return path.Join(home, ".mizu", "config.yaml"), nil
+	return path.Join(home, ".mizu"), nil
+}
+
+func getConfigPath() (string, error) {
+	mizuFolderPath, mizuPathError := getMizuFolderPath()
+	if mizuPathError != nil {
+		return "", mizuPathError
+	}
+
+	return path.Join(mizuFolderPath, "config.yaml"), nil
 }
 
 func getProxyUrl(namespace string, service string) string {
@@ -72,6 +82,13 @@ func getDefaultTapCommandArgsWithRegex(regex string) []string {
 	return append([]string{tapCommand, regex}, defaultCmdArgs...)
 }
 
+func getDefaultLogsCommandArgs() []string {
+	logsCommand := "logs"
+	defaultCmdArgs := getDefaultCommandArgs()
+
+	return append([]string{logsCommand}, defaultCmdArgs...)
+}
+
 func getDefaultTapNamespace() []string {
 	return []string{"-n", "mizu-tests"}
 }
@@ -193,3 +210,33 @@ func getPods(tapStatusInterface interface{}) ([]map[string]interface{}, error) {
 
 	return pods, nil
 }
+
+func getLogsPath() (string, error) {
+	dir, filePathErr := os.Getwd()
+	if filePathErr != nil {
+		return "", filePathErr
+	}
+
+	logsPath := path.Join(dir, "mizu_logs.zip")
+	return logsPath, nil
+}
+
+func Contains(slice []string, containsValue string) bool {
+	for _, sliceValue := range slice {
+		if sliceValue == containsValue {
+			return true
+		}
+	}
+
+	return false
+}
+
+func ContainsPartOfValue(slice []string, containsValue string) bool {
+	for _, sliceValue := range slice {
+		if strings.Contains(sliceValue, containsValue) {
+			return true
+		}
+	}
+
+	return false
+}

cli/cmd/tap.go

@@ -2,11 +2,13 @@ package cmd
 
 import (
 	"errors"
+	"fmt"
+	"os"
+
 	"github.com/up9inc/mizu/cli/config"
 	"github.com/up9inc/mizu/cli/config/configStructs"
 	"github.com/up9inc/mizu/cli/logger"
 	"github.com/up9inc/mizu/cli/telemetry"
-	"os"
 
 	"github.com/creasty/defaults"
 	"github.com/spf13/cobra"
@@ -66,4 +68,7 @@ func init() {
 	tapCmd.Flags().String(configStructs.HumanMaxEntriesDBSizeTapName, defaultTapConfig.HumanMaxEntriesDBSize, "Override the default max entries db size")
 	tapCmd.Flags().Bool(configStructs.DryRunTapName, defaultTapConfig.DryRun, "Preview of all pods matching the regex, without tapping them")
 	tapCmd.Flags().String(configStructs.EnforcePolicyFile, defaultTapConfig.EnforcePolicyFile, "Yaml file with policy rules")
+
+	tapCmd.Flags().String(configStructs.EnforcePolicyFileDeprecated, defaultTapConfig.EnforcePolicyFileDeprecated, "Yaml file with policy rules")
+	tapCmd.Flags().MarkDeprecated(configStructs.EnforcePolicyFileDeprecated, fmt.Sprintf("Use --%s instead", configStructs.EnforcePolicyFile))
 }

cli/cmd/tapRunner.go

@@ -47,14 +47,23 @@ func RunMizuTap() {
 		logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Error parsing regex-masking: %v", errormessage.FormatError(err)))
 		return
 	}
+
 	var mizuValidationRules string
-	if config.Config.Tap.EnforcePolicyFile != "" {
-		mizuValidationRules, err = readValidationRules(config.Config.Tap.EnforcePolicyFile)
+	if config.Config.Tap.EnforcePolicyFile != "" || config.Config.Tap.EnforcePolicyFileDeprecated != "" {
+		var trafficValidation string
+		if config.Config.Tap.EnforcePolicyFile != "" {
+			trafficValidation = config.Config.Tap.EnforcePolicyFile
+		} else {
+			trafficValidation = config.Config.Tap.EnforcePolicyFileDeprecated
+		}
+
+		mizuValidationRules, err = readValidationRules(trafficValidation)
 		if err != nil {
 			logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Error reading policy file: %v", errormessage.FormatError(err)))
 			return
 		}
 	}
+
 	kubernetesProvider, err := kubernetes.NewProvider(config.Config.KubeConfigPath())
 	if err != nil {
 		logger.Log.Error(err)
@@ -109,7 +118,7 @@ func RunMizuTap() {
 	}
 
 	go goUtils.HandleExcWrapper(watchApiServerPod, ctx, kubernetesProvider, cancel)
-	go goUtils.HandleExcWrapper(watchPodsForTapping, ctx, kubernetesProvider, targetNamespaces, cancel)
+	go goUtils.HandleExcWrapper(watchPodsForTapping, ctx, kubernetesProvider, targetNamespaces, cancel, mizuApiFilteringOptions)
 
 	//block until exit signal or error
 	waitForFinish(ctx, cancel)
@@ -135,7 +144,7 @@ func createMizuResources(ctx context.Context, kubernetesProvider *kubernetes.Pro
 		return err
 	}
 
-	if err := updateMizuTappers(ctx, kubernetesProvider, nodeToTappedPodIPMap); err != nil {
+	if err := updateMizuTappers(ctx, kubernetesProvider, nodeToTappedPodIPMap, mizuApiFilteringOptions); err != nil {
 		return err
 	}
 
@@ -221,7 +230,7 @@ func getMizuApiFilteringOptions() (*api.TrafficFilteringOptions, error) {
 	}, nil
 }
 
-func updateMizuTappers(ctx context.Context, kubernetesProvider *kubernetes.Provider, nodeToTappedPodIPMap map[string][]string) error {
+func updateMizuTappers(ctx context.Context, kubernetesProvider *kubernetes.Provider, nodeToTappedPodIPMap map[string][]string, mizuApiFilteringOptions *api.TrafficFilteringOptions) error {
 	if len(nodeToTappedPodIPMap) > 0 {
 		var serviceAccountName string
 		if state.mizuServiceAccountExists {
@@ -241,6 +250,7 @@ func updateMizuTappers(ctx context.Context, kubernetesProvider *kubernetes.Provi
 			serviceAccountName,
 			config.Config.Tap.TapperResources,
 			config.Config.ImagePullPolicy(),
+			mizuApiFilteringOptions,
 		); err != nil {
 			return err
 		}
@@ -347,7 +357,7 @@ func waitUntilNamespaceDeleted(ctx context.Context, cancel context.CancelFunc, k
 	}
 }
 
-func watchPodsForTapping(ctx context.Context, kubernetesProvider *kubernetes.Provider, targetNamespaces []string, cancel context.CancelFunc) {
+func watchPodsForTapping(ctx context.Context, kubernetesProvider *kubernetes.Provider, targetNamespaces []string, cancel context.CancelFunc, mizuApiFilteringOptions *api.TrafficFilteringOptions) {
 	added, modified, removed, errorChan := kubernetes.FilteredWatch(ctx, kubernetesProvider, targetNamespaces, config.Config.Tap.PodRegex())
 
 	restartTappers := func() {
@@ -371,7 +381,7 @@ func watchPodsForTapping(ctx context.Context, kubernetesProvider *kubernetes.Pro
 			logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Error building node to ips map: %v", errormessage.FormatError(err)))
 			cancel()
 		}
-		if err := updateMizuTappers(ctx, kubernetesProvider, nodeToTappedPodIPMap); err != nil {
+		if err := updateMizuTappers(ctx, kubernetesProvider, nodeToTappedPodIPMap, mizuApiFilteringOptions); err != nil {
 			logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Error updating daemonset: %v", errormessage.FormatError(err)))
 			cancel()
 		}

cli/config/configStructs/tapConfig.go

@@ -16,7 +16,8 @@ const (
 	DisableRedactionTapName       = "no-redact"
 	HumanMaxEntriesDBSizeTapName  = "max-entries-db-size"
 	DryRunTapName                 = "dry-run"
-	EnforcePolicyFile             = "test-rules"
+	EnforcePolicyFile             = "traffic-validation"
+	EnforcePolicyFileDeprecated   = "test-rules"
 )
 
 type TapConfig struct {
@@ -32,7 +33,8 @@ type TapConfig struct {
 	DisableRedaction             bool      `yaml:"no-redact" default:"false"`
 	HumanMaxEntriesDBSize        string    `yaml:"max-entries-db-size" default:"200MB"`
 	DryRun                       bool      `yaml:"dry-run" default:"false"`
-	EnforcePolicyFile            string    `yaml:"test-rules"`
+	EnforcePolicyFile            string    `yaml:"traffic-validation"`
+	EnforcePolicyFileDeprecated  string    `yaml:"test-rules"`
 	ApiServerResources           Resources `yaml:"api-server-resources"`
 	TapperResources              Resources `yaml:"tapper-resources"`
 }

cli/kubernetes/provider.go

@@ -576,7 +576,7 @@ func (provider *Provider) CreateConfigMap(ctx context.Context, namespace string,
 	return nil
 }
 
-func (provider *Provider) ApplyMizuTapperDaemonSet(ctx context.Context, namespace string, daemonSetName string, podImage string, tapperPodName string, apiServerPodIp string, nodeToTappedPodIPMap map[string][]string, serviceAccountName string, resources configStructs.Resources, imagePullPolicy core.PullPolicy) error {
+func (provider *Provider) ApplyMizuTapperDaemonSet(ctx context.Context, namespace string, daemonSetName string, podImage string, tapperPodName string, apiServerPodIp string, nodeToTappedPodIPMap map[string][]string, serviceAccountName string, resources configStructs.Resources, imagePullPolicy core.PullPolicy, mizuApiFilteringOptions *api.TrafficFilteringOptions) error {
 	logger.Log.Debugf("Applying %d tapper daemon sets, ns: %s, daemonSetName: %s, podImage: %s, tapperPodName: %s", len(nodeToTappedPodIPMap), namespace, daemonSetName, podImage, tapperPodName)
 
 	if len(nodeToTappedPodIPMap) == 0 {
@@ -588,6 +588,11 @@ func (provider *Provider) ApplyMizuTapperDaemonSet(ctx context.Context, namespac
 		return err
 	}
 
+	marshaledFilteringOptions, err := json.Marshal(mizuApiFilteringOptions)
+	if err != nil {
+		return err
+	}
+
 	mizuCmd := []string{
 		"./mizuagent",
 		"-i", "any",
@@ -606,6 +611,7 @@ func (provider *Provider) ApplyMizuTapperDaemonSet(ctx context.Context, namespac
 		applyconfcore.EnvVar().WithName(shared.HostModeEnvVar).WithValue("1"),
 		applyconfcore.EnvVar().WithName(shared.TappedAddressesPerNodeDictEnvVar).WithValue(string(nodeToTappedPodIPMapJsonStr)),
 		applyconfcore.EnvVar().WithName(shared.GoGCEnvVar).WithValue("12800"),
+		applyconfcore.EnvVar().WithName(shared.MizuFilteringOptionsEnvVar).WithValue(string(marshaledFilteringOptions)),
 	)
 	agentContainer.WithEnv(
 		applyconfcore.EnvVar().WithName(shared.NodeNameEnvVar).WithValueFrom(

docs/POLICY_RULES.md

@@ -24,14 +24,14 @@ To use this feature - create simple rules file (see details below) and pass this
 
 
 ```shell
-mizu tap --test-rules rules.yaml PODNAME
+mizu tap --traffic-validation rules.yaml PODNAME
 ```
 
 
 
 ## Rules file structure
 
-The structure of the test-rules-file is:
+The structure of the traffic-validation-file is:
 
 * `name`: string, name of the rule
 * `type`: string, type of the rule, must be `json` or `header` or `latency`

tap/extensions/http/handlers.go

@@ -14,7 +14,9 @@ import (
 )
 
 func filterAndEmit(item *api.OutputChannelItem, emitter api.Emitter, options *api.TrafficFilteringOptions) {
-	FilterSensitiveData(item, options)
+	if !options.DisableRedaction {
+		FilterSensitiveData(item, options)
+	}
 	emitter.Emit(item)
 }
 

tap/passive_tapper.go

@@ -18,6 +18,7 @@ import (
 	"os"
 	"os/signal"
 	"runtime"
+	_debug "runtime/debug"
 	"runtime/pprof"
 	"strconv"
 	"strings"
@@ -217,10 +218,10 @@ func startMemoryProfiler() {
 }
 
 func closeTimedoutTcpStreamChannels() {
-	maxNumberOfGoroutines = GetMaxNumberOfGoroutines()
 	TcpStreamChannelTimeoutMs := GetTcpChannelTimeoutMs()
 	for {
 		time.Sleep(10 * time.Millisecond)
+		_debug.FreeOSMemory()
 		streams.Range(func(key interface{}, value interface{}) bool {
 			streamWrapper := value.(*tcpStreamWrapper)
 			stream := streamWrapper.stream

tap/settings.go

@@ -13,11 +13,9 @@ const (
 	MaxBufferedPagesTotalEnvVarName           = "MAX_BUFFERED_PAGES_TOTAL"
 	MaxBufferedPagesPerConnectionEnvVarName   = "MAX_BUFFERED_PAGES_PER_CONNECTION"
 	TcpStreamChannelTimeoutMsEnvVarName       = "TCP_STREAM_CHANNEL_TIMEOUT_MS"
-	MaxNumberOfGoroutinesEnvVarName           = "MAX_NUMBER_OF_GOROUTINES"
 	MaxBufferedPagesTotalDefaultValue         = 5000
 	MaxBufferedPagesPerConnectionDefaultValue = 5000
-	TcpStreamChannelTimeoutMsDefaultValue     = 5000
-	MaxNumberOfGoroutinesDefaultValue         = 4000
+	TcpStreamChannelTimeoutMsDefaultValue     = 10000
 )
 
 type globalSettings struct {
@@ -62,14 +60,6 @@ func GetTcpChannelTimeoutMs() time.Duration {
 	return time.Duration(valueFromEnv) * time.Millisecond
 }
 
-func GetMaxNumberOfGoroutines() int {
-	valueFromEnv, err := strconv.Atoi(os.Getenv(MaxNumberOfGoroutinesEnvVarName))
-	if err != nil {
-		return MaxNumberOfGoroutinesDefaultValue
-	}
-	return valueFromEnv
-}
-
 func GetMemoryProfilingEnabled() bool {
 	return os.Getenv(MemoryProfilingEnabledEnvVarName) == "1"
 }

tap/tcp_stream_factory.go

@@ -2,7 +2,6 @@ package tap
 
 import (
 	"fmt"
-	"runtime"
 	"sync"
 	"time"
 
@@ -33,8 +32,6 @@ type tcpStreamWrapper struct {
 var streams *sync.Map = &sync.Map{} // global
 var streamId int64 = 0
 
-var maxNumberOfGoroutines int
-
 func (factory *tcpStreamFactory) New(net, transport gopacket.Flow, tcp *layers.TCP, ac reassembly.AssemblerContext) reassembly.Stream {
 	rlog.Debugf("* NEW: %s %s", net, transport)
 	fsmOptions := reassembly.TCPSimpleFSMOptions{
@@ -61,11 +58,6 @@ func (factory *tcpStreamFactory) New(net, transport gopacket.Flow, tcp *layers.T
 		superIdentifier: &api.SuperIdentifier{},
 	}
 	if stream.isTapTarget {
-		if runtime.NumGoroutine() > maxNumberOfGoroutines {
-			appStats.IncDroppedTcpStreams()
-			rlog.Debugf("Dropped a TCP stream because of load. Total dropped: %d Total Goroutines: %d\n", appStats.DroppedTcpStreams, runtime.NumGoroutine())
-			return stream
-		}
 		streamId++
 		stream.id = streamId
 		for i, extension := range extensions {