Merge pull request #121 from up9inc/develop

Missing request body (#120)

.github/workflows/publish-docker.yml

@@ -1,39 +0,0 @@
-name: publish-docker
-on:
-  push:
-    branches:
-      - 'develop'
-      - 'main'
-jobs:
-  docker:
-    runs-on: ubuntu-latest
-    steps:
-      - name: Checkout
-        uses: actions/checkout@v2
-      - name: Get base image name
-        shell: bash
-        run: echo "##[set-output name=image;]$(echo gcr.io/up9-docker-hub/mizu/${GITHUB_REF#refs/heads/})"
-        id: base_image_step
-      - name: Docker meta
-        id: meta
-        uses: crazy-max/ghaction-docker-meta@v2
-        with:
-          images: ${{ steps.base_image_step.outputs.image }}
-          tags: |
-            type=sha
-            type=raw,${{ github.sha }}
-            type=raw,latest
-      - name: Login to DockerHub
-        uses: docker/login-action@v1
-        with:
-          registry: gcr.io
-          username: _json_key
-          password: ${{ secrets.GCR_JSON_KEY }}
-      - name: Build and push
-        uses: docker/build-push-action@v2
-        with:
-          context: .
-          push: true
-          tags: ${{ steps.meta.outputs.tags }}
-          labels: ${{ steps.meta.outputs.labels }}
-

.github/workflows/publish.yml

@@ -1,9 +1,9 @@
-name: public-cli
+name: publish
 on:
   push:
     branches:
-      - develop
-      - main
+      - 'develop'
+      - 'main'
 jobs:
   docker:
     runs-on: ubuntu-latest
@@ -27,12 +27,43 @@ jobs:
         with:
           releaseType: ${{ steps.condval.outputs.value }}
           github_token: ${{ secrets.GITHUB_TOKEN }}
-      - name: Get base image name
+      - name: Get version parameters
         shell: bash
         run: |
           echo "##[set-output name=build_timestamp;]$(echo $(date +%s))"
           echo "##[set-output name=branch;]$(echo ${GITHUB_REF#refs/heads/})"
         id: version_parameters
+      - name: Get base image name
+        shell: bash
+        run: echo "##[set-output name=image;]$(echo gcr.io/up9-docker-hub/mizu/${GITHUB_REF#refs/heads/})"
+        id: base_image_step
+      - name: Docker meta
+        id: meta
+        uses: crazy-max/ghaction-docker-meta@v2
+        with:
+          images: ${{ steps.base_image_step.outputs.image }}
+          tags: |
+            type=sha
+            type=raw,${{ github.sha }}
+            type=raw,${{ steps.versioning.outputs.version }}
+      - name: Login to DockerHub
+        uses: docker/login-action@v1
+        with:
+          registry: gcr.io
+          username: _json_key
+          password: ${{ secrets.GCR_JSON_KEY }}
+      - name: Build and push
+        uses: docker/build-push-action@v2
+        with:
+          context: .
+          push: true
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}
+          build-args: |
+            SEM_VER=${{ steps.versioning.outputs.version }}
+            BUILD_TIMESTAMP=${{ steps.version_parameters.outputs.build_timestamp }}
+            GIT_BRANCH=${{ steps.version_parameters.outputs.branch }}
+            COMMIT_HASH=${{ github.sha }}
       - name: Build and Push CLI
         run: make push-cli SEM_VER='${{ steps.versioning.outputs.version }}' BUILD_TIMESTAMP='${{ steps.version_parameters.outputs.build_timestamp }}'
       - name: publish

Dockerfile

@@ -23,11 +23,20 @@ RUN go mod download
 # cheap trick to make the build faster (As long as go.mod wasn't changes)
 RUN go list -f '{{.Path}}@{{.Version}}' -m all | sed 1d | grep -e 'go-cache' -e 'sqlite' | xargs go get
 
+ARG COMMIT_HASH
+ARG GIT_BRANCH
+ARG BUILD_TIMESTAMP
+ARG SEM_VER
+
 # Copy and build api code
 COPY shared ../shared
 COPY tap ../tap
 COPY api .
-RUN go build -ldflags="-s -w" -o mizuagent .
+RUN go build -ldflags="-s -w \
+     -X 'mizuserver/pkg/version.GitCommitHash=${COMMIT_HASH}' \
+     -X 'mizuserver/pkg/version.Branch=${GIT_BRANCH}' \
+     -X 'mizuserver/pkg/version.BuildTimestamp=${BUILD_TIMESTAMP}' \
+     -X 'mizuserver/pkg/version.SemVer=${SEM_VER}'" -o mizuagent .
 
 
 FROM alpine:3.13.5

README.md

