Fix spelling in docs (#745)

Co-authored-by: gadotroee <55343099+gadotroee@users.noreply.github.com>

.github/workflows/build.yml

@@ -1,4 +1,4 @@
-name: PR validation
+name: Build
 
 on:
   pull_request:
@@ -12,7 +12,7 @@ concurrency:
 
 jobs:
   build-cli:
-    name: Build CLI
+    name: CLI executable build
     runs-on: ubuntu-latest
     steps:
       - name: Set up Go 1.16
@@ -27,35 +27,11 @@ jobs:
         run: make cli
 
   build-agent:
-    name: Build Agent
+    name: Agent docker image build
     runs-on: ubuntu-latest
     steps:
-      - name: Set up Go 1.16
-        uses: actions/setup-go@v2
-        with:
-          go-version: '1.16'
-
       - name: Check out code into the Go module directory
         uses: actions/checkout@v2
 
-      - shell: bash
-        run: |
-          sudo apt-get install libpcap-dev
-
       - name: Build Agent
-        run: make agent
-
-  build-ui:
-    name: Build UI
-    runs-on: ubuntu-latest
-    steps:
-      - name: Set up Node 16
-        uses: actions/setup-node@v2
-        with:
-          node-version: '16'
-
-      - name: Check out code into the Go module directory
-        uses: actions/checkout@v2
-
-      - name: Build UI
-        run: make ui
+        run: make agent-docker

.github/workflows/static_code_analysis.yml

@@ -0,0 +1,90 @@
+name: Static code analysis
+
+on:
+  pull_request:
+    branches:
+      - 'develop'
+      - 'main'
+
+permissions:
+  contents: read
+
+jobs:
+  golangci:
+    name: Go lint
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - uses: actions/setup-go@v2
+        with:
+            go-version: '^1.16'
+
+      - name: Install dependencies
+        run: |
+          sudo apt update
+          sudo apt install -y libpcap-dev
+
+      - name: Go lint - agent
+        uses: golangci/golangci-lint-action@v2
+        with:
+          version: latest
+          working-directory: agent
+          args: --timeout=3m
+
+      - name: Go lint - shared
+        uses: golangci/golangci-lint-action@v2
+        with:
+          version: latest
+          working-directory: shared
+          args: --timeout=3m
+
+      - name: Go lint - tap
+        uses: golangci/golangci-lint-action@v2
+        with:
+          version: latest
+          working-directory: tap
+          args: --timeout=3m
+
+      - name: Go lint - CLI
+        uses: golangci/golangci-lint-action@v2
+        with:
+          version: latest
+          working-directory: cli
+          args: --timeout=3m
+
+      - name: Go lint - acceptanceTests
+        uses: golangci/golangci-lint-action@v2
+        with:
+          version: latest
+          working-directory: acceptanceTests
+          args: --timeout=3m
+
+      - name: Go lint - tap/api
+        uses: golangci/golangci-lint-action@v2
+        with:
+          version: latest
+          working-directory: tap/api
+
+      - name: Go lint - tap/extensions/amqp
+        uses: golangci/golangci-lint-action@v2
+        with:
+          version: latest
+          working-directory: tap/extensions/amqp
+
+      - name: Go lint - tap/extensions/http
+        uses: golangci/golangci-lint-action@v2
+        with:
+          version: latest
+          working-directory: tap/extensions/http
+
+      - name: Go lint - tap/extensions/kafka
+        uses: golangci/golangci-lint-action@v2
+        with:
+          version: latest
+          working-directory: tap/extensions/kafka
+
+      - name: Go lint - tap/extensions/redis
+        uses: golangci/golangci-lint-action@v2
+        with:
+          version: latest
+          working-directory: tap/extensions/redis

.github/workflows/test.yml

@@ -1,14 +1,10 @@
-name: tests validation
+name: Test
 
 on:
   pull_request:
     branches:
       - 'develop'
       - 'main'
-  push:
-    branches:
-      - 'develop'
-      - 'main'
 
 concurrency:
   group: mizu-tests-validation-${{ github.ref }}
@@ -16,7 +12,7 @@ concurrency:
 
 jobs:
   run-tests-cli:
-    name: Run CLI tests
+    name: CLI Tests
     runs-on: ubuntu-latest
     steps:
       - name: Set up Go 1.16
@@ -34,7 +30,7 @@ jobs:
         uses: codecov/codecov-action@v2
 
   run-tests-agent:
-    name: Run Agent tests
+    name: Agent Tests
     runs-on: ubuntu-latest
     steps:
       - name: Set up Go 1.16

Dockerfile

@@ -72,10 +72,10 @@ ARG SEM_VER=0.0.0
 WORKDIR /app/agent-build
 
 RUN go build -ldflags="-extldflags=-static -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 .
+    -X 'github.com/up9inc/mizu/agent/pkg/version.GitCommitHash=${COMMIT_HASH}' \
+    -X 'github.com/up9inc/mizu/agent/pkg/version.Branch=${GIT_BRANCH}' \
+    -X 'github.com/up9inc/mizu/agent/pkg/version.BuildTimestamp=${BUILD_TIMESTAMP}' \
+    -X 'github.com/up9inc/mizu/agent/pkg/version.SemVer=${SEM_VER}'" -o mizuagent .
 
 # Download Basenine executable, verify the sha1sum
 ADD https://github.com/up9inc/basenine/releases/download/v0.4.13/basenine_linux_${GOARCH} ./basenine_linux_${GOARCH}

Makefile

@@ -47,6 +47,10 @@ agent-debug: ## Build agent for debug.
 docker: ## Build and publish agent docker image.
 	$(MAKE) push-docker
 
+agent-docker: ## Build agent docker image.
+	@echo "Building agent docker image"
+	@docker build -t up9inc/mizu:devlatest .
+
 push: push-docker push-cli ## Build and publish agent docker image & CLI.
 
 push-docker: ## Build and publish agent docker image.

acceptanceTests/cypress.json

@@ -4,7 +4,7 @@
   "viewportHeight": 1080,
   "video": false,
   "screenshotOnRunFailure": false,
-  "defaultCommandTimeout": 5000,
+  "defaultCommandTimeout": 6000,
   "testFiles": [
     "tests/GuiPort.js",
     "tests/MultipleNamespaces.js",

acceptanceTests/cypress/integration/tests/UiTest.js

@@ -3,6 +3,7 @@ import {resizeToHugeMizu, resizeToNormalMizu} from "../testHelpers/TrafficHelper
 const greenFilterColor = 'rgb(210, 250, 210)';
 const redFilterColor = 'rgb(250, 214, 220)';
 const refreshWaitTimeout = 10000;
+const bodyJsonClass = '.hljs';
 
 it('opening mizu', function () {
     cy.visit(Cypress.env('testUrl'));
@@ -25,6 +26,37 @@ it('filtering guide check', function () {
     cy.get('#modal-modal-title').should('not.exist');
 });
 
+it('right side sanity test', function () {
+    cy.get('#entryDetailedTitleBodySize').then(sizeTopLine => {
+        const sizeOnTopLine = sizeTopLine.text().replace(' B', '');
+        cy.contains('Response').click();
+        cy.contains('Body Size (bytes)').parent().next().then(size => {
+            const bodySizeByDetails = size.text();
+            expect(sizeOnTopLine).to.equal(bodySizeByDetails, 'The body size in the top line should match the details in the response');
+
+            if (parseInt(bodySizeByDetails) < 0) {
+                throw new Error(`The body size cannot be negative. got the size: ${bodySizeByDetails}`)
+            }
+
+            cy.get('#entryDetailedTitleElapsedTime').then(timeInMs => {
+                const time = timeInMs.text();
+                if (time < '0ms') {
+                    throw new Error(`The time in the top line cannot be negative ${time}`);
+                }
+
+                cy.get('#rightSideContainer [title="Status Code"]').then(status => {
+                    const statusCode = status.text();
+                    cy.contains('Status').parent().next().then(statusInDetails => {
+                        const statusCodeInDetails = statusInDetails.text();
+
+                        expect(statusCode).to.equal(statusCodeInDetails, 'The status code in the top line should match the status code in details');
+                    });
+                });
+            });
+        });
+    });
+});
+
 checkIllegalFilter('invalid filter');
 
 checkFilter({
@@ -161,11 +193,12 @@ function checkFilter(filterDetails){
             if (!applyByEnter)
                 cy.get('[type="submit"]').click();
 
-            // only one entry in DOM after filtering, checking all four checks on it
+            // only one entry in DOM after filtering, checking all checks on it
             leftTextCheck(totalEntries - 1, leftSidePath, leftSideExpectedText);
             leftOnHoverCheck(totalEntries - 1, leftSidePath, name);
             rightTextCheck(rightSidePath, rightSideExpectedText);
             rightOnHoverCheck(rightSidePath, name);
+            checkRightSideResponseBody();
 
             cy.get('[title="Fetch old records"]').click();
             resizeToHugeMizu();
@@ -196,6 +229,7 @@ function deeperChcek(leftSidePath, rightSidePath, filterName, leftSideExpectedTe
         cy.get(`#list #entry-${entryNum}`).click();
         rightTextCheck(rightSidePath, rightSideExpectedText);
         rightOnHoverCheck(rightSidePath, filterName);
+        checkRightSideResponseBody();
     });
 }
 
@@ -216,3 +250,88 @@ function rightOnHoverCheck(path, expectedText) {
     cy.get(`.TrafficPage-Container > :nth-child(2) ${path}`).trigger('mouseover');
     cy.get(`.TrafficPage-Container > :nth-child(2) .Queryable-Tooltip`).should('have.text', expectedText);
 }
+
+
+function checkRightSideResponseBody() {
+    cy.contains('Response').click();
+    clickCheckbox('Decode Base64');
+
+    cy.get(`${bodyJsonClass}`).then(value => {
+        const encodedBody = value.text();
+        cy.log(encodedBody);
+
+        const decodedBody = atob(encodedBody);
+        const responseBody = JSON.parse(decodedBody);
+
+        const expectdJsonBody = {
+            args: RegExp({}),
+            url: RegExp('http://.*/get'),
+            headers: {
+                "User-Agent": RegExp('[REDACTED]'),
+                "Accept-Encoding": RegExp('gzip'),
+                "X-Forwarded-Uri": RegExp('/api/v1/namespaces/.*/services/.*/proxy/get')
+            }
+        };
+
+        expect(responseBody.args).to.match(expectdJsonBody.args);
+        expect(responseBody.url).to.match(expectdJsonBody.url);
+        expect(responseBody.headers['User-Agent']).to.match(expectdJsonBody.headers['User-Agent']);
+        expect(responseBody.headers['Accept-Encoding']).to.match(expectdJsonBody.headers['Accept-Encoding']);
+        expect(responseBody.headers['X-Forwarded-Uri']).to.match(expectdJsonBody.headers['X-Forwarded-Uri']);
+
+        cy.get(`${bodyJsonClass}`).should('have.text', encodedBody);
+        clickCheckbox('Decode Base64');
+
+        cy.get(`${bodyJsonClass} > `).its('length').should('be.gt', 1).then(linesNum => {
+            cy.get(`${bodyJsonClass} > >`).its('length').should('be.gt', linesNum).then(jsonItemsNum => {
+                checkPrettyAndLineNums(jsonItemsNum, decodedBody);
+
+                clickCheckbox('Line numbers');
+                checkPrettyOrNothing(jsonItemsNum, decodedBody);
+
+                clickCheckbox('Pretty');
+                checkPrettyOrNothing(jsonItemsNum, decodedBody);
+
+                clickCheckbox('Line numbers');
+                checkOnlyLineNumberes(jsonItemsNum, decodedBody);
+            });
+        });
+    });
+}
+
+function clickCheckbox(type) {
+    cy.contains(`${type}`).prev().children().click();
+}
+
+function checkPrettyAndLineNums(jsonItemsLen, decodedBody) {
+    decodedBody = decodedBody.replaceAll(' ', '');
+    cy.get(`${bodyJsonClass} >`).then(elements => {
+        const lines = Object.values(elements);
+        lines.forEach((line, index) => {
+            if (line.getAttribute) {
+                const cleanLine = getCleanLine(line);
+                const currentLineFromDecodedText = decodedBody.substring(0, cleanLine.length);
+
+                expect(cleanLine).to.equal(currentLineFromDecodedText, `expected the text in line number ${index + 1} to match the text that generated by the base64 decoding`)
+
+                decodedBody = decodedBody.substring(cleanLine.length);
+            }
+        });
+    });
+}
+
+function getCleanLine(lineElement) {
+    return (lineElement.innerText.substring(0, lineElement.innerText.length - 1)).replaceAll(' ', '');
+}
+
+function checkPrettyOrNothing(jsonItems, decodedBody) {
+    cy.get(`${bodyJsonClass} > `).should('have.length', jsonItems).then(text => {
+        const json = text.text();
+        expect(json).to.equal(decodedBody);
+    });
+}
+
+function checkOnlyLineNumberes(jsonItems, decodedText) {
+    cy.get(`${bodyJsonClass} >`).should('have.length', 1).and('have.text', decodedText);
+    cy.get(`${bodyJsonClass} > >`).should('have.length', jsonItems)
+}

acceptanceTests/testsUtils.go

@@ -3,7 +3,6 @@ package acceptanceTests
 import (
 	"bytes"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"io/ioutil"
 	"net/http"
@@ -11,12 +10,10 @@ import (
 	"os/exec"
 	"path"
 	"strings"
-	"sync"
 	"syscall"
 	"testing"
 	"time"
 
-	"github.com/gorilla/websocket"
 	"github.com/up9inc/mizu/shared"
 )
 
@@ -28,7 +25,6 @@ const (
 	defaultServiceName    = "httpbin"
 	defaultEntriesCount   = 50
 	waitAfterTapPodsReady = 3 * time.Second
-	cleanCommandTimeout   = 1 * time.Minute
 )
 
 type PodDescriptor struct {
@@ -36,18 +32,6 @@ type PodDescriptor struct {
 	Namespace string
 }
 
-func isPodDescriptorInPodArray(pods []map[string]interface{}, podDescriptor PodDescriptor) bool {
-	for _, pod := range pods {
-		podNamespace := pod["namespace"].(string)
-		podName := pod["name"].(string)
-
-		if podDescriptor.Namespace == podNamespace && strings.Contains(podName, podDescriptor.Name) {
-			return true
-		}
-	}
-	return false
-}
-
 func getCliPath() (string, error) {
 	dir, filePathErr := os.Getwd()
 	if filePathErr != nil {
@@ -84,10 +68,6 @@ func getApiServerUrl(port uint16) string {
 	return fmt.Sprintf("http://localhost:%v", port)
 }
 
-func getWebSocketUrl(port uint16) string {
-	return fmt.Sprintf("ws://localhost:%v/ws", port)
-}
-
 func getDefaultCommandArgs() []string {
 	setFlag := "--set"
 	telemetry := "telemetry=false"
@@ -130,20 +110,6 @@ func getDefaultConfigCommandArgs() []string {
 	return append([]string{configCommand}, defaultCmdArgs...)
 }
 
-func getDefaultCleanCommandArgs() []string {
-	cleanCommand := "clean"
-	defaultCmdArgs := getDefaultCommandArgs()
-
-	return append([]string{cleanCommand}, defaultCmdArgs...)
-}
-
-func getDefaultViewCommandArgs() []string {
-	viewCommand := "view"
-	defaultCmdArgs := getDefaultCommandArgs()
-
-	return append([]string{viewCommand}, defaultCmdArgs...)
-}
-
 func runCypressTests(t *testing.T, cypressRunCmd string) {
 	cypressCmd := exec.Command("bash", "-c", cypressRunCmd)
 	t.Logf("running command: %v", cypressCmd.String())
@@ -268,36 +234,6 @@ func executeHttpPostRequestWithHeaders(url string, headers map[string]string, bo
 	return executeHttpRequest(response, requestErr)
 }
 
-func runMizuClean() error {
-	cliPath, err := getCliPath()
-	if err != nil {
-		return err
-	}
-
-	cleanCmdArgs := getDefaultCleanCommandArgs()
-
-	cleanCmd := exec.Command(cliPath, cleanCmdArgs...)
-
-	commandDone := make(chan error)
-	go func() {
-		if err := cleanCmd.Run(); err != nil {
-			commandDone <- err
-		}
-		commandDone <- nil
-	}()
-
-	select {
-	case err = <-commandDone:
-		if err != nil {
-			return err
-		}
-	case <-time.After(cleanCommandTimeout):
-		return errors.New("clean command timed out")
-	}
-
-	return nil
-}
-
 func cleanupCommand(cmd *exec.Cmd) error {
 	if err := cmd.Process.Signal(syscall.SIGQUIT); err != nil {
 		return err
@@ -310,17 +246,6 @@ func cleanupCommand(cmd *exec.Cmd) error {
 	return nil
 }
 
-func getPods(tapStatusInterface interface{}) ([]map[string]interface{}, error) {
-	tapPodsInterface := tapStatusInterface.([]interface{})
-
-	var pods []map[string]interface{}
-	for _, podInterface := range tapPodsInterface {
-		pods = append(pods, podInterface.(map[string]interface{}))
-	}
-
-	return pods, nil
-}
-
 func getLogsPath() (string, error) {
 	dir, filePathErr := os.Getwd()
 	if filePathErr != nil {
@@ -331,77 +256,6 @@ func getLogsPath() (string, error) {
 	return logsPath, nil
 }
 
-// waitTimeout waits for the waitgroup for the specified max timeout.
-// Returns true if waiting timed out.
-func waitTimeout(wg *sync.WaitGroup, timeout time.Duration) bool {
-	channel := make(chan struct{})
-	go func() {
-		defer close(channel)
-		wg.Wait()
-	}()
-	select {
-	case <-channel:
-		return false // completed normally
-	case <-time.After(timeout):
-		return true // timed out
-	}
-}
-
-// checkEntriesAtLeast checks whether the number of entries greater than or equal to n
-func checkEntriesAtLeast(entries []map[string]interface{}, n int) error {
-	if len(entries) < n {
-		return fmt.Errorf("Unexpected entries result - Expected more than %d entries", n-1)
-	}
-	return nil
-}
-
-// getDBEntries retrieves the entries from the database before the given timestamp.
-// Also limits the results according to the limit parameter.
-// Timeout for the WebSocket connection is defined by the timeout parameter.
-func getDBEntries(timestamp int64, limit int, timeout time.Duration) (entries []map[string]interface{}, err error) {
-	query := fmt.Sprintf("timestamp < %d and limit(%d)", timestamp, limit)
-	webSocketUrl := getWebSocketUrl(defaultApiServerPort)
-
-	var connection *websocket.Conn
-	connection, _, err = websocket.DefaultDialer.Dial(webSocketUrl, nil)
-	if err != nil {
-		return
-	}
-	defer connection.Close()
-
-	handleWSConnection := func(wg *sync.WaitGroup) {
-		defer wg.Done()
-		for {
-			_, message, err := connection.ReadMessage()
-			if err != nil {
-				return
-			}
-
-			var data map[string]interface{}
-			if err = json.Unmarshal([]byte(message), &data); err != nil {
-				return
-			}
-
-			if data["messageType"] == "entry" {
-				entries = append(entries, data)
-			}
-		}
-	}
-
-	err = connection.WriteMessage(websocket.TextMessage, []byte(query))
-	if err != nil {
-		return
-	}
-
-	var wg sync.WaitGroup
-	go handleWSConnection(&wg)
-	wg.Add(1)
-
-	waitTimeout(&wg, timeout)
-
-	return
-}
-
 func Contains(slice []string, containsValue string) bool {
 	for _, sliceValue := range slice {
 		if sliceValue == containsValue {

agent/go.mod

@@ -1,4 +1,4 @@
-module mizuserver
+module github.com/up9inc/mizu/agent
 
 go 1.16
 
@@ -6,13 +6,13 @@ require (
 	github.com/antelman107/net-wait-go v0.0.0-20210623112055-cf684aebda7b
 	github.com/chanced/openapi v0.0.6
 	github.com/djherbis/atime v1.0.0
+	github.com/elastic/go-elasticsearch/v7 v7.16.0
 	github.com/getkin/kin-openapi v0.76.0
 	github.com/gin-contrib/static v0.0.1
 	github.com/gin-gonic/gin v1.7.7
 	github.com/go-playground/locales v0.13.0
 	github.com/go-playground/universal-translator v0.17.0
 	github.com/go-playground/validator/v10 v10.5.0
-	github.com/google/martian v2.1.0+incompatible
 	github.com/google/uuid v1.1.2
 	github.com/gorilla/websocket v1.4.2
 	github.com/op/go-logging v0.0.0-20160315200505-970db520ece7

agent/go.sum

@@ -126,6 +126,8 @@ github.com/docker/go-units v0.4.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDD
 github.com/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815/go.mod h1:WwZ+bS3ebgob9U8Nd0kOddGdZWjyMGR8Wziv+TBNwSE=
 github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21 h1:YEetp8/yCZMuEPMUDHG0CW/brkkEp8mzqk2+ODEitlw=
 github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21/go.mod h1:+020luEh2TKB4/GOp8oxxtq0Daoen/Cii55CzbTV6DU=
+github.com/elastic/go-elasticsearch/v7 v7.16.0 h1:GHsxDFXIAlhSleXun4kwA89P7kQFADRChqvgOPeYP5A=
+github.com/elastic/go-elasticsearch/v7 v7.16.0/go.mod h1:OJ4wdbtDNk5g503kvlHLyErCgQwwzmDtaFC4XyOxXA4=
 github.com/elazarl/goproxy v0.0.0-20180725130230-947c36da3153 h1:yUdfgN0XgIJw7foRItutHYUIhlcKzcSf5vDpdhQAKTc=
 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=

agent/main.go

@@ -6,16 +6,6 @@ import (
 	"flag"
 	"fmt"
 	"io/ioutil"
-	"mizuserver/pkg/api"
-	"mizuserver/pkg/config"
-	"mizuserver/pkg/controllers"
-	"mizuserver/pkg/middlewares"
-	"mizuserver/pkg/models"
-	"mizuserver/pkg/oas"
-	"mizuserver/pkg/routes"
-	"mizuserver/pkg/servicemap"
-	"mizuserver/pkg/up9"
-	"mizuserver/pkg/utils"
 	"net/http"
 	"os"
 	"os/signal"
@@ -25,6 +15,21 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/up9inc/mizu/agent/pkg/middlewares"
+	"github.com/up9inc/mizu/agent/pkg/models"
+	"github.com/up9inc/mizu/agent/pkg/oas"
+	"github.com/up9inc/mizu/agent/pkg/routes"
+	"github.com/up9inc/mizu/agent/pkg/servicemap"
+	"github.com/up9inc/mizu/agent/pkg/up9"
+	"github.com/up9inc/mizu/agent/pkg/utils"
+
+	"github.com/up9inc/mizu/agent/pkg/elastic"
+
+	"github.com/up9inc/mizu/agent/pkg/controllers"
+
+	"github.com/up9inc/mizu/agent/pkg/api"
+	"github.com/up9inc/mizu/agent/pkg/config"
+
 	v1 "k8s.io/api/core/v1"
 
 	"github.com/antelman107/net-wait-go/wait"
@@ -61,7 +66,6 @@ const (
 	socketConnectionRetries    = 30
 	socketConnectionRetryDelay = time.Second * 2
 	socketHandshakeTimeout     = time.Second * 2
-	uiIndexPath                = "./site/index.html"
 )
 
 func main() {
@@ -159,6 +163,7 @@ func enableExpFeatureIfNeeded() {
 	if config.Config.ServiceMap {
 		servicemap.GetInstance().SetConfig(config.Config)
 	}
+	elastic.GetInstance().Configure(config.Config.Elastic)
 }
 
 func configureBasenineServer(host string, port string) {
@@ -167,7 +172,7 @@ func configureBasenineServer(host string, port string) {
 		wait.WithWait(200*time.Millisecond),
 		wait.WithBreak(50*time.Millisecond),
 		wait.WithDeadline(5*time.Second),
-		wait.WithDebug(true),
+		wait.WithDebug(config.Config.LogLevel == logging.DEBUG),
 	).Do([]string{fmt.Sprintf("%s:%s", host, port)}) {
 		logger.Log.Panicf("Basenine is not available!")
 	}
@@ -246,16 +251,23 @@ func hostApi(socketHarOutputChannel chan<- *tapApi.OutputChannelItem) {
 
 	app.Use(DisableRootStaticCache())
 
-	if err := setUIFlags(); err != nil {
-		logger.Log.Errorf("Error setting ui mode, err: %v", err)
+	var staticFolder string
+	if config.Config.StandaloneMode {
+		staticFolder = "./site-standalone"
+	} else {
+		staticFolder = "./site"
 	}
 
-	if config.Config.StandaloneMode {
-		app.Use(static.ServeRoot("/", "./site-standalone"))
-	} else {
-		app.Use(static.ServeRoot("/", "./site"))
+	indexStaticFile := staticFolder + "/index.html"
+	if err := setUIFlags(indexStaticFile); err != nil {
+		logger.Log.Errorf("Error setting ui flags, err: %v", err)
 	}
 
+	app.Use(static.ServeRoot("/", staticFolder))
+	app.NoRoute(func(c *gin.Context) {
+		c.File(indexStaticFile)
+	})
+
 	app.Use(middlewares.CORSMiddleware()) // This has to be called after the static middleware, does not work if its called before
 
 	api.WebSocketRoutes(app, &eventHandlers, startTime)
@@ -276,7 +288,6 @@ func hostApi(socketHarOutputChannel chan<- *tapApi.OutputChannelItem) {
 	routes.EntriesRoutes(app)
 	routes.MetadataRoutes(app)
 	routes.StatusRoutes(app)
-	routes.NotFoundRoute(app)
 	utils.StartServer(app)
 }
 
@@ -291,7 +302,7 @@ func DisableRootStaticCache() gin.HandlerFunc {
 	}
 }
 
-func setUIFlags() error {
+func setUIFlags(uiIndexPath string) error {
 	read, err := ioutil.ReadFile(uiIndexPath)
 	if err != nil {
 		return err

agent/pkg/api/main.go

@@ -5,24 +5,27 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"mizuserver/pkg/har"
-	"mizuserver/pkg/holder"
-	"mizuserver/pkg/providers"
 	"os"
 	"path"
 	"sort"
 	"strings"
 	"time"
 
-	"mizuserver/pkg/servicemap"
+	"github.com/up9inc/mizu/agent/pkg/elastic"
+	"github.com/up9inc/mizu/agent/pkg/har"
+	"github.com/up9inc/mizu/agent/pkg/holder"
+	"github.com/up9inc/mizu/agent/pkg/providers"
+
+	"github.com/up9inc/mizu/agent/pkg/servicemap"
+
+	"github.com/up9inc/mizu/agent/pkg/models"
+	"github.com/up9inc/mizu/agent/pkg/oas"
+	"github.com/up9inc/mizu/agent/pkg/resolver"
+	"github.com/up9inc/mizu/agent/pkg/utils"
 
 	"github.com/up9inc/mizu/shared"
 	"github.com/up9inc/mizu/shared/logger"
 	tapApi "github.com/up9inc/mizu/tap/api"
-	"mizuserver/pkg/models"
-	"mizuserver/pkg/oas"
-	"mizuserver/pkg/resolver"
-	"mizuserver/pkg/utils"
 
 	basenine "github.com/up9inc/basenine/client/go"
 )
@@ -40,10 +43,8 @@ func StartResolving(namespace string) {
 	res.Start(ctx)
 	go func() {
 		for {
-			select {
-			case err := <-errOut:
-				logger.Log.Infof("name resolving error %s", err)
-			}
+			err := <-errOut
+			logger.Log.Infof("name resolving error %s", err)
 		}
 	}()
 
@@ -65,7 +66,7 @@ func startReadingFiles(workingDir string) {
 		return
 	}
 
-	for true {
+	for {
 		dir, _ := os.Open(workingDir)
 		dirFiles, _ := dir.Readdir(-1)
 
@@ -122,7 +123,9 @@ func startReadingChannel(outputItems <-chan *tapApi.OutputChannelItem, extension
 		if extension.Protocol.Name == "http" {
 			if !disableOASValidation {
 				var httpPair tapApi.HTTPRequestResponsePair
-				json.Unmarshal([]byte(mizuEntry.HTTPPair), &httpPair)
+				if err := json.Unmarshal([]byte(mizuEntry.HTTPPair), &httpPair); err != nil {
+					logger.Log.Error(err)
+				}
 
 				contract := handleOAS(ctx, doc, router, httpPair.Request.Payload.RawRequest, httpPair.Response.Payload.RawResponse, contractContent)
 				mizuEntry.ContractStatus = contract.Status
@@ -150,6 +153,7 @@ func startReadingChannel(outputItems <-chan *tapApi.OutputChannelItem, extension
 		connection.SendText(string(data))
 
 		servicemap.GetInstance().NewTCPEntry(mizuEntry.Source, mizuEntry.Destination, &item.Protocol)
+		elastic.GetInstance().PushEntry(mizuEntry)
 	}
 }
 

agent/pkg/api/socket_routes.go

@@ -2,18 +2,17 @@ package api
 
 import (
 	"encoding/json"
-	"errors"
 	"fmt"
-	"mizuserver/pkg/models"
 	"net/http"
 	"sync"
 	"time"
 
+	"github.com/up9inc/mizu/agent/pkg/models"
+
 	"github.com/gin-gonic/gin"
 	"github.com/gorilla/websocket"
 	basenine "github.com/up9inc/basenine/client/go"
 	"github.com/up9inc/mizu/shared"
-	"github.com/up9inc/mizu/shared/debounce"
 	"github.com/up9inc/mizu/shared/logger"
 	tapApi "github.com/up9inc/mizu/tap/api"
 )
@@ -42,7 +41,7 @@ var connectedWebsocketIdCounter = 0
 
 func init() {
 	websocketUpgrader.CheckOrigin = func(r *http.Request) bool { return true } // like cors for web socket
-	connectedWebsockets = make(map[int]*SocketConnection, 0)
+	connectedWebsockets = make(map[int]*SocketConnection)
 }
 
 func WebSocketRoutes(app *gin.Engine, eventHandlers EventHandlers, startTime int64) {
@@ -94,7 +93,10 @@ func websocketHandler(w http.ResponseWriter, r *http.Request, eventHandlers Even
 	eventHandlers.WebSocketConnect(socketId, isTapper)
 
 	startTimeBytes, _ := models.CreateWebsocketStartTimeMessage(startTime)
-	SendToSocket(socketId, startTimeBytes)
+
+	if err = SendToSocket(socketId, startTimeBytes); err != nil {
+		logger.Log.Error(err)
+	}
 
 	for {
 		_, msg, err := ws.ReadMessage()
@@ -117,7 +119,9 @@ func websocketHandler(w http.ResponseWriter, r *http.Request, eventHandlers Even
 					AutoClose: 5000,
 					Text:      fmt.Sprintf("Syntax error: %s", err.Error()),
 				})
-				SendToSocket(socketId, toastBytes)
+				if err := SendToSocket(socketId, toastBytes); err != nil {
+					logger.Log.Error(err)
+				}
 				break
 			}
 
@@ -137,7 +141,9 @@ func websocketHandler(w http.ResponseWriter, r *http.Request, eventHandlers Even
 					base := tapApi.Summarize(entry)
 
 					baseEntryBytes, _ := models.CreateBaseEntryWebSocketMessage(base)
-					SendToSocket(socketId, baseEntryBytes)
+					if err := SendToSocket(socketId, baseEntryBytes); err != nil {
+						logger.Log.Error(err)
+					}
 				}
 			}
 
@@ -156,7 +162,9 @@ func websocketHandler(w http.ResponseWriter, r *http.Request, eventHandlers Even
 					}
 
 					metadataBytes, _ := models.CreateWebsocketQueryMetadataMessage(metadata)
-					SendToSocket(socketId, metadataBytes)
+					if err := SendToSocket(socketId, metadataBytes); err != nil {
+						logger.Log.Error(err)
+					}
 				}
 			}
 
@@ -183,14 +191,10 @@ func socketCleanup(socketId int, socketConnection *SocketConnection) {
 	socketConnection.eventHandlers.WebSocketDisconnect(socketId, socketConnection.isTapper)
 }
 
-var db = debounce.NewDebouncer(time.Second*5, func() {
-	logger.Log.Error("Successfully sent to socket")
-})
-
 func SendToSocket(socketId int, message []byte) error {
 	socketObj := connectedWebsockets[socketId]
 	if socketObj == nil {
-		return errors.New("Socket is disconnected")
+		return fmt.Errorf("Socket %v is disconnected", socketId)
 	}
 
 	var sent = false
@@ -204,7 +208,10 @@ func SendToSocket(socketId int, message []byte) error {
 	socketObj.lock.Lock() // gorilla socket panics from concurrent writes to a single socket
 	err := socketObj.connection.WriteMessage(1, message)
 	socketObj.lock.Unlock()
-
 	sent = true
-	return err
+
+	if err != nil {
+		return fmt.Errorf("Failed to write message to socket %v, err: %w", socketId, err)
+	}
+	return nil
 }

agent/pkg/api/socket_server_handlers.go

@@ -3,12 +3,13 @@ package api
 import (
 	"encoding/json"
 	"fmt"
-	"mizuserver/pkg/models"
-	"mizuserver/pkg/providers"
-	"mizuserver/pkg/providers/tappers"
-	"mizuserver/pkg/up9"
 	"sync"
 
+	"github.com/up9inc/mizu/agent/pkg/models"
+	"github.com/up9inc/mizu/agent/pkg/providers"
+	"github.com/up9inc/mizu/agent/pkg/providers/tappers"
+	"github.com/up9inc/mizu/agent/pkg/up9"
+
 	tapApi "github.com/up9inc/mizu/tap/api"
 
 	"github.com/up9inc/mizu/shared"
@@ -54,9 +55,8 @@ func (h *RoutesEventHandlers) WebSocketDisconnect(socketId int, isTapper bool) {
 func BroadcastToBrowserClients(message []byte) {
 	for _, socketId := range browserClientSocketUUIDs {
 		go func(socketId int) {
-			err := SendToSocket(socketId, message)
-			if err != nil {
-				logger.Log.Errorf("error sending message to socket ID %d: %v", socketId, err)
+			if err := SendToSocket(socketId, message); err != nil {
+				logger.Log.Error(err)
 			}
 		}(socketId)
 	}

agent/pkg/controllers/config_controller.go

@@ -2,21 +2,22 @@ package controllers
 
 import (
 	"context"
+	"net/http"
+	"regexp"
+	"time"
+
 	"github.com/gin-gonic/gin"
+	"github.com/up9inc/mizu/agent/pkg/config"
+	"github.com/up9inc/mizu/agent/pkg/models"
+	"github.com/up9inc/mizu/agent/pkg/providers"
+	"github.com/up9inc/mizu/agent/pkg/providers/tapConfig"
+	"github.com/up9inc/mizu/agent/pkg/providers/tappedPods"
+	"github.com/up9inc/mizu/agent/pkg/providers/tappers"
 	"github.com/up9inc/mizu/shared"
 	"github.com/up9inc/mizu/shared/kubernetes"
 	"github.com/up9inc/mizu/shared/logger"
 	tapApi "github.com/up9inc/mizu/tap/api"
 	v1 "k8s.io/api/core/v1"
-	"mizuserver/pkg/config"
-	"mizuserver/pkg/models"
-	"mizuserver/pkg/providers"
-	"mizuserver/pkg/providers/tapConfig"
-	"mizuserver/pkg/providers/tappedPods"
-	"mizuserver/pkg/providers/tappers"
-	"net/http"
-	"regexp"
-	"time"
 )
 
 var cancelTapperSyncer context.CancelFunc

agent/pkg/controllers/entries_controller.go

@@ -2,13 +2,14 @@ package controllers
 
 import (
 	"encoding/json"
-	"mizuserver/pkg/har"
-	"mizuserver/pkg/models"
-	"mizuserver/pkg/validation"
 	"net/http"
 	"strconv"
 	"time"
 
+	"github.com/up9inc/mizu/agent/pkg/har"
+	"github.com/up9inc/mizu/agent/pkg/models"
+	"github.com/up9inc/mizu/agent/pkg/validation"
+
 	"github.com/gin-gonic/gin"
 
 	basenine "github.com/up9inc/basenine/client/go"
@@ -26,7 +27,7 @@ func InitExtensionsMap(ref map[string]*tapApi.Extension) {
 func Error(c *gin.Context, err error) bool {
 	if err != nil {
 		logger.Log.Errorf("Error getting entry: %v", err)
-		c.Error(err)
+		_ = c.Error(err)
 		c.AbortWithStatusJSON(http.StatusInternalServerError, gin.H{
 			"error":     true,
 			"type":      "error",
@@ -131,7 +132,9 @@ func GetEntry(c *gin.Context) {
 		_, rulesMatched, _isRulesEnabled := models.RunValidationRulesState(*harEntry, entry.Destination.Name)
 		isRulesEnabled = _isRulesEnabled
 		inrec, _ := json.Marshal(rulesMatched)
-		json.Unmarshal(inrec, &rules)
+		if err := json.Unmarshal(inrec, &rules); err != nil {
+			logger.Log.Error(err)
+		}
 	}
 
 	c.JSON(http.StatusOK, tapApi.EntryWrapper{

agent/pkg/controllers/install_controller.go

@@ -1,9 +1,10 @@
 package controllers
 
 import (
-	"mizuserver/pkg/providers"
 	"net/http"
 
+	"github.com/up9inc/mizu/agent/pkg/providers"
+
 	"github.com/gin-gonic/gin"
 	"github.com/up9inc/mizu/shared/logger"
 )

agent/pkg/controllers/metadata_controller.go

@@ -1,10 +1,11 @@
 package controllers
 
 import (
-	"github.com/gin-gonic/gin"
-	"github.com/up9inc/mizu/shared"
-	"mizuserver/pkg/version"
 	"net/http"
+
+	"github.com/gin-gonic/gin"
+	"github.com/up9inc/mizu/agent/pkg/version"
+	"github.com/up9inc/mizu/shared"
 )
 
 func GetVersion(c *gin.Context) {

agent/pkg/controllers/oas_controller.go

@@ -1,11 +1,12 @@
 package controllers
 
 import (
+	"net/http"
+
 	"github.com/chanced/openapi"
 	"github.com/gin-gonic/gin"
+	"github.com/up9inc/mizu/agent/pkg/oas"
 	"github.com/up9inc/mizu/shared/logger"
-	"mizuserver/pkg/oas"
-	"net/http"
 )
 
 func GetOASServers(c *gin.Context) {

agent/pkg/controllers/oas_controller_test.go

@@ -1,10 +1,12 @@
 package controllers
 
 import (
-	"github.com/gin-gonic/gin"
-	"mizuserver/pkg/oas"
 	"net/http/httptest"
 	"testing"
+
+	"github.com/up9inc/mizu/agent/pkg/oas"
+
+	"github.com/gin-gonic/gin"
 )
 
 func TestGetOASServers(t *testing.T) {
@@ -15,7 +17,6 @@ func TestGetOASServers(t *testing.T) {
 
 	GetOASServers(c)
 	t.Logf("Written body: %s", recorder.Body.String())
-	return
 }
 
 func TestGetOASAllSpecs(t *testing.T) {
@@ -26,7 +27,6 @@ func TestGetOASAllSpecs(t *testing.T) {
 
 	GetOASAllSpecs(c)
 	t.Logf("Written body: %s", recorder.Body.String())
-	return
 }
 
 func TestGetOASSpec(t *testing.T) {
@@ -39,5 +39,4 @@ func TestGetOASSpec(t *testing.T) {
 
 	GetOASSpec(c)
 	t.Logf("Written body: %s", recorder.Body.String())
-	return
 }

agent/pkg/controllers/service_map_controller.go

@@ -1,9 +1,10 @@
 package controllers
 
 import (
-	"mizuserver/pkg/servicemap"
 	"net/http"
 
+	"github.com/up9inc/mizu/agent/pkg/servicemap"
+
 	"github.com/gin-gonic/gin"
 )
 

agent/pkg/controllers/service_map_controller_test.go

@@ -7,7 +7,7 @@ import (
 	"net/http/httptest"
 	"testing"
 
-	"mizuserver/pkg/servicemap"
+	"github.com/up9inc/mizu/agent/pkg/servicemap"
 
 	"github.com/gin-gonic/gin"
 	"github.com/stretchr/testify/suite"

agent/pkg/controllers/status_controller.go

@@ -2,17 +2,18 @@ package controllers
 
 import (
 	"encoding/json"
+	"net/http"
+
 	"github.com/gin-gonic/gin"
+	"github.com/up9inc/mizu/agent/pkg/api"
+	"github.com/up9inc/mizu/agent/pkg/holder"
+	"github.com/up9inc/mizu/agent/pkg/providers"
+	"github.com/up9inc/mizu/agent/pkg/providers/tappedPods"
+	"github.com/up9inc/mizu/agent/pkg/providers/tappers"
+	"github.com/up9inc/mizu/agent/pkg/up9"
+	"github.com/up9inc/mizu/agent/pkg/validation"
 	"github.com/up9inc/mizu/shared"
 	"github.com/up9inc/mizu/shared/logger"
-	"mizuserver/pkg/api"
-	"mizuserver/pkg/holder"
-	"mizuserver/pkg/providers"
-	"mizuserver/pkg/providers/tappedPods"
-	"mizuserver/pkg/providers/tappers"
-	"mizuserver/pkg/up9"
-	"mizuserver/pkg/validation"
-	"net/http"
 )
 
 func HealthCheck(c *gin.Context) {

agent/pkg/controllers/user_controller.go

@@ -1,7 +1,7 @@
 package controllers
 
 import (
-	"mizuserver/pkg/providers"
+	"github.com/up9inc/mizu/agent/pkg/providers"
 
 	"github.com/gin-gonic/gin"
 	"github.com/up9inc/mizu/shared/logger"

agent/pkg/elastic/esClient.go

@@ -0,0 +1,120 @@
+package elastic
+
+import (
+	"bytes"
+	"crypto/tls"
+	"encoding/json"
+	"github.com/elastic/go-elasticsearch/v7"
+	"github.com/up9inc/mizu/shared"
+	"github.com/up9inc/mizu/shared/logger"
+	"github.com/up9inc/mizu/tap/api"
+	"net/http"
+	"sync"
+	"time"
+)
+
+type client struct {
+	es            *elasticsearch.Client
+	index         string
+	insertedCount int
+}
+
+var instance *client
+var once sync.Once
+
+func GetInstance() *client {
+	once.Do(func() {
+		instance = newClient()
+	})
+	return instance
+}
+
+func (client *client) Configure(config shared.ElasticConfig) {
+	if config.Url == "" || config.User == "" || config.Password == "" {
+		logger.Log.Infof("No elastic configuration was supplied, elastic exporter disabled")
+		return
+	}
+	transport := http.DefaultTransport
+	tlsClientConfig := &tls.Config{InsecureSkipVerify: true}
+	transport.(*http.Transport).TLSClientConfig = tlsClientConfig
+	cfg := elasticsearch.Config{
+		Addresses: []string{config.Url},
+		Username:  config.User,
+		Password:  config.Password,
+		Transport: transport,
+	}
+
+	es, err := elasticsearch.NewClient(cfg)
+	if err != nil {
+		logger.Log.Fatalf("Failed to initialize elastic client %v", err)
+	}
+
+	// Have the client instance return a response
+	res, err := es.Info()
+	if err != nil {
+		logger.Log.Fatalf("Elastic client.Info() ERROR: %v", err)
+	} else {
+		client.es = es
+		client.index = "mizu_traffic_http_" + time.Now().Format("2006_01_02_15_04")
+		client.insertedCount = 0
+		logger.Log.Infof("Elastic client configured, index: %s, cluster info: %v", client.index, res)
+	}
+	defer res.Body.Close()
+}
+
+func newClient() *client {
+	return &client{
+		es:    nil,
+		index: "",
+	}
+}
+
+type httpEntry struct {
+	Source      *api.TCP               `json:"src"`
+	Destination *api.TCP               `json:"dst"`
+	Outgoing    bool                   `json:"outgoing"`
+	CreatedAt   time.Time              `json:"createdAt"`
+	Request     map[string]interface{} `json:"request"`
+	Response    map[string]interface{} `json:"response"`
+	Summary     string                 `json:"summary"`
+	Method      string                 `json:"method"`
+	Status      int                    `json:"status"`
+	ElapsedTime int64                  `json:"elapsedTime"`
+	Path        string                 `json:"path"`
+}
+
+func (client *client) PushEntry(entry *api.Entry) {
+	if client.es == nil {
+		return
+	}
+
+	if entry.Protocol.Name != "http" {
+		return
+	}
+
+	entryToPush := httpEntry{
+		Source:      entry.Source,
+		Destination: entry.Destination,
+		Outgoing:    entry.Outgoing,
+		CreatedAt:   entry.StartTime,
+		Request:     entry.Request,
+		Response:    entry.Response,
+		Summary:     entry.Summary,
+		Method:      entry.Method,
+		Status:      entry.Status,
+		ElapsedTime: entry.ElapsedTime,
+		Path:        entry.Path,
+	}
+
+	entryJson, err := json.Marshal(entryToPush)
+	if err != nil {
+		logger.Log.Errorf("json.Marshal ERROR: %v", err)
+		return
+	}
+	var buffer bytes.Buffer
+	buffer.WriteString(string(entryJson))
+	res, _ := client.es.Index(client.index, &buffer)
+	if res.StatusCode == 201 {
+		client.insertedCount += 1
+	}
+}

agent/pkg/har/utils.go

@@ -123,11 +123,11 @@ func NewRequest(request map[string]interface{}) (harRequest *Request, err error)
 	cookies := make([]Cookie, 0) // BuildCookies(request["_cookies"].([]interface{}))
 
 	postData, _ := request["postData"].(map[string]interface{})
-	mimeType, _ := postData["mimeType"]
+	mimeType := postData["mimeType"]
 	if mimeType == nil || len(mimeType.(string)) == 0 {
 		mimeType = "text/html"
 	}
-	text, _ := postData["text"]
+	text := postData["text"]
 	postDataText := ""
 	if text != nil {
 		postDataText = text.(string)
@@ -176,12 +176,12 @@ func NewResponse(response map[string]interface{}) (harResponse *Response, err er
 	cookies := make([]Cookie, 0) // BuildCookies(response["_cookies"].([]interface{}))
 
 	content, _ := response["content"].(map[string]interface{})
-	mimeType, _ := content["mimeType"]
+	mimeType := content["mimeType"]
 	if mimeType == nil || len(mimeType.(string)) == 0 {
 		mimeType = "text/html"
 	}
-	encoding, _ := content["encoding"]
-	text, _ := content["text"]
+	encoding := content["encoding"]
+	text := content["text"]
 	bodyText := ""
 	if text != nil {
 		bodyText = text.(string)

agent/pkg/holder/main.go

@@ -1,6 +1,6 @@
 package holder
 
-import "mizuserver/pkg/resolver"
+import "github.com/up9inc/mizu/agent/pkg/resolver"
 
 var k8sResolver *resolver.Resolver
 
@@ -11,4 +11,3 @@ func SetResolver(param *resolver.Resolver) {
 func GetResolver() *resolver.Resolver {
 	return k8sResolver
 }
-

agent/pkg/middlewares/auth.go

@@ -1,8 +1,8 @@
 package middlewares
 
 import (
-	"mizuserver/pkg/config"
-	"mizuserver/pkg/providers"
+	"github.com/up9inc/mizu/agent/pkg/config"
+	"github.com/up9inc/mizu/agent/pkg/providers"
 
 	"github.com/gin-gonic/gin"
 	ory "github.com/ory/kratos-client-go"

agent/pkg/models/models.go

@@ -2,9 +2,10 @@ package models
 
 import (
 	"encoding/json"
+
+	"github.com/up9inc/mizu/agent/pkg/har"
+	"github.com/up9inc/mizu/agent/pkg/rules"
 	tapApi "github.com/up9inc/mizu/tap/api"
-	"mizuserver/pkg/har"
-	"mizuserver/pkg/rules"
 
 	basenine "github.com/up9inc/basenine/client/go"
 	"github.com/up9inc/mizu/shared"

agent/pkg/oas/feeder_test.go

@@ -4,19 +4,21 @@ import (
 	"bufio"
 	"encoding/json"
 	"errors"
-	"github.com/up9inc/mizu/shared/logger"
 	"io"
 	"io/ioutil"
-	"mizuserver/pkg/har"
 	"os"
 	"path/filepath"
 	"sort"
 	"strings"
 	"testing"
+
+	"github.com/up9inc/mizu/agent/pkg/har"
+
+	"github.com/up9inc/mizu/shared/logger"
 )
 
 func getFiles(baseDir string) (result []string, err error) {
-	result = make([]string, 0, 0)
+	result = make([]string, 0)
 	logger.Log.Infof("Reading files from tree: %s", baseDir)
 
 	// https://yourbasic.org/golang/list-files-in-directory/
@@ -121,8 +123,6 @@ func feedEntry(entry *har.Entry, isSync bool) {
 
 	if strings.Contains(entry.Request.URL, "taboola") {
 		logger.Log.Debugf("Interesting: %s", entry.Request.URL)
-	} else {
-		//return
 	}
 
 	if isSync {

agent/pkg/oas/gibberish.go

@@ -33,11 +33,7 @@ func IsGibberish(str string) bool {
 	}
 
 	noise := noiseLevel(str)
-	if noise >= 0.2 {
-		return true
-	}
-
-	return false
+	return noise >= 0.2
 }
 
 func noiseLevel(str string) (score float64) {

agent/pkg/oas/oas_generator.go

@@ -3,10 +3,11 @@ package oas
 import (
 	"context"
 	"encoding/json"
-	"github.com/up9inc/mizu/shared/logger"
-	"mizuserver/pkg/har"
 	"net/url"
 	"sync"
+
+	"github.com/up9inc/mizu/agent/pkg/har"
+	"github.com/up9inc/mizu/shared/logger"
 )
 
 var (

agent/pkg/oas/specgen.go

@@ -3,15 +3,17 @@ package oas
 import (
 	"encoding/json"
 	"errors"
-	"github.com/chanced/openapi"
-	"github.com/google/uuid"
-	"github.com/up9inc/mizu/shared/logger"
 	"mime"
-	"mizuserver/pkg/har"
 	"net/url"
 	"strconv"
 	"strings"
 	"sync"
+
+	"github.com/up9inc/mizu/agent/pkg/har"
+
+	"github.com/chanced/openapi"
+	"github.com/google/uuid"
+	"github.com/up9inc/mizu/shared/logger"
 )
 
 type reqResp struct { // hello, generics in Go
@@ -272,7 +274,7 @@ func handleRequest(req *har.Request, opObj *openapi.Operation, isSuccess bool) e
 
 		if reqBody != nil {
 			reqCtype := getReqCtype(req)
-			reqMedia, err := fillContent(reqResp{Req: req}, reqBody.Content, reqCtype, err)
+			reqMedia, err := fillContent(reqResp{Req: req}, reqBody.Content, reqCtype)
 			if err != nil {
 				return err
 			}
@@ -294,7 +296,7 @@ func handleResponse(resp *har.Response, opObj *openapi.Operation, isSuccess bool
 
 	respCtype := getRespCtype(resp)
 	respContent := respObj.Content
-	respMedia, err := fillContent(reqResp{Resp: resp}, respContent, respCtype, err)
+	respMedia, err := fillContent(reqResp{Resp: resp}, respContent, respCtype)
 	if err != nil {
 		return err
 	}
@@ -342,11 +344,9 @@ func handleRespHeaders(reqHeaders []har.Header, respObj *openapi.ResponseObj) {
 			}
 		}
 	}
-
-	return
 }
 
-func fillContent(reqResp reqResp, respContent openapi.Content, ctype string, err error) (*openapi.MediaType, error) {
+func fillContent(reqResp reqResp, respContent openapi.Content, ctype string) (*openapi.MediaType, error) {
 	content, found := respContent[ctype]
 	if !found {
 		respContent[ctype] = &openapi.MediaType{}
@@ -367,14 +367,16 @@ func fillContent(reqResp reqResp, respContent openapi.Content, ctype string, err
 		any, isJSON := anyJSON(text)
 		if isJSON {
 			// re-marshal with forced indent
-			exampleMsg, err = json.MarshalIndent(any, "", "\t")
-			if err != nil {
+			if msg, err := json.MarshalIndent(any, "", "\t"); err != nil {
 				panic("Failed to re-marshal value, super-strange")
+			} else {
+				exampleMsg = msg
 			}
 		} else {
-			exampleMsg, err = json.Marshal(text)
-			if err != nil {
+			if msg, err := json.Marshal(text); err != nil {
 				return nil, err
+			} else {
+				exampleMsg = msg
 			}
 		}
 

agent/pkg/oas/specgen_test.go

@@ -2,15 +2,16 @@ package oas
 
 import (
 	"encoding/json"
-	"github.com/chanced/openapi"
-	"github.com/op/go-logging"
-	"github.com/up9inc/mizu/shared/logger"
 	"io/ioutil"
-	"mizuserver/pkg/har"
 	"os"
 	"strings"
 	"testing"
 	"time"
+
+	"github.com/chanced/openapi"
+	"github.com/op/go-logging"
+	"github.com/up9inc/mizu/agent/pkg/har"
+	"github.com/up9inc/mizu/shared/logger"
 )
 
 // if started via env, write file into subdir
@@ -171,8 +172,6 @@ func loadStartingOAS() {
 	gen.StartFromSpec(doc)
 
 	GetOasGeneratorInstance().ServiceSpecs.Store("catalogue", gen)
-
-	return
 }
 
 func TestEntriesNegative(t *testing.T) {
@@ -224,5 +223,4 @@ func TestLoadValid3_1(t *testing.T) {
 		t.Log(err)
 		t.FailNow()
 	}
-	return
 }

agent/pkg/oas/tree.go

@@ -1,11 +1,12 @@
 package oas
 
 import (
-	"github.com/chanced/openapi"
-	"github.com/up9inc/mizu/shared/logger"
 	"net/url"
 	"strconv"
 	"strings"
+
+	"github.com/chanced/openapi"
+	"github.com/up9inc/mizu/shared/logger"
 )
 
 type NodePath = []string
@@ -162,9 +163,7 @@ func (n *Node) listPaths() *openapi.Paths {
 		strChunk = *n.constant
 	} else if n.pathParam != nil {
 		strChunk = "{" + n.pathParam.Name + "}"
-	} else {
-		// this is the root node
-	}
+	} // else -> this is the root node
 
 	// add self
 	if n.pathObj != nil {

agent/pkg/oas/utils.go

@@ -3,11 +3,13 @@ package oas
 import (
 	"encoding/json"
 	"errors"
-	"github.com/chanced/openapi"
-	"github.com/up9inc/mizu/shared/logger"
-	"mizuserver/pkg/har"
 	"strconv"
 	"strings"
+
+	"github.com/up9inc/mizu/agent/pkg/har"
+
+	"github.com/chanced/openapi"
+	"github.com/up9inc/mizu/shared/logger"
 )
 
 func exampleResolver(ref string) (*openapi.ExampleObj, error) {
@@ -32,16 +34,14 @@ func headerResolver(ref string) (*openapi.HeaderObj, error) {
 
 func initParams(obj **openapi.ParameterList) {
 	if *obj == nil {
-		var params openapi.ParameterList
-		params = make([]openapi.Parameter, 0)
+		var params openapi.ParameterList = make([]openapi.Parameter, 0)
 		*obj = &params
 	}
 }
 
 func initHeaders(respObj *openapi.ResponseObj) {
 	if respObj.Headers == nil {
-		var created openapi.Headers
-		created = map[string]openapi.Header{}
+		var created openapi.Headers = map[string]openapi.Header{}
 		respObj.Headers = created
 	}
 }
@@ -85,7 +85,7 @@ func findParamByName(params *openapi.ParameterList, in openapi.In, name string)
 			continue
 		}
 
-		if paramObj.Name == name || (caseInsensitive && strings.ToLower(paramObj.Name) == strings.ToLower(name)) {
+		if paramObj.Name == name || (caseInsensitive && strings.EqualFold(paramObj.Name, name)) {
 			pathParam = paramObj
 			break
 		}
@@ -102,7 +102,7 @@ func findHeaderByName(headers *openapi.Headers, name string) *openapi.HeaderObj
 			continue
 		}
 
-		if strings.ToLower(hname) == strings.ToLower(name) {
+		if strings.EqualFold(hname, name) {
 			return hdrObj
 		}
 	}

agent/pkg/providers/install_provider.go

@@ -3,7 +3,10 @@ package providers
 import (
 	"context"
 	"errors"
-	"mizuserver/pkg/config"
+
+	"github.com/up9inc/mizu/agent/pkg/config"
+
+	"github.com/up9inc/mizu/shared/logger"
 
 	ory "github.com/ory/kratos-client-go"
 )
@@ -38,7 +41,9 @@ func CreateAdminUser(password string, ctx context.Context) (token *string, err e
 
 	if err != nil {
 		//Delete the user to prevent a half-setup situation where admin user is created without admin privileges
-		DeleteUser(identityId, ctx)
+		if err := DeleteUser(identityId, ctx); err != nil {
+			logger.Log.Error(err)
+		}
 
 		return nil, err, nil
 	}

agent/pkg/providers/stats_provider_test.go

@@ -2,8 +2,9 @@ package providers_test
 
 import (
 	"fmt"
-	"mizuserver/pkg/providers"
 	"testing"
+
+	"github.com/up9inc/mizu/agent/pkg/providers"
 )
 
 func TestNoEntryAddedCount(t *testing.T) {

agent/pkg/providers/status_provider.go

@@ -3,12 +3,13 @@ package providers
 import (
 	"encoding/json"
 	"fmt"
-	"github.com/patrickmn/go-cache"
-	"github.com/up9inc/mizu/shared"
-	"github.com/up9inc/mizu/tap"
-	"mizuserver/pkg/models"
 	"os"
 	"time"
+
+	"github.com/patrickmn/go-cache"
+	"github.com/up9inc/mizu/agent/pkg/models"
+	"github.com/up9inc/mizu/shared"
+	"github.com/up9inc/mizu/tap"
 )
 
 const tlsLinkRetainmentTime = time.Minute * 15

agent/pkg/providers/tapConfig/tap_config_provider.go

@@ -1,12 +1,13 @@
 package tapConfig
 
 import (
-	"github.com/up9inc/mizu/shared"
-	"github.com/up9inc/mizu/shared/logger"
-	"mizuserver/pkg/models"
-	"mizuserver/pkg/utils"
 	"os"
 	"sync"
+
+	"github.com/up9inc/mizu/agent/pkg/models"
+	"github.com/up9inc/mizu/agent/pkg/utils"
+	"github.com/up9inc/mizu/shared"
+	"github.com/up9inc/mizu/shared/logger"
 )
 
 const FilePath = shared.DataDirPath + "tap-config.json"

agent/pkg/providers/tappedPods/tapped_pods_provider.go

@@ -1,13 +1,14 @@
 package tappedPods
 
 import (
-	"github.com/up9inc/mizu/shared"
-	"github.com/up9inc/mizu/shared/logger"
-	"mizuserver/pkg/providers/tappers"
-	"mizuserver/pkg/utils"
 	"os"
 	"strings"
 	"sync"
+
+	"github.com/up9inc/mizu/agent/pkg/providers/tappers"
+	"github.com/up9inc/mizu/agent/pkg/utils"
+	"github.com/up9inc/mizu/shared"
+	"github.com/up9inc/mizu/shared/logger"
 )
 
 const FilePath = shared.DataDirPath + "tapped-pods.json"

agent/pkg/providers/tappers/tappers_provider.go

@@ -1,11 +1,12 @@
 package tappers
 
 import (
-	"github.com/up9inc/mizu/shared"
-	"github.com/up9inc/mizu/shared/logger"
-	"mizuserver/pkg/utils"
 	"os"
 	"sync"
+
+	"github.com/up9inc/mizu/agent/pkg/utils"
+	"github.com/up9inc/mizu/shared"
+	"github.com/up9inc/mizu/shared/logger"
 )
 
 const FilePath = shared.DataDirPath + "tappers-status.json"

agent/pkg/providers/user_provider.go

@@ -85,11 +85,11 @@ func DeleteUser(identityId string, ctx context.Context) error {
 		return err
 	}
 	if result == nil {
-		return errors.New("unknown error occured during user deletion")
+		return fmt.Errorf("unknown error occured during user deletion %v", identityId)
 	}
 
 	if result.StatusCode < 200 || result.StatusCode > 299 {
-		return errors.New(fmt.Sprintf("user deletion returned bad status %d", result.StatusCode))
+		return fmt.Errorf("user deletion %v returned bad status %d", identityId, result.StatusCode)
 	} else {
 		return nil
 	}

agent/pkg/providers/user_role_provider.go

@@ -15,7 +15,8 @@ more on keto here: https://www.ory.sh/keto/docs/
 
 import (
 	"fmt"
-	"mizuserver/pkg/utils"
+
+	"github.com/up9inc/mizu/agent/pkg/utils"
 
 	ketoClient "github.com/ory/keto-client-go/client"
 	ketoRead "github.com/ory/keto-client-go/client/read"

agent/pkg/routes/config_routes.go

@@ -1,8 +1,8 @@
 package routes
 
 import (
-	"mizuserver/pkg/controllers"
-	"mizuserver/pkg/middlewares"
+	"github.com/up9inc/mizu/agent/pkg/controllers"
+	"github.com/up9inc/mizu/agent/pkg/middlewares"
 
 	"github.com/gin-gonic/gin"
 )

agent/pkg/routes/entries_routes.go

@@ -1,8 +1,8 @@
 package routes
 
 import (
-	"mizuserver/pkg/controllers"
-	"mizuserver/pkg/middlewares"
+	"github.com/up9inc/mizu/agent/pkg/controllers"
+	"github.com/up9inc/mizu/agent/pkg/middlewares"
 
 	"github.com/gin-gonic/gin"
 )

agent/pkg/routes/install_routes.go

@@ -1,7 +1,7 @@
 package routes
 
 import (
-	"mizuserver/pkg/controllers"
+	"github.com/up9inc/mizu/agent/pkg/controllers"
 
 	"github.com/gin-gonic/gin"
 )

agent/pkg/routes/metadata_routes.go

@@ -1,7 +1,7 @@
 package routes
 
 import (
-	"mizuserver/pkg/controllers"
+	"github.com/up9inc/mizu/agent/pkg/controllers"
 
 	"github.com/gin-gonic/gin"
 )

agent/pkg/routes/not_found_route.go

@@ -1,18 +0,0 @@
-package routes
-
-import (
-	"github.com/gin-gonic/gin"
-	"net/http"
-)
-
-// NotFoundRoute defines the 404 Error route.
-func NotFoundRoute(app *gin.Engine) {
-	app.Use(
-		func(c *gin.Context) {
-			c.JSON(http.StatusNotFound, map[string]interface{}{
-				"error": true,
-				"msg":   "sorry, endpoint is not found",
-			})
-		},
-	)
-}

agent/pkg/routes/oas_routes.go

@@ -1,8 +1,8 @@
 package routes
 
 import (
-	"mizuserver/pkg/controllers"
-	"mizuserver/pkg/middlewares"
+	"github.com/up9inc/mizu/agent/pkg/controllers"
+	"github.com/up9inc/mizu/agent/pkg/middlewares"
 
 	"github.com/gin-gonic/gin"
 )

agent/pkg/routes/query_routes.go

@@ -1,8 +1,8 @@
 package routes
 
 import (
-	"mizuserver/pkg/controllers"
-	"mizuserver/pkg/middlewares"
+	"github.com/up9inc/mizu/agent/pkg/controllers"
+	"github.com/up9inc/mizu/agent/pkg/middlewares"
 
 	"github.com/gin-gonic/gin"
 )

agent/pkg/routes/service_map_routes.go

@@ -1,8 +1,8 @@
 package routes
 
 import (
-	"mizuserver/pkg/controllers"
-	"mizuserver/pkg/middlewares"
+	"github.com/up9inc/mizu/agent/pkg/controllers"
+	"github.com/up9inc/mizu/agent/pkg/middlewares"
 
 	"github.com/gin-gonic/gin"
 )

agent/pkg/routes/status_routes.go

@@ -1,8 +1,8 @@
 package routes
 
 import (
-	"mizuserver/pkg/controllers"
-	"mizuserver/pkg/middlewares"
+	"github.com/up9inc/mizu/agent/pkg/controllers"
+	"github.com/up9inc/mizu/agent/pkg/middlewares"
 
 	"github.com/gin-gonic/gin"
 )

agent/pkg/routes/user_routes.go

@@ -1,7 +1,7 @@
 package routes
 
 import (
-	"mizuserver/pkg/controllers"
+	"github.com/up9inc/mizu/agent/pkg/controllers"
 
 	"github.com/gin-gonic/gin"
 )

agent/pkg/rules/rulesHTTP.go

@@ -4,11 +4,12 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
-	"mizuserver/pkg/har"
 	"reflect"
 	"regexp"
 	"strings"
 
+	"github.com/up9inc/mizu/agent/pkg/har"
+
 	"github.com/up9inc/mizu/shared/logger"
 
 	"github.com/up9inc/mizu/shared"
@@ -108,7 +109,7 @@ func PassedValidationRules(rulesMatched []RulesMatched) (bool, int64, int) {
 	}
 
 	for _, rule := range rulesMatched {
-		if rule.Matched == false {
+		if !rule.Matched {
 			return false, responseTime, numberOfRulesMatched
 		} else {
 			if strings.ToLower(rule.Rule.Type) == "slo" {

agent/pkg/up9/main.go

@@ -6,8 +6,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
-	"mizuserver/pkg/har"
-	"mizuserver/pkg/utils"
 	"net/http"
 	"net/url"
 	"regexp"
@@ -15,6 +13,9 @@ import (
 	"sync"
 	"time"
 
+	"github.com/up9inc/mizu/agent/pkg/har"
+	"github.com/up9inc/mizu/agent/pkg/utils"
+
 	basenine "github.com/up9inc/basenine/client/go"
 	"github.com/up9inc/mizu/shared"
 	"github.com/up9inc/mizu/shared/logger"

agent/pkg/utils/utils.go

@@ -31,10 +31,14 @@ func StartServer(app *gin.Engine) {
 	}
 
 	go func() {
-		_ = <-signals
+		<-signals
 		logger.Log.Infof("Shutting down...")
-		ctx, _ := context.WithTimeout(context.Background(), 5*time.Second)
-		_ = srv.Shutdown(ctx)
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		err := srv.Shutdown(ctx)
+		if err != nil {
+			logger.Log.Errorf("%v", err)
+		}
 		os.Exit(0)
 	}()
 
@@ -91,7 +95,7 @@ func UniqueStringSlice(s []string) []string {
 	uniqueMap := map[string]bool{}
 
 	for _, val := range s {
-		if uniqueMap[val] == true {
+		if uniqueMap[val] {
 			continue
 		}
 		uniqueMap[val] = true

cli/apiserver/provider.go

@@ -23,8 +23,8 @@ type Provider struct {
 	client  *http.Client
 }
 
-const DefaultRetries = 20
-const DefaultTimeout = 5 * time.Second
+const DefaultRetries = 3
+const DefaultTimeout = 2 * time.Second
 
 func NewProvider(url string, retries int, timeout time.Duration) *Provider {
 	return &Provider{
@@ -36,16 +36,6 @@ func NewProvider(url string, retries int, timeout time.Duration) *Provider {
 	}
 }
 
-func NewProviderWithoutRetries(url string, timeout time.Duration) *Provider {
-	return &Provider{
-		url:     url,
-		retries: 1,
-		client: &http.Client{
-			Timeout: timeout,
-		},
-	}
-}
-
 func (provider *Provider) TestConnection() error {
 	retriesLeft := provider.retries
 	for retriesLeft > 0 {

cli/cmd/checkRunner.go

@@ -3,13 +3,14 @@ package cmd
 import (
 	"context"
 	"fmt"
+	"regexp"
+
 	"github.com/up9inc/mizu/cli/apiserver"
 	"github.com/up9inc/mizu/cli/config"
 	"github.com/up9inc/mizu/cli/uiUtils"
 	"github.com/up9inc/mizu/shared/kubernetes"
 	"github.com/up9inc/mizu/shared/logger"
 	"github.com/up9inc/mizu/shared/semver"
-	"regexp"
 )
 
 func runMizuCheck() {
@@ -96,7 +97,7 @@ func checkServerConnection(kubernetesProvider *kubernetes.Provider) bool {
 
 	serverUrl := GetApiServerUrl()
 
-	apiServerProvider := apiserver.NewProviderWithoutRetries(serverUrl, apiserver.DefaultTimeout)
+	apiServerProvider := apiserver.NewProvider(serverUrl, 1, apiserver.DefaultTimeout)
 	if err := apiServerProvider.TestConnection(); err == nil {
 		logger.Log.Infof("%v found Mizu server tunnel available and connected successfully to API server", fmt.Sprintf(uiUtils.Green, "√"))
 		return true
@@ -169,32 +170,32 @@ func checkAllResourcesExist(ctx context.Context, kubernetesProvider *kubernetes.
 	allResourcesExist := checkResourceExist(config.Config.MizuResourcesNamespace, "namespace", exist, err)
 
 	exist, err = kubernetesProvider.DoesConfigMapExist(ctx, config.Config.MizuResourcesNamespace, kubernetes.ConfigMapName)
-	allResourcesExist = checkResourceExist(kubernetes.ConfigMapName, "config map", exist, err)
+	allResourcesExist = checkResourceExist(kubernetes.ConfigMapName, "config map", exist, err) && allResourcesExist
 
 	exist, err = kubernetesProvider.DoesServiceAccountExist(ctx, config.Config.MizuResourcesNamespace, kubernetes.ServiceAccountName)
-	allResourcesExist = checkResourceExist(kubernetes.ServiceAccountName, "service account", exist, err)
+	allResourcesExist = checkResourceExist(kubernetes.ServiceAccountName, "service account", exist, err) && allResourcesExist
 
 	if config.Config.IsNsRestrictedMode() {
 		exist, err = kubernetesProvider.DoesRoleExist(ctx, config.Config.MizuResourcesNamespace, kubernetes.RoleName)
-		allResourcesExist = checkResourceExist(kubernetes.RoleName, "role", exist, err)
+		allResourcesExist = checkResourceExist(kubernetes.RoleName, "role", exist, err) && allResourcesExist
 
 		exist, err = kubernetesProvider.DoesRoleBindingExist(ctx, config.Config.MizuResourcesNamespace, kubernetes.RoleBindingName)
-		allResourcesExist = checkResourceExist(kubernetes.RoleBindingName, "role binding", exist, err)
+		allResourcesExist = checkResourceExist(kubernetes.RoleBindingName, "role binding", exist, err) && allResourcesExist
 	} else {
 		exist, err = kubernetesProvider.DoesClusterRoleExist(ctx, kubernetes.ClusterRoleName)
-		allResourcesExist = checkResourceExist(kubernetes.ClusterRoleName, "cluster role", exist, err)
+		allResourcesExist = checkResourceExist(kubernetes.ClusterRoleName, "cluster role", exist, err) && allResourcesExist
 
 		exist, err = kubernetesProvider.DoesClusterRoleBindingExist(ctx, kubernetes.ClusterRoleBindingName)
-		allResourcesExist = checkResourceExist(kubernetes.ClusterRoleBindingName, "cluster role binding", exist, err)
+		allResourcesExist = checkResourceExist(kubernetes.ClusterRoleBindingName, "cluster role binding", exist, err) && allResourcesExist
 	}
 
 	exist, err = kubernetesProvider.DoesServiceExist(ctx, config.Config.MizuResourcesNamespace, kubernetes.ApiServerPodName)
-	allResourcesExist = checkResourceExist(kubernetes.ApiServerPodName, "service", exist, err)
+	allResourcesExist = checkResourceExist(kubernetes.ApiServerPodName, "service", exist, err) && allResourcesExist
 
 	if isInstallCommand {
-		allResourcesExist = checkInstallResourcesExist(ctx, kubernetesProvider)
+		allResourcesExist = checkInstallResourcesExist(ctx, kubernetesProvider) && allResourcesExist
 	} else {
-		allResourcesExist = checkTapResourcesExist(ctx, kubernetesProvider)
+		allResourcesExist = checkTapResourcesExist(ctx, kubernetesProvider) && allResourcesExist
 	}
 
 	return allResourcesExist
@@ -205,13 +206,13 @@ func checkInstallResourcesExist(ctx context.Context, kubernetesProvider *kuberne
 	installResourcesExist := checkResourceExist(kubernetes.DaemonRoleName, "role", exist, err)
 
 	exist, err = kubernetesProvider.DoesRoleBindingExist(ctx, config.Config.MizuResourcesNamespace, kubernetes.DaemonRoleBindingName)
-	installResourcesExist = checkResourceExist(kubernetes.DaemonRoleBindingName, "role binding", exist, err)
+	installResourcesExist = checkResourceExist(kubernetes.DaemonRoleBindingName, "role binding", exist, err) && installResourcesExist
 
 	exist, err = kubernetesProvider.DoesPersistentVolumeClaimExist(ctx, config.Config.MizuResourcesNamespace, kubernetes.PersistentVolumeClaimName)
-	installResourcesExist = checkResourceExist(kubernetes.PersistentVolumeClaimName, "persistent volume claim", exist, err)
+	installResourcesExist = checkResourceExist(kubernetes.PersistentVolumeClaimName, "persistent volume claim", exist, err) && installResourcesExist
 
 	exist, err = kubernetesProvider.DoesDeploymentExist(ctx, config.Config.MizuResourcesNamespace, kubernetes.ApiServerPodName)
-	installResourcesExist = checkResourceExist(kubernetes.ApiServerPodName, "deployment", exist, err)
+	installResourcesExist = checkResourceExist(kubernetes.ApiServerPodName, "deployment", exist, err) && installResourcesExist
 
 	return installResourcesExist
 }

cli/cmd/common.go

@@ -5,6 +5,10 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"path"
+	"regexp"
+	"time"
+
 	"github.com/up9inc/mizu/cli/apiserver"
 	"github.com/up9inc/mizu/cli/config/configStructs"
 	"github.com/up9inc/mizu/cli/errormessage"
@@ -13,9 +17,6 @@ import (
 	"github.com/up9inc/mizu/cli/resources"
 	"github.com/up9inc/mizu/cli/uiUtils"
 	"github.com/up9inc/mizu/shared"
-	"path"
-	"regexp"
-	"time"
 
 	"github.com/up9inc/mizu/cli/config"
 	"github.com/up9inc/mizu/shared/kubernetes"
@@ -35,10 +36,10 @@ func startProxyReportErrorIfAny(kubernetesProvider *kubernetes.Provider, ctx con
 		return
 	}
 
-	apiProvider = apiserver.NewProviderWithoutRetries(GetApiServerUrl(), time.Second) // short check for proxy
+	apiProvider = apiserver.NewProvider(GetApiServerUrl(), apiserver.DefaultRetries, apiserver.DefaultTimeout)
 	if err := apiProvider.TestConnection(); err != nil {
 		logger.Log.Debugf("Couldn't connect using proxy, stopping proxy and trying to create port-forward")
-		if err := httpServer.Shutdown(context.Background()); err != nil {
+		if err := httpServer.Shutdown(ctx); err != nil {
 			logger.Log.Debugf("Error occurred while stopping proxy %v", errormessage.FormatError(err))
 		}
 
@@ -50,7 +51,7 @@ func startProxyReportErrorIfAny(kubernetesProvider *kubernetes.Provider, ctx con
 			return
 		}
 
-		apiProvider = apiserver.NewProvider(GetApiServerUrl(), apiserver.DefaultRetries, apiserver.DefaultTimeout) // long check for port-forward
+		apiProvider = apiserver.NewProvider(GetApiServerUrl(), apiserver.DefaultRetries, apiserver.DefaultTimeout)
 		if err := apiProvider.TestConnection(); err != nil {
 			logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Couldn't connect to API server, for more info check logs at %s", fsUtils.GetLogFilePath()))
 			cancel()

cli/cmd/config.go

@@ -50,7 +50,9 @@ func init() {
 	rootCmd.AddCommand(configCmd)
 
 	defaultConfig := config.ConfigStruct{}
-	defaults.Set(&defaultConfig)
+	if err := defaults.Set(&defaultConfig); err != nil {
+		logger.Log.Debug(err)
+	}
 
 	configCmd.Flags().BoolP(configStructs.RegenerateConfigName, "r", defaultConfig.Config.Regenerate, fmt.Sprintf("Regenerate the config file with default values to path %s or to chosen path using --%s", defaultConfig.ConfigFilePath, config.ConfigFilePathCommandName))
 }

cli/cmd/installRunner.go

@@ -31,7 +31,9 @@ func runMizuInstall() {
 	var defaultMaxEntriesDBSizeBytes int64 = 200 * 1000 * 1000
 
 	defaultResources := shared.Resources{}
-	defaults.Set(&defaultResources)
+	if err := defaults.Set(&defaultResources); err != nil {
+		logger.Log.Debug(err)
+	}
 
 	mizuAgentConfig := getInstallMizuAgentConfig(defaultMaxEntriesDBSizeBytes, defaultResources)
 	serializedMizuConfig, err := getSerializedMizuAgentConfig(mizuAgentConfig)
@@ -74,6 +76,7 @@ func getInstallMizuAgentConfig(maxDBSizeBytes int64, tapperResources shared.Reso
 		StandaloneMode:         true,
 		ServiceMap:             config.Config.ServiceMap,
 		OAS:                    config.Config.OAS,
+		Elastic:                config.Config.Elastic,
 	}
 
 	return &mizuAgentConfig

cli/cmd/logs.go

@@ -23,7 +23,7 @@ var logsCmd = &cobra.Command{
 		if err != nil {
 			return nil
 		}
-		ctx, _ := context.WithCancel(context.Background())
+		ctx := context.Background()
 
 		if validationErr := config.Config.Logs.Validate(); validationErr != nil {
 			return errormessage.FormatError(validationErr)
@@ -43,7 +43,9 @@ func init() {
 	rootCmd.AddCommand(logsCmd)
 
 	defaultLogsConfig := configStructs.LogsConfig{}
-	defaults.Set(&defaultLogsConfig)
+	if err := defaults.Set(&defaultLogsConfig); err != nil {
+		logger.Log.Debug(err)
+	}
 
 	logsCmd.Flags().StringP(configStructs.FileLogsName, "f", defaultLogsConfig.FileStr, "Path for zip file (default current <pwd>\\mizu_logs.zip)")
 }

cli/cmd/root.go

@@ -30,7 +30,9 @@ Further info is available at https://github.com/up9inc/mizu`,
 
 func init() {
 	defaultConfig := config.ConfigStruct{}
-	defaults.Set(&defaultConfig)
+	if err := defaults.Set(&defaultConfig); err != nil {
+		logger.Log.Debug(err)
+	}
 
 	rootCmd.PersistentFlags().StringSlice(config.SetCommandName, []string{}, fmt.Sprintf("Override values using --%s", config.SetCommandName))
 	rootCmd.PersistentFlags().String(config.ConfigFilePathCommandName, defaultConfig.ConfigFilePath, fmt.Sprintf("Override config file path using --%s", config.ConfigFilePathCommandName))

cli/cmd/tap.go

@@ -104,7 +104,9 @@ func init() {
 	rootCmd.AddCommand(tapCmd)
 
 	defaultTapConfig := configStructs.TapConfig{}
-	defaults.Set(&defaultTapConfig)
+	if err := defaults.Set(&defaultTapConfig); err != nil {
+		logger.Log.Debug(err)
+	}
 
 	tapCmd.Flags().Uint16P(configStructs.GuiPortTapName, "p", defaultTapConfig.GuiPort, "Provide a custom port for the web interface webserver")
 	tapCmd.Flags().StringSliceP(configStructs.NamespacesTapName, "n", defaultTapConfig.Namespaces, "Namespaces selector")

cli/cmd/tapRunner.go

@@ -164,6 +164,7 @@ func getTapMizuAgentConfig() *shared.MizuAgentConfig {
 		AgentDatabasePath:      shared.DataDirPath,
 		ServiceMap:             config.Config.ServiceMap,
 		OAS:                    config.Config.OAS,
+		Elastic:                config.Config.Elastic,
 	}
 
 	return &mizuAgentConfig
@@ -342,7 +343,7 @@ func watchApiServerPod(ctx context.Context, kubernetesProvider *kubernetes.Provi
 
 				if modifiedPod.Status.Phase == core.PodRunning && !isPodReady {
 					isPodReady = true
-					postApiServerStarted(ctx, kubernetesProvider, cancel, err)
+					postApiServerStarted(ctx, kubernetesProvider, cancel)
 				}
 			case kubernetes.EventBookmark:
 				break
@@ -405,7 +406,7 @@ func watchApiServerEvents(ctx context.Context, kubernetesProvider *kubernetes.Pr
 			case "FailedScheduling", "Failed":
 				logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Mizu API Server status: %s - %s", event.Reason, event.Note))
 				cancel()
-				break
+				
 			}
 		case err, ok := <-errorChan:
 			if !ok {
@@ -421,11 +422,11 @@ func watchApiServerEvents(ctx context.Context, kubernetesProvider *kubernetes.Pr
 	}
 }
 
-func postApiServerStarted(ctx context.Context, kubernetesProvider *kubernetes.Provider, cancel context.CancelFunc, err error) {
+func postApiServerStarted(ctx context.Context, kubernetesProvider *kubernetes.Provider, cancel context.CancelFunc) {
 	startProxyReportErrorIfAny(kubernetesProvider, ctx, cancel)
 
 	options, _ := getMizuApiFilteringOptions()
-	if err = startTapperSyncer(ctx, cancel, kubernetesProvider, state.targetNamespaces, *options, state.startTime); err != nil {
+	if err := startTapperSyncer(ctx, cancel, kubernetesProvider, state.targetNamespaces, *options, state.startTime); err != nil {
 		logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Error starting mizu tapper syncer: %v", err))
 		cancel()
 	}

cli/cmd/version.go

@@ -36,7 +36,9 @@ func init() {
 	rootCmd.AddCommand(versionCmd)
 
 	defaultVersionConfig := configStructs.VersionConfig{}
-	defaults.Set(&defaultVersionConfig)
+	if err := defaults.Set(&defaultVersionConfig); err != nil {
+		logger.Log.Debug(err)
+	}
 
 	versionCmd.Flags().BoolP(configStructs.DebugInfoVersionName, "d", defaultVersionConfig.DebugInfo, "Provide all information about version")
 

cli/cmd/view.go

@@ -6,6 +6,7 @@ import (
 	"github.com/up9inc/mizu/cli/config"
 	"github.com/up9inc/mizu/cli/config/configStructs"
 	"github.com/up9inc/mizu/cli/telemetry"
+	"github.com/up9inc/mizu/shared/logger"
 )
 
 var viewCmd = &cobra.Command{
@@ -22,10 +23,14 @@ func init() {
 	rootCmd.AddCommand(viewCmd)
 
 	defaultViewConfig := configStructs.ViewConfig{}
-	defaults.Set(&defaultViewConfig)
+	if err := defaults.Set(&defaultViewConfig); err != nil {
+		logger.Log.Debug(err)
+	}
 
 	viewCmd.Flags().Uint16P(configStructs.GuiPortViewName, "p", defaultViewConfig.GuiPort, "Provide a custom port for the web interface webserver")
 	viewCmd.Flags().StringP(configStructs.UrlViewName, "u", defaultViewConfig.Url, "Provide a custom host")
 
-	viewCmd.Flags().MarkHidden(configStructs.UrlViewName)
+	if err := viewCmd.Flags().MarkHidden(configStructs.UrlViewName); err != nil {
+		logger.Log.Debug(err)
+	}
 }

cli/config/configStruct.go

@@ -38,8 +38,9 @@ type ConfigStruct struct {
 	ConfigFilePath         string                      `yaml:"config-path,omitempty" readonly:""`
 	HeadlessMode           bool                        `yaml:"headless" default:"false"`
 	LogLevelStr            string                      `yaml:"log-level,omitempty" default:"INFO" readonly:""`
-	ServiceMap             bool                        `yaml:"service-map,omitempty" default:"false" readonly:""`
+	ServiceMap             bool                        `yaml:"service-map" default:"false"`
 	OAS                    bool                        `yaml:"oas,omitempty" default:"false" readonly:""`
+	Elastic                shared.ElasticConfig        `yaml:"elastic"`
 }
 
 func (config *ConfigStruct) validate() error {

cli/config/configStructs/tapConfig.go

@@ -60,12 +60,12 @@ func (config *TapConfig) MaxEntriesDBSizeBytes() int64 {
 func (config *TapConfig) Validate() error {
 	_, compileErr := regexp.Compile(config.PodRegexStr)
 	if compileErr != nil {
-		return errors.New(fmt.Sprintf("%s is not a valid regex %s", config.PodRegexStr, compileErr))
+		return fmt.Errorf("%s is not a valid regex %s", config.PodRegexStr, compileErr)
 	}
 
 	_, parseHumanDataSizeErr := units.HumanReadableToBytes(config.HumanMaxEntriesDBSize)
 	if parseHumanDataSizeErr != nil {
-		return errors.New(fmt.Sprintf("Could not parse --%s value %s", HumanMaxEntriesDBSizeTapName, config.HumanMaxEntriesDBSize))
+		return fmt.Errorf("Could not parse --%s value %s", HumanMaxEntriesDBSizeTapName, config.HumanMaxEntriesDBSize)
 	}
 
 	if config.Workspace != "" {
@@ -76,7 +76,7 @@ func (config *TapConfig) Validate() error {
 	}
 
 	if config.Analysis && config.Workspace != "" {
-		return errors.New(fmt.Sprintf("Can't run with both --%s and --%s flags", AnalysisTapName, WorkspaceTapName))
+		return fmt.Errorf("Can't run with both --%s and --%s flags", AnalysisTapName, WorkspaceTapName)
 	}
 
 	return nil

cli/mizu/fsUtils/dirUtils.go

@@ -1,7 +1,6 @@
 package fsUtils
 
 import (
-	"errors"
 	"fmt"
 	"os"
 )
@@ -18,7 +17,7 @@ func EnsureDir(dirName string) error {
 			return err
 		}
 		if !info.IsDir() {
-			return errors.New(fmt.Sprintf("path exists but is not a directory: %s", dirName))
+			return fmt.Errorf("path exists but is not a directory: %s", dirName)
 		}
 		return nil
 	}

cli/up9/provider.go

@@ -23,9 +23,5 @@ func IsTokenValid(tokenString string, envName string) bool {
 	}
 	defer response.Body.Close()
 
-	if response.StatusCode != http.StatusOK {
-		return false
-	}
-
-	return true
+	return response.StatusCode == http.StatusOK
 }

docs/CHANGELOG.md

@@ -2,6 +2,35 @@
 This document summarizes main and fixes changes published in stable (aka `main`) branch of this project.
 Ongoing work and development releases are under `develop` branch.
 
+## 0.24.0
+
+### main features
+* ARM64 support -- Mizu is now available for ARM 64bit architecture
+  * Now you can run Mizu with `minikube` on your Apple M1 laptop or any other ARM-based hosts
+* New command helps user verify Mizu deployment
+  * Run `mizu check` to verify Mizu was deployed successfully
+  * `mizu check` verifies version compatibility, resources and permissions required by Mizu
+* EXPERIMENTAL: Service Map - graph of all service interactions
+  * Arrow direction show client to server connection
+  * Graph edge width reflects volume of traffic captured between the services
+  * to enable this experimental feature use `--set service-map=true` flag
+
+### improvements
+* Mizu container images are now served from [Docker Hub](https://hub.docker.com/r/up9inc/mizu), as multi-architecture images (arm64, amd64)
+* in Mizu GUI the filter query can now be applied by pressing CONTROL/COMMAND + ENTER
+* try port-forwarding if http-proxy connection to Mizu API server is not available
+
+### notable bug fixes
+* Fixed HTTP/1.0 presentation which was shown as HTTP/1.1
+* Fixed handling of long-living TCP connections, improves capturing gRPC and HTTP/2 traffic, and helps in service-mesh setups (istio, linkerd)
+
+
+## 0.23.0
+### notable bug fixes
+* fixed errors in Redis protocol parser (better handling of Array and Bulk String message types)
+
+
+
 ## 0.22.0
 
 ### main features

docs/PERMISSIONS.md

@@ -4,7 +4,7 @@
 
 This document describes in details all permissions required for full and correct operation of Mizu.
 
-## Editting permissions
+## Editing permissions
 
 During installation, Mizu creates a `ServiceAccount` and the roles it requires. No further action is required.
 However, if there is a need, it is possible to make changes to Mizu permissions.

shared/debounce/debounce.go

@@ -44,7 +44,7 @@ func (d *Debouncer) SetOn() error {
 	if d.canceled {
 		return fmt.Errorf("debouncer cancelled")
 	}
-	if d.running == true {
+	if d.running {
 		return nil
 	}
 

shared/go.sum

@@ -93,6 +93,7 @@ github.com/docker/go-units v0.3.3/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDD
 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 h1:yUdfgN0XgIJw7foRItutHYUIhlcKzcSf5vDpdhQAKTc=
 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=
 github.com/emicklei/go-restful v2.9.5+incompatible/go.mod h1:otzb+WCGbkyDHkqmQmT5YD2WR4BBwUdeQoFo8l/7tVs=
@@ -307,6 +308,7 @@ github.com/mitchellh/gox v0.4.0/go.mod h1:Sd9lOJ0+aimLBi73mGofS1ycjY8lL3uZM3JPS4
 github.com/mitchellh/iochan v1.0.0/go.mod h1:JwYml1nuB7xOzsp52dPpHFffvOCDupsG0QubkSMEySY=
 github.com/mitchellh/mapstructure v0.0.0-20160808181253-ca63d7c062ee/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
 github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
+github.com/moby/spdystream v0.2.0 h1:cjW1zVyyoiM0T7b6UoySUFqzXMoqRckQtXwGPiBhOM8=
 github.com/moby/spdystream v0.2.0/go.mod h1:f7i0iNDQJ059oMTcWxx8MA/zKFIuD/lY+0GqbN2Wy8c=
 github.com/moby/term v0.0.0-20201216013528-df9cb8a40635/go.mod h1:FBS0z0QWA44HXygs7VXDUOGoN/1TV3RuWkLO04am3wc=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=

shared/kubernetes/eventWatchHelper.go

@@ -34,7 +34,7 @@ func (wh *EventWatchHelper) Filter(wEvent *WatchEvent) (bool, error) {
 		return false, nil
 	}
 
-	if strings.ToLower(event.Regarding.Kind) != strings.ToLower(wh.Kind) {
+	if strings.EqualFold(event.Regarding.Kind, wh.Kind) {
 		return false, nil
 	}
 

shared/kubernetes/mizuTapperSyncer.go

@@ -3,21 +3,22 @@ package kubernetes
 import (
 	"context"
 	"fmt"
+	"regexp"
+	"time"
+
 	"github.com/op/go-logging"
 	"github.com/up9inc/mizu/shared"
 	"github.com/up9inc/mizu/shared/debounce"
 	"github.com/up9inc/mizu/shared/logger"
 	"github.com/up9inc/mizu/tap/api"
 	core "k8s.io/api/core/v1"
-	"regexp"
-	"time"
 )
 
 const updateTappersDelay = 5 * time.Second
 
 type TappedPodChangeEvent struct {
-	Added                []core.Pod
-	Removed              []core.Pod
+	Added   []core.Pod
+	Removed []core.Pod
 }
 
 // MizuTapperSyncer uses a k8s pod watch to update tapper daemonsets when targeted pods are removed or created
@@ -222,10 +223,14 @@ func (tapperSyncer *MizuTapperSyncer) watchPodsForTapping() {
 			switch wEvent.Type {
 			case EventAdded:
 				logger.Log.Debugf("Added matching pod %s, ns: %s", pod.Name, pod.Namespace)
-				restartTappersDebouncer.SetOn()
+				if err := restartTappersDebouncer.SetOn(); err != nil {
+					logger.Log.Error(err)
+				}
 			case EventDeleted:
 				logger.Log.Debugf("Removed matching pod %s, ns: %s", pod.Name, pod.Namespace)
-				restartTappersDebouncer.SetOn()
+				if err := restartTappersDebouncer.SetOn(); err != nil {
+					logger.Log.Error(err)
+				}
 			case EventModified:
 				logger.Log.Debugf("Modified matching pod %s, ns: %s, phase: %s, ip: %s", pod.Name, pod.Namespace, pod.Status.Phase, pod.Status.PodIP)
 				// Act only if the modified pod has already obtained an IP address.
@@ -235,7 +240,9 @@ func (tapperSyncer *MizuTapperSyncer) watchPodsForTapping() {
 				// - Pod reaches ready state
 				// Ready/unready transitions might also trigger this event.
 				if pod.Status.PodIP != "" {
-					restartTappersDebouncer.SetOn()
+					if err := restartTappersDebouncer.SetOn(); err != nil {
+						logger.Log.Error(err)
+					}
 				}
 			case EventBookmark:
 				break
@@ -285,8 +292,8 @@ func (tapperSyncer *MizuTapperSyncer) updateCurrentlyTappedPods() (err error, ch
 			tapperSyncer.CurrentlyTappedPods = podsToTap
 			tapperSyncer.nodeToTappedPodMap = GetNodeHostToTappedPodsMap(tapperSyncer.CurrentlyTappedPods)
 			tapperSyncer.TapPodChangesOut <- TappedPodChangeEvent{
-				Added:                addedPods,
-				Removed:              removedPods,
+				Added:   addedPods,
+				Removed: removedPods,
 			}
 			return nil, true
 		}

shared/kubernetes/provider.go

@@ -194,19 +194,19 @@ func (provider *Provider) GetMizuApiServerPodObject(opts *ApiServerOptions, moun
 
 	cpuLimit, err := resource.ParseQuantity(opts.Resources.CpuLimit)
 	if err != nil {
-		return nil, errors.New(fmt.Sprintf("invalid cpu limit for %s container", opts.PodName))
+		return nil, fmt.Errorf("invalid cpu limit for %s container", opts.PodName)
 	}
 	memLimit, err := resource.ParseQuantity(opts.Resources.MemoryLimit)
 	if err != nil {
-		return nil, errors.New(fmt.Sprintf("invalid memory limit for %s container", opts.PodName))
+		return nil, fmt.Errorf("invalid memory limit for %s container", opts.PodName)
 	}
 	cpuRequests, err := resource.ParseQuantity(opts.Resources.CpuRequests)
 	if err != nil {
-		return nil, errors.New(fmt.Sprintf("invalid cpu request for %s container", opts.PodName))
+		return nil, fmt.Errorf("invalid cpu request for %s container", opts.PodName)
 	}
 	memRequests, err := resource.ParseQuantity(opts.Resources.MemoryRequests)
 	if err != nil {
-		return nil, errors.New(fmt.Sprintf("invalid memory request for %s container", opts.PodName))
+		return nil, fmt.Errorf("invalid memory request for %s container", opts.PodName)
 	}
 
 	command := []string{"./mizuagent", "--api-server"}
@@ -395,7 +395,7 @@ func (provider *Provider) CreatePod(ctx context.Context, namespace string, podSp
 }
 
 func (provider *Provider) CreateDeployment(ctx context.Context, namespace string, deploymentName string, podSpec *core.Pod) (*v1.Deployment, error) {
-	if _, keyExists := podSpec.ObjectMeta.Labels["app"]; keyExists == false {
+	if _, keyExists := podSpec.ObjectMeta.Labels["app"]; !keyExists {
 		return nil, errors.New("pod spec must contain 'app' label")
 	}
 	podTemplate := &core.PodTemplateSpec{
@@ -764,7 +764,7 @@ func (provider *Provider) handleRemovalError(err error) error {
 }
 
 func (provider *Provider) CreateConfigMap(ctx context.Context, namespace string, configMapName string, serializedValidationRules string, serializedContract string, serializedMizuConfig string) error {
-	configMapData := make(map[string]string, 0)
+	configMapData := make(map[string]string)
 	if serializedValidationRules != "" {
 		configMapData[shared.ValidationRulesFileName] = serializedValidationRules
 	}
@@ -854,19 +854,19 @@ func (provider *Provider) ApplyMizuTapperDaemonSet(ctx context.Context, namespac
 	)
 	cpuLimit, err := resource.ParseQuantity(resources.CpuLimit)
 	if err != nil {
-		return errors.New(fmt.Sprintf("invalid cpu limit for %s container", tapperPodName))
+		return fmt.Errorf("invalid cpu limit for %s container", tapperPodName)
 	}
 	memLimit, err := resource.ParseQuantity(resources.MemoryLimit)
 	if err != nil {
-		return errors.New(fmt.Sprintf("invalid memory limit for %s container", tapperPodName))
+		return fmt.Errorf("invalid memory limit for %s container", tapperPodName)
 	}
 	cpuRequests, err := resource.ParseQuantity(resources.CpuRequests)
 	if err != nil {
-		return errors.New(fmt.Sprintf("invalid cpu request for %s container", tapperPodName))
+		return fmt.Errorf("invalid cpu request for %s container", tapperPodName)
 	}
 	memRequests, err := resource.ParseQuantity(resources.MemoryRequests)
 	if err != nil {
-		return errors.New(fmt.Sprintf("invalid memory request for %s container", tapperPodName))
+		return fmt.Errorf("invalid memory request for %s container", tapperPodName)
 	}
 	agentResourceLimits := core.ResourceList{
 		"cpu":    cpuLimit,

shared/kubernetes/utils.go

@@ -10,7 +10,7 @@ import (
 )
 
 func GetNodeHostToTappedPodsMap(tappedPods []core.Pod) map[string][]core.Pod {
-	nodeToTappedPodMap := make(map[string][]core.Pod, 0)
+	nodeToTappedPodMap := make(map[string][]core.Pod)
 	for _, pod := range tappedPods {
 		minimizedPod := getMinimizedPod(pod)
 

shared/kubernetes/watch.go

@@ -4,11 +4,12 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"github.com/up9inc/mizu/shared/debounce"
-	"github.com/up9inc/mizu/shared/logger"
 	"sync"
 	"time"
 
+	"github.com/up9inc/mizu/shared/debounce"
+	"github.com/up9inc/mizu/shared/logger"
+
 	"k8s.io/apimachinery/pkg/watch"
 )
 
@@ -31,7 +32,7 @@ func FilteredWatch(ctx context.Context, watcherCreator WatchCreator, targetNames
 
 		go func(targetNamespace string) {
 			defer wg.Done()
-			watchRestartDebouncer := debounce.NewDebouncer(1 * time.Minute, func() {})
+			watchRestartDebouncer := debounce.NewDebouncer(1*time.Minute, func() {})
 
 			for {
 				watcher, err := watcherCreator.NewWatcher(ctx, targetNamespace)
@@ -44,7 +45,7 @@ func FilteredWatch(ctx context.Context, watcherCreator WatchCreator, targetNames
 				watcher.Stop()
 
 				select {
-				case <- ctx.Done():
+				case <-ctx.Done():
 					return
 				default:
 					break
@@ -55,7 +56,9 @@ func FilteredWatch(ctx context.Context, watcherCreator WatchCreator, targetNames
 					break
 				} else {
 					if !watchRestartDebouncer.IsOn() {
-						watchRestartDebouncer.SetOn()
+						if err := watchRestartDebouncer.SetOn(); err != nil {
+							logger.Log.Error(err)
+						}
 						logger.Log.Debug("k8s watch channel closed, restarting watcher")
 						time.Sleep(time.Second * 5)
 						continue

shared/models.go

@@ -43,6 +43,13 @@ type MizuAgentConfig struct {
 	StandaloneMode         bool          `json:"standaloneMode"`
 	ServiceMap             bool          `json:"serviceMap"`
 	OAS                    bool          `json:"oas"`
+	Elastic                ElasticConfig `json:"elastic"`
+}
+
+type ElasticConfig struct {
+	User     string `yaml:"user,omitempty" default:"" readonly:""`
+	Password string `yaml:"password,omitempty" default:"" readonly:""`
+	Url      string `yaml:"url,omitempty" default:"" readonly:""`
 }
 
 type WebSocketMessageMetadata struct {

tap/api/api.go

@@ -6,11 +6,12 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"github.com/google/martian/har"
 	"io/ioutil"
 	"net/http"
 	"sync"
 	"time"
+
+	"github.com/google/martian/har"
 )
 
 type Protocol struct {
@@ -282,7 +283,7 @@ func (h HTTPPayload) MarshalJSON() ([]byte, error) {
 			RawResponse: &HTTPResponseWrapper{Response: h.Data.(*http.Response)},
 		})
 	default:
-		panic(fmt.Sprintf("HTTP payload cannot be marshaled: %s", h.Type))
+		panic(fmt.Sprintf("HTTP payload cannot be marshaled: %v", h.Type))
 	}
 }
 
@@ -312,7 +313,7 @@ type HTTPRequestWrapper struct {
 func (r *HTTPRequestWrapper) MarshalJSON() ([]byte, error) {
 	body, _ := ioutil.ReadAll(r.Request.Body)
 	r.Request.Body = ioutil.NopCloser(bytes.NewBuffer(body))
-	return json.Marshal(&struct {
+	return json.Marshal(&struct { //nolint
 		Body    string `json:"Body,omitempty"`
 		GetBody string `json:"GetBody,omitempty"`
 		Cancel  string `json:"Cancel,omitempty"`
@@ -330,7 +331,7 @@ type HTTPResponseWrapper struct {
 func (r *HTTPResponseWrapper) MarshalJSON() ([]byte, error) {
 	body, _ := ioutil.ReadAll(r.Response.Body)
 	r.Response.Body = ioutil.NopCloser(bytes.NewBuffer(body))
-	return json.Marshal(&struct {
+	return json.Marshal(&struct { //nolint
 		Body    string `json:"Body,omitempty"`
 		GetBody string `json:"GetBody,omitempty"`
 		Cancel  string `json:"Cancel,omitempty"`

tap/extensions/amqp/helpers.go

@@ -24,14 +24,14 @@ var connectionMethodMap = map[int]string{
 	61: "connection unblocked",
 }
 
-var channelMethodMap = map[int]string{
-	10: "channel open",
-	11: "channel open-ok",
-	20: "channel flow",
-	21: "channel flow-ok",
-	40: "channel close",
-	41: "channel close-ok",
-}
+// var channelMethodMap = map[int]string{
+// 	10: "channel open",
+// 	11: "channel open-ok",
+// 	20: "channel flow",
+// 	21: "channel flow-ok",
+// 	40: "channel close",
+// 	41: "channel close-ok",
+// }
 
 var exchangeMethodMap = map[int]string{
 	10: "exchange declare",
@@ -78,14 +78,14 @@ var basicMethodMap = map[int]string{
 	120: "basic nack",
 }
 
-var txMethodMap = map[int]string{
-	10: "tx select",
-	11: "tx select-ok",
-	20: "tx commit",
-	21: "tx commit-ok",
-	30: "tx rollback",
-	31: "tx rollback-ok",
-}
+// var txMethodMap = map[int]string{
+// 	10: "tx select",
+// 	11: "tx select-ok",
+// 	20: "tx commit",
+// 	21: "tx commit-ok",
+// 	30: "tx rollback",
+// 	31: "tx rollback-ok",
+// }
 
 type AMQPWrapper struct {
 	Method  string      `json:"method"`
@@ -550,14 +550,12 @@ func representConnectionStart(event map[string]interface{}) []interface{} {
 		headers := make([]api.TableData, 0)
 		for name, value := range event["serverProperties"].(map[string]interface{}) {
 			var outcome string
-			switch value.(type) {
+			switch v := value.(type) {
 			case string:
-				outcome = value.(string)
-				break
+				outcome = v
 			case map[string]interface{}:
 				x, _ := json.Marshal(value)
 				outcome = string(x)
-				break
 			default:
 				panic("Unknown data type for the server property!")
 			}

tap/extensions/amqp/main.go

@@ -47,7 +47,6 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
 
 	var remaining int
 	var header *HeaderFrame
-	var body []byte
 
 	connectionInfo := &api.ConnectionInfo{
 		ClientIP:   tcpID.SrcIP,
@@ -102,13 +101,10 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
 				eventBasicPublish.Properties = header.Properties
 			case *BasicDeliver:
 				eventBasicDeliver.Properties = header.Properties
-			default:
-				frame = nil
 			}
 
 		case *BodyFrame:
 			// continue until terminated
-			body = append(body, f.Body...)
 			remaining -= len(f.Body)
 			switch lastMethodFrameMessage.(type) {
 			case *BasicPublish:
@@ -119,9 +115,6 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
 				eventBasicDeliver.Body = f.Body
 				superIdentifier.Protocol = &protocol
 				emitAMQP(*eventBasicDeliver, amqpRequest, basicMethodMap[60], connectionInfo, superTimer.CaptureTime, emitter)
-			default:
-				body = nil
-				frame = nil
 			}
 
 		case *MethodFrame:
@@ -211,10 +204,6 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
 				}
 				superIdentifier.Protocol = &protocol
 				emitAMQP(*eventConnectionClose, amqpRequest, connectionMethodMap[50], connectionInfo, superTimer.CaptureTime, emitter)
-
-			default:
-				frame = nil
-
 			}
 
 		default:
@@ -231,32 +220,24 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, resolvedSource string,
 	switch request["method"] {
 	case basicMethodMap[40]:
 		summary = reqDetails["exchange"].(string)
-		break
 	case basicMethodMap[60]:
 		summary = reqDetails["exchange"].(string)
-		break
 	case exchangeMethodMap[10]:
 		summary = reqDetails["exchange"].(string)
-		break
 	case queueMethodMap[10]:
 		summary = reqDetails["queue"].(string)
-		break
 	case connectionMethodMap[10]:
 		summary = fmt.Sprintf(
 			"%s.%s",
 			strconv.Itoa(int(reqDetails["versionMajor"].(float64))),
 			strconv.Itoa(int(reqDetails["versionMinor"].(float64))),
 		)
-		break
 	case connectionMethodMap[50]:
 		summary = reqDetails["replyText"].(string)
-		break
 	case queueMethodMap[20]:
 		summary = reqDetails["queue"].(string)
-		break
 	case basicMethodMap[20]:
 		summary = reqDetails["queue"].(string)
-		break
 	}
 
 	request["url"] = summary
@@ -288,33 +269,25 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, resolvedSource string,
 
 func (d dissecting) Represent(request map[string]interface{}, response map[string]interface{}) (object []byte, bodySize int64, err error) {
 	bodySize = 0
-	representation := make(map[string]interface{}, 0)
+	representation := make(map[string]interface{})
 	var repRequest []interface{}
 	switch request["method"].(string) {
 	case basicMethodMap[40]:
 		repRequest = representBasicPublish(request)
-		break
 	case basicMethodMap[60]:
 		repRequest = representBasicDeliver(request)
-		break
 	case queueMethodMap[10]:
 		repRequest = representQueueDeclare(request)
-		break
 	case exchangeMethodMap[10]:
 		repRequest = representExchangeDeclare(request)
-		break
 	case connectionMethodMap[10]:
 		repRequest = representConnectionStart(request)
-		break
 	case connectionMethodMap[50]:
 		repRequest = representConnectionClose(request)
-		break
 	case queueMethodMap[20]:
 		repRequest = representQueueBind(request)
-		break
 	case basicMethodMap[20]:
 		repRequest = representBasicConsume(request)
-		break
 	}
 	representation["request"] = repRequest
 	object, err = json.Marshal(representation)

tap/extensions/amqp/read.go

@@ -135,20 +135,6 @@ func readDecimal(r io.Reader) (v Decimal, err error) {
 	return
 }
 
-func readFloat32(r io.Reader) (v float32, err error) {
-	if err = binary.Read(r, binary.BigEndian, &v); err != nil {
-		return
-	}
-	return
-}
-
-func readFloat64(r io.Reader) (v float64, err error) {
-	if err = binary.Read(r, binary.BigEndian, &v); err != nil {
-		return
-	}
-	return
-}
-
 func readTimestamp(r io.Reader) (v time.Time, err error) {
 	var sec int64
 	if err = binary.Read(r, binary.BigEndian, &sec); err != nil {

tap/extensions/amqp/types.go

@@ -78,15 +78,6 @@ type Error struct {
 	Recover bool   // true when this error can be recovered by retrying later or with different parameters
 }
 
-func newError(code uint16, text string) *Error {
-	return &Error{
-		Code:    int(code),
-		Reason:  text,
-		Recover: isSoftExceptionCode(int(code)),
-		Server:  true,
-	}
-}
-
 func (e Error) Error() string {
 	return fmt.Sprintf("Exception (%d) Reason: %q", e.Code, e.Reason)
 }
@@ -262,19 +253,6 @@ func (t Table) Validate() error {
 	return validateField(t)
 }
 
-// Heap interface for maintaining delivery tags
-type tagSet []uint64
-
-func (set tagSet) Len() int              { return len(set) }
-func (set tagSet) Less(i, j int) bool    { return (set)[i] < (set)[j] }
-func (set tagSet) Swap(i, j int)         { (set)[i], (set)[j] = (set)[j], (set)[i] }
-func (set *tagSet) Push(tag interface{}) { *set = append(*set, tag.(uint64)) }
-func (set *tagSet) Pop() interface{} {
-	val := (*set)[len(*set)-1]
-	*set = (*set)[:len(*set)-1]
-	return val
-}
-
 type Message interface {
 	id() (uint16, uint16)
 	wait() bool
@@ -282,12 +260,6 @@ type Message interface {
 	write(io.Writer) error
 }
 
-type messageWithContent interface {
-	Message
-	getContent() (Properties, []byte)
-	setContent(Properties, []byte)
-}
-
 /*
 The base interface implemented as:
 
@@ -322,22 +294,6 @@ type AmqpReader struct {
 	R io.Reader
 }
 
-type writer struct {
-	w io.Writer
-}
-
-// Implements the frame interface for Connection RPC
-type protocolHeader struct{}
-
-func (protocolHeader) write(w io.Writer) error {
-	_, err := w.Write([]byte{'A', 'M', 'Q', 'P', 0, 0, 9, 1})
-	return err
-}
-
-func (protocolHeader) channel() uint16 {
-	panic("only valid as initial handshake")
-}
-
 /*
 Method frames carry the high-level protocol commands (which we call "methods").
 One method frame carries one command.  The method frame payload has this format:

tap/extensions/amqp/write.go

@@ -6,7 +6,6 @@
 package amqp
 
 import (
-	"bufio"
 	"bytes"
 	"encoding/binary"
 	"errors"
@@ -15,18 +14,6 @@ import (
 	"time"
 )
 
-func (w *writer) WriteFrame(frame frame) (err error) {
-	if err = frame.write(w.w); err != nil {
-		return
-	}
-
-	if buf, ok := w.w.(*bufio.Writer); ok {
-		err = buf.Flush()
-	}
-
-	return
-}
-
 func (f *MethodFrame) write(w io.Writer) (err error) {
 	var payload bytes.Buffer
 
@@ -412,5 +399,5 @@ func writeTable(w io.Writer, table Table) (err error) {
 		}
 	}
 
-	return writeLongstr(w, string(buf.Bytes()))
+	return writeLongstr(w, buf.String())
 }

tap/extensions/http/helpers.go

@@ -81,7 +81,7 @@ func representSliceAsTable(slice []interface{}, selectorPrefix string) (represen
 		selector := fmt.Sprintf("%s[%d]", selectorPrefix, i)
 		table = append(table, api.TableData{
 			Name:     strconv.Itoa(i),
-			Value:    item.(interface{}),
+			Value:    item,
 			Selector: selector,
 		})
 	}

tap/extensions/http/main.go

@@ -92,21 +92,30 @@ func (d dissecting) Ping() {
 }
 
 func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, superIdentifier *api.SuperIdentifier, emitter api.Emitter, options *api.TrafficFilteringOptions) error {
-	isHTTP2, err := checkIsHTTP2Connection(b, isClient)
+	var err error
+	isHTTP2, _ := checkIsHTTP2Connection(b, isClient)
 
 	var http2Assembler *Http2Assembler
 	if isHTTP2 {
-		prepareHTTP2Connection(b, isClient)
+		err = prepareHTTP2Connection(b, isClient)
+		if err != nil {
+			return err
+		}
 		http2Assembler = createHTTP2Assembler(b)
 	}
 
-	dissected := false
 	switchingProtocolsHTTP2 := false
 	for {
 		if switchingProtocolsHTTP2 {
 			switchingProtocolsHTTP2 = false
 			isHTTP2, err = checkIsHTTP2Connection(b, isClient)
-			prepareHTTP2Connection(b, isClient)
+			if err != nil {
+				break
+			}
+			err = prepareHTTP2Connection(b, isClient)
+			if err != nil {
+				break
+			}
 			http2Assembler = createHTTP2Assembler(b)
 		}
 
@@ -121,7 +130,7 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
 			} else if err != nil {
 				continue
 			}
-			dissected = true
+			superIdentifier.Protocol = &http11protocol
 		} else if isClient {
 			var req *http.Request
 			switchingProtocolsHTTP2, req, err = handleHTTP1ClientStream(b, tcpID, counterPair, superTimer, emitter, options)
@@ -130,7 +139,7 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
 			} else if err != nil {
 				continue
 			}
-			dissected = true
+			superIdentifier.Protocol = &http11protocol
 
 			// In case of an HTTP2 upgrade, duplicate the HTTP1 request into HTTP2 with stream ID 1
 			if switchingProtocolsHTTP2 {
@@ -161,14 +170,14 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
 			} else if err != nil {
 				continue
 			}
-			dissected = true
+			superIdentifier.Protocol = &http11protocol
 		}
 	}
 
-	if !dissected {
+	if superIdentifier.Protocol == nil {
 		return err
 	}
-	superIdentifier.Protocol = &http11protocol
+
 	return nil
 }
 
@@ -341,11 +350,11 @@ func representRequest(request map[string]interface{}) (repRequest []interface{})
 	})
 
 	postData, _ := request["postData"].(map[string]interface{})
-	mimeType, _ := postData["mimeType"]
+	mimeType := postData["mimeType"]
 	if mimeType == nil || len(mimeType.(string)) == 0 {
 		mimeType = "text/html"
 	}
-	text, _ := postData["text"]
+	text := postData["text"]
 	if text != nil {
 		repRequest = append(repRequest, api.SectionData{
 			Type:     api.BODY,
@@ -425,12 +434,12 @@ func representResponse(response map[string]interface{}) (repResponse []interface
 	})
 
 	content, _ := response["content"].(map[string]interface{})
-	mimeType, _ := content["mimeType"]
+	mimeType := content["mimeType"]
 	if mimeType == nil || len(mimeType.(string)) == 0 {
 		mimeType = "text/html"
 	}
-	encoding, _ := content["encoding"]
-	text, _ := content["text"]
+	encoding := content["encoding"]
+	text := content["text"]
 	if text != nil {
 		repResponse = append(repResponse, api.SectionData{
 			Type:     api.BODY,
@@ -446,7 +455,7 @@ func representResponse(response map[string]interface{}) (repResponse []interface
 }
 
 func (d dissecting) Represent(request map[string]interface{}, response map[string]interface{}) (object []byte, bodySize int64, err error) {
-	representation := make(map[string]interface{}, 0)
+	representation := make(map[string]interface{})
 	repRequest := representRequest(request)
 	repResponse, bodySize := representResponse(response)
 	representation["request"] = repRequest

tap/extensions/http/sensitive_data_cleaner.go

@@ -89,7 +89,7 @@ func filterResponseBody(response *http.Response, options *api.TrafficFilteringOp
 }
 
 func filterHeaders(headers *http.Header) {
-	for key, _ := range *headers {
+	for key := range *headers {
 		if strings.ToLower(key) == userAgent {
 			continue
 		}
@@ -103,7 +103,7 @@ func filterHeaders(headers *http.Header) {
 }
 
 func getContentTypeHeaderValue(headers http.Header) string {
-	for key, _ := range headers {
+	for key := range headers {
 		if strings.ToLower(key) == "content-type" {
 			return headers.Get(key)
 		}

tap/extensions/kafka/buffer.go

@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"fmt"
 	"io"
-	"math"
 	"sync"
 	"sync/atomic"
 )
@@ -201,25 +200,6 @@ func newPageBuffer() *pageBuffer {
 	return b
 }
 
-func (pb *pageBuffer) refTo(ref *pageRef, begin, end int64) {
-	length := end - begin
-
-	if length > math.MaxUint32 {
-		panic("reference to contiguous buffer pages exceeds the maximum size of 4 GB")
-	}
-
-	ref.pages = append(ref.buffer[:0], pb.pages.slice(begin, end)...)
-	ref.pages.ref()
-	ref.offset = begin
-	ref.length = uint32(length)
-}
-
-func (pb *pageBuffer) ref(begin, end int64) *pageRef {
-	ref := new(pageRef)
-	pb.refTo(ref, begin, end)
-	return ref
-}
-
 func (pb *pageBuffer) unref() {
 	pb.refc.unref(func() {
 		pb.pages.unref()
@@ -353,12 +333,12 @@ func (pb *pageBuffer) Write(b []byte) (int, error) {
 		free := tail.Cap() - tail.Len()
 
 		if len(b) <= free {
-			tail.Write(b)
+			_, _ = tail.Write(b)
 			pb.length += len(b)
 			break
 		}
 
-		tail.Write(b[:free])
+		_, _ = tail.Write(b[:free])
 		b = b[free:]
 
 		pb.length += free
@@ -374,7 +354,7 @@ func (pb *pageBuffer) WriteAt(b []byte, off int64) (int, error) {
 		return n, err
 	}
 	if n < len(b) {
-		pb.Write(b[n:])
+		_, _ = pb.Write(b[n:])
 	}
 	return len(b), nil
 }
@@ -406,12 +386,6 @@ var (
 
 type contiguousPages []*page
 
-func (pages contiguousPages) ref() {
-	for _, p := range pages {
-		p.ref()
-	}
-}
-
 func (pages contiguousPages) unref() {
 	for _, p := range pages {
 		p.unref()
@@ -480,7 +454,6 @@ var (
 )
 
 type pageRef struct {
-	buffer [2]*page
 	pages  contiguousPages
 	offset int64
 	cursor int64
@@ -590,28 +563,6 @@ var (
 	_ io.WriterTo = (*pageRef)(nil)
 )
 
-type pageRefAllocator struct {
-	refs []pageRef
-	head int
-	size int
-}
-
-func (a *pageRefAllocator) newPageRef() *pageRef {
-	if a.head == len(a.refs) {
-		a.refs = make([]pageRef, a.size)
-		a.head = 0
-	}
-	ref := &a.refs[a.head]
-	a.head++
-	return ref
-}
-
-func unref(x interface{}) {
-	if r, _ := x.(interface{ unref() }); r != nil {
-		r.unref()
-	}
-}
-
 func seek(cursor, limit, offset int64, whence int) (int64, error) {
 	switch whence {
 	case io.SeekStart:
@@ -631,15 +582,3 @@ func seek(cursor, limit, offset int64, whence int) (int64, error) {
 	}
 	return offset, nil
 }
-
-func closeBytes(b Bytes) {
-	if b != nil {
-		b.Close()
-	}
-}
-
-func resetBytes(b Bytes) {
-	if r, _ := b.(interface{ Reset() }); r != nil {
-		r.Reset()
-	}
-}

tap/extensions/kafka/compression.go

@@ -1,8 +1,6 @@
 package kafka
 
 import (
-	"errors"
-
 	"github.com/segmentio/kafka-go/compress"
 )
 
@@ -12,19 +10,3 @@ type CompressionCodec = compress.Codec
 
 // TODO: this file should probably go away once the internals of the package
 // have moved to use the protocol package.
-const (
-	compressionCodecMask = 0x07
-)
-
-var (
-	errUnknownCodec = errors.New("the compression code is invalid or its codec has not been imported")
-)
-
-// resolveCodec looks up a codec by Code()
-func resolveCodec(code int8) (CompressionCodec, error) {
-	codec := compress.Compression(code).Codec()
-	if codec == nil {
-		return nil, errUnknownCodec
-	}
-	return codec, nil
-}

tap/extensions/kafka/decode.go

@@ -58,14 +58,6 @@ func (d *decoder) ReadByte() (byte, error) {
 	return c, d.err
 }
 
-func (d *decoder) done() bool {
-	return d.remain == 0 || d.err != nil
-}
-
-func (d *decoder) setCRC(table *crc32.Table) {
-	d.table, d.crc32 = table, 0
-}
-
 func (d *decoder) decodeBool(v value) {
 	v.setBool(d.readBool())
 }
@@ -199,19 +191,6 @@ func (d *decoder) read(n int) []byte {
 	return b
 }
 
-func (d *decoder) writeTo(w io.Writer, n int) {
-	limit := d.remain
-	if n < limit {
-		d.remain = n
-	}
-	c, err := io.Copy(w, d)
-	if int(c) < n && err == nil {
-		err = io.ErrUnexpectedEOF
-	}
-	d.remain = limit - int(c)
-	d.setError(err)
-}
-
 func (d *decoder) setError(err error) {
 	if d.err == nil && err != nil {
 		d.err = err
@@ -272,14 +251,6 @@ func (d *decoder) readString() string {
 	}
 }
 
-func (d *decoder) readVarString() string {
-	if n := d.readVarInt(); n < 0 {
-		return ""
-	} else {
-		return bytesToString(d.read(int(n)))
-	}
-}
-
 func (d *decoder) readCompactString() string {
 	if n := d.readUnsignedVarInt(); n < 1 {
 		return ""
@@ -296,32 +267,6 @@ func (d *decoder) readBytes() []byte {
 	}
 }
 
-func (d *decoder) readBytesTo(w io.Writer) bool {
-	if n := d.readInt32(); n < 0 {
-		return false
-	} else {
-		d.writeTo(w, int(n))
-		return d.err == nil
-	}
-}
-
-func (d *decoder) readVarBytes() []byte {
-	if n := d.readVarInt(); n < 0 {
-		return nil
-	} else {
-		return d.read(int(n))
-	}
-}
-
-func (d *decoder) readVarBytesTo(w io.Writer) bool {
-	if n := d.readVarInt(); n < 0 {
-		return false
-	} else {
-		d.writeTo(w, int(n))
-		return d.err == nil
-	}
-}
-
 func (d *decoder) readCompactBytes() []byte {
 	if n := d.readUnsignedVarInt(); n < 1 {
 		return nil
@@ -330,15 +275,6 @@ func (d *decoder) readCompactBytes() []byte {
 	}
 }
 
-func (d *decoder) readCompactBytesTo(w io.Writer) bool {
-	if n := d.readUnsignedVarInt(); n < 1 {
-		return false
-	} else {
-		d.writeTo(w, int(n-1))
-		return d.err == nil
-	}
-}
-
 func (d *decoder) readVarInt() int64 {
 	n := 11 // varints are at most 11 bytes
 

tap/extensions/kafka/discard.go

@@ -14,37 +14,3 @@ func discardN(r *bufio.Reader, sz int, n int) (int, error) {
 	}
 	return sz - n, err
 }
-
-func discardInt8(r *bufio.Reader, sz int) (int, error) {
-	return discardN(r, sz, 1)
-}
-
-func discardInt16(r *bufio.Reader, sz int) (int, error) {
-	return discardN(r, sz, 2)
-}
-
-func discardInt32(r *bufio.Reader, sz int) (int, error) {
-	return discardN(r, sz, 4)
-}
-
-func discardInt64(r *bufio.Reader, sz int) (int, error) {
-	return discardN(r, sz, 8)
-}
-
-func discardString(r *bufio.Reader, sz int) (int, error) {
-	return readStringWith(r, sz, func(r *bufio.Reader, sz int, n int) (int, error) {
-		if n < 0 {
-			return sz, nil
-		}
-		return discardN(r, sz, n)
-	})
-}
-
-func discardBytes(r *bufio.Reader, sz int) (int, error) {
-	return readBytesWith(r, sz, func(r *bufio.Reader, sz int, n int) (int, error) {
-		if n < 0 {
-			return sz, nil
-		}
-		return discardN(r, sz, n)
-	})
-}

tap/extensions/kafka/encode.go

@@ -3,7 +3,6 @@ package kafka
 import (
 	"bytes"
 	"encoding/binary"
-	"fmt"
 	"hash/crc32"
 	"io"
 	"reflect"
@@ -95,10 +94,6 @@ func (e *encoder) WriteString(s string) (int, error) {
 	return n, nil
 }
 
-func (e *encoder) setCRC(table *crc32.Table) {
-	e.table, e.crc32 = table, 0
-}
-
 func (e *encoder) update(b []byte) {
 	if e.table != nil {
 		e.crc32 = crc32.Update(e.crc32, e.table, b)
@@ -133,10 +128,6 @@ func (e *encoder) encodeString(v value) {
 	e.writeString(v.string())
 }
 
-func (e *encoder) encodeVarString(v value) {
-	e.writeVarString(v.string())
-}
-
 func (e *encoder) encodeCompactString(v value) {
 	e.writeCompactString(v.string())
 }
@@ -145,10 +136,6 @@ func (e *encoder) encodeNullString(v value) {
 	e.writeNullString(v.string())
 }
 
-func (e *encoder) encodeVarNullString(v value) {
-	e.writeVarNullString(v.string())
-}
-
 func (e *encoder) encodeCompactNullString(v value) {
 	e.writeCompactNullString(v.string())
 }
@@ -157,10 +144,6 @@ func (e *encoder) encodeBytes(v value) {
 	e.writeBytes(v.bytes())
 }
 
-func (e *encoder) encodeVarBytes(v value) {
-	e.writeVarBytes(v.bytes())
-}
-
 func (e *encoder) encodeCompactBytes(v value) {
 	e.writeCompactBytes(v.bytes())
 }
@@ -169,10 +152,6 @@ func (e *encoder) encodeNullBytes(v value) {
 	e.writeNullBytes(v.bytes())
 }
 
-func (e *encoder) encodeVarNullBytes(v value) {
-	e.writeVarNullBytes(v.bytes())
-}
-
 func (e *encoder) encodeCompactNullBytes(v value) {
 	e.writeCompactNullBytes(v.bytes())
 }
@@ -228,37 +207,32 @@ func (e *encoder) encodeCompactNullArray(v value, elemType reflect.Type, encodeE
 
 func (e *encoder) writeInt8(i int8) {
 	writeInt8(e.buffer[:1], i)
-	e.Write(e.buffer[:1])
+	_, _ = e.Write(e.buffer[:1])
 }
 
 func (e *encoder) writeInt16(i int16) {
 	writeInt16(e.buffer[:2], i)
-	e.Write(e.buffer[:2])
+	_, _ = e.Write(e.buffer[:2])
 }
 
 func (e *encoder) writeInt32(i int32) {
 	writeInt32(e.buffer[:4], i)
-	e.Write(e.buffer[:4])
+	_, _ = e.Write(e.buffer[:4])
 }
 
 func (e *encoder) writeInt64(i int64) {
 	writeInt64(e.buffer[:8], i)
-	e.Write(e.buffer[:8])
+	_, _ = e.Write(e.buffer[:8])
 }
 
 func (e *encoder) writeString(s string) {
 	e.writeInt16(int16(len(s)))
-	e.WriteString(s)
-}
-
-func (e *encoder) writeVarString(s string) {
-	e.writeVarInt(int64(len(s)))
-	e.WriteString(s)
+	_, _ = e.WriteString(s)
 }
 
 func (e *encoder) writeCompactString(s string) {
 	e.writeUnsignedVarInt(uint64(len(s)) + 1)
-	e.WriteString(s)
+	_, _ = e.WriteString(s)
 }
 
 func (e *encoder) writeNullString(s string) {
@@ -266,16 +240,7 @@ func (e *encoder) writeNullString(s string) {
 		e.writeInt16(-1)
 	} else {
 		e.writeInt16(int16(len(s)))
-		e.WriteString(s)
-	}
-}
-
-func (e *encoder) writeVarNullString(s string) {
-	if s == "" {
-		e.writeVarInt(-1)
-	} else {
-		e.writeVarInt(int64(len(s)))
-		e.WriteString(s)
+		_, _ = e.WriteString(s)
 	}
 }
 
@@ -284,23 +249,18 @@ func (e *encoder) writeCompactNullString(s string) {
 		e.writeUnsignedVarInt(0)
 	} else {
 		e.writeUnsignedVarInt(uint64(len(s)) + 1)
-		e.WriteString(s)
+		_, _ = e.WriteString(s)
 	}
 }
 
 func (e *encoder) writeBytes(b []byte) {
 	e.writeInt32(int32(len(b)))
-	e.Write(b)
-}
-
-func (e *encoder) writeVarBytes(b []byte) {
-	e.writeVarInt(int64(len(b)))
-	e.Write(b)
+	_, _ = e.Write(b)
 }
 
 func (e *encoder) writeCompactBytes(b []byte) {
 	e.writeUnsignedVarInt(uint64(len(b)) + 1)
-	e.Write(b)
+	_, _ = e.Write(b)
 }
 
 func (e *encoder) writeNullBytes(b []byte) {
@@ -308,16 +268,7 @@ func (e *encoder) writeNullBytes(b []byte) {
 		e.writeInt32(-1)
 	} else {
 		e.writeInt32(int32(len(b)))
-		e.Write(b)
-	}
-}
-
-func (e *encoder) writeVarNullBytes(b []byte) {
-	if b == nil {
-		e.writeVarInt(-1)
-	} else {
-		e.writeVarInt(int64(len(b)))
-		e.Write(b)
+		_, _ = e.Write(b)
 	}
 }
 
@@ -326,69 +277,10 @@ func (e *encoder) writeCompactNullBytes(b []byte) {
 		e.writeUnsignedVarInt(0)
 	} else {
 		e.writeUnsignedVarInt(uint64(len(b)) + 1)
-		e.Write(b)
+		_, _ = e.Write(b)
 	}
 }
 
-func (e *encoder) writeBytesFrom(b Bytes) error {
-	size := int64(b.Len())
-	e.writeInt32(int32(size))
-	n, err := io.Copy(e, b)
-	if err == nil && n != size {
-		err = fmt.Errorf("size of bytes does not match the number of bytes that were written (size=%d, written=%d): %w", size, n, io.ErrUnexpectedEOF)
-	}
-	return err
-}
-
-func (e *encoder) writeNullBytesFrom(b Bytes) error {
-	if b == nil {
-		e.writeInt32(-1)
-		return nil
-	} else {
-		size := int64(b.Len())
-		e.writeInt32(int32(size))
-		n, err := io.Copy(e, b)
-		if err == nil && n != size {
-			err = fmt.Errorf("size of nullable bytes does not match the number of bytes that were written (size=%d, written=%d): %w", size, n, io.ErrUnexpectedEOF)
-		}
-		return err
-	}
-}
-
-func (e *encoder) writeVarNullBytesFrom(b Bytes) error {
-	if b == nil {
-		e.writeVarInt(-1)
-		return nil
-	} else {
-		size := int64(b.Len())
-		e.writeVarInt(size)
-		n, err := io.Copy(e, b)
-		if err == nil && n != size {
-			err = fmt.Errorf("size of nullable bytes does not match the number of bytes that were written (size=%d, written=%d): %w", size, n, io.ErrUnexpectedEOF)
-		}
-		return err
-	}
-}
-
-func (e *encoder) writeCompactNullBytesFrom(b Bytes) error {
-	if b == nil {
-		e.writeUnsignedVarInt(0)
-		return nil
-	} else {
-		size := int64(b.Len())
-		e.writeUnsignedVarInt(uint64(size + 1))
-		n, err := io.Copy(e, b)
-		if err == nil && n != size {
-			err = fmt.Errorf("size of compact nullable bytes does not match the number of bytes that were written (size=%d, written=%d): %w", size, n, io.ErrUnexpectedEOF)
-		}
-		return err
-	}
-}
-
-func (e *encoder) writeVarInt(i int64) {
-	e.writeUnsignedVarInt(uint64((i << 1) ^ (i >> 63)))
-}
-
 func (e *encoder) writeUnsignedVarInt(i uint64) {
 	b := e.buffer[:]
 	n := 0
@@ -404,7 +296,7 @@ func (e *encoder) writeUnsignedVarInt(i uint64) {
 		n++
 	}
 
-	e.Write(b[:n])
+	_, _ = e.Write(b[:n])
 }
 
 type encodeFunc func(*encoder, value)
@@ -530,7 +422,7 @@ func structEncodeFuncOf(typ reflect.Type, version int16, flexible bool) encodeFu
 				se := &encoder{writer: buf}
 				f.encode(se, v.fieldByIndex(f.index))
 				e.writeUnsignedVarInt(uint64(buf.Len()))
-				e.Write(buf.Bytes())
+				_, _ = e.Write(buf.Bytes())
 			}
 		}
 	}

tap/extensions/kafka/helpers.go

@@ -346,7 +346,13 @@ func representProduceRequest(data map[string]interface{}) []interface{} {
 			})
 
 			obj, err := oj.ParseString(string(partitionsJson))
+			if err != nil {
+				return rep
+			}
 			recordBatchPath, err := jp.ParseString(`partitionData.records.recordBatch`)
+			if err != nil {
+				return rep
+			}
 			recordBatchresults := recordBatchPath.Get(obj)
 			if len(recordBatchresults) > 0 {
 				rep = append(rep, api.SectionData{
@@ -357,6 +363,9 @@ func representProduceRequest(data map[string]interface{}) []interface{} {
 			}
 
 			recordsPath, err := jp.ParseString(`partitionData.records.recordBatch.record`)
+			if err != nil {
+				return rep
+			}
 			recordsResults := recordsPath.Get(obj)
 			if len(recordsResults) > 0 {
 				records := recordsResults[0].([]interface{})

tap/extensions/kafka/main.go

@@ -81,10 +81,8 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, resolvedSource string,
 		if len(summary) > 0 {
 			summary = summary[:len(summary)-2]
 		}
-		break
 	case ApiVersions:
 		summary = reqDetails["clientID"].(string)
-		break
 	case Produce:
 		_topics := reqDetails["payload"].(map[string]interface{})["topicData"]
 		if _topics == nil {
@@ -97,7 +95,6 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, resolvedSource string,
 		if len(summary) > 0 {
 			summary = summary[:len(summary)-2]
 		}
-		break
 	case Fetch:
 		_topics := reqDetails["payload"].(map[string]interface{})["topics"]
 		if _topics == nil {
@@ -110,7 +107,6 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, resolvedSource string,
 		if len(summary) > 0 {
 			summary = summary[:len(summary)-2]
 		}
-		break
 	case ListOffsets:
 		_topics := reqDetails["payload"].(map[string]interface{})["topics"]
 		if _topics == nil {
@@ -123,7 +119,6 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, resolvedSource string,
 		if len(summary) > 0 {
 			summary = summary[:len(summary)-2]
 		}
-		break
 	case CreateTopics:
 		topics := reqDetails["payload"].(map[string]interface{})["topics"].([]interface{})
 		for _, topic := range topics {
@@ -132,13 +127,11 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, resolvedSource string,
 		if len(summary) > 0 {
 			summary = summary[:len(summary)-2]
 		}
-		break
 	case DeleteTopics:
 		topicNames := reqDetails["topicNames"].([]string)
 		for _, name := range topicNames {
 			summary += fmt.Sprintf("%s, ", name)
 		}
-		break
 	}
 
 	request["url"] = summary
@@ -173,7 +166,7 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, resolvedSource string,
 
 func (d dissecting) Represent(request map[string]interface{}, response map[string]interface{}) (object []byte, bodySize int64, err error) {
 	bodySize = 0
-	representation := make(map[string]interface{}, 0)
+	representation := make(map[string]interface{})
 
 	apiKey := ApiKey(request["apiKey"].(float64))
 
@@ -183,31 +176,24 @@ func (d dissecting) Represent(request map[string]interface{}, response map[strin
 	case Metadata:
 		repRequest = representMetadataRequest(request)
 		repResponse = representMetadataResponse(response)
-		break
 	case ApiVersions:
 		repRequest = representApiVersionsRequest(request)
 		repResponse = representApiVersionsResponse(response)
-		break
 	case Produce:
 		repRequest = representProduceRequest(request)
 		repResponse = representProduceResponse(response)
-		break
 	case Fetch:
 		repRequest = representFetchRequest(request)
 		repResponse = representFetchResponse(response)
-		break
 	case ListOffsets:
 		repRequest = representListOffsetsRequest(request)
 		repResponse = representListOffsetsResponse(response)
-		break
 	case CreateTopics:
 		repRequest = representCreateTopicsRequest(request)
 		repResponse = representCreateTopicsResponse(response)
-		break
 	case DeleteTopics:
 		repRequest = representDeleteTopicsRequest(request)
 		repResponse = representDeleteTopicsResponse(response)
-		break
 	}
 
 	representation["request"] = repRequest

tap/extensions/kafka/matcher.go

@@ -26,9 +26,9 @@ func CreateResponseRequestMatcher() requestResponseMatcher {
 func (matcher *requestResponseMatcher) registerRequest(key string, request *Request) *RequestResponsePair {
 	if response, found := matcher.openMessagesMap.LoadAndDelete(key); found {
 		// Check for a situation that only occurs when a Kafka broker is initiating
-		switch response.(type) {
+		switch v := response.(type) {
 		case *Response:
-			return matcher.preparePair(request, response.(*Response))
+			return matcher.preparePair(request, v)
 		}
 	}
 

tap/extensions/kafka/protocol.go

@@ -167,10 +167,6 @@ type messageType struct {
 	encode   encodeFunc
 }
 
-func (t *messageType) new() Message {
-	return reflect.New(t.gotype).Interface().(Message)
-}
-
 type apiType struct {
 	requests  []messageType
 	responses []messageType
@@ -389,16 +385,16 @@ func (b Broker) String() string {
 func (b Broker) Format(w fmt.State, v rune) {
 	switch v {
 	case 'd':
-		io.WriteString(w, itoa(b.ID))
+		_, _ = io.WriteString(w, itoa(b.ID))
 	case 's':
-		io.WriteString(w, b.String())
+		_, _ = io.WriteString(w, b.String())
 	case 'v':
-		io.WriteString(w, itoa(b.ID))
-		io.WriteString(w, " ")
-		io.WriteString(w, b.String())
+		_, _ = io.WriteString(w, itoa(b.ID))
+		_, _ = io.WriteString(w, " ")
+		_, _ = io.WriteString(w, b.String())
 		if b.Rack != "" {
-			io.WriteString(w, " ")
-			io.WriteString(w, b.Rack)
+			_, _ = io.WriteString(w, " ")
+			_, _ = io.WriteString(w, b.Rack)
 		}
 	}
 }