Fix resource limits (#110)

Adding the upload interval as parameter to tap function

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/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/up9/main.go

@@ -36,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)
 }
 
@@ -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
@@ -112,13 +112,13 @@ 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
 

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/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/passive_tapper.go

@@ -426,7 +426,7 @@ func startPassiveTapper(harWriter *HarWriter, outboundLinkWriter *OutboundLinkWr
 				CaptureInfo: packet.Metadata().CaptureInfo,
 			}
 			stats.totalsz += len(tcp.Payload)
-			rlog.Debugf(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()

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