TRA-4169 apply query by enter (#733)

* start dev commands

* env files

* no message

.github/workflows/pr_validation.yml

@@ -12,7 +12,7 @@ concurrency:
 
 jobs:
   build-cli:
-    name: Build CLI
+    name: Build CLI executable
     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: Build Agent docker image
     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

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.

agent/go.mod

@@ -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

@@ -9,6 +9,7 @@ import (
 	"mizuserver/pkg/api"
 	"mizuserver/pkg/config"
 	"mizuserver/pkg/controllers"
+	"mizuserver/pkg/elastic"
 	"mizuserver/pkg/middlewares"
 	"mizuserver/pkg/models"
 	"mizuserver/pkg/oas"
@@ -61,7 +62,6 @@ const (
 	socketConnectionRetries    = 30
 	socketConnectionRetryDelay = time.Second * 2
 	socketHandshakeTimeout     = time.Second * 2
-	uiIndexPath                = "./site/index.html"
 )
 
 func main() {
@@ -159,6 +159,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 +168,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 +247,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 +284,6 @@ func hostApi(socketHarOutputChannel chan<- *tapApi.OutputChannelItem) {
 	routes.EntriesRoutes(app)
 	routes.MetadataRoutes(app)
 	routes.StatusRoutes(app)
-	routes.NotFoundRoute(app)
 	utils.StartServer(app)
 }
 
@@ -291,7 +298,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,6 +5,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"mizuserver/pkg/elastic"
 	"mizuserver/pkg/har"
 	"mizuserver/pkg/holder"
 	"mizuserver/pkg/providers"
@@ -16,14 +17,15 @@ import (
 
 	"mizuserver/pkg/servicemap"
 
-	"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"
 
+	"github.com/up9inc/mizu/shared"
+	"github.com/up9inc/mizu/shared/logger"
+	tapApi "github.com/up9inc/mizu/tap/api"
+
 	basenine "github.com/up9inc/basenine/client/go"
 )
 
@@ -150,6 +152,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/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/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",
-			})
-		},
-	)
-}

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

@@ -74,6 +74,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)

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/config/configStruct.go

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

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

@@ -100,7 +100,6 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
 		http2Assembler = createHTTP2Assembler(b)
 	}
 
