Merge pull request #297 from up9inc/develop

Develop -> main

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
@@ -152,12 +152,12 @@ Web interface is now available at http://localhost:8899
 ```
 Any request that contains `User-Agent` header with one of the specified values (`kube-probe` or `prometheus`) will not be captured
 
-### API Rules validation
+### Traffic validation rules
 
-This feature allows you to define set of simple rules, and test the API against them.
+This feature allows you to define set of simple rules, and test the traffic against them.
 Such validation may test response for specific JSON fields, headers, etc.
 
-Please see [API RULES](docs/POLICY_RULES.md) page for more details and syntax.
+Please see [TRAFFIC RULES](docs/POLICY_RULES.md) page for more details and syntax.
 
 
 ## How to Run local UI

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"
@@ -758,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
+}

agent/main.go

@@ -61,10 +61,10 @@ func main() {
 
 		go filterItems(outputItemsChannel, filteredOutputItemsChannel, filteringOptions)
 		go api.StartReadingEntries(filteredOutputItemsChannel, nil, extensionsMap)
-		// go api.StartReadingOutbound(outboundLinkOutputChannel)
 
 		hostApi(nil)
 	} else if *tapperMode {
+		rlog.Infof("Starting tapper, websocket address: %s", *apiServerAddress)
 		if *apiServerAddress == "" {
 			panic("API server address must be provided with --api-server-address when using --tap")
 		}
@@ -77,13 +77,13 @@ func main() {
 
 		filteredOutputItemsChannel := make(chan *tapApi.OutputChannelItem)
 		tap.StartPassiveTapper(tapOpts, filteredOutputItemsChannel, extensions, filteringOptions)
-		socketConnection, err := shared.ConnectToSocketServer(*apiServerAddress, shared.DEFAULT_SOCKET_RETRIES, shared.DEFAULT_SOCKET_RETRY_SLEEP_TIME, false)
+		socketConnection, err := utils.ConnectToSocketServer(*apiServerAddress)
 		if err != nil {
 			panic(fmt.Sprintf("Error connecting to socket server at %s %v", *apiServerAddress, err))
 		}
+		rlog.Infof("Connected successfully to websocket %s", *apiServerAddress)
 
 		go pipeTapChannelToSocket(socketConnection, filteredOutputItemsChannel)
-		// go pipeOutboundLinksChannelToSocket(socketConnection, outboundLinkOutputChannel)
 	} else if *apiServerMode {
 		api.StartResolving(*namespace)
 
@@ -122,7 +122,7 @@ func loadExtensions() {
 	extensionsMap = make(map[string]*tapApi.Extension)
 	for i, file := range files {
 		filename := file.Name()
-		log.Printf("Loading extension: %s\n", filename)
+		rlog.Infof("Loading extension: %s\n", filename)
 		extension := &tapApi.Extension{
 			Path: path.Join(extensionsDir, filename),
 		}
@@ -290,7 +290,7 @@ func pipeTapChannelToSocket(connection *websocket.Conn, messageDataChannel <-cha
 	for messageData := range messageDataChannel {
 		marshaledData, err := models.CreateWebsocketTappedEntryMessage(messageData)
 		if err != nil {
-			rlog.Infof("error converting message to json %s, (%v,%+v)\n", err, err, err)
+			rlog.Errorf("error converting message to json %v, err: %s, (%v,%+v)", messageData, err, err, err)
 			continue
 		}
 
@@ -298,26 +298,8 @@ func pipeTapChannelToSocket(connection *websocket.Conn, messageDataChannel <-cha
 		// and goes into the intermediate WebSocket.
 		err = connection.WriteMessage(websocket.TextMessage, marshaledData)
 		if err != nil {
-			rlog.Infof("error sending message through socket server %s, (%v,%+v)\n", err, err, err)
+			rlog.Errorf("error sending message through socket server %v, err: %s, (%v,%+v)", messageData, err, err, err)
 			continue
 		}
 	}
 }
-
-func pipeOutboundLinksChannelToSocket(connection *websocket.Conn, outboundLinkChannel <-chan *tap.OutboundLink) {
-	for outboundLink := range outboundLinkChannel {
-		if outboundLink.SuggestedProtocol == tap.TLSProtocol {
-			marshaledData, err := models.CreateWebsocketOutboundLinkMessage(outboundLink)
-			if err != nil {
-				rlog.Infof("Error converting outbound link to json %s, (%v,%+v)", err, err, err)
-				continue
-			}
-
-			err = connection.WriteMessage(websocket.TextMessage, marshaledData)
-			if err != nil {
-				rlog.Infof("error sending outbound link message through socket server %s, (%v,%+v)", err, err, err)
-				continue
-			}
-		}
-	}
-}

agent/pkg/api/main.go

@@ -7,7 +7,7 @@ import (
 	"fmt"
 	"mizuserver/pkg/database"
 	"mizuserver/pkg/holder"
-	"net/url"
+	"mizuserver/pkg/providers"
 	"os"
 	"path"
 	"sort"
@@ -18,7 +18,6 @@ import (
 
 	"github.com/google/martian/har"
 	"github.com/romana/rlog"
-	"github.com/up9inc/mizu/tap"
 	tapApi "github.com/up9inc/mizu/tap/api"
 
 	"mizuserver/pkg/models"
@@ -59,8 +58,10 @@ func StartReadingEntries(harChannel <-chan *tapApi.OutputChannelItem, workingDir
 }
 
 func startReadingFiles(workingDir string) {
-	err := os.MkdirAll(workingDir, os.ModePerm)
-	utils.CheckErr(err)
+	if err := os.MkdirAll(workingDir, os.ModePerm); err != nil {
+		rlog.Errorf("Failed to make dir: %s, err: %v", workingDir, err)
+		return
+	}
 
 	for true {
 		dir, _ := os.Open(workingDir)
@@ -88,17 +89,6 @@ func startReadingFiles(workingDir string) {
 		decErr := json.NewDecoder(bufio.NewReader(file)).Decode(&inputHar)
 		utils.CheckErr(decErr)
 
-		// for _, entry := range inputHar.Log.Entries {
-		// 	time.Sleep(time.Millisecond * 250)
-		// 	// connectionInfo := &tap.ConnectionInfo{
-		// 	// 	ClientIP:   fileInfo.Name(),
-		// 	// 	ClientPort: "",
-		// 	// 	ServerIP:   "",
-		// 	// 	ServerPort: "",
-		// 	// 	IsOutgoing: false,
-		// 	// }
-		// 	// saveHarToDb(entry, connectionInfo)
-		// }
 		rmErr := os.Remove(inputFilePath)
 		utils.CheckErr(rmErr)
 	}
@@ -110,6 +100,8 @@ func startReadingChannel(outputItems <-chan *tapApi.OutputChannelItem, extension
 	}
 
 	for item := range outputItems {
+		providers.EntryAdded()
+
 		extension := extensionsMap[item.Protocol.Name]
 		resolvedSource, resolvedDestionation := resolveIP(item.ConnectionInfo)
 		mizuEntry := extension.Dissector.Analyze(item, primitive.NewObjectID().Hex(), resolvedSource, resolvedDestionation)
@@ -119,10 +111,12 @@ func startReadingChannel(outputItems <-chan *tapApi.OutputChannelItem, extension
 		if extension.Protocol.Name == "http" {
 			var pair tapApi.RequestResponsePair
 			json.Unmarshal([]byte(mizuEntry.Entry), &pair)
-			harEntry, _ := utils.NewEntry(&pair)
-			rules, _ := models.RunValidationRulesState(*harEntry, mizuEntry.Service)
-			baseEntry.Rules = rules
-			baseEntry.Latency = mizuEntry.ElapsedTime
+			harEntry, err := utils.NewEntry(&pair)
+			if err == nil {
+				rules, _ := models.RunValidationRulesState(*harEntry, mizuEntry.Service)
+				baseEntry.Rules = rules
+				baseEntry.Latency = mizuEntry.ElapsedTime
+			}
 		}
 
 		baseEntryBytes, _ := models.CreateBaseEntryWebSocketMessage(baseEntry)
@@ -130,13 +124,6 @@ func startReadingChannel(outputItems <-chan *tapApi.OutputChannelItem, extension
 	}
 }
 
-func StartReadingOutbound(outboundLinkChannel <-chan *tap.OutboundLink) {
-	// tcpStreamFactory will block on write to channel. Empty channel to unblock.
-	// TODO: Make write to channel optional.
-	for range outboundLinkChannel {
-	}
-}
-
 func resolveIP(connectionInfo *tapApi.ConnectionInfo) (resolvedSource string, resolvedDestination string) {
 	if k8sResolver != nil {
 		unresolvedSource := connectionInfo.ClientIP
@@ -159,12 +146,6 @@ func resolveIP(connectionInfo *tapApi.ConnectionInfo) (resolvedSource string, re
 	return resolvedSource, resolvedDestination
 }
 
-func getServiceNameFromUrl(inputUrl string) (string, string) {
-	parsed, err := url.Parse(inputUrl)
-	utils.CheckErr(err)
-	return fmt.Sprintf("%s://%s", parsed.Scheme, parsed.Host), parsed.Path
-}
-
 func CheckIsServiceIP(address string) bool {
 	if k8sResolver == nil {
 		return false

agent/pkg/database/main.go

@@ -61,7 +61,7 @@ func GetEntriesFromDb(timestampFrom int64, timestampTo int64, protocolName *stri
 	order := OrderDesc
 	protocolNameCondition := "1 = 1"
 	if protocolName != nil {
-		protocolNameCondition = fmt.Sprintf("protocolKey = '%s'", *protocolName)
+		protocolNameCondition = fmt.Sprintf("protocolName = '%s'", *protocolName)
 	}
 
 	var entries []tapApi.MizuEntry

agent/pkg/rules/rulesHTTP.go

@@ -4,6 +4,7 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
+	"github.com/romana/rlog"
 	"reflect"
 	"regexp"
 	"strings"
@@ -65,7 +66,7 @@ func MatchRequestPolicy(harEntry har.Entry, service string) []RulesMatched {
 				if err != nil {
 					continue
 				}
-				fmt.Println(matchValue, rule.Value)
+				rlog.Info(matchValue, rule.Value)
 			} else {
 				val := fmt.Sprint(out)
 				matchValue, err = regexp.MatchString(rule.Value, val)

agent/pkg/utils/har.go

@@ -4,16 +4,14 @@ import (
 	"bytes"
 	"errors"
 	"fmt"
+	"github.com/google/martian/har"
+	"github.com/romana/rlog"
+	"github.com/up9inc/mizu/tap/api"
 	"strconv"
 	"strings"
 	"time"
-
-	"github.com/google/martian/har"
-	"github.com/up9inc/mizu/tap"
-	"github.com/up9inc/mizu/tap/api"
 )
 
-
 // Keep it because we might want cookies in the future
 //func BuildCookies(rawCookies []interface{}) []har.Cookie {
 //	cookies := make([]har.Cookie, 0, len(rawCookies))
@@ -82,7 +80,7 @@ func BuildHeaders(rawHeaders []interface{}) ([]har.Header, string, string, strin
 			path = h["value"].(string)
 		}
 		if h["name"] == ":status" {
-			path = h["value"].(string)
+			status = h["value"].(string)
 		}
 	}
 
@@ -205,7 +203,7 @@ func NewResponse(response *api.GenericMessage) (harResponse *har.Response, err e
 	if strings.HasPrefix(mimeType.(string), "application/grpc") {
 		status, err = strconv.Atoi(_status)
 		if err != nil {
-			tap.SilentError("convert-response-status-for-har", "Failed converting status to int %s (%v,%+v)", err, err, err)
+			rlog.Errorf("Failed converting status to int %s (%v,%+v)", err, err, err)
 			return nil, errors.New("failed converting response status to int for HAR")
 		}
 	}
@@ -226,14 +224,13 @@ func NewResponse(response *api.GenericMessage) (harResponse *har.Response, err e
 func NewEntry(pair *api.RequestResponsePair) (*har.Entry, error) {
 	harRequest, err := NewRequest(&pair.Request)
 	if err != nil {
-		tap.SilentError("convert-request-to-har", "Failed converting request to HAR %s (%v,%+v)", err, err, err)
+		rlog.Errorf("Failed converting request to HAR %s (%v,%+v)", err, err, err)
 		return nil, errors.New("failed converting request to HAR")
 	}
 
 	harResponse, err := NewResponse(&pair.Response)
 	if err != nil {
-		fmt.Printf("err: %+v\n", err)
-		tap.SilentError("convert-response-to-har", "Failed converting response to HAR %s (%v,%+v)", err, err, err)
+		rlog.Errorf("Failed converting response to HAR %s (%v,%+v)", err, err, err)
 		return nil, errors.New("failed converting response to HAR")
 	}
 

agent/pkg/utils/socket_client.go

@@ -1,8 +1,8 @@
-package shared
+package utils
 
 import (
-	"fmt"
 	"github.com/gorilla/websocket"
+	"github.com/romana/rlog"
 	"time"
 )
 
@@ -11,24 +11,23 @@ const (
 	DEFAULT_SOCKET_RETRY_SLEEP_TIME = time.Second * 10
 )
 
-func ConnectToSocketServer(address string, retries int, retrySleepTime time.Duration, hideTimeoutErrors bool) (*websocket.Conn, error) {
+func ConnectToSocketServer(address string) (*websocket.Conn, error) {
 	var err error
 	var connection *websocket.Conn
 	try := 0
 
 	// Connection to server fails if client pod is up before server.
 	// Retries solve this issue.
-	for try < retries {
+	for try < DEFAULT_SOCKET_RETRIES {
+		rlog.Infof("Trying to connect to websocket: %s, attempt: %v/%v", address, try, DEFAULT_SOCKET_RETRIES)
 		connection, _, err = websocket.DefaultDialer.Dial(address, nil)
 		if err != nil {
+			rlog.Warnf("Failed connecting to websocket: %s, attempt: %v/%v, err: %s, (%v,%+v)", address, try, DEFAULT_SOCKET_RETRIES, err, err, err)
 			try++
-			if !hideTimeoutErrors {
-				fmt.Printf("Failed connecting to websocket server: %s, (%v,%+v)\n", err, err, err)
-			}
 		} else {
 			break
 		}
-		time.Sleep(retrySleepTime)
+		time.Sleep(DEFAULT_SOCKET_RETRY_SLEEP_TIME)
 	}
 
 	if err != nil {

agent/pkg/utils/utils.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"github.com/gin-gonic/gin"
 	"github.com/romana/rlog"
-	"log"
 	"net/http"
 	"net/url"
 	"os"
@@ -18,8 +17,8 @@ import (
 func StartServer(app *gin.Engine) {
 	signals := make(chan os.Signal, 2)
 	signal.Notify(signals,
-		os.Interrupt,  	  // this catch ctrl + c
-		syscall.SIGTSTP,  // this catch ctrl + z
+		os.Interrupt,    // this catch ctrl + c
+		syscall.SIGTSTP, // this catch ctrl + z
 	)
 
 	srv := &http.Server{
@@ -36,8 +35,9 @@ func StartServer(app *gin.Engine) {
 	}()
 
 	// Run server.
+	rlog.Infof("Starting the server...")
 	if err := app.Run(":8899"); err != nil {
-		log.Printf("Oops... Server is not running! Reason: %v", err)
+		rlog.Errorf("Server is not running! Reason: %v", err)
 	}
 }
 
@@ -54,15 +54,14 @@ func ReverseSlice(data interface{}) {
 
 func CheckErr(e error) {
 	if e != nil {
-		log.Printf("%v", e)
-		//panic(e)
+		rlog.Errorf("%v", e)
 	}
 }
 
 func SetHostname(address, newHostname string) string {
 	replacedUrl, err := url.Parse(address)
-	if err != nil{
-		log.Printf("error replacing hostname to %s in address %s, returning original %v",newHostname, address, err)
+	if err != nil {
+		rlog.Errorf("error replacing hostname to %s in address %s, returning original %v", newHostname, address, err)
 		return address
 	}
 	replacedUrl.Host = newHostname

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"
@@ -65,5 +67,8 @@ func init() {
 	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().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.EnforcePolicyFile, defaultTapConfig.EnforcePolicyFile, "Yaml file path 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-file"
+	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-file"`
+	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(

cli/mizu/controlSocket.go

@@ -1,42 +0,0 @@
-package mizu
-
-import (
-	"encoding/json"
-	"github.com/gorilla/websocket"
-	"github.com/up9inc/mizu/shared"
-	core "k8s.io/api/core/v1"
-	"time"
-)
-
-type ControlSocket struct {
-	connection *websocket.Conn
-}
-
-func CreateControlSocket(socketServerAddress string) (*ControlSocket, error) {
-	connection, err := shared.ConnectToSocketServer(socketServerAddress, 30, 2 * time.Second, true)
-	if err != nil {
-		return nil, err
-	} else {
-		return &ControlSocket{connection: connection}, nil
-	}
-}
-
-func (controlSocket *ControlSocket) SendNewTappedPodsListMessage(pods []core.Pod) error {
-	podInfos := make([]shared.PodInfo, 0)
-	for _, pod := range pods {
-		podInfos = append(podInfos, shared.PodInfo{Name: pod.Name, Namespace: pod.Namespace})
-	}
-	tapStatus := shared.TapStatus{Pods: podInfos}
-	socketMessage := shared.CreateWebSocketStatusMessage(tapStatus)
-
-	jsonMessage, err := json.Marshal(socketMessage)
-	if err != nil {
-		return err
-	}
-	err = controlSocket.connection.WriteMessage(websocket.TextMessage, jsonMessage)
-	if err != nil {
-		return err
-	}
-
-	return nil
-}

docs/POLICY_RULES.md

@@ -1,37 +1,34 @@
 
-# API rules validation
+# Traffic validation rules
 
-This feature allows you to define set of simple rules, and test the API against them.
+This feature allows you to define set of simple rules, and test the traffic against them.
 Such validation may test response for specific JSON fields, headers, etc.
 
 ## Examples
 
-
-Example 1: HTTP request (REST API call) that didn’t pass validation is highlighted in red
+Example 1: HTTP request (REST API call) that didn't pass validation is highlighted in red
 
 ![Simple UI](../assets/validation-example1.png)
 
 - - -
 
-
 Example 2: Details pane shows the validation rule details and whether it passed or failed
 
 ![Simple UI](../assets/validation-example2.png)
 
 
 ## How to use
+
 To use this feature - create simple rules file (see details below) and pass this file as parameter to `mizu tap` command. For example, if rules are stored in file named `rules.yaml` — run the following command:
 
-
 ```shell
-mizu tap --test-rules rules.yaml PODNAME
+mizu tap --traffic-validation-file rules.yaml
 ```
 
 
-
 ## 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`
@@ -62,6 +59,7 @@ rules:
   service: "carts.*"
 ```
 
+
 ### Explanation:
 
 * First rule `holy-in-name-property`:
@@ -74,5 +72,4 @@ rules:
 
 * Third rule `latency-test`:
 
-  > This rule will be applied to all request made to `carts.*` services. If the latency of the response is greater than `1` will be marked as failure, marked as success otherwise.
-
+  > This rule will be applied to all request made to `carts.*` services. If the latency of the response is greater than `1ms` will be marked as failure, marked as success otherwise.

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)
 }