Return 404 instead of 500 if the entry could not be found and display a toast message (#464)

* Update main.go, status_controller.go, and 2 more files...

* Update status_controller.go and mizuTapperSyncer.go

acceptanceTests/tap_test.go

@@ -427,9 +427,10 @@ func TestTapRedact(t *testing.T) {
 	}
 
 	proxyUrl := getProxyUrl(defaultNamespaceName, defaultServiceName)
+	requestHeaders := map[string]string{"User-Header": "Mizu"}
 	requestBody := map[string]string{"User": "Mizu"}
 	for i := 0; i < defaultEntriesCount; i++ {
-		if _, requestErr := executeHttpPostRequest(fmt.Sprintf("%v/post", proxyUrl), requestBody); requestErr != nil {
+		if _, requestErr := executeHttpPostRequestWithHeaders(fmt.Sprintf("%v/post", proxyUrl), requestHeaders, requestBody); requestErr != nil {
 			t.Errorf("failed to send proxy request, err: %v", requestErr)
 			return
 		}
@@ -460,12 +461,12 @@ func TestTapRedact(t *testing.T) {
 		headers := request["_headers"].([]interface{})
 		for _, headerInterface := range headers {
 			header := headerInterface.(map[string]interface{})
-			if header["name"].(string) != "User-Agent" {
+			if header["name"].(string) != "User-Header" {
 				continue
 			}
 
-			userAgent := header["value"].(string)
-			if userAgent != "[REDACTED]" {
+			userHeader := header["value"].(string)
+			if userHeader != "[REDACTED]" {
 				return fmt.Errorf("unexpected result - user agent is not redacted")
 			}
 		}
@@ -530,9 +531,10 @@ func TestTapNoRedact(t *testing.T) {
 	}
 
 	proxyUrl := getProxyUrl(defaultNamespaceName, defaultServiceName)
+	requestHeaders := map[string]string{"User-Header": "Mizu"}
 	requestBody := map[string]string{"User": "Mizu"}
 	for i := 0; i < defaultEntriesCount; i++ {
-		if _, requestErr := executeHttpPostRequest(fmt.Sprintf("%v/post", proxyUrl), requestBody); requestErr != nil {
+		if _, requestErr := executeHttpPostRequestWithHeaders(fmt.Sprintf("%v/post", proxyUrl), requestHeaders, requestBody); requestErr != nil {
 			t.Errorf("failed to send proxy request, err: %v", requestErr)
 			return
 		}
@@ -563,12 +565,12 @@ func TestTapNoRedact(t *testing.T) {
 		headers := request["_headers"].([]interface{})
 		for _, headerInterface := range headers {
 			header := headerInterface.(map[string]interface{})
-			if header["name"].(string) != "User-Agent" {
+			if header["name"].(string) != "User-Header" {
 				continue
 			}
 
-			userAgent := header["value"].(string)
-			if userAgent == "[REDACTED]" {
+			userHeader := header["value"].(string)
+			if userHeader == "[REDACTED]" {
 				return fmt.Errorf("unexpected result - user agent is redacted")
 			}
 		}

acceptanceTests/testsUtils.go

@@ -240,13 +240,24 @@ func executeHttpGetRequest(url string) (interface{}, error) {
 	return executeHttpRequest(response, requestErr)
 }
 
-func executeHttpPostRequest(url string, body interface{}) (interface{}, error) {
+func executeHttpPostRequestWithHeaders(url string, headers map[string]string, body interface{}) (interface{}, error) {
 	requestBody, jsonErr := json.Marshal(body)
 	if jsonErr != nil {
 		return nil, jsonErr
 	}
 
-	response, requestErr := http.Post(url, "application/json", bytes.NewBuffer(requestBody))
+	request, err := http.NewRequest(http.MethodPost, url, bytes.NewBuffer(requestBody))
+	if err != nil {
+		return nil, err
+	}
+
+	request.Header.Add("Content-Type", "application/json")
+	for headerKey, headerValue := range headers {
+		request.Header.Add(headerKey, headerValue)
+	}
+
+	client := &http.Client{}
+	response, requestErr := client.Do(request)
 	return executeHttpRequest(response, requestErr)
 }
 

agent/main.go

@@ -207,7 +207,7 @@ func loadExtensions() {
 	extensionsMap = make(map[string]*tapApi.Extension)
 	for i, file := range files {
 		filename := file.Name()
-		logger.Log.Infof("Loading extension: %s\n", filename)
+		logger.Log.Infof("Loading extension: %s", filename)
 		extension := &tapApi.Extension{
 			Path: path.Join(extensionsDir, filename),
 		}
@@ -219,7 +219,7 @@ func loadExtensions() {
 		var ok bool
 		dissector, ok = symDissector.(tapApi.Dissector)
 		if err != nil || !ok {
-			panic(fmt.Sprintf("Failed to load the extension: %s\n", extension.Path))
+			panic(fmt.Sprintf("Failed to load the extension: %s", extension.Path))
 		}
 		dissector.Register(extension)
 		extension.Dissector = dissector
@@ -232,7 +232,7 @@ func loadExtensions() {
 	})
 
 	for _, extension := range extensions {
-		logger.Log.Infof("Extension Properties: %+v\n", extension)
+		logger.Log.Infof("Extension Properties: %+v", extension)
 	}
 
 	controllers.InitExtensionsMap(extensionsMap)
@@ -459,7 +459,7 @@ func startMizuTapperSyncer(ctx context.Context) (*kubernetes.MizuTapperSyncer, e
 					return
 				}
 				logger.Log.Fatalf("fatal tap syncer error: %v", syncerErr)
-			case _, ok := <-tapperSyncer.TapPodChangesOut:
+			case tapPodChangeEvent, ok := <-tapperSyncer.TapPodChangesOut:
 				if !ok {
 					logger.Log.Debug("mizuTapperSyncer pod changes channel closed, ending listener loop")
 					return
@@ -472,6 +472,7 @@ func startMizuTapperSyncer(ctx context.Context) (*kubernetes.MizuTapperSyncer, e
 				}
 				api.BroadcastToBrowserClients(serializedTapStatus)
 				providers.TapStatus.Pods = tapStatus.Pods
+				providers.ExpectedTapperAmount = tapPodChangeEvent.ExpectedTapperAmount
 			case <-ctx.Done():
 				logger.Log.Debug("mizuTapperSyncer event listener loop exiting due to context done")
 				return

agent/pkg/api/main.go

@@ -76,7 +76,7 @@ func startReadingFiles(workingDir string) {
 		sort.Sort(utils.ByModTime(harFiles))
 
 		if len(harFiles) == 0 {
-			logger.Log.Infof("Waiting for new files\n")
+			logger.Log.Infof("Waiting for new files")
 			time.Sleep(3 * time.Second)
 			continue
 		}
@@ -109,7 +109,7 @@ func startReadingChannel(outputItems <-chan *tapApi.OutputChannelItem, extension
 	ctx := context.Background()
 	doc, contractContent, router, err := loadOAS(ctx)
 	if err != nil {
-		logger.Log.Infof("Disabled OAS validation: %s\n", err.Error())
+		logger.Log.Infof("Disabled OAS validation: %s", err.Error())
 		disableOASValidation = true
 	}
 
@@ -154,7 +154,7 @@ func resolveIP(connectionInfo *tapApi.ConnectionInfo) (resolvedSource string, re
 		unresolvedSource := connectionInfo.ClientIP
 		resolvedSource = k8sResolver.Resolve(unresolvedSource)
 		if resolvedSource == "" {
-			logger.Log.Debugf("Cannot find resolved name to source: %s\n", unresolvedSource)
+			logger.Log.Debugf("Cannot find resolved name to source: %s", unresolvedSource)
 			if os.Getenv("SKIP_NOT_RESOLVED_SOURCE") == "1" {
 				return
 			}
@@ -162,7 +162,7 @@ func resolveIP(connectionInfo *tapApi.ConnectionInfo) (resolvedSource string, re
 		unresolvedDestination := fmt.Sprintf("%s:%s", connectionInfo.ServerIP, connectionInfo.ServerPort)
 		resolvedDestination = k8sResolver.Resolve(unresolvedDestination)
 		if resolvedDestination == "" {
-			logger.Log.Debugf("Cannot find resolved name to dest: %s\n", unresolvedDestination)
+			logger.Log.Debugf("Cannot find resolved name to dest: %s", unresolvedDestination)
 			if os.Getenv("SKIP_NOT_RESOLVED_DEST") == "1" {
 				return
 			}

agent/pkg/api/socket_routes.go

@@ -146,7 +146,7 @@ func websocketHandler(w http.ResponseWriter, r *http.Request, eventHandlers Even
 					var metadata *basenine.Metadata
 					err = json.Unmarshal(bytes, &metadata)
 					if err != nil {
-						logger.Log.Debugf("Error recieving metadata: %v\n", err.Error())
+						logger.Log.Debugf("Error recieving metadata: %v", err.Error())
 					}
 
 					metadataBytes, _ := models.CreateWebsocketQueryMetadataMessage(metadata)
@@ -167,7 +167,7 @@ func websocketHandler(w http.ResponseWriter, r *http.Request, eventHandlers Even
 func socketCleanup(socketId int, socketConnection *SocketConnection) {
 	err := socketConnection.connection.Close()
 	if err != nil {
-		logger.Log.Errorf("Error closing socket connection for socket id %d: %v\n", socketId, err)
+		logger.Log.Errorf("Error closing socket connection for socket id %d: %v", socketId, err)
 	}
 
 	websocketIdsLock.Lock()

agent/pkg/api/socket_server_handlers.go

@@ -65,14 +65,14 @@ func (h *RoutesEventHandlers) WebSocketMessage(_ int, message []byte) {
 	var socketMessageBase shared.WebSocketMessageMetadata
 	err := json.Unmarshal(message, &socketMessageBase)
 	if err != nil {
-		logger.Log.Infof("Could not unmarshal websocket message %v\n", err)
+		logger.Log.Infof("Could not unmarshal websocket message %v", err)
 	} else {
 		switch socketMessageBase.MessageType {
 		case shared.WebSocketMessageTypeTappedEntry:
 			var tappedEntryMessage models.WebSocketTappedEntryMessage
 			err := json.Unmarshal(message, &tappedEntryMessage)
 			if err != nil {
-				logger.Log.Infof("Could not unmarshal message of message type %s %v\n", socketMessageBase.MessageType, err)
+				logger.Log.Infof("Could not unmarshal message of message type %s %v", socketMessageBase.MessageType, err)
 			} else {
 				// NOTE: This is where the message comes back from the intermediate WebSocket to code.
 				h.SocketOutChannel <- tappedEntryMessage.Data
@@ -81,7 +81,7 @@ func (h *RoutesEventHandlers) WebSocketMessage(_ int, message []byte) {
 			var statusMessage shared.WebSocketStatusMessage
 			err := json.Unmarshal(message, &statusMessage)
 			if err != nil {
-				logger.Log.Infof("Could not unmarshal message of message type %s %v\n", socketMessageBase.MessageType, err)
+				logger.Log.Infof("Could not unmarshal message of message type %s %v", socketMessageBase.MessageType, err)
 			} else {
 				providers.TapStatus.Pods = statusMessage.TappingStatus.Pods
 				BroadcastToBrowserClients(message)
@@ -90,7 +90,7 @@ func (h *RoutesEventHandlers) WebSocketMessage(_ int, message []byte) {
 			var outboundLinkMessage models.WebsocketOutboundLinkMessage
 			err := json.Unmarshal(message, &outboundLinkMessage)
 			if err != nil {
-				logger.Log.Infof("Could not unmarshal message of message type %s %v\n", socketMessageBase.MessageType, err)
+				logger.Log.Infof("Could not unmarshal message of message type %s %v", socketMessageBase.MessageType, err)
 			} else {
 				handleTLSLink(outboundLinkMessage)
 			}

agent/pkg/controllers/entries_controller.go

@@ -25,7 +25,12 @@ func Error(c *gin.Context, err error) bool {
 	if err != nil {
 		logger.Log.Errorf("Error getting entry: %v", err)
 		c.Error(err)
-		c.AbortWithStatusJSON(http.StatusInternalServerError, gin.H{"error": true, "msg": err.Error()})
+		c.AbortWithStatusJSON(http.StatusInternalServerError, gin.H{
+			"error":     true,
+			"type":      "error",
+			"autoClose": "5000",
+			"msg":       err.Error(),
+		})
 		return true // signal that there was an error and the caller should return
 	}
 	return false // no error, can continue
@@ -39,7 +44,13 @@ func GetEntry(c *gin.Context) {
 		return // exit
 	}
 	err = json.Unmarshal(bytes, &entry)
-	if Error(c, err) {
+	if err != nil {
+		c.JSON(http.StatusNotFound, gin.H{
+			"error":     true,
+			"type":      "error",
+			"autoClose": "5000",
+			"msg":       string(bytes),
+		})
 		return // exit
 	}
 

agent/pkg/controllers/status_controller.go

@@ -2,7 +2,9 @@ package controllers
 
 import (
 	"encoding/json"
+	"fmt"
 	"mizuserver/pkg/api"
+	"mizuserver/pkg/config"
 	"mizuserver/pkg/holder"
 	"mizuserver/pkg/providers"
 	"mizuserver/pkg/up9"
@@ -15,6 +17,13 @@ import (
 )
 
 func HealthCheck(c *gin.Context) {
+	if config.Config.DaemonMode {
+		if providers.ExpectedTapperAmount != providers.TappersCount {
+			c.JSON(http.StatusInternalServerError, fmt.Sprintf("expecting more tappers than are actually connected (%d expected, %d connected)", providers.ExpectedTapperAmount, providers.TappersCount))
+			return
+		}
+	}
+
 	response := shared.HealthResponse{
 		TapStatus:    providers.TapStatus,
 		TappersCount: providers.TappersCount,
@@ -37,7 +46,7 @@ func PostTappedPods(c *gin.Context) {
 	providers.TapStatus.Pods = tapStatus.Pods
 	message := shared.CreateWebSocketStatusMessage(*tapStatus)
 	if jsonBytes, err := json.Marshal(message); err != nil {
-		logger.Log.Errorf("Could not Marshal message %v\n", err)
+		logger.Log.Errorf("Could not Marshal message %v", err)
 	} else {
 		api.BroadcastToBrowserClients(jsonBytes)
 	}

agent/pkg/providers/status_provider.go

@@ -19,7 +19,7 @@ var (
 	TapStatus      shared.TapStatus
 	authStatus     *models.AuthStatus
 	RecentTLSLinks = cache.New(tlsLinkRetainmentTime, tlsLinkRetainmentTime)
-
+	ExpectedTapperAmount = -1 //only relevant in daemon mode as cli manages tappers otherwise
 	tappersCountLock = sync.Mutex{}
 )
 

agent/pkg/resolver/resolver.go

@@ -164,10 +164,10 @@ func (resolver *Resolver) watchServices(ctx context.Context) error {
 func (resolver *Resolver) saveResolvedName(key string, resolved string, eventType watch.EventType) {
 	if eventType == watch.Deleted {
 		resolver.nameMap.Remove(key)
-		logger.Log.Infof("setting %s=nil\n", key)
+		logger.Log.Infof("setting %s=nil", key)
 	} else {
 		resolver.nameMap.Set(key, resolved)
-		logger.Log.Infof("setting %s=%s\n", key, resolved)
+		logger.Log.Infof("setting %s=%s", key, resolved)
 	}
 }
 
@@ -188,7 +188,7 @@ func (resolver *Resolver) infiniteErrorHandleRetryFunc(ctx context.Context, fun
 			var statusError *k8serrors.StatusError
 			if errors.As(err, &statusError) {
 				if statusError.ErrStatus.Reason == metav1.StatusReasonForbidden {
-					logger.Log.Infof("Resolver loop encountered permission error, aborting event listening - %v\n", err)
+					logger.Log.Infof("Resolver loop encountered permission error, aborting event listening - %v", err)
 					return
 				}
 			}

agent/pkg/up9/main.go

@@ -112,14 +112,14 @@ func GetAnalyzeInfo() *shared.AnalyzeStatus {
 }
 
 func SyncEntries(syncEntriesConfig *shared.SyncEntriesConfig) error {
-	logger.Log.Infof("Sync entries - started\n")
+	logger.Log.Infof("Sync entries - started")
 
 	var (
 		token, model string
 		guestMode    bool
 	)
 	if syncEntriesConfig.Token == "" {
-		logger.Log.Infof("Sync entries - creating anonymous token. env %s\n", syncEntriesConfig.Env)
+		logger.Log.Infof("Sync entries - creating anonymous token. env %s", syncEntriesConfig.Env)
 		guestToken, err := createAnonymousToken(syncEntriesConfig.Env)
 		if err != nil {
 			return fmt.Errorf("failed creating anonymous token, err: %v", err)
@@ -133,7 +133,7 @@ func SyncEntries(syncEntriesConfig *shared.SyncEntriesConfig) error {
 		model = syncEntriesConfig.Workspace
 		guestMode = false
 
-		logger.Log.Infof("Sync entries - upserting model. env %s, model %s\n", syncEntriesConfig.Env, model)
+		logger.Log.Infof("Sync entries - upserting model. env %s, model %s", syncEntriesConfig.Env, model)
 		if err := upsertModel(token, model, syncEntriesConfig.Env); err != nil {
 			return fmt.Errorf("failed upserting model, err: %v", err)
 		}
@@ -144,7 +144,7 @@ func SyncEntries(syncEntriesConfig *shared.SyncEntriesConfig) error {
 		return fmt.Errorf("invalid model name, model name: %s", model)
 	}
 
-	logger.Log.Infof("Sync entries - syncing. token: %s, model: %s, guest mode: %v\n", token, model, guestMode)
+	logger.Log.Infof("Sync entries - syncing. token: %s, model: %s, guest mode: %v", token, model, guestMode)
 	go syncEntriesImpl(token, model, syncEntriesConfig.Env, syncEntriesConfig.UploadIntervalSec, guestMode)
 
 	return nil
@@ -209,7 +209,7 @@ func syncEntriesImpl(token string, model string, envPrefix string, uploadInterva
 	// "http or grpc" filter indicates that we're only interested in HTTP and gRPC entries
 	query := "http or grpc"
 
-	logger.Log.Infof("Getting entries from the database\n")
+	logger.Log.Infof("Getting entries from the database")
 
 	var connection *basenine.Connection
 	var err error

cli/cmd/tapRunner.go

@@ -282,7 +282,7 @@ func createMizuResources(ctx context.Context, cancel context.CancelFunc, kuberne
 	}
 
 	if err := createMizuConfigmap(ctx, kubernetesProvider, serializedValidationRules, serializedContract, serializedMizuConfig); err != nil {
-		logger.Log.Warningf(uiUtils.Warning, fmt.Sprintf("Failed to create resources required for policy validation. Mizu will not validate policy rules. error: %v\n", errormessage.FormatError(err)))
+		logger.Log.Warningf(uiUtils.Warning, fmt.Sprintf("Failed to create resources required for policy validation. Mizu will not validate policy rules. error: %v", errormessage.FormatError(err)))
 	}
 
 	var err error
@@ -441,7 +441,7 @@ func dumpLogsIfNeeded(ctx context.Context, kubernetesProvider *kubernetes.Provid
 }
 
 func cleanUpMizuResources(ctx context.Context, cancel context.CancelFunc, kubernetesProvider *kubernetes.Provider) {
-	logger.Log.Infof("\nRemoving mizu resources\n")
+	logger.Log.Infof("\nRemoving mizu resources")
 
 	var leftoverResources []string
 
@@ -626,7 +626,7 @@ func watchApiServerPod(ctx context.Context, kubernetesProvider *kubernetes.Provi
 					break
 				}
 
-				logger.Log.Infof("Mizu is available at %s\n", url)
+				logger.Log.Infof("Mizu is available at %s", url)
 				if !config.Config.HeadlessMode {
 					uiUtils.OpenBrowser(url)
 				}

cli/cmd/viewRunner.go

@@ -56,7 +56,7 @@ func runMizuView() {
 		return
 	}
 
-	logger.Log.Infof("Mizu is available at %s\n", url)
+	logger.Log.Infof("Mizu is available at %s", url)
 
 	if !config.Config.HeadlessMode {
 		uiUtils.OpenBrowser(url)

cli/mizu/fsUtils/mizuLogsUtils.go

@@ -78,6 +78,6 @@ func DumpLogs(ctx context.Context, provider *kubernetes.Provider, filePath strin
 		logger.Log.Debugf("Successfully added file %s", GetLogFilePath())
 	}
 
-	logger.Log.Infof("You can find the zip file with all logs in %s\n", filePath)
+	logger.Log.Infof("You can find the zip file with all logs in %s", filePath)
 	return nil
 }

shared/kubernetes/mizuTapperSyncer.go

@@ -16,18 +16,20 @@ import (
 const updateTappersDelay = 5 * time.Second
 
 type TappedPodChangeEvent struct {
-	Added   []core.Pod
-	Removed []core.Pod
+	Added                []core.Pod
+	Removed              []core.Pod
+	ExpectedTapperAmount int
 }
 
 // MizuTapperSyncer uses a k8s pod watch to update tapper daemonsets when targeted pods are removed or created
 type MizuTapperSyncer struct {
-	context             context.Context
-	CurrentlyTappedPods []core.Pod
-	config              TapperSyncerConfig
-	kubernetesProvider  *Provider
-	TapPodChangesOut    chan TappedPodChangeEvent
-	ErrorOut            chan K8sTapManagerError
+	context              context.Context
+	CurrentlyTappedPods  []core.Pod
+	config               TapperSyncerConfig
+	kubernetesProvider   *Provider
+	TapPodChangesOut     chan TappedPodChangeEvent
+	ErrorOut             chan K8sTapManagerError
+	nodeToTappedPodIPMap map[string][]string
 }
 
 type TapperSyncerConfig struct {
@@ -160,9 +162,11 @@ func (tapperSyncer *MizuTapperSyncer) updateCurrentlyTappedPods() (err error, ch
 		}
 		if len(addedPods) > 0 || len(removedPods) > 0 {
 			tapperSyncer.CurrentlyTappedPods = podsToTap
+			tapperSyncer.nodeToTappedPodIPMap = GetNodeHostToTappedPodIpsMap(tapperSyncer.CurrentlyTappedPods)
 			tapperSyncer.TapPodChangesOut <- TappedPodChangeEvent{
-				Added:   addedPods,
-				Removed: removedPods,
+				Added:                addedPods,
+				Removed:              removedPods,
+				ExpectedTapperAmount: len(tapperSyncer.nodeToTappedPodIPMap),
 			}
 			return nil, true
 		}
@@ -171,9 +175,7 @@ func (tapperSyncer *MizuTapperSyncer) updateCurrentlyTappedPods() (err error, ch
 }
 
 func (tapperSyncer *MizuTapperSyncer) updateMizuTappers() error {
-	nodeToTappedPodIPMap := GetNodeHostToTappedPodIpsMap(tapperSyncer.CurrentlyTappedPods)
-
-	if len(nodeToTappedPodIPMap) > 0 {
+	if len(tapperSyncer.nodeToTappedPodIPMap) > 0 {
 		var serviceAccountName string
 		if tapperSyncer.config.MizuServiceAccountExists {
 			serviceAccountName = ServiceAccountName
@@ -188,7 +190,7 @@ func (tapperSyncer *MizuTapperSyncer) updateMizuTappers() error {
 			tapperSyncer.config.AgentImage,
 			TapperPodName,
 			fmt.Sprintf("%s.%s.svc.cluster.local", ApiServerPodName, tapperSyncer.config.MizuResourcesNamespace),
-			nodeToTappedPodIPMap,
+			tapperSyncer.nodeToTappedPodIPMap,
 			serviceAccountName,
 			tapperSyncer.config.TapperResources,
 			tapperSyncer.config.ImagePullPolicy,
@@ -197,7 +199,7 @@ func (tapperSyncer *MizuTapperSyncer) updateMizuTappers() error {
 		); err != nil {
 			return err
 		}
-		logger.Log.Debugf("Successfully created %v tappers", len(nodeToTappedPodIPMap))
+		logger.Log.Debugf("Successfully created %v tappers", len(tapperSyncer.nodeToTappedPodIPMap))
 	} else {
 		if err := tapperSyncer.kubernetesProvider.RemoveDaemonSet(tapperSyncer.context, tapperSyncer.config.MizuResourcesNamespace, TapperDaemonSetName); err != nil {
 			return err

shared/logger/logger.go

@@ -9,7 +9,7 @@ import (
 var Log = logging.MustGetLogger("mizu")
 
 var format = logging.MustStringFormatter(
-	`%{time} %{level:.5s} ▶ %{message} ▶ %{pid} %{shortfile} %{shortfunc}`,
+	`%{time:2006-01-02T15:04:05.999Z-07:00} %{level:-5s} ▶ %{message} ▶ %{pid} %{shortfile} %{shortfunc}`,
 )
 
 func InitLogger(logPath string) {

shared/models.go

@@ -2,9 +2,9 @@ package shared
 
 import (
 	"github.com/op/go-logging"
+	"github.com/up9inc/mizu/shared/logger"
 	"github.com/up9inc/mizu/tap/api"
 	"io/ioutil"
-	"log"
 	"strings"
 
 	"gopkg.in/yaml.v3"
@@ -141,14 +141,12 @@ func (r *RulePolicy) validateType() bool {
 	permitedTypes := []string{"json", "header", "slo"}
 	_, found := Find(permitedTypes, r.Type)
 	if !found {
-		log.Printf("Error: %s. ", r.Name)
-		log.Printf("Only json, header and slo types are supported on rule definition. This rule will be ignored\n")
+		logger.Log.Errorf("Only json, header and slo types are supported on rule definition. This rule will be ignored. rule name: %s", r.Name)
 		found = false
 	}
 	if strings.ToLower(r.Type) == "slo" {
 		if r.ResponseTime <= 0 {
-			log.Printf("Error: %s. ", r.Name)
-			log.Printf("When type=slo, the field response-time should be specified and have a value >= 1\n\n")
+			logger.Log.Errorf("When rule type is slo, the field response-time should be specified and have a value >= 1. rule name: %s", r.Name)
 			found = false
 		}
 	}

tap/api/api.go

@@ -287,7 +287,7 @@ func (h HTTPPayload) MarshalJSON() ([]byte, error) {
 			RawResponse: &HTTPResponseWrapper{Response: h.Data.(*http.Response)},
 		})
 	default:
-		panic(fmt.Sprintf("HTTP payload cannot be marshaled: %s\n", h.Type))
+		panic(fmt.Sprintf("HTTP payload cannot be marshaled: %s", h.Type))
 	}
 }
 

tap/diagnose/diagnose.go

@@ -39,7 +39,7 @@ func StartMemoryProfiler(envDumpPath string, envTimeInterval string) {
 
 			filename := fmt.Sprintf("%s/%s__mem.prof", dumpPath, t.Format("15_04_05"))
 
-			logger.Log.Infof("Writing memory profile to %s\n", filename)
+			logger.Log.Infof("Writing memory profile to %s", filename)
 
 			f, err := os.Create(filename)
 			if err != nil {

tap/extensions/amqp/main.go

@@ -37,7 +37,7 @@ func (d dissecting) Register(extension *api.Extension) {
 }
 
 func (d dissecting) Ping() {
-	log.Printf("pong %s\n", protocol.Name)
+	log.Printf("pong %s", protocol.Name)
 }
 
 const amqpRequest string = "amqp_request"
@@ -218,7 +218,7 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
 			}
 
 		default:
-			// log.Printf("unexpected frame: %+v\n", f)
+			// log.Printf("unexpected frame: %+v", f)
 		}
 	}
 }

tap/extensions/http/main.go

@@ -58,7 +58,7 @@ func (d dissecting) Register(extension *api.Extension) {
 }
 
 func (d dissecting) Ping() {
-	log.Printf("pong %s\n", protocol.Name)
+	log.Printf("pong %s", protocol.Name)
 }
 
 func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, superIdentifier *api.SuperIdentifier, emitter api.Emitter, options *api.TrafficFilteringOptions) error {

tap/extensions/kafka/main.go

@@ -37,7 +37,7 @@ func (d dissecting) Register(extension *api.Extension) {
 }
 
 func (d dissecting) Ping() {
-	log.Printf("pong %s\n", _protocol.Name)
+	log.Printf("pong %s", _protocol.Name)
 }
 
 func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, superIdentifier *api.SuperIdentifier, emitter api.Emitter, options *api.TrafficFilteringOptions) error {

tap/extensions/redis/main.go

@@ -36,7 +36,7 @@ func (d dissecting) Register(extension *api.Extension) {
 }
 
 func (d dissecting) Ping() {
-	log.Printf("pong %s\n", protocol.Name)
+	log.Printf("pong %s", protocol.Name)
 }
 
 func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, superIdentifier *api.SuperIdentifier, emitter api.Emitter, options *api.TrafficFilteringOptions) error {

tap/extensions/redis/read.go

@@ -313,7 +313,7 @@ func (p *RedisProtocol) Read() (packet *RedisPacket, err error) {
 				packet.Value = fmt.Sprintf("%s]", packet.Value)
 			}
 		default:
-			msg := fmt.Sprintf("Unrecognized element in Redis array: %v\n", reflect.TypeOf(array[0]))
+			msg := fmt.Sprintf("Unrecognized element in Redis array: %v", reflect.TypeOf(array[0]))
 			err = errors.New(msg)
 			return
 		}
@@ -333,7 +333,7 @@ func (p *RedisProtocol) Read() (packet *RedisPacket, err error) {
 	case int64:
 		packet.Value = fmt.Sprintf("%d", x.(int64))
 	default:
-		msg := fmt.Sprintf("Unrecognized Redis data type: %v\n", reflect.TypeOf(x))
+		msg := fmt.Sprintf("Unrecognized Redis data type: %v", reflect.TypeOf(x))
 		err = errors.New(msg)
 		return
 	}

tap/passive_tapper.go

@@ -193,7 +193,7 @@ func startPassiveTapper(outputItems chan *api.OutputChannelItem) {
 	}
 
 	if err := diagnose.DumpMemoryProfile(*memprofile); err != nil {
-		logger.Log.Errorf("Error dumping memory profile %v\n", err)
+		logger.Log.Errorf("Error dumping memory profile %v", err)
 	}
 
 	assembler.waitAndDump()

tap/tcp_assembler.go

@@ -78,7 +78,7 @@ func (a *tcpAssembler) processPackets(dumpPacket bool, packets <-chan source.Tcp
 			if *checksum {
 				err := tcp.SetNetworkLayerForChecksum(packet.NetworkLayer())
 				if err != nil {
-					logger.Log.Fatalf("Failed to set network layer for checksum: %s\n", err)
+					logger.Log.Fatalf("Failed to set network layer for checksum: %s", err)
 				}
 			}
 			c := context{

tap/tcp_reader.go

@@ -66,7 +66,7 @@ func (h *tcpReader) Read(p []byte) (int, error) {
 			clientHello := tlsx.ClientHello{}
 			err := clientHello.Unmarshall(msg.bytes)
 			if err == nil {
-				logger.Log.Debugf("Detected TLS client hello with SNI %s\n", clientHello.SNI)
+				logger.Log.Debugf("Detected TLS client hello with SNI %s", clientHello.SNI)
 				// TODO: Throws `panic: runtime error: invalid memory address or nil pointer dereference` error.
 				// numericPort, _ := strconv.Atoi(h.tcpID.DstPort)
 				// h.outboundLinkWriter.WriteOutboundLink(h.tcpID.SrcIP, h.tcpID.DstIP, numericPort, clientHello.SNI, TLSProtocol)

tap/tcp_streams_map.go

@@ -46,7 +46,7 @@ func (streamMap *tcpStreamMap) closeTimedoutTcpStreamChannels() {
 				if !stream.isClosed && time.Now().After(streamWrapper.createdAt.Add(tcpStreamChannelTimeout)) {
 					stream.Close()
 					diagnose.AppStats.IncDroppedTcpStreams()
-					logger.Log.Debugf("Dropped an unidentified TCP stream because of timeout. Total dropped: %d Total Goroutines: %d Timeout (ms): %d\n",
+					logger.Log.Debugf("Dropped an unidentified TCP stream because of timeout. Total dropped: %d Total Goroutines: %d Timeout (ms): %d",
 						diagnose.AppStats.DroppedTcpStreams, runtime.NumGoroutine(), tcpStreamChannelTimeout/1000000)
 				}
 			} else {

tap/tester/tester.go

@@ -33,7 +33,7 @@ func loadExtensions() ([]*tapApi.Extension, error) {
 			continue
 		}
 
-		logger.Log.Infof("Loading extension: %s\n", filename)
+		logger.Log.Infof("Loading extension: %s", filename)
 
 		extension := &tapApi.Extension{
 			Path: path.Join(extensionsDir, filename),
@@ -55,7 +55,7 @@ func loadExtensions() ([]*tapApi.Extension, error) {
 		dissector, ok := symDissector.(tapApi.Dissector)
 
 		if !ok {
-			return nil, errors.Errorf("Symbol Dissector type error: %v %T\n", file, symDissector)
+			return nil, errors.Errorf("Symbol Dissector type error: %v %T", file, symDissector)
 		}
 
 		dissector.Register(extension)
@@ -68,7 +68,7 @@ func loadExtensions() ([]*tapApi.Extension, error) {
 	})
 
 	for _, extension := range extensions {
-		logger.Log.Infof("Extension Properties: %+v\n", extension)
+		logger.Log.Infof("Extension Properties: %+v", extension)
 	}
 
 	return extensions, nil
@@ -92,7 +92,7 @@ func internalRun() error {
 
 	tap.StartPassiveTapper(&opts, outputItems, extenssions, &tapOpts)
 
-	logger.Log.Infof("Tapping, press enter to exit...\n")
+	logger.Log.Infof("Tapping, press enter to exit...")
 	reader := bufio.NewReader(os.Stdin)
 	reader.ReadLine()
 	return nil
@@ -104,9 +104,9 @@ func main() {
 	if err != nil {
 		switch err := err.(type) {
 		case *errors.Error:
-			logger.Log.Errorf("Error: %v\n", err.ErrorStack())
+			logger.Log.Errorf("Error: %v", err.ErrorStack())
 		default:
-			logger.Log.Errorf("Error: %v\n", err)
+			logger.Log.Errorf("Error: %v", err)
 		}
 
 		os.Exit(1)

ui/src/components/TrafficPage.tsx

@@ -189,6 +189,16 @@ export const TrafficPage: React.FC<TrafficPageProps> = ({setAnalyzeStatus, onTLS
                 const entryData = await api.getEntry(focusedEntryId);
                 setSelectedEntryData(entryData);
             } catch (error) {
+                toast[error.response.data.type](`Entry[${focusedEntryId}]: ${error.response.data.msg}`, {
+                    position: "bottom-right",
+                    theme: "colored",
+                    autoClose: error.response.data.autoClose,
+                    hideProgressBar: false,
+                    closeOnClick: true,
+                    pauseOnHover: true,
+                    draggable: true,
+                    progress: undefined,
+                });
                 console.error(error);
             }
         })()