-	dissected := false
 	switchingProtocolsHTTP2 := false
 	for {
 		if switchingProtocolsHTTP2 {
@@ -121,7 +120,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 +129,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 +160,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
 }
 

tap/extensions/kafka/response.go

@@ -251,10 +251,10 @@ func ReadResponse(r io.Reader, tcpID *api.TcpID, superTimer *api.SuperTimer, emi
 	}
 
 	connectionInfo := &api.ConnectionInfo{
-		ClientIP:   tcpID.SrcIP,
-		ClientPort: tcpID.SrcPort,
-		ServerIP:   tcpID.DstIP,
-		ServerPort: tcpID.DstPort,
+		ClientIP:   tcpID.DstIP,
+		ClientPort: tcpID.DstPort,
+		ServerIP:   tcpID.SrcIP,
+		ServerPort: tcpID.SrcPort,
 		IsOutgoing: true,
 	}
 

ui/.env.basic

@@ -1,2 +1 @@
-REACT_APP_OVERRIDE_WS_URL="ws://localhost:8899/ws"
-REACT_APP_OVERRIDE_API_URL="http://localhost:8899/"
+REACT_APP_OVERRIDE_IS_ENTERPRISE="false"

ui/.env.dev.basic

@@ -0,0 +1,3 @@
+REACT_APP_OVERRIDE_WS_URL="ws://localhost:8899/ws"
+REACT_APP_OVERRIDE_API_URL="http://localhost:8899/"
+REACT_APP_OVERRIDE_IS_ENTERPRISE="false"

ui/.env.dev.enterprise

@@ -0,0 +1,3 @@
+REACT_APP_OVERRIDE_WS_URL="ws://localhost:8899/ws"
+REACT_APP_OVERRIDE_API_URL="http://localhost:8899/"
+REACT_APP_OVERRIDE_IS_ENTERPRISE="true"

ui/.env.enterprise

@@ -1,3 +1 @@
-REACT_APP_OVERRIDE_WS_URL="ws://localhost:8899/ws"
-REACT_APP_OVERRIDE_API_URL="http://localhost:8899/"
 REACT_APP_OVERRIDE_IS_ENTERPRISE="true"

ui/package.json

@@ -49,7 +49,8 @@
   },
   "scripts": {
     "start": "craco start",
-    "start-ent": "./node_modules/.bin/env-cmd -f .env.enterprise craco start",
+    "start-dev": "./node_modules/.bin/env-cmd -f .env.dev.basic craco start",
+    "start-dev-ent": "./node_modules/.bin/env-cmd -f .env.dev.enterprise craco start",
     "build": "./node_modules/.bin/env-cmd -f .env.basic craco build",
     "build-ent": "BUILD_PATH='./build-ent' ./node_modules/.bin/env-cmd -f .env.enterprise craco build",
     "test": "craco test",

ui/src/components/EntryDetailed/EntrySections.tsx

@@ -137,12 +137,12 @@ export const EntryBodySection: React.FC<EntryBodySectionProps> = ({
         const chunk = body.slice(0, MAXIMUM_BYTES_TO_FORMAT);
         const bodyBuf = isBase64Encoding ? atob(chunk) : chunk;
 
-        if (!isPretty) return bodyBuf;
-
         try {
             if (jsonLikeFormats.some(format => contentType?.indexOf(format) > -1)) {
+                if (!isPretty) return bodyBuf;
                 return jsonBeautify(JSON.parse(bodyBuf), null, 2, 80);
             }  else if (xmlLikeFormats.some(format => contentType?.indexOf(format) > -1)) {
+                if (!isPretty) return bodyBuf;
                 return xmlBeautify(bodyBuf, {
                     indentation: '  ',
                     filter: (node) => node.type !== 'Comment',
@@ -152,7 +152,9 @@ export const EntryBodySection: React.FC<EntryBodySectionProps> = ({
             } else if (protobufFormats.some(format => contentType?.indexOf(format) > -1)) {
                 // Replace all non printable characters (ASCII)
                 const protobufDecoder = new ProtobufDecoder(bodyBuf, true);
-                return jsonBeautify(protobufDecoder.decode().toSimple(), null, 2, 80);
+                const protobufDecoded = protobufDecoder.decode().toSimple();
+                if (!isPretty) return JSON.stringify(protobufDecoded);
+                return jsonBeautify(protobufDecoded, null, 2, 80);
             }
         } catch (error) {
             console.error(error);

ui/src/components/OasModal/OasModal.tsx

@@ -21,7 +21,6 @@ const OasModal = ({ openModal, handleCloseModal }) => {
         const services = await api.getOasServices();
         setOasServices(services);
       } catch (e) {
-        toast.error("Error occurred while fetching services list");
         console.error(e);
       }
     })();

ui/src/helpers/api.js

@@ -1,14 +1,12 @@
 import * as axios from "axios";
 
-// When working locally cp `cp .env.example .env`
 export const MizuWebsocketURL = process.env.REACT_APP_OVERRIDE_WS_URL ? process.env.REACT_APP_OVERRIDE_WS_URL :
-                        window.location.protocol === 'https:' ? `wss://${window.location.host}/ws` : `ws://${window.location.host}/ws`;
+    window.location.protocol === 'https:' ? `wss://${window.location.host}/ws` : `ws://${window.location.host}/ws`;
 
 export const FormValidationErrorType = "formError";
 
 const CancelToken = axios.CancelToken;
 
-// When working locally cp `cp .env.example .env`
 const apiURL = process.env.REACT_APP_OVERRIDE_API_URL ? process.env.REACT_APP_OVERRIDE_API_URL : `${window.location.origin}/`;
 
 export default class Api {

ui/src/hooks/useKeyPress.ts

@@ -10,8 +10,11 @@ const useKeyPress = (eventConfigs, callback, node = null) => {
   // handle what happens on key press
   const handleKeyPress = useCallback(
     (event) => {
+      
       // check if one of the key is part of the ones we want
       if (eventConfigs.some((eventConfig) => Object.keys(eventConfig).every(nameKey => eventConfig[nameKey] === event[nameKey]))) {
+        event.stopPropagation()
+        event.preventDefault();
         callbackRef.current(event);
       }
     },