Fix resource limits (#110)

Adding the upload interval as parameter to tap function

api/go.mod

@@ -14,6 +14,7 @@ require (
 	github.com/google/martian v2.1.0+incompatible
 	github.com/gorilla/websocket v1.4.2
 	github.com/leodido/go-urn v1.2.1 // indirect
+	github.com/romana/rlog v0.0.0-20171115192701-f018bc92e7d7
 	github.com/up9inc/mizu/shared v0.0.0
 	github.com/up9inc/mizu/tap v0.0.0
 	go.mongodb.org/mongo-driver v1.5.1

api/go.sum

@@ -263,6 +263,8 @@ github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:
 github.com/rogpeppe/go-internal v1.1.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/rogpeppe/go-internal v1.2.2/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
+github.com/romana/rlog v0.0.0-20171115192701-f018bc92e7d7 h1:jkvpcEatpwuMF5O5LVxTnehj6YZ/aEZN4NWD/Xml4pI=
+github.com/romana/rlog v0.0.0-20171115192701-f018bc92e7d7/go.mod h1:KTrHyWpO1sevuXPZwyeZc72ddWRFqNSKDFl7uVWKpg0=
 github.com/savsgio/gotils v0.0.0-20200117113501-90175b0fbe3f/go.mod h1:lHhJedqxCoHN+zMtwGNTXWmF0u9Jt363FYRhV6g0CdY=
 github.com/savsgio/gotils v0.0.0-20200616100644-13ff1fd2c28c h1:KKqhycXW1WVNkX7r4ekTV2gFkbhdyihlWD8c0/FiWmk=
 github.com/savsgio/gotils v0.0.0-20200616100644-13ff1fd2c28c/go.mod h1:TWNAOTaVzGOXq8RbEvHnhzA/A2sLZzgn0m6URjnukY8=

api/main.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"github.com/gofiber/fiber/v2"
 	"github.com/gorilla/websocket"
+	"github.com/romana/rlog"
 	"github.com/up9inc/mizu/shared"
 	"github.com/up9inc/mizu/tap"
 	"mizuserver/pkg/api"
@@ -50,7 +51,7 @@ func main() {
 		tapTargets := getTapTargets()
 		if tapTargets != nil {
 			tap.SetFilterAuthorities(tapTargets)
-			fmt.Println("Filtering for the following authorities:", tap.GetFilterIPs())
+			rlog.Infof("Filtering for the following authorities: %v", tap.GetFilterIPs())
 		}
 
 		harOutputChannel, outboundLinkOutputChannel := tap.StartPassiveTapper(tapOpts)
@@ -76,7 +77,7 @@ func main() {
 	signal.Notify(signalChan, os.Interrupt)
 	<-signalChan
 
-	fmt.Println("Exiting")
+	rlog.Info("Exiting")
 }
 
 func hostApi(socketHarOutputChannel chan<- *tap.OutputChannelItem) {
@@ -149,13 +150,13 @@ func pipeChannelToSocket(connection *websocket.Conn, messageDataChannel <-chan *
 	for messageData := range messageDataChannel {
 		marshaledData, err := models.CreateWebsocketTappedEntryMessage(messageData)
 		if err != nil {
-			fmt.Printf("error converting message to json %s, (%v,%+v)\n", err, err, err)
+			rlog.Infof("error converting message to json %s, (%v,%+v)\n", err, err, err)
 			continue
 		}
 
 		err = connection.WriteMessage(websocket.TextMessage, marshaledData)
 		if err != nil {
-			fmt.Printf("error sending message through socket server %s, (%v,%+v)\n", err, err, err)
+			rlog.Infof("error sending message through socket server %s, (%v,%+v)\n", err, err, err)
 			continue
 		}
 	}

api/pkg/api/main.go

@@ -5,6 +5,11 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"github.com/google/martian/har"
+	"github.com/romana/rlog"
+	"github.com/up9inc/mizu/tap"
+	"go.mongodb.org/mongo-driver/bson/primitive"
+	"mizuserver/pkg/holder"
 	"net/url"
 	"os"
 	"path"
@@ -12,10 +17,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/google/martian/har"
-	"github.com/up9inc/mizu/tap"
-	"go.mongodb.org/mongo-driver/bson/primitive"
-
 	"mizuserver/pkg/database"
 	"mizuserver/pkg/models"
 	"mizuserver/pkg/resolver"
@@ -28,7 +29,7 @@ func init() {
 	errOut := make(chan error, 100)
 	res, err := resolver.NewFromInCluster(errOut)
 	if err != nil {
-		fmt.Printf("error creating k8s resolver %s", err)
+		rlog.Infof("error creating k8s resolver %s", err)
 		return
 	}
 	ctx := context.Background()
@@ -37,12 +38,13 @@ func init() {
 		for {
 			select {
 			case err := <-errOut:
-				fmt.Printf("name resolving error %s", err)
+				rlog.Infof("name resolving error %s", err)
 			}
 		}
 	}()
 
 	k8sResolver = res
+	holder.SetResolver(res)
 }
 
 func StartReadingEntries(harChannel <-chan *tap.OutputChannelItem, workingDir *string) {
@@ -70,7 +72,7 @@ func startReadingFiles(workingDir string) {
 		sort.Sort(utils.ByModTime(harFiles))
 
 		if len(harFiles) == 0 {
-			fmt.Printf("Waiting for new files\n")
+			rlog.Infof("Waiting for new files\n")
 			time.Sleep(3 * time.Second)
 			continue
 		}
@@ -126,9 +128,24 @@ func saveHarToDb(entry *har.Entry, connectionInfo *tap.ConnectionInfo) {
 		resolvedDestination string
 	)
 	if k8sResolver != nil {
-		resolvedSource = k8sResolver.Resolve(connectionInfo.ClientIP)
-		resolvedDestination = k8sResolver.Resolve(fmt.Sprintf("%s:%s", connectionInfo.ServerIP, connectionInfo.ServerPort))
+		unresolvedSource := connectionInfo.ClientIP
+		resolvedSource = k8sResolver.Resolve(unresolvedSource)
+		if resolvedSource == "" {
+			rlog.Debugf("Cannot find resolved name to source: %s\n", unresolvedSource)
+			if os.Getenv("SKIP_NOT_RESOLVED_SOURCE") == "1" {
+				return
+			}
+		}
+		unresolvedDestination := fmt.Sprintf("%s:%s", connectionInfo.ServerIP, connectionInfo.ServerPort)
+		resolvedDestination = k8sResolver.Resolve(unresolvedDestination)
+		if resolvedDestination == "" {
+			rlog.Debugf("Cannot find resolved name to dest: %s\n", unresolvedDestination)
+			if os.Getenv("SKIP_NOT_RESOLVED_DEST") == "1" {
+				return
+			}
+		}
 	}
+
 	mizuEntry := models.MizuEntry{
 		EntryId:             entryId,
 		Entry:               string(entryBytes), // simple way to store it and not convert to bytes

api/pkg/api/socket_server_handlers.go

@@ -2,8 +2,8 @@ package api
 
 import (
 	"encoding/json"
-	"fmt"
 	"github.com/antoniodipinto/ikisocket"
+	"github.com/romana/rlog"
 	"github.com/up9inc/mizu/shared"
 	"github.com/up9inc/mizu/tap"
 	"mizuserver/pkg/controllers"
@@ -25,18 +25,18 @@ func init() {
 
 func (h *RoutesEventHandlers) WebSocketConnect(ep *ikisocket.EventPayload) {
 	if ep.Kws.GetAttribute("is_tapper") == true {
-		fmt.Println(fmt.Sprintf("Websocket Connection event - Tapper connected: %s", ep.SocketUUID))
+		rlog.Infof("Websocket Connection event - Tapper connected: %s", ep.SocketUUID)
 	} else {
-		fmt.Println(fmt.Sprintf("Websocket Connection event - Browser socket connected: %s", ep.SocketUUID))
+		rlog.Infof("Websocket Connection event - Browser socket connected: %s", ep.SocketUUID)
 		browserClientSocketUUIDs = append(browserClientSocketUUIDs, ep.SocketUUID)
 	}
 }
 
 func (h *RoutesEventHandlers) WebSocketDisconnect(ep *ikisocket.EventPayload) {
 	if ep.Kws.GetAttribute("is_tapper") == true {
-		fmt.Println(fmt.Sprintf("Disconnection event - Tapper connected: %s", ep.SocketUUID))
+		rlog.Infof("Disconnection event - Tapper connected: %s", ep.SocketUUID)
 	} else {
-		fmt.Println(fmt.Sprintf("Disconnection event - Browser socket connected: %s", ep.SocketUUID))
+		rlog.Infof("Disconnection event - Browser socket connected: %s", ep.SocketUUID)
 		removeSocketUUIDFromBrowserSlice(ep.SocketUUID)
 	}
 }
@@ -47,29 +47,29 @@ func broadcastToBrowserClients(message []byte) {
 
 func (h *RoutesEventHandlers) WebSocketClose(ep *ikisocket.EventPayload) {
 	if ep.Kws.GetAttribute("is_tapper") == true {
-		fmt.Println(fmt.Sprintf("Websocket Close event - Tapper connected: %s", ep.SocketUUID))
+		rlog.Infof("Websocket Close event - Tapper connected: %s", ep.SocketUUID)
 	} else {
-		fmt.Println(fmt.Sprintf("Websocket  Close event - Browser socket connected: %s", ep.SocketUUID))
+		rlog.Infof("Websocket  Close event - Browser socket connected: %s", ep.SocketUUID)
 		removeSocketUUIDFromBrowserSlice(ep.SocketUUID)
 	}
 }
 
 func (h *RoutesEventHandlers) WebSocketError(ep *ikisocket.EventPayload) {
-	fmt.Println(fmt.Sprintf("Socket error - Socket uuid : %s %v", ep.SocketUUID, ep.Error))
+	rlog.Infof("Socket error - Socket uuid : %s %v", ep.SocketUUID, ep.Error)
 }
 
 func (h *RoutesEventHandlers) WebSocketMessage(ep *ikisocket.EventPayload) {
 	var socketMessageBase shared.WebSocketMessageMetadata
 	err := json.Unmarshal(ep.Data, &socketMessageBase)
 	if err != nil {
-		fmt.Printf("Could not unmarshal websocket message %v\n", err)
+		rlog.Infof("Could not unmarshal websocket message %v\n", err)
 	} else {
 		switch socketMessageBase.MessageType {
 		case shared.WebSocketMessageTypeTappedEntry:
 			var tappedEntryMessage models.WebSocketTappedEntryMessage
 			err := json.Unmarshal(ep.Data, &tappedEntryMessage)
 			if err != nil {
-				fmt.Printf("Could not unmarshal message of message type %s %v\n", socketMessageBase.MessageType, err)
+				rlog.Infof("Could not unmarshal message of message type %s %v\n", socketMessageBase.MessageType, err)
 			} else {
 				h.SocketHarOutChannel <- tappedEntryMessage.Data
 			}
@@ -77,13 +77,13 @@ func (h *RoutesEventHandlers) WebSocketMessage(ep *ikisocket.EventPayload) {
 			var statusMessage shared.WebSocketStatusMessage
 			err := json.Unmarshal(ep.Data, &statusMessage)
 			if err != nil {
-				fmt.Printf("Could not unmarshal message of message type %s %v\n", socketMessageBase.MessageType, err)
+				rlog.Infof("Could not unmarshal message of message type %s %v\n", socketMessageBase.MessageType, err)
 			} else {
 				controllers.TapStatus = statusMessage.TappingStatus
 				broadcastToBrowserClients(ep.Data)
 			}
 		default:
-			fmt.Printf("Received socket message of type %s for which no handlers are defined", socketMessageBase.MessageType)
+			rlog.Infof("Received socket message of type %s for which no handlers are defined", socketMessageBase.MessageType)
 		}
 	}
 }

api/pkg/controllers/entries_controller.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"github.com/gofiber/fiber/v2"
 	"github.com/google/martian/har"
+	"github.com/romana/rlog"
 	"mizuserver/pkg/database"
 	"mizuserver/pkg/models"
 	"mizuserver/pkg/up9"
@@ -140,6 +141,8 @@ func GetHARs(c *fiber.Ctx) error {
 }
 
 func UploadEntries(c *fiber.Ctx) error {
+	rlog.Infof("Upload entries - started\n")
+
 	uploadRequestBody := &models.UploadEntriesRequestBody{}
 	if err := c.QueryParser(uploadRequestBody); err != nil {
 		return c.Status(fiber.StatusBadRequest).JSON(err)
@@ -150,9 +153,13 @@ func UploadEntries(c *fiber.Ctx) error {
 	if up9.GetAnalyzeInfo().IsAnalyzing {
 		return c.Status(fiber.StatusBadRequest).SendString("Cannot analyze, mizu is already analyzing")
 	}
-
-	token, _ := up9.CreateAnonymousToken(uploadRequestBody.Dest)
-	go up9.UploadEntriesImpl(token.Token, token.Model, uploadRequestBody.Dest)
+	rlog.Infof("Upload entries - creating token. dest %s\n", uploadRequestBody.Dest)
+	token, err := up9.CreateAnonymousToken(uploadRequestBody.Dest)
+	if err != nil {
+		return c.Status(fiber.StatusServiceUnavailable).SendString("Can't get token")
+	}
+	rlog.Infof("Upload entries - uploading. token: %s model: %s\n", token.Token, token.Model)
+	go up9.UploadEntriesImpl(token.Token, token.Model, uploadRequestBody.Dest, uploadRequestBody.SleepIntervalSec)
 	return c.Status(fiber.StatusOK).SendString("OK")
 }
 

api/pkg/controllers/resolving_controller.go

@@ -0,0 +1,11 @@
+package controllers
+
+import (
+	"github.com/gofiber/fiber/v2"
+	"mizuserver/pkg/holder"
+)
+
+func GetCurrentResolvingInformation(c *fiber.Ctx) error {
+	return c.Status(fiber.StatusOK).JSON(holder.GetResolver().GetMap())
+}
+

api/pkg/holder/main.go

@@ -0,0 +1,14 @@
+package holder
+
+import "mizuserver/pkg/resolver"
+
+var k8sResolver *resolver.Resolver
+
+func SetResolver(param *resolver.Resolver) {
+	k8sResolver = param
+}
+
+func GetResolver() *resolver.Resolver {
+	return k8sResolver
+}
+

api/pkg/models/models.go

@@ -100,8 +100,6 @@ func (fedex *FullEntryDetailsExtra) UnmarshalData(entry *MizuEntry) error {
 	return nil
 }
 
-
-
 type EntryData struct {
 	Entry               string `json:"entry,omitempty"`
 	ResolvedDestination string `json:"resolvedDestination,omitempty" gorm:"column:resolvedDestination"`
@@ -114,7 +112,8 @@ type EntriesFilter struct {
 }
 
 type UploadEntriesRequestBody struct {
-	Dest string `query:"dest"`
+	Dest             string `query:"dest"`
+	SleepIntervalSec int    `query:"interval"`
 }
 
 type HarFetchRequestBody struct {

api/pkg/resolver/resolver.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"github.com/romana/rlog"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 
 	corev1 "k8s.io/api/core/v1"
@@ -42,6 +43,10 @@ func (resolver *Resolver) Resolve(name string) string {
 	return resolvedName
 }
 
+func (resolver *Resolver) GetMap() map[string]string {
+	return resolver.nameMap
+}
+
 func (resolver *Resolver) CheckIsServiceIP(address string) bool {
 	_, isFound := resolver.serviceMap[address]
 	return isFound
@@ -147,10 +152,10 @@ func (resolver *Resolver) watchServices(ctx context.Context) error {
 func (resolver *Resolver) saveResolvedName(key string, resolved string, eventType watch.EventType) {
 	if eventType == watch.Deleted {
 		delete(resolver.nameMap, key)
-		// fmt.Printf("setting %s=nil\n", key)
+		rlog.Infof("setting %s=nil\n", key)
 	} else {
 		resolver.nameMap[key] = resolved
-		// fmt.Printf("setting %s=%s\n", key, resolved)
+		rlog.Infof("setting %s=%s\n", key, resolved)
 	}
 }
 
@@ -171,7 +176,7 @@ func (resolver *Resolver) infiniteErrorHandleRetryFunc(ctx context.Context, fun
 			var statusError *k8serrors.StatusError
 			if errors.As(err, &statusError) {
 				if statusError.ErrStatus.Reason == metav1.StatusReasonForbidden {
-					fmt.Printf("Resolver loop encountered permission error, aborting event listening - %v\n", err)
+					rlog.Infof("Resolver loop encountered permission error, aborting event listening - %v\n", err)
 					return
 				}
 			}

api/pkg/routes/public_routes.go

@@ -13,6 +13,7 @@ func EntriesRoutes(fiberApp *fiber.App) {
 	routeGroup.Get("/entries/:entryId", controllers.GetEntry) // get single (full) entry
 	routeGroup.Get("/exportEntries", controllers.GetFullEntries)
 	routeGroup.Get("/uploadEntries", controllers.UploadEntries)
+	routeGroup.Get("/resolving", controllers.GetCurrentResolvingInformation)
 
 	routeGroup.Get("/har", controllers.GetHARs)
 

api/pkg/up9/main.go

@@ -5,6 +5,7 @@ import (
 	"compress/zlib"
 	"encoding/json"
 	"fmt"
+	"github.com/romana/rlog"
 	"github.com/up9inc/mizu/shared"
 	"io/ioutil"
 	"log"
@@ -35,7 +36,7 @@ func getGuestToken(url string, target *GuestToken) error {
 		return err
 	}
 	defer resp.Body.Close()
-
+	rlog.Infof("Got token from the server, starting to json decode... status code: %v", resp.StatusCode)
 	return json.NewDecoder(resp.Body).Decode(target)
 }
 
@@ -46,7 +47,7 @@ func CreateAnonymousToken(envPrefix string) (*GuestToken, error) {
 	}
 	token := &GuestToken{}
 	if err := getGuestToken(tokenUrl, token); err != nil {
-		fmt.Println(err)
+		rlog.Infof("Failed to get token, %s", err)
 		return nil, err
 	}
 	return token, nil
@@ -111,19 +112,19 @@ func GetAnalyzeInfo() *shared.AnalyzeStatus {
 	}
 }
 
-func UploadEntriesImpl(token string, model string, envPrefix string) {
+func UploadEntriesImpl(token string, model string, envPrefix string, sleepIntervalSec int) {
 	analyzeInformation.IsAnalyzing = true
 	analyzeInformation.AnalyzedModel = model
 	analyzeInformation.AnalyzeToken = token
 	analyzeInformation.AnalyzeDestination = envPrefix
 
-	sleepTime := time.Second * 10
+	sleepTime := time.Second * time.Duration(sleepIntervalSec)
 
 	var timestampFrom int64 = 0
 
 	for {
 		timestampTo := time.Now().UnixNano() / int64(time.Millisecond)
-		fmt.Printf("Getting entries from %v, to %v\n", timestampFrom, timestampTo)
+		rlog.Infof("Getting entries from %v, to %v\n", timestampFrom, timestampTo)
 		entriesArray := database.GetEntriesFromDb(timestampFrom, timestampTo)
 
 		if len(entriesArray) > 0 {
@@ -136,12 +137,12 @@ func UploadEntriesImpl(token string, model string, envPrefix string) {
 				}
 				fullEntriesExtra = append(fullEntriesExtra, harEntry)
 			}
-			fmt.Printf("About to upload %v entries\n", len(fullEntriesExtra))
+			rlog.Infof("About to upload %v entries\n", len(fullEntriesExtra))
 
 			body, jMarshalErr := json.Marshal(fullEntriesExtra)
 			if jMarshalErr != nil {
 				analyzeInformation.Reset()
-				fmt.Println("Stopping analyzing")
+				rlog.Infof("Stopping analyzing")
 				log.Fatal(jMarshalErr)
 			}
 
@@ -164,16 +165,16 @@ func UploadEntriesImpl(token string, model string, envPrefix string) {
 
 			if _, postErr := http.DefaultClient.Do(req); postErr != nil {
 				analyzeInformation.Reset()
-				log.Println("Stopping analyzing")
+				rlog.Info("Stopping analyzing")
 				log.Fatal(postErr)
 			}
-			fmt.Printf("Finish uploading %v entries to %s\n", len(entriesArray), GetTrafficDumpUrl(envPrefix, model))
+			rlog.Infof("Finish uploading %v entries to %s\n", len(entriesArray), GetTrafficDumpUrl(envPrefix, model))
 
 		} else {
-			fmt.Println("Nothing to upload")
+			rlog.Infof("Nothing to upload")
 		}
 
-		fmt.Printf("Sleeping for %v...\n", sleepTime)
+		rlog.Infof("Sleeping for %v...\n", sleepTime)
 		time.Sleep(sleepTime)
 		timestampFrom = timestampTo
 	}

api/pkg/utils/utils.go

@@ -1,8 +1,8 @@
 package utils
 
 import (
-	"fmt"
 	"github.com/gofiber/fiber/v2"
+	"github.com/romana/rlog"
 	"log"
 	"net/url"
 	"os"
@@ -21,7 +21,7 @@ func StartServer(app *fiber.App) {
 
 	go func() {
 		_ = <-signals
-		fmt.Println("Shutting down...")
+		rlog.Infof("Shutting down...")
 		_ = app.Shutdown()
 	}()
 

cli/cmd/tap.go

@@ -21,6 +21,7 @@ type MizuTapOptions struct {
 	MizuImage              string
 	PlainTextFilterRegexes []string
 	TapOutgoing            bool
+	SleepIntervalSec       uint16
 }
 
 var mizuTapOptions = &MizuTapOptions{}
@@ -64,6 +65,7 @@ func init() {
 	tapCmd.Flags().StringVarP(&mizuTapOptions.Namespace, "namespace", "n", "", "Namespace selector")
 	tapCmd.Flags().BoolVar(&mizuTapOptions.Analyze, "analyze", false, "Uploads traffic to UP9 for further analysis (Beta)")
 	tapCmd.Flags().StringVar(&mizuTapOptions.AnalyzeDestination, "dest", "up9.app", "Destination environment")
+	tapCmd.Flags().Uint16VarP(&mizuTapOptions.SleepIntervalSec, "upload-interval", "", 10, "Interval in seconds for uploading data to UP9")
 	tapCmd.Flags().BoolVarP(&mizuTapOptions.AllNamespaces, "all-namespaces", "A", false, "Tap all namespaces")
 	tapCmd.Flags().StringVarP(&mizuTapOptions.KubeConfigPath, "kube-config", "k", "", "Path to kube-config file")
 	tapCmd.Flags().StringVarP(&mizuTapOptions.MizuImage, "mizu-image", "", fmt.Sprintf("gcr.io/up9-docker-hub/mizu/%s:latest", mizu.Branch), "Custom image for mizu collector")

cli/cmd/tapRunner.go

@@ -3,6 +3,7 @@ package cmd
 import (
 	"context"
 	"fmt"
+	"github.com/romana/rlog"
 	"github.com/up9inc/mizu/cli/debounce"
 	"github.com/up9inc/mizu/cli/kubernetes"
 	"github.com/up9inc/mizu/cli/mizu"
@@ -252,13 +253,14 @@ func portForwardApiPod(ctx context.Context, kubernetesProvider *kubernetes.Provi
 
 				time.Sleep(time.Second * 5) // Waiting to be sure the proxy is ready
 				if tappingOptions.Analyze {
-					url_path := fmt.Sprintf("http://%s/api/uploadEntries?dest=%s", mizuProxiedUrl, tappingOptions.AnalyzeDestination)
+					url_path := fmt.Sprintf("http://%s/api/uploadEntries?dest=%s&interval=%v", mizuProxiedUrl, url.QueryEscape(tappingOptions.AnalyzeDestination), tappingOptions.SleepIntervalSec)
 					u, err := url.ParseRequestURI(url_path)
 					if err != nil {
 						log.Fatal(fmt.Sprintf("Failed parsing the URL %v\n", err))
 					}
-					if response, err := http.Get(u.String()); err != nil && response.StatusCode != 200 {
-						fmt.Printf("error sending upload entries req %v\n", err)
+					rlog.Debugf("Sending get request to %v\n", u.String())
+					if response, err := http.Get(u.String()); err != nil || response.StatusCode != 200 {
+						fmt.Printf("error sending upload entries req, status code: %v, err: %v\n", response.StatusCode, err)
 					} else {
 						fmt.Printf(mizu.Purple, "Traffic is uploading to UP9 for further analsys")
 						fmt.Println()
@@ -324,7 +326,8 @@ func waitForFinish(ctx context.Context, cancel context.CancelFunc) {
 }
 
 func syncApiStatus(ctx context.Context, cancel context.CancelFunc, tappingOptions *MizuTapOptions) {
-	controlSocket, err := mizu.CreateControlSocket(fmt.Sprintf("ws://%s/ws", kubernetes.GetMizuCollectorProxiedHostAndPath(tappingOptions.GuiPort, mizu.ResourcesNamespace, mizu.AggregatorPodName)))
+	controlSocketStr := fmt.Sprintf("ws://%s/ws", kubernetes.GetMizuCollectorProxiedHostAndPath(tappingOptions.GuiPort, mizu.ResourcesNamespace, mizu.AggregatorPodName))
+	controlSocket, err := mizu.CreateControlSocket(controlSocketStr)
 	if err != nil {
 		fmt.Printf("error establishing control socket connection %s\n", err)
 		cancel()
@@ -337,7 +340,7 @@ func syncApiStatus(ctx context.Context, cancel context.CancelFunc, tappingOption
 		default:
 			err = controlSocket.SendNewTappedPodsListMessage(currentlyTappedPods)
 			if err != nil {
-				fmt.Printf("error Sending message via control socket %s\n", err)
+				rlog.Debugf("error Sending message via control socket %v, error: %s\n", controlSocketStr, err)
 			}
 			time.Sleep(10 * time.Second)
 		}

cli/go.mod

@@ -4,6 +4,7 @@ go 1.16
 
 require (
 	github.com/gorilla/websocket v1.4.2
+	github.com/romana/rlog v0.0.0-20171115192701-f018bc92e7d7
 	github.com/spf13/cobra v1.1.3
 	github.com/up9inc/mizu/shared v0.0.0
 	k8s.io/api v0.21.2

cli/go.sum

@@ -550,6 +550,8 @@ github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af h1:gu+uRPtBe88sK
 github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6SoW27p1b0cqNHllgS5HIMJraePCO15w5zCzIWYg=
 github.com/rogpeppe/go-internal v1.3.0 h1:RR9dF3JtopPvtkroDZuVD7qquD0bnHlKSqaQhgwt8yk=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
+github.com/romana/rlog v0.0.0-20171115192701-f018bc92e7d7 h1:jkvpcEatpwuMF5O5LVxTnehj6YZ/aEZN4NWD/Xml4pI=
+github.com/romana/rlog v0.0.0-20171115192701-f018bc92e7d7/go.mod h1:KTrHyWpO1sevuXPZwyeZc72ddWRFqNSKDFl7uVWKpg0=
 github.com/russross/blackfriday v1.5.2 h1:HyvC0ARfnZBqnXwABFeSZHpKvJHJJfPz81GNueLj0oo=
 github.com/russross/blackfriday v1.5.2/go.mod h1:JO/DiYxRf+HjHt06OyowR9PTA263kcR/rfWxYHBV53g=
 github.com/russross/blackfriday/v2 v2.0.1 h1:lPqVAte+HuHNfhJ/0LC98ESWRz8afy9tM/0RK8m9o+Q=

cli/kubernetes/provider.go

@@ -14,6 +14,7 @@ import (
 	rbac "k8s.io/api/rbac/v1"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	resource "k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/apimachinery/pkg/watch"
 	applyconfapp "k8s.io/client-go/applyconfigurations/apps/v1"
@@ -75,6 +76,24 @@ func (provider *Provider) CreateMizuAggregatorPod(ctx context.Context, namespace
 	if err != nil {
 		return nil, err
 	}
+
+	cpuLimit, err := resource.ParseQuantity("750m")
+	if err != nil {
+		return nil, errors.New("invalid cpu limit for aggregator container")
+	}
+	memLimit, err := resource.ParseQuantity("512Mi")
+	if err != nil {
+		return nil, errors.New("invalid memory limit for aggregator container")
+	}
+	cpuRequests, err := resource.ParseQuantity("50m")
+	if err != nil {
+		return nil, errors.New("invalid cpu request for aggregator container")
+	}
+	memRequests, err := resource.ParseQuantity("50Mi")
+	if err != nil {
+		return nil, errors.New("invalid memory request for aggregator container")
+	}
+
 	pod := &core.Pod{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      podName,
@@ -98,6 +117,16 @@ func (provider *Provider) CreateMizuAggregatorPod(ctx context.Context, namespace
 							Value: string(marshaledFilteringOptions),
 						},
 					},
+					Resources: core.ResourceRequirements{
+						Limits: core.ResourceList{
+							"cpu": cpuLimit,
+							"memory": memLimit,
+						},
+						Requests: core.ResourceList{
+							"cpu": cpuRequests,
+							"memory": memRequests,
+						},
+					},
 				},
 			},
 			DNSPolicy:                     core.DNSClusterFirstWithHostNet,
@@ -336,6 +365,32 @@ func (provider *Provider) ApplyMizuTapperDaemonSet(ctx context.Context, namespac
 			),
 		),
 	)
+	cpuLimit, err := resource.ParseQuantity("500m")
+	if err != nil {
+		return errors.New("invalid cpu limit for tapper container")
+	}
+	memLimit, err := resource.ParseQuantity("1Gi")
+	if err != nil {
+		return errors.New("invalid memory limit for tapper container")
+	}
+	cpuRequests, err := resource.ParseQuantity("50m")
+	if err != nil {
+		return errors.New("invalid cpu request for tapper container")
+	}
+	memRequests, err := resource.ParseQuantity("50Mi")
+	if err != nil {
+		return errors.New("invalid memory request for tapper container")
+	}
+	agentResourceLimits := core.ResourceList{
+		"cpu": cpuLimit,
+		"memory": memLimit,
+	}
+	agentResourceRequests := core.ResourceList{
+		"cpu": cpuRequests,
+		"memory": memRequests,
+	}
+	agentResources := applyconfcore.ResourceRequirements().WithRequests(agentResourceRequests).WithLimits(agentResourceLimits)
+	agentContainer.WithResources(agentResources)
 
 	nodeNames := make([]string, 0, len(nodeToTappedPodIPMap))
 	for nodeName := range nodeToTappedPodIPMap {

tap/go.mod

@@ -5,8 +5,7 @@ go 1.16
 require (
 	github.com/google/gopacket v1.1.19
 	github.com/google/martian v2.1.0+incompatible
-	github.com/gorilla/websocket v1.4.2
 	github.com/orcaman/concurrent-map v0.0.0-20210106121528-16402b402231
-	github.com/patrickmn/go-cache v2.1.0+incompatible
+	github.com/romana/rlog v0.0.0-20171115192701-f018bc92e7d7
 	golang.org/x/net v0.0.0-20210421230115-4e50805a0758
 )

tap/go.sum

@@ -2,26 +2,24 @@ github.com/google/gopacket v1.1.19 h1:ves8RnFZPGiFnTS0uPQStjwru6uO6h+nlr9j6fL7kF
 github.com/google/gopacket v1.1.19/go.mod h1:iJ8V8n6KS+z2U1A8pUwu8bW5SyEMkXJB8Yo/Vo+TKTo=
 github.com/google/martian v2.1.0+incompatible h1:/CP5g8u/VJHijgedC/Legn3BAbAaWPgecwXBIDzw5no=
 github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=
-github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc=
-github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
-github.com/orcaman/concurrent-map v0.0.0-20210501183033-44dafcb38ecc h1:Ak86L+yDSOzKFa7WM5bf5itSOo1e3Xh8bm5YCMUXIjQ=
-github.com/orcaman/concurrent-map v0.0.0-20210501183033-44dafcb38ecc/go.mod h1:Lu3tH6HLW3feq74c2GC+jIMS/K2CFcDWnWD9XkenwhI=
-github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
-github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
+github.com/orcaman/concurrent-map v0.0.0-20210106121528-16402b402231 h1:fa50YL1pzKW+1SsBnJDOHppJN9stOEwS+CRWyUtyYGU=
+github.com/orcaman/concurrent-map v0.0.0-20210106121528-16402b402231/go.mod h1:Lu3tH6HLW3feq74c2GC+jIMS/K2CFcDWnWD9XkenwhI=
+github.com/romana/rlog v0.0.0-20171115192701-f018bc92e7d7 h1:jkvpcEatpwuMF5O5LVxTnehj6YZ/aEZN4NWD/Xml4pI=
+github.com/romana/rlog v0.0.0-20171115192701-f018bc92e7d7/go.mod h1:KTrHyWpO1sevuXPZwyeZc72ddWRFqNSKDFl7uVWKpg0=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/lint v0.0.0-20200302205851-738671d3881b/go.mod h1:3xt1FjdF8hUf6vQPIChWIBhFzV8gjjsPE/fR3IyQdNY=
 golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
-golang.org/x/net v0.0.0-20210525063256-abc453219eb5 h1:wjuX4b5yYQnEQHzd+CBcrcC6OVR2J1CN6mUy0oSxIPo=
-golang.org/x/net v0.0.0-20210525063256-abc453219eb5/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
+golang.org/x/net v0.0.0-20210421230115-4e50805a0758 h1:aEpZnXcAmXkd6AvLb2OPt+EN1Zu/8Ne3pCqPjja5PXY=
+golang.org/x/net v0.0.0-20210421230115-4e50805a0758/go.mod h1:72T/g9IO56b78aLF+1Kcs5dz7/ng1VjMUvfKvpfy+jM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20210423082822-04245dca01da h1:b3NXsE2LusjYGGjL5bxEVZZORm/YEFFrWFjR8eFrw/c=
-golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210420072515-93ed5bcd2bfe h1:WdX7u8s3yOigWAhHEaDl8r9G+4XwFQEQFtBMYyN+kXQ=
+golang.org/x/sys v0.0.0-20210420072515-93ed5bcd2bfe/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.6 h1:aRYxNxv6iGQlyVaZmk6ZgYEDa+Jg18DxebPSrd6bg1M=

tap/http_matcher.go

@@ -49,12 +49,12 @@ func (matcher *requestResponseMatcher) registerRequest(ident string, request *ht
 			SilentError("Request-Duplicate", "Got duplicate request with same identifier")
 			return nil
 		}
-		Debug("Matched open Response for %s", key)
+		Trace("Matched open Response for %s", key)
 		return matcher.preparePair(&requestHTTPMessage, responseHTTPMessage)
 	}
 
 	matcher.openMessagesMap.Set(key, &requestHTTPMessage)
-	Debug("Registered open Request for %s", key)
+	Trace("Registered open Request for %s", key)
 	return nil
 }
 
@@ -75,12 +75,12 @@ func (matcher *requestResponseMatcher) registerResponse(ident string, response *
 			SilentError("Response-Duplicate", "Got duplicate response with same identifier")
 			return nil
 		}
-		Debug("Matched open Request for %s", key)
+		Trace("Matched open Request for %s", key)
 		return matcher.preparePair(requestHTTPMessage, &responseHTTPMessage)
 	}
 
 	matcher.openMessagesMap.Set(key, &responseHTTPMessage)
-	Debug("Registered open Response for %s", key)
+	Trace("Registered open Response for %s", key)
 	return nil
 }
 

tap/http_reader.go

@@ -184,13 +184,13 @@ func (h *httpReader) handleHTTP1ClientStream(b *bufio.Reader) error {
 	if err != nil {
 		SilentError("HTTP-request-body", "stream %s Got body err: %s", h.ident, err)
 	} else if h.hexdump {
-		Info("Body(%d/0x%x) - %s", len(body), len(body), hex.Dump(body))
+		Debug("Body(%d/0x%x) - %s", len(body), len(body), hex.Dump(body))
 	}
 	if err := req.Body.Close(); err != nil {
 		SilentError("HTTP-request-body-close", "stream %s Failed to close request body: %s", h.ident, err)
 	}
 	encoding := req.Header["Content-Encoding"]
-	Info("HTTP/1 Request: %s %s %s (Body:%d) -> %s", h.ident, req.Method, req.URL, s, encoding)
+	Debug("HTTP/1 Request: %s %s %s (Body:%d) -> %s", h.ident, req.Method, req.URL, s, encoding)
 
 	ident := fmt.Sprintf("%s->%s %s->%s %d", h.tcpID.srcIP, h.tcpID.dstIP, h.tcpID.srcPort, h.tcpID.dstPort, h.messageCount)
 	reqResPair := reqResMatcher.registerRequest(ident, req, h.captureTime)
@@ -242,7 +242,7 @@ func (h *httpReader) handleHTTP1ServerStream(b *bufio.Reader) error {
 		SilentError("HTTP-response-body", "HTTP/%s: failed to get body(parsed len:%d): %s", h.ident, s, err)
 	}
 	if h.hexdump {
-		Info("Body(%d/0x%x) - %s", len(body), len(body), hex.Dump(body))
+		Debug("Body(%d/0x%x) - %s", len(body), len(body), hex.Dump(body))
 	}
 	if err := res.Body.Close(); err != nil {
 		SilentError("HTTP-response-body-close", "HTTP/%s: failed to close body(parsed len:%d): %s", h.ident, s, err)
@@ -256,7 +256,7 @@ func (h *httpReader) handleHTTP1ServerStream(b *bufio.Reader) error {
 		contentType = []string{http.DetectContentType(body)}
 	}
 	encoding := res.Header["Content-Encoding"]
-	Info("HTTP/1 Response: %s %s URL:%s (%d%s%d%s) -> %s", h.ident, res.Status, req, res.ContentLength, sym, s, contentType, encoding)
+	Debug("HTTP/1 Response: %s %s URL:%s (%d%s%d%s) -> %s", h.ident, res.Status, req, res.ContentLength, sym, s, contentType, encoding)
 
 	ident := fmt.Sprintf("%s->%s %s->%s %d", h.tcpID.dstIP, h.tcpID.srcIP, h.tcpID.dstPort, h.tcpID.srcPort, h.messageCount)
 	reqResPair := reqResMatcher.registerResponse(ident, res, h.captureTime)

tap/passive_tapper.go

@@ -12,6 +12,7 @@ import (
 	"encoding/hex"
 	"flag"
 	"fmt"
+	"github.com/romana/rlog"
 	"log"
 	"os"
 	"os/signal"
@@ -131,9 +132,10 @@ func logError(minOutputLevel int, t string, s string, a ...interface{}) {
 	nb, _ := errorsMap[t]
 	errorsMap[t] = nb + 1
 	errorsMapMutex.Unlock()
+
 	if outputLevel >= minOutputLevel {
 		formatStr := fmt.Sprintf("%s: %s", t, s)
-		log.Printf(formatStr, a...)
+		rlog.Errorf(formatStr, a...)
 	}
 }
 func Error(t string, s string, a ...interface{}) {
@@ -142,15 +144,11 @@ func Error(t string, s string, a ...interface{}) {
 func SilentError(t string, s string, a ...interface{}) {
 	logError(2, t, s, a...)
 }
-func Info(s string, a ...interface{}) {
-	if outputLevel >= 1 {
-		log.Printf(s, a...)
-	}
-}
 func Debug(s string, a ...interface{}) {
-	if outputLevel >= 2 {
-		log.Printf(s, a...)
-	}
+	rlog.Debugf(s, a...)
+}
+func Trace(s string, a ...interface{}) {
+	rlog.Tracef(1, s, a...)
 }
 
 func inArrayInt(arr []int, valueToCheck int) bool {
@@ -214,8 +212,8 @@ func startPassiveTapper(harWriter *HarWriter, outboundLinkWriter *OutboundLinkWr
 
 	if localhostIPs, err := getLocalhostIPs(); err != nil {
 		// TODO: think this over
-		log.Println("Failed to get self IP addresses")
-		Error("Getting-Self-Address", "Error getting self ip address: %s (%v,%+v)", err, err, err)
+		rlog.Info("Failed to get self IP addresses")
+		rlog.Errorf("Getting-Self-Address", "Error getting self ip address: %s (%v,%+v)", err, err, err)
 		ownIps = make([]string, 0)
 	} else {
 		ownIps = localhostIPs
@@ -224,7 +222,7 @@ func startPassiveTapper(harWriter *HarWriter, outboundLinkWriter *OutboundLinkWr
 	appPortsStr := os.Getenv(AppPortsEnvVar)
 	var appPorts []int
 	if appPortsStr == "" {
-		log.Println("Received empty/no APP_PORTS env var! only listening to http on port 80!")
+		rlog.Info("Received empty/no APP_PORTS env var! only listening to http on port 80!")
 		appPorts = make([]int, 0)
 	} else {
 		appPorts = parseAppPorts(appPortsStr)
@@ -232,14 +230,14 @@ func startPassiveTapper(harWriter *HarWriter, outboundLinkWriter *OutboundLinkWr
 	SetFilterPorts(appPorts)
 	envVal := os.Getenv(maxHTTP2DataLenEnvVar)
 	if envVal == "" {
-		log.Println("Received empty/no HTTP2_DATA_SIZE_LIMIT env var! falling back to", maxHTTP2DataLenDefault)
+		rlog.Infof("Received empty/no HTTP2_DATA_SIZE_LIMIT env var! falling back to %v", maxHTTP2DataLenDefault)
 		maxHTTP2DataLen = maxHTTP2DataLenDefault
 	} else {
 		if convertedInt, err := strconv.Atoi(envVal); err != nil {
-			log.Println("Received invalid HTTP2_DATA_SIZE_LIMIT env var! falling back to", maxHTTP2DataLenDefault)
+			rlog.Infof("Received invalid HTTP2_DATA_SIZE_LIMIT env var! falling back to %v", maxHTTP2DataLenDefault)
 			maxHTTP2DataLen = maxHTTP2DataLenDefault
 		} else {
-			log.Println("Received HTTP2_DATA_SIZE_LIMIT env var:", maxHTTP2DataLenDefault)
+			rlog.Infof("Received HTTP2_DATA_SIZE_LIMIT env var: %v", maxHTTP2DataLenDefault)
 			maxHTTP2DataLen = convertedInt
 		}
 	}
@@ -282,7 +280,7 @@ func startPassiveTapper(harWriter *HarWriter, outboundLinkWriter *OutboundLinkWr
 	}
 	if len(flag.Args()) > 0 {
 		bpffilter := strings.Join(flag.Args(), " ")
-		Info("Using BPF filter %q", bpffilter)
+		rlog.Infof("Using BPF filter %q", bpffilter)
 		if err = handle.SetBPFFilter(bpffilter); err != nil {
 			log.Fatalf("BPF filter error: %v", err)
 		}
@@ -306,7 +304,7 @@ func startPassiveTapper(harWriter *HarWriter, outboundLinkWriter *OutboundLinkWr
 	source := gopacket.NewPacketSource(handle, dec)
 	source.Lazy = *lazy
 	source.NoCopy = true
-	Info("Starting to read packets")
+	rlog.Info("Starting to read packets")
 	count := 0
 	bytes := int64(0)
 	start := time.Now()
@@ -381,11 +379,11 @@ func startPassiveTapper(harWriter *HarWriter, outboundLinkWriter *OutboundLinkWr
 
 	for packet := range source.Packets() {
 		count++
-		Debug("PACKET #%d", count)
+		rlog.Debugf("PACKET #%d", count)
 		data := packet.Data()
 		bytes += int64(len(data))
 		if *hexdumppkt {
-			Debug("Packet content (%d/0x%x) - %s", len(data), len(data), hex.Dump(data))
+			rlog.Debugf("Packet content (%d/0x%x) - %s", len(data), len(data), hex.Dump(data))
 		}
 
 		// defrag the IPv4 packet if required
@@ -400,12 +398,12 @@ func startPassiveTapper(harWriter *HarWriter, outboundLinkWriter *OutboundLinkWr
 			if err != nil {
 				log.Fatalln("Error while de-fragmenting", err)
 			} else if newip4 == nil {
-				Debug("Fragment...")
+				rlog.Debugf("Fragment...")
 				continue // packet fragment, we don't have whole packet yet.
 			}
 			if newip4.Length != l {
 				stats.ipdefrag++
-				Debug("Decoding re-assembled packet: %s", newip4.NextLayerType())
+				rlog.Debugf("Decoding re-assembled packet: %s", newip4.NextLayerType())
 				pb, ok := packet.(gopacket.PacketBuilder)
 				if !ok {
 					log.Panic("Not a PacketBuilder")
@@ -428,7 +426,7 @@ func startPassiveTapper(harWriter *HarWriter, outboundLinkWriter *OutboundLinkWr
 				CaptureInfo: packet.Metadata().CaptureInfo,
 			}
 			stats.totalsz += len(tcp.Payload)
-			// log.Println(packet.NetworkLayer().NetworkFlow().Src(), ":", tcp.SrcPort, " -> ", packet.NetworkLayer().NetworkFlow().Dst(), ":", tcp.DstPort)
+			rlog.Debugf("%s : %v -> %s : %v", packet.NetworkLayer().NetworkFlow().Src(), tcp.SrcPort, packet.NetworkLayer().NetworkFlow().Dst(), tcp.DstPort)
 			assemblerMutex.Lock()
 			assembler.AssembleWithContext(packet.NetworkLayer().NetworkFlow(), tcp, &c)
 			assemblerMutex.Unlock()
@@ -456,7 +454,7 @@ func startPassiveTapper(harWriter *HarWriter, outboundLinkWriter *OutboundLinkWr
 	assemblerMutex.Lock()
 	closed := assembler.FlushAll()
 	assemblerMutex.Unlock()
-	Debug("Final flush: %d closed", closed)
+	rlog.Debugf("Final flush: %d closed", closed)
 	if outputLevel >= 2 {
 		streamPool.Dump()
 	}
@@ -472,7 +470,7 @@ func startPassiveTapper(harWriter *HarWriter, outboundLinkWriter *OutboundLinkWr
 
 	streamFactory.WaitGoRoutines()
 	assemblerMutex.Lock()
-	Debug("%s", assembler.Dump())
+	rlog.Debugf("%s", assembler.Dump())
 	assemblerMutex.Unlock()
 	if !*nodefrag {
 		log.Printf("IPdefrag:\t\t%d", stats.ipdefrag)

tap/tcp_stream.go

@@ -106,7 +106,7 @@ func (t *tcpStream) ReassembledSG(sg reassembly.ScatterGather, ac reassembly.Ass
 	} else {
 		ident = fmt.Sprintf("%v %v(%s): ", t.net.Reverse(), t.transport.Reverse(), dir)
 	}
-	Debug("%s: SG reassembled packet with %d bytes (start:%v,end:%v,skip:%d,saved:%d,nb:%d,%d,overlap:%d,%d)", ident, length, start, end, skip, saved, sgStats.Packets, sgStats.Chunks, sgStats.OverlapBytes, sgStats.OverlapPackets)
+	Trace("%s: SG reassembled packet with %d bytes (start:%v,end:%v,skip:%d,saved:%d,nb:%d,%d,overlap:%d,%d)", ident, length, start, end, skip, saved, sgStats.Packets, sgStats.Chunks, sgStats.OverlapBytes, sgStats.OverlapPackets)
 	if skip == -1 && *allowmissinginit {
 		// this is allowed
 	} else if skip != 0 {
@@ -125,9 +125,9 @@ func (t *tcpStream) ReassembledSG(sg reassembly.ScatterGather, ac reassembly.Ass
 		}
 		dnsSize := binary.BigEndian.Uint16(data[:2])
 		missing := int(dnsSize) - len(data[2:])
-		Debug("dnsSize: %d, missing: %d", dnsSize, missing)
+		Trace("dnsSize: %d, missing: %d", dnsSize, missing)
 		if missing > 0 {
-			Info("Missing some bytes: %d", missing)
+			Debug("Missing some bytes: %d", missing)
 			sg.KeepFrom(0)
 			return
 		}
@@ -136,7 +136,7 @@ func (t *tcpStream) ReassembledSG(sg reassembly.ScatterGather, ac reassembly.Ass
 		if err != nil {
 			SilentError("DNS-parser", "Failed to decode DNS: %v", err)
 		} else {
-			Debug("DNS: %s", gopacket.LayerDump(dns))
+			Trace("DNS: %s", gopacket.LayerDump(dns))
 		}
 		if len(data) > 2+int(dnsSize) {
 			sg.KeepFrom(2 + int(dnsSize))
@@ -144,7 +144,7 @@ func (t *tcpStream) ReassembledSG(sg reassembly.ScatterGather, ac reassembly.Ass
 	} else if t.isHTTP {
 		if length > 0 {
 			if *hexdump {
-				Debug("Feeding http with:%s", hex.Dump(data))
+				Trace("Feeding http with:%s", hex.Dump(data))
 			}
 			// This is where we pass the reassembled information onwards
 			// This channel is read by an httpReader object
@@ -158,7 +158,7 @@ func (t *tcpStream) ReassembledSG(sg reassembly.ScatterGather, ac reassembly.Ass
 }
 
 func (t *tcpStream) ReassemblyComplete(ac reassembly.AssemblerContext) bool {
-	Debug("%s: Connection closed", t.ident)
+	Trace("%s: Connection closed", t.ident)
 	if t.isHTTP {
 		close(t.client.msgQueue)
 		close(t.server.msgQueue)

tap/tcp_stream_factory.go

@@ -2,6 +2,7 @@ package tap
 
 import (
 	"fmt"
+	"github.com/romana/rlog"
 	"sync"
 
 	"github.com/google/gopacket"
@@ -22,11 +23,11 @@ type tcpStreamFactory struct {
 }
 
 func (factory *tcpStreamFactory) New(net, transport gopacket.Flow, tcp *layers.TCP, ac reassembly.AssemblerContext) reassembly.Stream {
-	Debug("* NEW: %s %s", net, transport)
+	rlog.Debugf("* NEW: %s %s", net, transport)
 	fsmOptions := reassembly.TCPSimpleFSMOptions{
 		SupportMissingEstablishment: *allowmissinginit,
 	}
-	Debug("Current App Ports: %v", gSettings.filterPorts)
+	rlog.Debugf("Current App Ports: %v", gSettings.filterPorts)
 	srcIp := net.Src().String()
 	dstIp := net.Dst().String()
 	dstPort := int(tcp.DstPort)
@@ -91,25 +92,31 @@ func (factory *tcpStreamFactory) WaitGoRoutines() {
 func (factory *tcpStreamFactory) getStreamProps(srcIP string, dstIP string, dstPort int) *streamProps {
 	if hostMode {
 		if inArrayString(gSettings.filterAuthorities, fmt.Sprintf("%s:%d", dstIP, dstPort)) == true {
+			rlog.Debugf("getStreamProps %s", fmt.Sprintf("+ host1 %s:%d", dstIP, dstPort))
 			return &streamProps{isTapTarget: true, isOutgoing: false}
 		} else if inArrayString(gSettings.filterAuthorities, dstIP) == true {
+			rlog.Debugf("getStreamProps %s", fmt.Sprintf("+ host2 %s", dstIP))
 			return &streamProps{isTapTarget: true, isOutgoing: false}
 		} else if *anydirection && inArrayString(gSettings.filterAuthorities, srcIP) == true {
+			rlog.Debugf("getStreamProps %s", fmt.Sprintf("+ host3 %s", srcIP))
 			return &streamProps{isTapTarget: true, isOutgoing: true}
 		}
 		return &streamProps{isTapTarget: false}
 	} else {
 		isTappedPort := dstPort == 80 || (gSettings.filterPorts != nil && (inArrayInt(gSettings.filterPorts, dstPort)))
 		if !isTappedPort {
+			rlog.Debugf("getStreamProps %s", fmt.Sprintf("- notHost1 %d", dstPort))
 			return &streamProps{isTapTarget: false, isOutgoing: false}
 		}
 
 		isOutgoing := !inArrayString(ownIps, dstIP)
 
 		if !*anydirection && isOutgoing {
+			rlog.Debugf("getStreamProps %s", fmt.Sprintf("- notHost2"))
 			return &streamProps{isTapTarget: false, isOutgoing: isOutgoing}
 		}
 
+		rlog.Debugf("getStreamProps %s", fmt.Sprintf("+ notHost3 %s -> %s:%d", srcIP, dstIP, dstPort))
 		return &streamProps{isTapTarget: true}
 	}
 }

ui/public/index.html

@@ -2,7 +2,7 @@
 <html lang="en">
   <head>
     <meta charset="utf-8" />
-    <link rel="icon" href="%PUBLIC_URL%/favicon.ico" />
+    <link rel="icon" href="%PUBLIC_URL%/fav.svg" />
     <meta name="viewport" content="width=device-width, initial-scale=1" />
     <meta name="theme-color" content="#000000" />
     <meta

ui/src/App.sass

@@ -6,23 +6,20 @@
   width: 100%
 
   .header
-    height: 80px
+    height: 60px
     display: flex
     align-items: center
-    padding-left: 24px
-    padding-right: 24px
+    padding: 5px 24px
     justify-content: space-between
 
     .title
-      font-size: 45px
       letter-spacing: 2px
 
       img
-        height: 40px
+        height: 45px
 
     .description
       margin-left: 10px
-      padding-top: 10px
-      font-size: 14px
+      font-size: 11px
       font-weight: bold
       color: $light-blue-color

ui/src/App.tsx

@@ -1,6 +1,6 @@
 import React, {useState} from 'react';
 import './App.sass';
-import logo from './components/assets/Mizu.svg';
+import logo from './components/assets/Mizu-logo.svg';
 import {Button} from "@material-ui/core";
 import {HarPage} from "./components/HarPage";
 

ui/src/components/HarPage.tsx

@@ -5,7 +5,7 @@ import {makeStyles} from "@material-ui/core";
 import "./style/HarPage.sass";
 import styles from './style/HarEntriesList.module.sass';
 import {HAREntryDetailed} from "./HarEntryDetailed";
-import playIcon from './assets/play.svg';
+import playIcon from './assets/run.svg';
 import pauseIcon from './assets/pause.svg';
 import variables from './style/variables.module.scss';
 import {StatusBar} from "./StatusBar";
@@ -127,6 +127,7 @@ export const HarPage: React.FC<HarPageProps> = ({setAnalyzeStatus}) => {
         fetch(`${mizuApiUrl}/api/analyzeStatus`)
             .then(response => response.json())
             .then(data => setAnalyzeStatus(data));
+        // eslint-disable-next-line
     }, []);
 
 

ui/src/components/assets/play.svg

@@ -1,4 +0,0 @@
-<svg width="30" height="30" viewBox="0 0 30 30" fill="none" xmlns="http://www.w3.org/2000/svg">
-<rect width="30" height="30" rx="15" fill="#205CF5"/>
-<path d="M17.0747 15L12.9876 12.6433V17.3567L17.0747 15ZM20 15C20 15.3167 19.8392 15.6335 19.5175 15.8189L12.5051 19.8624C11.8427 20.2444 11 19.7858 11 19.0435V10.9565C11 10.2142 11.8427 9.75564 12.5051 10.1376L19.5175 14.1811C19.8392 14.3665 20 14.6833 20 15Z" fill="white"/>
-</svg>

ui/src/components/assets/run.svg

@@ -0,0 +1,4 @@
+<svg width="30" height="30" viewBox="0 0 30 30" fill="none" xmlns="http://www.w3.org/2000/svg">
+<circle cx="15" cy="15" r="13.5" stroke="#205CF5" stroke-width="3"/>
+<path d="M20 15C20 15.3167 19.8392 15.6335 19.5175 15.8189L12.5051 19.8624C11.8427 20.2444 11 19.7858 11 19.0435V10.9565C11 10.2142 11.8427 9.75564 12.5051 10.1376L19.5175 14.1811C19.8392 14.3665 20 14.6833 20 15Z" fill="#205CF5"/>
+</svg>

ui/src/components/style/HarPage.sass

@@ -6,7 +6,7 @@
     flex-direction: column
     overflow: hidden
     flex-grow: 1
-    height: calc(100vh - 80px)
+    height: calc(100vh - 70px)
 
     .harPageHeader
         padding: 20px 24px

ui/src/index.sass

@@ -14,6 +14,7 @@ body
   -moz-osx-font-smoothing: grayscale
   margin: 0
   padding: 0
+  overflow: hidden
 
 code
   font-family: source-code-pro, Menlo, Monaco, Consolas, "Courier New", monospace