Merge branch 'develop'

Conflicts:
	cli/cmd/tapRunner.go

api/main.go

@@ -52,7 +52,7 @@ func main() {
 		tapTargets := getTapTargets()
 		if tapTargets != nil {
 			tap.SetFilterAuthorities(tapTargets)
-			rlog.Info("Filtering for the following authorities:", tap.GetFilterIPs())
+			rlog.Infof("Filtering for the following authorities: %v", tap.GetFilterIPs())
 		}
 
 		harOutputChannel, outboundLinkOutputChannel := tap.StartPassiveTapper(tapOpts)

api/pkg/api/main.go

@@ -131,7 +131,7 @@ func saveHarToDb(entry *har.Entry, connectionInfo *tap.ConnectionInfo) {
 		unresolvedSource := connectionInfo.ClientIP
 		resolvedSource = k8sResolver.Resolve(unresolvedSource)
 		if resolvedSource == "" {
-			rlog.Debug("Cannot find resolved name to source: %s\n", unresolvedSource)
+			rlog.Debugf("Cannot find resolved name to source: %s\n", unresolvedSource)
 			if os.Getenv("SKIP_NOT_RESOLVED_SOURCE") == "1" {
 				return
 			}
@@ -139,7 +139,7 @@ func saveHarToDb(entry *har.Entry, connectionInfo *tap.ConnectionInfo) {
 		unresolvedDestination := fmt.Sprintf("%s:%s", connectionInfo.ServerIP, connectionInfo.ServerPort)
 		resolvedDestination = k8sResolver.Resolve(unresolvedDestination)
 		if resolvedDestination == "" {
-			rlog.Debug("Cannot find resolved name to dest: %s\n", unresolvedDestination)
+			rlog.Debugf("Cannot find resolved name to dest: %s\n", unresolvedDestination)
 			if os.Getenv("SKIP_NOT_RESOLVED_DEST") == "1" {
 				return
 			}

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.Debugf("Upload entries - started\n")
+
 	uploadRequestBody := &models.UploadEntriesRequestBody{}
 	if err := c.QueryParser(uploadRequestBody); err != nil {
 		return c.Status(fiber.StatusBadRequest).JSON(err)
@@ -150,8 +153,12 @@ 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)
+	rlog.Debugf("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)
 	return c.Status(fiber.StatusOK).SendString("OK")
 }

api/pkg/up9/main.go

@@ -36,7 +36,7 @@ func getGuestToken(url string, target *GuestToken) error {
 		return err
 	}
 	defer resp.Body.Close()
-
+	rlog.Debugf("Got token from the server, starting to json decode... status code: %v", resp.StatusCode)
 	return json.NewDecoder(resp.Body).Decode(target)
 }
 
@@ -47,7 +47,7 @@ func CreateAnonymousToken(envPrefix string) (*GuestToken, error) {
 	}
 	token := &GuestToken{}
 	if err := getGuestToken(tokenUrl, token); err != nil {
-		rlog.Infof("%s", err)
+		rlog.Infof("Failed to get token, %s", err)
 		return nil, err
 	}
 	return token, nil

cli/cmd/tapRunner.go

@@ -3,6 +3,7 @@ package cmd
 import (
 	"context"
 	"fmt"
+	"github.com/romana/rlog"
 	"github.com/up9inc/mizu/cli/kubernetes"
 	"github.com/up9inc/mizu/cli/mizu"
 	"github.com/up9inc/mizu/shared"
@@ -252,11 +253,12 @@ 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", mizuProxiedUrl, url.QueryEscape(tappingOptions.AnalyzeDestination))
 					u, err := url.ParseRequestURI(url_path)
 					if err != nil {
 						log.Fatal(fmt.Sprintf("Failed parsing the URL %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 %v\n", err)
 					} else {
@@ -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/kubernetes/provider.go

@@ -15,6 +15,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"
@@ -76,6 +77,24 @@ func (provider *Provider) CreateMizuAggregatorPod(ctx context.Context, namespace
 	if err != nil {
 		return nil, err
 	}
+
+	cpuLimit, err := resource.ParseQuantity("750")
+	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,
@@ -103,6 +122,16 @@ func (provider *Provider) CreateMizuAggregatorPod(ctx context.Context, namespace
 							Value: strconv.FormatInt(maxEntriesDBSizeBytes, 10),
 						},
 					},
+					Resources: core.ResourceRequirements{
+						Limits: core.ResourceList{
+							"cpu": cpuLimit,
+							"memory": memLimit,
+						},
+						Requests: core.ResourceList{
+							"cpu": cpuRequests,
+							"memory": memRequests,
+						},
+					},
 				},
 			},
 			DNSPolicy:                     core.DNSClusterFirstWithHostNet,
@@ -341,6 +370,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/passive_tapper.go

@@ -213,7 +213,7 @@ func startPassiveTapper(harWriter *HarWriter, outboundLinkWriter *OutboundLinkWr
 	if localhostIPs, err := getLocalhostIPs(); err != nil {
 		// TODO: think this over
 		rlog.Info("Failed to get self IP addresses")
-		rlog.Error("Getting-Self-Address", "Error getting self ip address: %s (%v,%+v)", err, err, err)
+		rlog.Errorf("Getting-Self-Address", "Error getting self ip address: %s (%v,%+v)", err, err, err)
 		ownIps = make([]string, 0)
 	} else {
 		ownIps = localhostIPs
@@ -230,14 +230,14 @@ func startPassiveTapper(harWriter *HarWriter, outboundLinkWriter *OutboundLinkWr
 	SetFilterPorts(appPorts)
 	envVal := os.Getenv(maxHTTP2DataLenEnvVar)
 	if envVal == "" {
-		rlog.Info("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 {
-			rlog.Info("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 {
-			rlog.Info("Received HTTP2_DATA_SIZE_LIMIT env var:", maxHTTP2DataLenDefault)
+			rlog.Infof("Received HTTP2_DATA_SIZE_LIMIT env var: %v", maxHTTP2DataLenDefault)
 			maxHTTP2DataLen = convertedInt
 		}
 	}
@@ -379,11 +379,11 @@ func startPassiveTapper(harWriter *HarWriter, outboundLinkWriter *OutboundLinkWr
 
 	for packet := range source.Packets() {
 		count++
-		rlog.Debug("PACKET #%d", count)
+		rlog.Debugf("PACKET #%d", count)
 		data := packet.Data()
 		bytes += int64(len(data))
 		if *hexdumppkt {
-			rlog.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
@@ -398,12 +398,12 @@ func startPassiveTapper(harWriter *HarWriter, outboundLinkWriter *OutboundLinkWr
 			if err != nil {
 				log.Fatalln("Error while de-fragmenting", err)
 			} else if newip4 == nil {
-				rlog.Debug("Fragment...")
+				rlog.Debugf("Fragment...")
 				continue // packet fragment, we don't have whole packet yet.
 			}
 			if newip4.Length != l {
 				stats.ipdefrag++
-				rlog.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")
@@ -426,7 +426,7 @@ func startPassiveTapper(harWriter *HarWriter, outboundLinkWriter *OutboundLinkWr
 				CaptureInfo: packet.Metadata().CaptureInfo,
 			}
 			stats.totalsz += len(tcp.Payload)
-			rlog.Debug(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()
@@ -454,7 +454,7 @@ func startPassiveTapper(harWriter *HarWriter, outboundLinkWriter *OutboundLinkWr
 	assemblerMutex.Lock()
 	closed := assembler.FlushAll()
 	assemblerMutex.Unlock()
-	rlog.Debug("Final flush: %d closed", closed)
+	rlog.Debugf("Final flush: %d closed", closed)
 	if outputLevel >= 2 {
 		streamPool.Dump()
 	}
@@ -470,7 +470,7 @@ func startPassiveTapper(harWriter *HarWriter, outboundLinkWriter *OutboundLinkWr
 
 	streamFactory.WaitGoRoutines()
 	assemblerMutex.Lock()
-	rlog.Debug("%s", assembler.Dump())
+	rlog.Debugf("%s", assembler.Dump())
 	assemblerMutex.Unlock()
 	if !*nodefrag {
 		log.Printf("IPdefrag:\t\t%d", stats.ipdefrag)

tap/tcp_stream_factory.go

@@ -23,11 +23,11 @@ type tcpStreamFactory struct {
 }
 
 func (factory *tcpStreamFactory) New(net, transport gopacket.Flow, tcp *layers.TCP, ac reassembly.AssemblerContext) reassembly.Stream {
-	rlog.Debug("* NEW: %s %s", net, transport)
+	rlog.Debugf("* NEW: %s %s", net, transport)
 	fsmOptions := reassembly.TCPSimpleFSMOptions{
 		SupportMissingEstablishment: *allowmissinginit,
 	}
-	rlog.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)
@@ -92,31 +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.Debug("getStreamProps %s", fmt.Sprintf("+ host1 %s:%d", dstIP, dstPort))
+			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.Debug("getStreamProps %s", fmt.Sprintf("+ host2 %s", dstIP))
+			rlog.Debugf("getStreamProps %s", fmt.Sprintf("+ host2 %s", dstIP))
 			return &streamProps{isTapTarget: true, isOutgoing: false}
 		} else if *anydirection && inArrayString(gSettings.filterAuthorities, srcIP) == true {
-			rlog.Debug("getStreamProps %s", fmt.Sprintf("+ host3 %s", srcIP))
+			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.Debug("getStreamProps %s", fmt.Sprintf("- notHost1 %d", dstPort))
+			rlog.Debugf("getStreamProps %s", fmt.Sprintf("- notHost1 %d", dstPort))
 			return &streamProps{isTapTarget: false, isOutgoing: false}
 		}
 
 		isOutgoing := !inArrayString(ownIps, dstIP)
 
 		if !*anydirection && isOutgoing {
-			rlog.Debug("getStreamProps %s", fmt.Sprintf("- notHost2"))
+			rlog.Debugf("getStreamProps %s", fmt.Sprintf("- notHost2"))
 			return &streamProps{isTapTarget: false, isOutgoing: isOutgoing}
 		}
 
-		rlog.Debug("getStreamProps %s", fmt.Sprintf("+ notHost3 %s -> %s:%d", srcIP, dstIP, dstPort))
+		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