@@ -26,6 +26,49 @@ SHA256 checksums are available on the [Releases](https://github.com/up9inc/mizu/
 ### Development (unstable) build
 Pick one from the [Releases](https://github.com/up9inc/mizu/releases) page.
 
+## Prerequisites
+1. Set `KUBECONFIG` environment variable to your kubernetes configuration. If this is not set, mizu assumes that configuration is at `${HOME}/.kube/config`
+2. mizu needs following permissions on your kubernetes cluster to run
+```
+- apiGroups:
+  - ""
+  - apps
+  resources:
+  - pods
+  - services
+  verbs:
+  - list
+  - get
+  - create
+  - delete
+- apiGroups:
+  - ""
+  - apps
+  resources:
+  - daemonsets
+  verbs:
+  - list
+  - get
+  - create
+  - patch
+  - delete
+```
+3. Optionally, for resolving traffic ip to kubernetes service name, mizu needs below permissions
+```
+- apiGroups:
+  - ""
+  - apps
+  - "rbac.authorization.k8s.io"
+  resources:
+  - clusterroles
+  - clusterrolebindings
+  - serviceaccounts
+  verbs:
+  - get
+  - create
+  - delete
+```
+
 ## How to run
 
 1. Find pod you'd like to tap to in your Kubernetes cluster

api/go.mod

@@ -23,6 +23,7 @@ require (
 	k8s.io/api v0.21.0
 	k8s.io/apimachinery v0.21.0
 	k8s.io/client-go v0.21.0
+	github.com/fsnotify/fsnotify v1.4.9
 )
 
 replace github.com/up9inc/mizu/shared v0.0.0 => ../shared

api/go.sum

@@ -61,6 +61,8 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/djherbis/atime v1.0.0 h1:ySLvBAM0EvOGaX7TI4dAM5lWj+RdJUCKtGSEHN8SGBg=
 github.com/djherbis/atime v1.0.0/go.mod h1:5W+KBIuTwVGcqjIfaTwt+KSYX1o6uep8dtevevQP/f8=
+github.com/docker/go-units v0.4.0 h1:3uh0PgVws3nIA0Q+MwDC8yjEPf9zjRfZZWXZYDct3Tw=
+github.com/docker/go-units v0.4.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
 github.com/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815/go.mod h1:WwZ+bS3ebgob9U8Nd0kOddGdZWjyMGR8Wziv+TBNwSE=
 github.com/elazarl/goproxy v0.0.0-20180725130230-947c36da3153/go.mod h1:/Zj4wYkgs4iZTTu3o/KG3Itv/qCCa8VVMlb3i9OVuzc=
 github.com/emicklei/go-restful v0.0.0-20170410110728-ff4f55a20633/go.mod h1:otzb+WCGbkyDHkqmQmT5YD2WR4BBwUdeQoFo8l/7tVs=
@@ -73,6 +75,8 @@ github.com/fasthttp/websocket v1.4.3-beta.1/go.mod h1:JGrgLaT02bL9NuJkZbHN8mVV2t
 github.com/form3tech-oss/jwt-go v3.2.2+incompatible h1:TcekIExNqud5crz4xD2pavyTgWiPvpYe4Xau31I0PRk=
 github.com/form3tech-oss/jwt-go v3.2.2+incompatible/go.mod h1:pbq4aXjuKjdthFRnoDwaVPLA+WlJuPGy+QneDUgJi2k=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
+github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4=
+github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
 github.com/go-gl/glfw v0.0.0-20190409004039-e6da0acd62b1/go.mod h1:vR7hzQXu2zJy9AVAgeJqvqgH9Q5CA+iKCZ2gyEVpxRU=
 github.com/go-gl/glfw/v3.3/glfw v0.0.0-20191125211704-12ad95a8df72/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
 github.com/go-gl/glfw/v3.3/glfw v0.0.0-20200222043503-6f7a984d4dc4/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
@@ -411,6 +415,7 @@ golang.org/x/sys v0.0.0-20190616124812-15dcb6c0061f/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20190624142023-c5567b49c5d0/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190726091711-fc99dfbffb4e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191001151750-bb3f8db39f24/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20191005200804-aed5e4c7ecf9/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191204072324-ce4227a45e2e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191228213918-04cbcbbfeed8/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=

api/main.go

@@ -18,6 +18,7 @@ import (
 	"mizuserver/pkg/utils"
 	"os"
 	"os/signal"
+	"strings"
 )
 
 var shouldTap = flag.Bool("tap", false, "Run in tapper mode without API")
@@ -25,13 +26,12 @@ var aggregator = flag.Bool("aggregator", false, "Run in aggregator mode with API
 var standalone = flag.Bool("standalone", false, "Run in standalone tapper and API mode")
 var aggregatorAddress = flag.String("aggregator-address", "", "Address of mizu collector for tapping")
 
-
 func main() {
 	flag.Parse()
 	hostMode := os.Getenv(shared.HostModeEnvVar) == "1"
 	tapOpts := &tap.TapOpts{HostMode: hostMode}
 
-	if !*shouldTap && !*aggregator && !*standalone{
+	if !*shouldTap && !*aggregator && !*standalone {
 		panic("One of the flags --tap, --api or --standalone must be provided")
 	}
 
@@ -89,7 +89,6 @@ func hostApi(socketHarOutputChannel chan<- *tap.OutputChannelItem) {
 		AllowMethods: "*",
 		AllowHeaders: "*",
 	}))
-
 	middleware.FiberMiddleware(app) // Register Fiber's middleware for app.
 	app.Static("/", "./site")
 
@@ -102,12 +101,12 @@ func hostApi(socketHarOutputChannel chan<- *tap.OutputChannelItem) {
 	}
 	routes.WebSocketRoutes(app, &eventHandlers)
 	routes.EntriesRoutes(app)
+	routes.MetadataRoutes(app)
 	routes.NotFoundRoute(app)
 
 	utils.StartServer(app)
 }
 
-
 func getTapTargets() []string {
 	nodeName := os.Getenv(shared.NodeNameEnvVar)
 	var tappedAddressesPerNodeDict map[string][]string
@@ -132,11 +131,17 @@ func getTrafficFilteringOptions() *shared.TrafficFilteringOptions {
 	return &filteringOptions
 }
 
-func filterHarItems(inChannel <- chan *tap.OutputChannelItem, outChannel chan *tap.OutputChannelItem, filterOptions *shared.TrafficFilteringOptions) {
+var userAgentsToFilter = []string{"kube-probe", "prometheus"}
+
+func filterHarItems(inChannel <-chan *tap.OutputChannelItem, outChannel chan *tap.OutputChannelItem, filterOptions *shared.TrafficFilteringOptions) {
 	for message := range inChannel {
 		if message.ConnectionInfo.IsOutgoing && api.CheckIsServiceIP(message.ConnectionInfo.ServerIP) {
 			continue
 		}
+		// TODO: move this to tappers https://up9.atlassian.net/browse/TRA-3441
+		if filterOptions.HideHealthChecks && isHealthCheckByUserAgent(message) {
+			continue
+		}
 
 		sensitiveDataFiltering.FilterSensitiveInfoFromHarRequest(message, filterOptions)
 
@@ -144,6 +149,20 @@ func filterHarItems(inChannel <- chan *tap.OutputChannelItem, outChannel chan *t
 	}
 }
 
+func isHealthCheckByUserAgent(message *tap.OutputChannelItem) bool {
+	for _, header := range message.HarEntry.Request.Headers {
+		if strings.ToLower(header.Name) == "user-agent" {
+			for _, userAgent := range userAgentsToFilter {
+				if strings.Contains(strings.ToLower(header.Value), userAgent) {
+					return true
+				}
+			}
+			return false
+		}
+	}
+	return false
+}
+
 func pipeChannelToSocket(connection *websocket.Conn, messageDataChannel <-chan *tap.OutputChannelItem) {
 	if connection == nil {
 		panic("Websocket connection is nil")

api/pkg/api/main.go

@@ -160,7 +160,8 @@ func saveHarToDb(entry *har.Entry, connectionInfo *tap.ConnectionInfo) {
 		ResolvedDestination: resolvedDestination,
 		IsOutgoing:          connectionInfo.IsOutgoing,
 	}
-	database.GetEntriesTable().Create(&mizuEntry)
+	mizuEntry.EstimatedSizeBytes = getEstimatedEntrySizeBytes(mizuEntry)
+	database.CreateEntry(&mizuEntry)
 
 	baseEntry := models.BaseEntryDetails{}
 	if err := models.GetEntry(&mizuEntry, &baseEntry); err != nil {
@@ -179,3 +180,22 @@ func getServiceNameFromUrl(inputUrl string) (string, string) {
 func CheckIsServiceIP(address string) bool {
 	return k8sResolver.CheckIsServiceIP(address)
 }
+
+// gives a rough estimate of the size this will take up in the db, good enough for maintaining db size limit accurately
+func getEstimatedEntrySizeBytes(mizuEntry models.MizuEntry) int {
+	sizeBytes := len(mizuEntry.Entry)
+	sizeBytes += len(mizuEntry.EntryId)
+	sizeBytes += len(mizuEntry.Service)
+	sizeBytes += len(mizuEntry.Url)
+	sizeBytes += len(mizuEntry.Method)
+	sizeBytes += len(mizuEntry.RequestSenderIp)
+	sizeBytes += len(mizuEntry.ResolvedDestination)
+	sizeBytes += len(mizuEntry.ResolvedSource)
+	sizeBytes += 8 // Status bytes (sqlite integer is always 8 bytes)
+	sizeBytes += 8 // Timestamp bytes
+	sizeBytes += 8 // SizeBytes bytes
+	sizeBytes += 1 // IsOutgoing bytes
+
+
+	return sizeBytes
+}

api/pkg/controllers/metadata_controller.go

@@ -0,0 +1,12 @@
+package controllers
+
+import (
+	"github.com/gofiber/fiber/v2"
+	"github.com/up9inc/mizu/shared"
+	"mizuserver/pkg/version"
+)
+
+func GetVersion(c *fiber.Ctx) error {
+	resp := shared.VersionResponse{SemVer: version.SemVer}
+	return c.Status(fiber.StatusOK).JSON(resp)
+}

api/pkg/database/main.go

@@ -4,19 +4,14 @@ import (
 	"fmt"
 	"gorm.io/driver/sqlite"
 	"gorm.io/gorm"
+	"gorm.io/gorm/logger"
 	"mizuserver/pkg/models"
 	"mizuserver/pkg/utils"
+	"time"
 )
 
 const (
 	DBPath = "./entries.db"
-)
-
-var (
-	DB = initDataBase(DBPath)
-)
-
-const (
 	OrderDesc = "desc"
 	OrderAsc  = "asc"
 	LT        = "lt"
@@ -24,6 +19,8 @@ const (
 )
 
 var (
+	DB *gorm.DB
+	IsDBLocked = false
 	OperatorToSymbolMapping = map[string]string{
 		LT: "<",
 		GT: ">",
@@ -34,12 +31,26 @@ var (
 	}
 )
 
+func init() {
+	DB = initDataBase(DBPath)
+	go StartEnforcingDatabaseSize()
+}
+
 func GetEntriesTable() *gorm.DB {
 	return DB.Table("mizu_entries")
 }
 
+func CreateEntry(entry *models.MizuEntry) {
+	if IsDBLocked {
+		return
+	}
+	GetEntriesTable().Create(entry)
+}
+
 func initDataBase(databasePath string) *gorm.DB {
-	temp, _ := gorm.Open(sqlite.Open(databasePath), &gorm.Config{})
+	temp, _ := gorm.Open(sqlite.Open(databasePath), &gorm.Config{
+		Logger: &utils.TruncatingLogger{LogLevel: logger.Warn, SlowThreshold: 500 * time.Millisecond},
+	})
 	_ = temp.AutoMigrate(&models.MizuEntry{}) // this will ensure table is created
 	return temp
 }

api/pkg/database/size_enforcer.go

@@ -0,0 +1,121 @@
+package database
+
+import (
+	"fmt"
+	"github.com/fsnotify/fsnotify"
+	"github.com/up9inc/mizu/shared"
+	"github.com/up9inc/mizu/shared/debounce"
+	"github.com/up9inc/mizu/shared/units"
+	"log"
+	"mizuserver/pkg/models"
+	"os"
+	"strconv"
+	"time"
+)
+
+const percentageOfMaxSizeBytesToPrune = 15
+const defaultMaxDatabaseSizeBytes int64 = 200 * 1000 * 1000
+
+func StartEnforcingDatabaseSize() {
+	watcher, err := fsnotify.NewWatcher()
+	if err != nil {
+		log.Fatalf("Error creating filesystem watcher for db size enforcement: %v\n", err)
+		return
+	}
+
+	maxEntriesDBByteSize, err := getMaxEntriesDBByteSize()
+	if err != nil {
+		log.Fatalf("Error parsing max db size: %v\n", err)
+		return
+	}
+
+	checkFileSizeDebouncer := debounce.NewDebouncer(5*time.Second, func() {
+		checkFileSize(maxEntriesDBByteSize)
+	})
+
+	go func() {
+		for {
+			select {
+			case event, ok := <-watcher.Events:
+				if !ok {
+					return // closed channel
+				}
+				if event.Op == fsnotify.Write {
+					checkFileSizeDebouncer.SetOn()
+				}
+			case err, ok := <-watcher.Errors:
+				if !ok {
+					return // closed channel
+				}
+				fmt.Printf("filesystem watcher encountered error:%v\n", err)
+			}
+		}
+	}()
+
+	err = watcher.Add(DBPath)
+	if err != nil {
+		log.Fatalf("Error adding %s to filesystem watcher for db size enforcement: %v\n", DBPath, err)
+	}
+}
+
+func getMaxEntriesDBByteSize() (int64, error) {
+	maxEntriesDBByteSize := defaultMaxDatabaseSizeBytes
+	var err error
+
+	maxEntriesDBSizeByteSEnvVarValue := os.Getenv(shared.MaxEntriesDBSizeByteSEnvVar)
+	if maxEntriesDBSizeByteSEnvVarValue != "" {
+		maxEntriesDBByteSize, err = strconv.ParseInt(maxEntriesDBSizeByteSEnvVarValue, 10, 64)
+	}
+	return maxEntriesDBByteSize, err
+}
+
+func checkFileSize(maxSizeBytes int64) {
+	fileStat, err := os.Stat(DBPath)
+	if err != nil {
+		fmt.Printf("Error checking %s file size: %v\n", DBPath, err)
+	} else {
+		if fileStat.Size() > maxSizeBytes {
+			pruneOldEntries(fileStat.Size())
+		}
+	}
+}
+
+func pruneOldEntries(currentFileSize int64) {
+	// sqlite locks the database while delete or VACUUM are running and sqlite is terrible at handling its own db lock while a lot of inserts are attempted, we prevent a significant bottleneck by handling the db lock ourselves here
+	IsDBLocked = true
+	defer func() {IsDBLocked = false}()
+
+	amountOfBytesToTrim := currentFileSize / (100 / percentageOfMaxSizeBytesToPrune)
+
+	rows, err := GetEntriesTable().Limit(10000).Order("id").Rows()
+	if err != nil {
+		fmt.Printf("Error getting 10000 first db rows: %v\n", err)
+		return
+	}
+
+	entryIdsToRemove := make([]uint, 0)
+	bytesToBeRemoved := int64(0)
+	for rows.Next() {
+		if bytesToBeRemoved >= amountOfBytesToTrim {
+			break
+		}
+		var entry models.MizuEntry
+		err = DB.ScanRows(rows, &entry)
+		if err != nil {
+			fmt.Printf("Error scanning db row: %v\n", err)
+			continue
+		}
+
+		entryIdsToRemove = append(entryIdsToRemove, entry.ID)
+		bytesToBeRemoved += int64(entry.EstimatedSizeBytes)
+	}
+
+	if len(entryIdsToRemove) > 0 {
+		GetEntriesTable().Where(entryIdsToRemove).Delete(models.MizuEntry{})
+		// VACUUM causes sqlite to shrink the db file after rows have been deleted, the db file will not shrink without this
+		DB.Exec("VACUUM")
+		fmt.Printf("Removed %d rows and cleared %s\n", len(entryIdsToRemove), units.BytesToHumanReadable(bytesToBeRemoved))
+	} else {
+		fmt.Println("Found no rows to remove when pruning")
+	}
+}

api/pkg/models/models.go

@@ -33,6 +33,7 @@ type MizuEntry struct {
 	ResolvedSource      string `json:"resolvedSource,omitempty" gorm:"column:resolvedSource"`
 	ResolvedDestination string `json:"resolvedDestination,omitempty" gorm:"column:resolvedDestination"`
 	IsOutgoing          bool   `json:"isOutgoing,omitempty" gorm:"column:isOutgoing"`
+	EstimatedSizeBytes           int `json:"-" gorm:"column:estimatedSizeBytes"`
 }
 
 type BaseEntryDetails struct {

api/pkg/resolver/loader.go

@@ -10,6 +10,7 @@ import (
 	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/client-go/util/homedir"
 	"path/filepath"
+	"os"
 )
 
 func NewFromInCluster(errOut chan error) (*Resolver, error) {
@@ -26,8 +27,13 @@ func NewFromInCluster(errOut chan error) (*Resolver, error) {
 
 func NewFromOutOfCluster(kubeConfigPath string, errOut chan error) (*Resolver, error) {
 	if kubeConfigPath == "" {
-		home := homedir.HomeDir()
-		kubeConfigPath = filepath.Join(home, ".kube", "config")
+		env := os.Getenv("KUBECONFIG") 
+		if env != "" {
+			kubeConfigPath = env
+		} else {
+			home := homedir.HomeDir()
+			kubeConfigPath = filepath.Join(home, ".kube", "config")
+		}
 	}
 
 	configPathList := filepath.SplitList(kubeConfigPath)

api/pkg/routes/entries_routes.go

@@ -5,7 +5,7 @@ import (
 	"mizuserver/pkg/controllers"
 )
 
-// EntriesRoutes func for describe group of public routes.
+// EntriesRoutes defines the group of har entries routes.
 func EntriesRoutes(fiberApp *fiber.App) {
 	routeGroup := fiberApp.Group("/api")
 

api/pkg/routes/metadata_routes.go

@@ -0,0 +1,13 @@
+package routes
+
+import (
+	"github.com/gofiber/fiber/v2"
+	"mizuserver/pkg/controllers"
+)
+
+// MetadataRoutes defines the group of metadata routes.
+func MetadataRoutes(fiberApp *fiber.App) {
+	routeGroup := fiberApp.Group("/metadata")
+
+	routeGroup.Get("/version", controllers.GetVersion)
+}

api/pkg/routes/not_found_route.go

@@ -2,7 +2,7 @@ package routes
 
 import "github.com/gofiber/fiber/v2"
 
-// NotFoundRoute func for describe 404 Error route.
+// NotFoundRoute defines the 404 Error route.
 func NotFoundRoute(fiberApp *fiber.App) {
 	fiberApp.Use(
 		func(c *fiber.Ctx) error {

api/pkg/utils/truncating_logger.go

@@ -0,0 +1,58 @@
+package utils
+
+import (
+	"context"
+	"fmt"
+	"gorm.io/gorm/logger"
+	"gorm.io/gorm/utils"
+	"time"
+)
+
+// TruncatingLogger implements the gorm logger.Interface interface. Its purpose is to act as gorm's logger while truncating logs to a max of 50 characters to minimise the performance impact
+type TruncatingLogger struct {
+	LogLevel logger.LogLevel
+	SlowThreshold time.Duration
+}
+
+func (truncatingLogger *TruncatingLogger) LogMode(logLevel logger.LogLevel) logger.Interface {
+	truncatingLogger.LogLevel = logLevel
+	return truncatingLogger
+}
+
+func (truncatingLogger *TruncatingLogger) Info(_ context.Context, message string, __ ...interface{}) {
+	if truncatingLogger.LogLevel < logger.Info {
+		return
+	}
+	fmt.Printf("gorm info: %.150s\n", message)
+}
+
+func (truncatingLogger *TruncatingLogger) Warn(_ context.Context, message string, __ ...interface{}) {
+	if truncatingLogger.LogLevel < logger.Warn {
+		return
+	}
+	fmt.Printf("gorm warning: %.150s\n", message)
+}
+
+func (truncatingLogger *TruncatingLogger) Error(_ context.Context, message string, __ ...interface{}) {
+	if truncatingLogger.LogLevel < logger.Error {
+		return
+	}
+	fmt.Printf("gorm error: %.150s\n", message)
+}
+
+func (truncatingLogger *TruncatingLogger) Trace(ctx context.Context, begin time.Time, fc func() (string, int64), err error) {
+	if truncatingLogger.LogLevel == logger.Silent {
+		return
+	}
+	elapsed := time.Since(begin)
+	if err != nil {
+		sql, rows := fc() // copied into every condition as this is a potentially heavy operation best done only when necessary
+		truncatingLogger.Error(ctx, fmt.Sprintf("Error in %s: %v - elapsed: %fs affected rows: %d, sql: %s", utils.FileWithLineNum(), err, elapsed.Seconds(), rows, sql))
+	} else if truncatingLogger.LogLevel >= logger.Warn && elapsed > truncatingLogger.SlowThreshold {
+		sql, rows := fc()
+		truncatingLogger.Warn(ctx, fmt.Sprintf("Slow sql query - elapse: %fs rows: %d, sql: %s", elapsed.Seconds(), rows, sql))
+	} else if truncatingLogger.LogLevel >= logger.Info {
+		sql, rows := fc()
+		truncatingLogger.Info(ctx, fmt.Sprintf("Sql query - elapse: %fs rows: %d, sql: %s", elapsed.Seconds(), rows, sql))
+	}
+}

api/pkg/version/consts.go

@@ -0,0 +1,8 @@
+package version
+
+var (
+	SemVer         = "0.0.1"
+	Branch         = "develop"
+	GitCommitHash  = "" // this var is overridden using ldflags in makefile when building
+	BuildTimestamp = "" // this var is overridden using ldflags in makefile when building
+)

build-push-featurebranch.sh

@@ -14,7 +14,7 @@ then
 fi
 
 echo "building $DOCKER_TAGGED_BUILD"
-docker build -t "$DOCKER_TAGGED_BUILD" .
+docker build -t "$DOCKER_TAGGED_BUILD" --build-arg SEM_VER=${SEM_VER} --build-arg BUILD_TIMESTAMP=${BUILD_TIMESTAMP} --build-arg GIT_BRANCH=${GIT_BRANCH} --build-arg COMMIT_HASH=${COMMIT_HASH} .
 
 echo pushing to "$REPOSITORY"
 docker push "$DOCKER_TAGGED_BUILD"

cli/cmd/fetch.go

@@ -2,6 +2,7 @@ package cmd
 
 import (
 	"github.com/spf13/cobra"
+	"github.com/up9inc/mizu/cli/mizu"
 )
 
 type MizuFetchOptions struct {
@@ -17,6 +18,11 @@ var fetchCmd = &cobra.Command{
 	Use:   "fetch",
 	Short: "Download recorded traffic to files",
 	RunE: func(cmd *cobra.Command, args []string) error {
+		if isCompatible, err := mizu.CheckVersionCompatibility(mizuFetchOptions.MizuPort); err != nil {
+			return err
+		} else if !isCompatible {
+			return nil
+		}
 		RunMizuFetch(&mizuFetchOptions)
 		return nil
 	},

cli/cmd/tap.go

@@ -3,13 +3,13 @@ package cmd
 import (
 	"errors"
 	"fmt"
+	"github.com/spf13/cobra"
 	"github.com/up9inc/mizu/cli/mizu"
 	"github.com/up9inc/mizu/cli/uiUtils"
+	"github.com/up9inc/mizu/shared/units"
 	"os"
 	"regexp"
 	"strings"
-
-	"github.com/spf13/cobra"
 )
 
 type MizuTapOptions struct {
@@ -22,12 +22,17 @@ type MizuTapOptions struct {
 	MizuImage              string
 	PlainTextFilterRegexes []string
 	TapOutgoing            bool
+	HideHealthChecks       bool
+	MaxEntriesDBSizeBytes  int64
 	SleepIntervalSec       uint16
 }
 
 var mizuTapOptions = &MizuTapOptions{}
 var direction string
+var humanMaxEntriesDBSize string
 var regex *regexp.Regexp
+const maxEntriesDBSizeFlagName = "max-entries-db-size"
+
 
 const analysisMessageToConfirm = `NOTE: running mizu with --analysis flag will upload recorded traffic
 to UP9 cloud for further analysis and enriched presentation options.
@@ -55,6 +60,13 @@ Supported protocols are HTTP and gRPC.`,
 			return errors.New(fmt.Sprintf("%s is not a valid regex %s", args[0], compileErr))
 		}
 
+		var parseHumanDataSizeErr error
+		mizuTapOptions.MaxEntriesDBSizeBytes, parseHumanDataSizeErr = units.HumanReadableToBytes(humanMaxEntriesDBSize)
+		if parseHumanDataSizeErr != nil {
+			return errors.New(fmt.Sprintf("Could not parse --max-entries-db-size value %s", humanMaxEntriesDBSize))
+		}
+		fmt.Printf("Mizu will store up to %s of traffic, old traffic will be cleared once the limit is reached.\n", units.BytesToHumanReadable(mizuTapOptions.MaxEntriesDBSizeBytes))
+
 		directionLowerCase := strings.ToLower(direction)
 		if directionLowerCase == "any" {
 			mizuTapOptions.TapOutgoing = true
@@ -85,7 +97,9 @@ func init() {
 	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")
+	tapCmd.Flags().StringVarP(&mizuTapOptions.MizuImage, "mizu-image", "", fmt.Sprintf("gcr.io/up9-docker-hub/mizu/%s:%s", mizu.Branch, mizu.SemVer), "Custom image for mizu collector")
 	tapCmd.Flags().StringArrayVarP(&mizuTapOptions.PlainTextFilterRegexes, "regex-masking", "r", nil, "List of regex expressions that are used to filter matching values from text/plain http bodies")
 	tapCmd.Flags().StringVarP(&direction, "direction", "", "in", "Record traffic that goes in this direction (relative to the tapped pod): in/any")
+	tapCmd.Flags().BoolVar(&mizuTapOptions.HideHealthChecks, "hide-healthchecks", false, "hides requests with kube-probe or prometheus user-agent headers")
+	tapCmd.Flags().StringVarP(&humanMaxEntriesDBSize, maxEntriesDBSizeFlagName, "", "200MB", "override the default max entries db size of 200mb")
 }

cli/cmd/tapRunner.go

@@ -4,10 +4,10 @@ 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"
 	"github.com/up9inc/mizu/shared"
+	"github.com/up9inc/mizu/shared/debounce"
 	core "k8s.io/api/core/v1"
 	"log"
 	"net/http"
@@ -96,7 +96,7 @@ func createMizuAggregator(ctx context.Context, kubernetesProvider *kubernetes.Pr
 	var err error
 
 	mizuServiceAccountExists = createRBACIfNecessary(ctx, kubernetesProvider)
-	_, err = kubernetesProvider.CreateMizuAggregatorPod(ctx, mizu.ResourcesNamespace, mizu.AggregatorPodName, tappingOptions.MizuImage, mizuServiceAccountExists, mizuApiFilteringOptions)
+	_, err = kubernetesProvider.CreateMizuAggregatorPod(ctx, mizu.ResourcesNamespace, mizu.AggregatorPodName, tappingOptions.MizuImage, mizuServiceAccountExists, mizuApiFilteringOptions, tappingOptions.MaxEntriesDBSizeBytes)
 	if err != nil {
 		fmt.Printf("Error creating mizu collector pod: %v\n", err)
 		return err
@@ -112,21 +112,21 @@ func createMizuAggregator(ctx context.Context, kubernetesProvider *kubernetes.Pr
 }
 
 func getMizuApiFilteringOptions(tappingOptions *MizuTapOptions) (*shared.TrafficFilteringOptions, error) {
-	if tappingOptions.PlainTextFilterRegexes == nil || len(tappingOptions.PlainTextFilterRegexes) == 0 {
-		return nil, nil
-	}
+	var compiledRegexSlice []*shared.SerializableRegexp
 
-	compiledRegexSlice := make([]*shared.SerializableRegexp, 0)
-	for _, regexStr := range tappingOptions.PlainTextFilterRegexes {
-		compiledRegex, err := shared.CompileRegexToSerializableRegexp(regexStr)
-		if err != nil {
-			fmt.Printf("Regex %s is invalid: %v", regexStr, err)
-			return nil, err
+	if tappingOptions.PlainTextFilterRegexes != nil && len(tappingOptions.PlainTextFilterRegexes) > 0 {
+		compiledRegexSlice = make([]*shared.SerializableRegexp, 0)
+		for _, regexStr := range tappingOptions.PlainTextFilterRegexes {
+			compiledRegex, err := shared.CompileRegexToSerializableRegexp(regexStr)
+			if err != nil {
+				fmt.Printf("Regex %s is invalid: %v", regexStr, err)
+				return nil, err
+			}
+			compiledRegexSlice = append(compiledRegexSlice, compiledRegex)
 		}
-		compiledRegexSlice = append(compiledRegexSlice, compiledRegex)
 	}
 
-	return &shared.TrafficFilteringOptions{PlainTextMaskingRegexes: compiledRegexSlice}, nil
+	return &shared.TrafficFilteringOptions{PlainTextMaskingRegexes: compiledRegexSlice, HideHealthChecks: tappingOptions.HideHealthChecks}, nil
 }
 
 func updateMizuTappers(ctx context.Context, kubernetesProvider *kubernetes.Provider, nodeToTappedPodIPMap map[string][]string, tappingOptions *MizuTapOptions) error {

cli/cmd/view.go

@@ -2,6 +2,7 @@ package cmd
 
 import (
 	"github.com/spf13/cobra"
+	"github.com/up9inc/mizu/cli/mizu"
 )
 
 type MizuViewOptions struct {
@@ -14,9 +15,15 @@ var viewCmd = &cobra.Command{
 	Use:   "view",
 	Short: "Open GUI in browser",
 	RunE: func(cmd *cobra.Command, args []string) error {
+		if isCompatible, err := mizu.CheckVersionCompatibility(mizuFetchOptions.MizuPort); err != nil {
+			return err
+		} else if !isCompatible {
+			return nil
+		}
 		runMizuView(mizuViewOptions)
 		return nil
 	},
+
 }
 
 func init() {

cli/go.sum

@@ -90,6 +90,7 @@ github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZm
 github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954/go.mod h1:vAd38F8PWV+bWy6jNmig1y/TA+kYO4g3RSRF0IAv0no=
 github.com/docker/distribution v2.7.1+incompatible/go.mod h1:J2gT2udsDAN96Uj4KfcMRqY0/ypR+oyYUYmja8H+y+w=
 github.com/docker/go-units v0.3.3/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
+github.com/docker/go-units v0.4.0 h1:3uh0PgVws3nIA0Q+MwDC8yjEPf9zjRfZZWXZYDct3Tw=
 github.com/docker/go-units v0.4.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
 github.com/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815/go.mod h1:WwZ+bS3ebgob9U8Nd0kOddGdZWjyMGR8Wziv+TBNwSE=
 github.com/elazarl/goproxy v0.0.0-20180725130230-947c36da3153/go.mod h1:/Zj4wYkgs4iZTTu3o/KG3Itv/qCCa8VVMlb3i9OVuzc=

cli/kubernetes/provider.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"path/filepath"
 	"regexp"
+	"strconv"
 
 	"github.com/up9inc/mizu/shared"
 	core "k8s.io/api/core/v1"
@@ -71,7 +72,7 @@ func (provider *Provider) GetPodWatcher(ctx context.Context, namespace string) w
 	return watcher
 }
 
-func (provider *Provider) CreateMizuAggregatorPod(ctx context.Context, namespace string, podName string, podImage string, linkServiceAccount bool, mizuApiFilteringOptions *shared.TrafficFilteringOptions) (*core.Pod, error) {
+func (provider *Provider) CreateMizuAggregatorPod(ctx context.Context, namespace string, podName string, podImage string, linkServiceAccount bool, mizuApiFilteringOptions *shared.TrafficFilteringOptions, maxEntriesDBSizeBytes int64) (*core.Pod, error) {
 	marshaledFilteringOptions, err := json.Marshal(mizuApiFilteringOptions)
 	if err != nil {
 		return nil, err
@@ -116,6 +117,10 @@ func (provider *Provider) CreateMizuAggregatorPod(ctx context.Context, namespace
 							Name:  shared.MizuFilteringOptionsEnvVar,
 							Value: string(marshaledFilteringOptions),
 						},
+						{
+							Name:  shared.MaxEntriesDBSizeByteSEnvVar,
+							Value: strconv.FormatInt(maxEntriesDBSizeBytes, 10),
+						},
 					},
 					Resources: core.ResourceRequirements{
 						Limits: core.ResourceList{

cli/mizu/versionCheck.go

@@ -0,0 +1,46 @@
+package mizu
+
+import (
+	"encoding/json"
+	"fmt"
+	"github.com/up9inc/mizu/shared"
+	"github.com/up9inc/mizu/shared/semver"
+	"net/http"
+	"net/url"
+)
+
+
+func getApiVersion(port uint16) (string, error) {
+	versionUrl, _ := url.Parse(fmt.Sprintf("http://localhost:%d/mizu/metadata/version", port))
+	req := &http.Request{
+		Method: http.MethodGet,
+		URL:    versionUrl,
+	}
+	statusResp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return "", err
+	}
+	defer statusResp.Body.Close()
+
+	versionResponse := &shared.VersionResponse{}
+	if err := json.NewDecoder(statusResp.Body).Decode(&versionResponse); err != nil {
+		return "", err
+	}
+
+	return versionResponse.SemVer, nil
+}
+
+func CheckVersionCompatibility(port uint16) (bool, error) {
+	apiSemVer, err := getApiVersion(port)
+	if err != nil {
+		return false, err
+	}
+
+	if semver.SemVersion(apiSemVer).Major() == semver.SemVersion(SemVer).Major() &&
+		semver.SemVersion(apiSemVer).Minor() == semver.SemVersion(SemVer).Minor() {
+		return true, nil
+	}
+
+	fmt.Printf(Red, fmt.Sprintf("cli version (%s) is not compatible with api version (%s)\n", SemVer, apiSemVer))
+	return false, nil
+}

shared/consts.go

@@ -1,8 +1,9 @@
 package shared
 
 const (
-	MizuFilteringOptionsEnvVar = "SENSITIVE_DATA_FILTERING_OPTIONS"
-	HostModeEnvVar = "HOST_MODE"
-	NodeNameEnvVar = "NODE_NAME"
+	MizuFilteringOptionsEnvVar       = "SENSITIVE_DATA_FILTERING_OPTIONS"
+	HostModeEnvVar                   = "HOST_MODE"
+	NodeNameEnvVar                   = "NODE_NAME"
 	TappedAddressesPerNodeDictEnvVar = "TAPPED_ADDRESSES_PER_HOST"
+	MaxEntriesDBSizeByteSEnvVar      = "MAX_ENTRIES_DB_BYTES"
 )

shared/go.mod

@@ -2,4 +2,8 @@ module github.com/up9inc/mizu/shared
 
 go 1.16
 
-require github.com/gorilla/websocket v1.4.2
+require (
+	github.com/gorilla/websocket v1.4.2
+	github.com/docker/go-units v0.4.0
+)
+

shared/go.sum

@@ -1,2 +1,4 @@
+github.com/docker/go-units v0.4.0 h1:3uh0PgVws3nIA0Q+MwDC8yjEPf9zjRfZZWXZYDct3Tw=
+github.com/docker/go-units v0.4.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
 github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc=
 github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=

shared/models.go

@@ -59,4 +59,10 @@ func CreateWebSocketMessageTypeAnalyzeStatus(analyzeStatus AnalyzeStatus) WebSoc
 
 type TrafficFilteringOptions struct {
 	PlainTextMaskingRegexes []*SerializableRegexp
+	HideHealthChecks        bool
 }
+
+type VersionResponse struct {
+	SemVer string `json:"semver"`
+}
+

shared/semver/semver.go

@@ -0,0 +1,28 @@
+package semver
+
+import (
+	"regexp"
+)
+
+type SemVersion string
+
+func (v SemVersion) Breakdown() (string, string, string) {
+	re := regexp.MustCompile(`\d+`)
+	breakdown := re.FindAllString(string(v), 3)
+	return breakdown[0], breakdown[1], breakdown[2]
+}
+
+func (v SemVersion) Major() string {
+	major, _, _ := v.Breakdown()
+	return major
+}
+
+func (v SemVersion) Minor() string {
+	_, minor, _ := v.Breakdown()
+	return minor
+}
+
+func (v SemVersion) Patch() string {
+	_, _, patch := v.Breakdown()
+	return patch
+}

shared/socket_client.go

@@ -7,7 +7,7 @@ import (
 )
 
 const (
-	DEFAULT_SOCKET_RETRIES = 3
+	DEFAULT_SOCKET_RETRIES          = 3
 	DEFAULT_SOCKET_RETRY_SLEEP_TIME = time.Second * 10
 )
 

shared/units/human_data_sizes.go

@@ -0,0 +1,11 @@
+package units
+
+import "github.com/docker/go-units"
+
+func BytesToHumanReadable(bytes int64) string {
+	return units.HumanSize(float64(bytes))
+}
+
+func HumanReadableToBytes(humanReadableSize string) (int64, error) {
+	return units.FromHumanSize(humanReadableSize)
+}

tap/har_writer.go

@@ -1,9 +1,11 @@
 package tap
 
 import (
+	"bytes"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io/ioutil"
 	"log"
 	"net/http"
 	"os"
@@ -46,12 +48,27 @@ type HarFile struct {
 }
 
 func NewEntry(request *http.Request, requestTime time.Time, response *http.Response, responseTime time.Time) (*har.Entry, error) {
-	harRequest, err := har.NewRequest(request, true)
+	harRequest, err := har.NewRequest(request, false)
 	if err != nil {
 		SilentError("convert-request-to-har", "Failed converting request to HAR %s (%v,%+v)", err, err, err)
 		return nil, errors.New("Failed converting request to HAR")
 	}
 
+	// For requests with multipart/form-data or application/x-www-form-urlencoded Content-Type,
+	// martian/har will parse the request body and place the parameters in harRequest.PostData.Params
+	// instead of harRequest.PostData.Text (as the HAR spec requires it).
+	// Mizu currently only looks at PostData.Text. Therefore, instead of letting martian/har set the content of
+	// PostData, always copy the request body to PostData.Text.
+	if (request.ContentLength > 0) {
+		reqBody, err := ioutil.ReadAll(request.Body)
+		if err != nil {
+			SilentError("read-request-body", "Failed converting request to HAR %s (%v,%+v)", err, err, err)
+			return nil, errors.New("Failed reading request body")
+		}
+		request.Body = ioutil.NopCloser(bytes.NewReader(reqBody))
+		harRequest.PostData.Text = string(reqBody)
+	}
+
 	harResponse, err := har.NewResponse(response, true)
 	if err != nil {
 		SilentError("convert-response-to-har", "Failed converting response to HAR %s (%v,%+v)", err, err, err)

ui/src/components/style/HarPage.sass

@@ -14,6 +14,12 @@
         align-items: center
         background-color: $header-background-color
 
+    .harPageHeader
+        padding: 20px 24px
+        display: flex
+        align-items: center
+        background-color: $header-background-color
+
 .HarPage-Header
     display: flex
     height: 2.5%