Hide Encoding field if it's undefined or empty in the UI (#511)

* Add HTTP2 Over Cleartext (H2C) support

* Remove a parameter which is a remnant of debugging

.gitignore

@@ -32,3 +32,6 @@ pprof/*
 # Database Files
 *.bin
 *.gob
+
+# Nohup Files - https://man7.org/linux/man-pages/man1/nohup.1p.html
+nohup.*

Dockerfile

@@ -42,8 +42,8 @@ RUN go build -ldflags="-s -w \
      -X 'mizuserver/pkg/version.SemVer=${SEM_VER}'" -o mizuagent .
 
 # Download Basenine executable, verify the sha1sum and move it to a directory in $PATH
-ADD https://github.com/up9inc/basenine/releases/download/v0.2.10/basenine_linux_amd64 ./basenine_linux_amd64
-ADD https://github.com/up9inc/basenine/releases/download/v0.2.10/basenine_linux_amd64.sha256 ./basenine_linux_amd64.sha256
+ADD https://github.com/up9inc/basenine/releases/download/v0.2.11/basenine_linux_amd64 ./basenine_linux_amd64
+ADD https://github.com/up9inc/basenine/releases/download/v0.2.11/basenine_linux_amd64.sha256 ./basenine_linux_amd64.sha256
 RUN shasum -a 256 -c basenine_linux_amd64.sha256
 RUN chmod +x ./basenine_linux_amd64
 

README.md

@@ -4,7 +4,7 @@
 
 A simple-yet-powerful API traffic viewer for Kubernetes enabling you to view all API communication between microservices to help your debug and troubleshoot regressions.
 
-Think TCPDump and Chrome Dev Tools combined.
+Think TCPDump and Wireshark re-invented for Kubernetes.
 
 ![Simple UI](assets/mizu-ui.png)
 
@@ -12,13 +12,13 @@ Think TCPDump and Chrome Dev Tools combined.
 
 - Simple and powerful CLI
 - Monitoring network traffic in real-time. Supported protocols:
-  - [HTTP/1.1](https://datatracker.ietf.org/doc/html/rfc2616)
+  - [HTTP/1.1](https://datatracker.ietf.org/doc/html/rfc2616) (REST, etc.)
   - [HTTP/2](https://datatracker.ietf.org/doc/html/rfc7540) (gRPC)
   - [AMQP](https://www.rabbitmq.com/amqp-0-9-1-reference.html) (RabbitMQ, Apache Qpid, etc.)
   - [Apache Kafka](https://kafka.apache.org/protocol)
   - [Redis](https://redis.io/topics/protocol)
-- No installation or code instrumentation
-- Works completely on premises
+- Works with Kubernetes APIs. No installation or code instrumentation
+- Rich filtering
 
 ## Requirements
 
@@ -185,15 +185,3 @@ Mizu can be run detached from the cli using the daemon flag: `mizu tap --daemon`
 indefinitely in the cluster.
 
 For more information please refer to [DAEMON MODE](docs/DAEMON_MODE.md)
-
-## How to Run local UI
-
-- run from mizu/agent `go run main.go --hars-read --hars-dir <folder>`
-
-- copy Har files into the folder from last command
-
-- change `MizuWebsocketURL` and `apiURL` in `api.js` file
-
-- run from mizu/ui - `npm run start`
-
-- open browser on `localhost:3000`

agent/go.mod

@@ -16,7 +16,7 @@ require (
 	github.com/op/go-logging v0.0.0-20160315200505-970db520ece7
 	github.com/orcaman/concurrent-map v0.0.0-20210106121528-16402b402231
 	github.com/patrickmn/go-cache v2.1.0+incompatible
-	github.com/up9inc/basenine/client/go v0.0.0-20211118123155-7ed075f85c73
+	github.com/up9inc/basenine/client/go v0.0.0-20211121072216-04366911881c
 	github.com/up9inc/mizu/shared v0.0.0
 	github.com/up9inc/mizu/tap v0.0.0
 	github.com/up9inc/mizu/tap/api v0.0.0

agent/go.sum

@@ -450,8 +450,8 @@ github.com/ugorji/go v1.1.7 h1:/68gy2h+1mWMrwZFeD1kQialdSzAb432dtpeJ42ovdo=
 github.com/ugorji/go v1.1.7/go.mod h1:kZn38zHttfInRq0xu/PH0az30d+z6vm202qpg1oXVMw=
 github.com/ugorji/go/codec v1.1.7 h1:2SvQaVZ1ouYrrKKwoSk2pzd4A9evlKJb9oTL+OaLUSs=
 github.com/ugorji/go/codec v1.1.7/go.mod h1:Ax+UKWsSmolVDwsd+7N3ZtXu+yMGCf907BLYF3GoBXY=
-github.com/up9inc/basenine/client/go v0.0.0-20211118123155-7ed075f85c73 h1:FJUM7w7E0jRGFPcSMa7cVy+jr5zcpbyT6qA30dEtGGI=
-github.com/up9inc/basenine/client/go v0.0.0-20211118123155-7ed075f85c73/go.mod h1:SvJGPoa/6erhUQV7kvHBwM/0x5LyO6XaG2lUaCaKiUI=
+github.com/up9inc/basenine/client/go v0.0.0-20211121072216-04366911881c h1:GJsCVhDKjV/k3mNG255VN7hAQ7fxyNgX5T+VJyzoOQ0=
+github.com/up9inc/basenine/client/go v0.0.0-20211121072216-04366911881c/go.mod h1:SvJGPoa/6erhUQV7kvHBwM/0x5LyO6XaG2lUaCaKiUI=
 github.com/vektah/gqlparser v1.1.2/go.mod h1:1ycwN7Ij5njmMkPPAOaRFY4rET2Enx7IkVv3vaXspKw=
 github.com/vishvananda/netns v0.0.0-20210104183010-2eb08e3e575f h1:p4VB7kIXpOQvVn1ZaTIVp+3vuYAXFe3OJEvjbUYJLaA=
 github.com/vishvananda/netns v0.0.0-20210104183010-2eb08e3e575f/go.mod h1:DD4vA1DwXk04H54A1oHXtwZmA0grkVMdPxx/VGLCah0=

agent/main.go

@@ -22,6 +22,7 @@ import (
 	"path"
 	"path/filepath"
 	"plugin"
+	"regexp"
 	"sort"
 	"syscall"
 	"time"
@@ -264,9 +265,16 @@ func hostApi(socketHarOutputChannel chan<- *tapApi.OutputChannelItem) {
 		ctx, cancel := context.WithCancel(context.Background())
 		defer cancel()
 
-		if _, err := startMizuTapperSyncer(ctx); err != nil {
+		kubernetesProvider, err := kubernetes.NewProviderInCluster()
+		if err != nil {
+			logger.Log.Fatalf("error creating k8s provider: %+v", err)
+		}
+
+		if _, err := startMizuTapperSyncer(ctx, kubernetesProvider); err != nil {
 			logger.Log.Fatalf("error initializing tapper syncer: %+v", err)
 		}
+
+		go watchMizuEvents(ctx, kubernetesProvider, cancel) 
 	}
 
 	utils.StartServer(app)
@@ -426,12 +434,7 @@ func dialSocketWithRetry(socketAddress string, retryAmount int, retryDelay time.
 	return nil, lastErr
 }
 
-func startMizuTapperSyncer(ctx context.Context) (*kubernetes.MizuTapperSyncer, error) {
-	provider, err := kubernetes.NewProviderInCluster()
-	if err != nil {
-		return nil, err
-	}
-
+func startMizuTapperSyncer(ctx context.Context, provider *kubernetes.Provider) (*kubernetes.MizuTapperSyncer, error) {
 	tapperSyncer, err := kubernetes.CreateAndStartMizuTapperSyncer(ctx, provider, kubernetes.TapperSyncerConfig{
 		TargetNamespaces:         config.Config.TargetNamespaces,
 		PodFilterRegex:           config.Config.TapTargetRegex.Regexp,
@@ -483,3 +486,48 @@ func startMizuTapperSyncer(ctx context.Context) (*kubernetes.MizuTapperSyncer, e
 
 	return tapperSyncer, nil
 }
+
+func watchMizuEvents(ctx context.Context, kubernetesProvider *kubernetes.Provider, cancel context.CancelFunc) {
+	// Round down because k8s CreationTimestamp is given in 1 sec resolution.
+	startTime := time.Now().Truncate(time.Second)
+
+	mizuResourceRegex := regexp.MustCompile(fmt.Sprintf("^%s.*", kubernetes.MizuResourcesPrefix))
+	eventWatchHelper := kubernetes.NewEventWatchHelper(kubernetesProvider, mizuResourceRegex)
+	eventChan, errorChan := kubernetes.FilteredWatch(ctx, eventWatchHelper, []string{config.Config.MizuResourcesNamespace}, eventWatchHelper)
+
+	for {
+		select {
+		case wEvent, ok := <-eventChan:
+			if !ok {
+				eventChan = nil
+				continue
+			}
+
+			event, err := wEvent.ToEvent()
+			if err != nil {
+				logger.Log.Errorf("error parsing Mizu resource event: %+v", err)
+				cancel()
+			}
+
+			if startTime.After(event.CreationTimestamp.Time) {
+				continue
+			}
+
+			if event.Type == v1.EventTypeWarning {
+				logger.Log.Warningf("resource %s in state %s - %s", event.Regarding.Name, event.Reason, event.Note)
+			}
+		case err, ok := <-errorChan:
+			if !ok {
+				errorChan = nil
+				continue
+			}
+
+			logger.Log.Errorf("error in watch mizu resource events loop: %+v", err)
+			cancel()
+
+		case <-ctx.Done():
+			logger.Log.Debugf("watching Mizu resource events loop, ctx done")
+			return
+		}
+	}
+}

agent/pkg/controllers/entries_controller.go

@@ -55,7 +55,7 @@ func GetEntry(c *gin.Context) {
 	}
 
 	extension := extensionsMap[entry.Protocol.Name]
-	protocol, representation, bodySize, _ := extension.Dissector.Represent(entry.Protocol, entry.Request, entry.Response)
+	representation, bodySize, _ := extension.Dissector.Represent(entry.Request, entry.Response)
 
 	var rules []map[string]interface{}
 	var isRulesEnabled bool
@@ -68,7 +68,7 @@ func GetEntry(c *gin.Context) {
 	}
 
 	c.JSON(http.StatusOK, tapApi.MizuEntryWrapper{
-		Protocol:       protocol,
+		Protocol:       entry.Protocol,
 		Representation: string(representation),
 		BodySize:       bodySize,
 		Data:           entry,

cli/apiserver/provider.go

@@ -3,11 +3,12 @@ package apiserver
 import (
 	"bytes"
 	"encoding/json"
-	"errors"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"net/http"
 	"net/url"
+	"strings"
 	"time"
 
 	"github.com/up9inc/mizu/shared/kubernetes"
@@ -61,7 +62,14 @@ func (provider *Provider) GetHealthStatus() (*shared.HealthResponse, error) {
 	if response, err := provider.client.Get(healthUrl); err != nil {
 		return nil, err
 	} else if response.StatusCode > 299 {
-		return nil, errors.New(fmt.Sprintf("status code: %d", response.StatusCode))
+		responseBody := new(strings.Builder)
+
+		if _, err := io.Copy(responseBody, response.Body); err != nil {
+			return nil, fmt.Errorf("status code: %d - (bad response - %v)", response.StatusCode, err)
+		} else {
+			singleLineResponse := strings.ReplaceAll(responseBody.String(), "\n", "")
+			return nil, fmt.Errorf("status code: %d - (response - %v)", response.StatusCode, singleLineResponse)
+		}
 	} else {
 		defer response.Body.Close()
 

cli/cmd/tapRunner.go

@@ -42,6 +42,8 @@ var state tapState
 var apiProvider *apiserver.Provider
 
 func RunMizuTap() {
+	startTime := time.Now()
+
 	mizuApiFilteringOptions, err := getMizuApiFilteringOptions()
 	apiProvider = apiserver.NewProvider(GetApiServerUrl(), apiserver.DefaultRetries, apiserver.DefaultTimeout)
 	if err != nil {
@@ -150,6 +152,7 @@ func RunMizuTap() {
 
 		go goUtils.HandleExcWrapper(watchApiServerPod, ctx, kubernetesProvider, cancel)
 		go goUtils.HandleExcWrapper(watchTapperPod, ctx, kubernetesProvider, cancel)
+		go goUtils.HandleExcWrapper(watchMizuEvents, ctx, kubernetesProvider, cancel, startTime)
 
 		// block until exit signal or error
 		waitForFinish(ctx, cancel)
@@ -556,76 +559,73 @@ func waitUntilNamespaceDeleted(ctx context.Context, cancel context.CancelFunc, k
 func watchApiServerPod(ctx context.Context, kubernetesProvider *kubernetes.Provider, cancel context.CancelFunc) {
 	podExactRegex := regexp.MustCompile(fmt.Sprintf("^%s$", kubernetes.ApiServerPodName))
 	podWatchHelper := kubernetes.NewPodWatchHelper(kubernetesProvider, podExactRegex)
-	added, modified, removed, errorChan := kubernetes.FilteredWatch(ctx, podWatchHelper, []string{config.Config.MizuResourcesNamespace}, podWatchHelper)
+	eventChan, errorChan := kubernetes.FilteredWatch(ctx, podWatchHelper, []string{config.Config.MizuResourcesNamespace}, podWatchHelper)
 	isPodReady := false
 	timeAfter := time.After(25 * time.Second)
 	for {
 		select {
-		case _, ok := <-added:
+		case wEvent, ok := <-eventChan:
 			if !ok {
-				added = nil
+				eventChan = nil
 				continue
 			}
 
-			logger.Log.Debugf("Watching API Server pod loop, added")
-		case _, ok := <-removed:
-			if !ok {
-				removed = nil
-				continue
-			}
-
-			logger.Log.Infof("%s removed", kubernetes.ApiServerPodName)
-			cancel()
-			return
-		case wEvent, ok := <-modified:
-			if !ok {
-				modified = nil
-				continue
-			}
-
-			modifiedPod, err := wEvent.ToPod()
-			if err != nil {
-				logger.Log.Errorf(uiUtils.Error, err)
+			switch wEvent.Type {
+			case kubernetes.EventAdded:
+				logger.Log.Debugf("Watching API Server pod loop, added")
+			case kubernetes.EventDeleted:
+				logger.Log.Infof("%s removed", kubernetes.ApiServerPodName)
 				cancel()
-				continue
-			}
-
-			logger.Log.Debugf("Watching API Server pod loop, modified: %v", modifiedPod.Status.Phase)
-
-			if modifiedPod.Status.Phase == core.PodPending {
-				if modifiedPod.Status.Conditions[0].Type == core.PodScheduled && modifiedPod.Status.Conditions[0].Status != core.ConditionTrue {
-					logger.Log.Debugf("Wasn't able to deploy the API server. Reason: \"%s\"", modifiedPod.Status.Conditions[0].Message)
-					logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Wasn't able to deploy the API server, for more info check logs at %s", fsUtils.GetLogFilePath()))
+				return
+			case kubernetes.EventModified:
+				modifiedPod, err := wEvent.ToPod()
+				if err != nil {
+					logger.Log.Errorf(uiUtils.Error, err)
 					cancel()
-					break
+					continue
 				}
 
-				if len(modifiedPod.Status.ContainerStatuses) > 0 && modifiedPod.Status.ContainerStatuses[0].State.Waiting != nil && modifiedPod.Status.ContainerStatuses[0].State.Waiting.Reason == "ErrImagePull" {
-					logger.Log.Debugf("Wasn't able to deploy the API server. (ErrImagePull) Reason: \"%s\"", modifiedPod.Status.ContainerStatuses[0].State.Waiting.Message)
-					logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Wasn't able to deploy the API server: failed to pull the image, for more info check logs at %v", fsUtils.GetLogFilePath()))
-					cancel()
-					break
-				}
-			}
+				logger.Log.Debugf("Watching API Server pod loop, modified: %v", modifiedPod.Status.Phase)
 
-			if modifiedPod.Status.Phase == core.PodRunning && !isPodReady {
-				isPodReady = true
-				go startProxyReportErrorIfAny(kubernetesProvider, cancel)
+				if modifiedPod.Status.Phase == core.PodPending {
+					if modifiedPod.Status.Conditions[0].Type == core.PodScheduled && modifiedPod.Status.Conditions[0].Status != core.ConditionTrue {
+						logger.Log.Debugf("Wasn't able to deploy the API server. Reason: \"%s\"", modifiedPod.Status.Conditions[0].Message)
+						logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Wasn't able to deploy the API server, for more info check logs at %s", fsUtils.GetLogFilePath()))
+						cancel()
+						break
+					}
 
-				url := GetApiServerUrl()
-				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()
-					break
+					if len(modifiedPod.Status.ContainerStatuses) > 0 && modifiedPod.Status.ContainerStatuses[0].State.Waiting != nil && modifiedPod.Status.ContainerStatuses[0].State.Waiting.Reason == "ErrImagePull" {
+						logger.Log.Debugf("Wasn't able to deploy the API server. (ErrImagePull) Reason: \"%s\"", modifiedPod.Status.ContainerStatuses[0].State.Waiting.Message)
+						logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Wasn't able to deploy the API server: failed to pull the image, for more info check logs at %v", fsUtils.GetLogFilePath()))
+						cancel()
+						break
+					}
 				}
 
-				logger.Log.Infof("Mizu is available at %s", url)
-				if !config.Config.HeadlessMode {
-					uiUtils.OpenBrowser(url)
-				}
-				if err := apiProvider.ReportTappedPods(state.tapperSyncer.CurrentlyTappedPods); err != nil {
-					logger.Log.Debugf("[Error] failed update tapped pods %v", err)
+				if modifiedPod.Status.Phase == core.PodRunning && !isPodReady {
+					isPodReady = true
+					go startProxyReportErrorIfAny(kubernetesProvider, cancel)
+
+					url := GetApiServerUrl()
+					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()
+						break
+					}
+
+					logger.Log.Infof("Mizu is available at %s", url)
+					if !config.Config.HeadlessMode {
+						uiUtils.OpenBrowser(url)
+					}
+					if err := apiProvider.ReportTappedPods(state.tapperSyncer.CurrentlyTappedPods); err != nil {
+						logger.Log.Debugf("[Error] failed update tapped pods %v", err)
+					}
 				}
+			case kubernetes.EventBookmark:
+				break
+			case kubernetes.EventError:
+				break
 			}
 		case err, ok := <-errorChan:
 			if !ok {
@@ -651,83 +651,60 @@ func watchApiServerPod(ctx context.Context, kubernetesProvider *kubernetes.Provi
 func watchTapperPod(ctx context.Context, kubernetesProvider *kubernetes.Provider, cancel context.CancelFunc) {
 	podExactRegex := regexp.MustCompile(fmt.Sprintf("^%s.*", kubernetes.TapperDaemonSetName))
 	podWatchHelper := kubernetes.NewPodWatchHelper(kubernetesProvider, podExactRegex)
-	added, modified, removed, errorChan := kubernetes.FilteredWatch(ctx, podWatchHelper, []string{config.Config.MizuResourcesNamespace}, podWatchHelper)
-	var prevPodPhase core.PodPhase
+	eventChan, errorChan := kubernetes.FilteredWatch(ctx, podWatchHelper, []string{config.Config.MizuResourcesNamespace}, podWatchHelper)
+
 	for {
 		select {
-		case wEvent, ok := <-added:
+		case wEvent, ok := <-eventChan:
 			if !ok {
-				added = nil
+				eventChan = nil
 				continue
 			}
 
-			addedPod, err := wEvent.ToPod()
+			pod, err := wEvent.ToPod()
 			if err != nil {
 				logger.Log.Errorf(uiUtils.Error, err)
 				cancel()
 				continue
 			}
 
-			logger.Log.Debugf("Tapper is created [%s]", addedPod.Name)
-		case wEvent, ok := <-removed:
-			if !ok {
-				removed = nil
-				continue
-			}
+			switch wEvent.Type {
+			case kubernetes.EventAdded:
+				logger.Log.Debugf("Tapper is created [%s]", pod.Name)
+			case kubernetes.EventDeleted:
+				logger.Log.Debugf("Tapper is removed [%s]", pod.Name)
+			case kubernetes.EventModified:
+				if pod.Status.Phase == core.PodPending && pod.Status.Conditions[0].Type == core.PodScheduled && pod.Status.Conditions[0].Status != core.ConditionTrue {
+					logger.Log.Infof(uiUtils.Red, fmt.Sprintf("Wasn't able to deploy the tapper %s. Reason: \"%s\"", pod.Name, pod.Status.Conditions[0].Message))
+					cancel()
+					continue
+				}
 
-			removedPod, err := wEvent.ToPod()
-			if err != nil {
-				logger.Log.Errorf(uiUtils.Error, err)
-				cancel()
-				continue
-			}
+				podStatus := pod.Status
 
-
-			logger.Log.Debugf("Tapper is removed [%s]", removedPod.Name)
-		case wEvent, ok := <-modified:
-			if !ok {
-				modified = nil
-				continue
-			}
-
-			modifiedPod, err := wEvent.ToPod()
-			if err != nil {
-				logger.Log.Errorf(uiUtils.Error, err)
-				cancel()
-				continue
-			}
-
-			if modifiedPod.Status.Phase == core.PodPending && modifiedPod.Status.Conditions[0].Type == core.PodScheduled && modifiedPod.Status.Conditions[0].Status != core.ConditionTrue {
-				logger.Log.Infof(uiUtils.Red, fmt.Sprintf("Wasn't able to deploy the tapper %s. Reason: \"%s\"", modifiedPod.Name, modifiedPod.Status.Conditions[0].Message))
-				cancel()
-				continue
-			}
-
-			podStatus := modifiedPod.Status
-			if podStatus.Phase == core.PodPending && prevPodPhase == podStatus.Phase {
-				logger.Log.Debugf("Tapper %s is %s", modifiedPod.Name, strings.ToLower(string(podStatus.Phase)))
-				continue
-			}
-			prevPodPhase = podStatus.Phase
-
-			if podStatus.Phase == core.PodRunning {
-				state := podStatus.ContainerStatuses[0].State
-				if state.Terminated != nil {
-					switch state.Terminated.Reason {
-					case "OOMKilled":
-						logger.Log.Infof(uiUtils.Red, fmt.Sprintf("Tapper %s was terminated (reason: OOMKilled). You should consider increasing machine resources.", modifiedPod.Name))
+				if podStatus.Phase == core.PodRunning {
+					state := podStatus.ContainerStatuses[0].State
+					if state.Terminated != nil {
+						switch state.Terminated.Reason {
+						case "OOMKilled":
+							logger.Log.Infof(uiUtils.Red, fmt.Sprintf("Tapper %s was terminated (reason: OOMKilled). You should consider increasing machine resources.", pod.Name))
+						}
 					}
 				}
-			}
 
-			logger.Log.Debugf("Tapper %s is %s", modifiedPod.Name, strings.ToLower(string(podStatus.Phase)))
+				logger.Log.Debugf("Tapper %s is %s", pod.Name, strings.ToLower(string(podStatus.Phase)))
+			case kubernetes.EventBookmark:
+				break
+			case kubernetes.EventError:
+				break
+			}
 		case err, ok := <-errorChan:
 			if !ok {
 				errorChan = nil
 				continue
 			}
 
-			logger.Log.Errorf("[Error] Error in mizu tapper watch, err: %v", err)
+			logger.Log.Errorf("[Error] Error in mizu tapper pod watch, err: %v", err)
 			cancel()
 
 		case <-ctx.Done():
@@ -737,6 +714,51 @@ func watchTapperPod(ctx context.Context, kubernetesProvider *kubernetes.Provider
 	}
 }
 
+func watchMizuEvents(ctx context.Context, kubernetesProvider *kubernetes.Provider, cancel context.CancelFunc, startTime time.Time) {
+	// Round down because k8s CreationTimestamp is given in 1 sec resolution.
+	startTime = startTime.Truncate(time.Second)
+
+	mizuResourceRegex := regexp.MustCompile(fmt.Sprintf("^%s.*", kubernetes.MizuResourcesPrefix))
+	eventWatchHelper := kubernetes.NewEventWatchHelper(kubernetesProvider, mizuResourceRegex)
+	eventChan, errorChan := kubernetes.FilteredWatch(ctx, eventWatchHelper, []string{config.Config.MizuResourcesNamespace}, eventWatchHelper)
+
+	for {
+		select {
+		case wEvent, ok := <-eventChan:
+			if !ok {
+				eventChan = nil
+				continue
+			}
+
+			event, err := wEvent.ToEvent()
+			if err != nil {
+				logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("error parsing Mizu resource event: %+v", err))
+				cancel()
+			}
+
+			if startTime.After(event.CreationTimestamp.Time) {
+				continue
+			}
+
+			if event.Type == core.EventTypeWarning {
+				logger.Log.Warningf(uiUtils.Warning, fmt.Sprintf("Resource %s in state %s - %s", event.Regarding.Name, event.Reason, event.Note))
+			}
+		case err, ok := <-errorChan:
+			if !ok {
+				errorChan = nil
+				continue
+			}
+
+			logger.Log.Errorf("error in watch mizu resource events loop: %+v", err)
+			cancel()
+
+		case <-ctx.Done():
+			logger.Log.Debugf("watching Mizu resource events loop, ctx done")
+			return
+		}
+	}
+}
+
 func getNamespaces(kubernetesProvider *kubernetes.Provider) []string {
 	if config.Config.Tap.AllNamespaces {
 		return []string{kubernetes.K8sAllNamespaces}

debug.Dockerfile

@@ -37,8 +37,8 @@ COPY agent .
 RUN go build -gcflags="all=-N -l" -o mizuagent .
 
 # Download Basenine executable, verify the sha1sum and move it to a directory in $PATH
-ADD https://github.com/up9inc/basenine/releases/download/v0.2.10/basenine_linux_amd64 ./basenine_linux_amd64
-ADD https://github.com/up9inc/basenine/releases/download/v0.2.10/basenine_linux_amd64.sha256 ./basenine_linux_amd64.sha256
+ADD https://github.com/up9inc/basenine/releases/download/v0.2.11/basenine_linux_amd64 ./basenine_linux_amd64
+ADD https://github.com/up9inc/basenine/releases/download/v0.2.11/basenine_linux_amd64.sha256 ./basenine_linux_amd64.sha256
 RUN shasum -a 256 -c basenine_linux_amd64.sha256
 RUN chmod +x ./basenine_linux_amd64
 

examples/roles/permissions-all-namespaces-daemon.yaml

@@ -49,6 +49,9 @@ rules:
   - apiGroups: ["", "apps", "extensions"]
     resources: ["endpoints"]
     verbs: ["get", "list", "watch"]
+  - apiGroups: ["events.k8s.io"]
+    resources: ["events"]
+    verbs: ["list", "watch"]
 ---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1

examples/roles/permissions-all-namespaces-without-ip-resolution.yaml

@@ -23,6 +23,9 @@ rules:
 - apiGroups: [""]
   resources: ["configmaps"]
   verbs: ["get", "create", "delete"]
+  - apiGroups: ["events.k8s.io"]
+    resources: ["events"]
+    verbs: ["list", "watch"]
 ---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1

examples/roles/permissions-all-namespaces.yaml

@@ -46,6 +46,9 @@ rules:
 - apiGroups: ["", "apps", "extensions"]
   resources: ["endpoints"]
   verbs: ["get", "list", "watch"]
+- apiGroups: ["events.k8s.io"]
+  resources: ["events"]
+  verbs: ["list", "watch"]
 ---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1

examples/roles/permissions-ns-daemon.yaml

@@ -41,6 +41,9 @@ rules:
 - apiGroups: ["", "apps", "extensions"]
   resources: ["endpoints"]
   verbs: ["get", "list", "watch"]
+- apiGroups: ["events.k8s.io"]
+  resources: ["events"]
+  verbs: ["list", "watch"]
 ---
 kind: RoleBinding
 apiVersion: rbac.authorization.k8s.io/v1

examples/roles/permissions-ns-with-validation.yaml

@@ -38,6 +38,9 @@ rules:
 - apiGroups: ["", "apps", "extensions"]
   resources: ["endpoints"]
   verbs: ["get", "list", "watch"]
+- apiGroups: ["events.k8s.io"]
+  resources: ["events"]
+  verbs: ["list", "watch"]
 ---
 kind: RoleBinding
 apiVersion: rbac.authorization.k8s.io/v1

examples/roles/permissions-ns-without-ip-resolution.yaml

@@ -20,6 +20,9 @@ rules:
 - apiGroups: [""]
   resources: ["configmaps"]
   verbs: ["get", "create", "delete"]
+- apiGroups: ["events.k8s.io"]
+  resources: ["events"]
+  verbs: ["list", "watch"]
 ---
 kind: RoleBinding
 apiVersion: rbac.authorization.k8s.io/v1

examples/roles/permissions-ns.yaml

@@ -38,6 +38,9 @@ rules:
 - apiGroups: ["", "apps", "extensions"]
   resources: ["endpoints"]
   verbs: ["get", "list", "watch"]
+- apiGroups: ["events.k8s.io"]
+  resources: ["events"]
+  verbs: ["list", "watch"]
 ---
 kind: RoleBinding
 apiVersion: rbac.authorization.k8s.io/v1

shared/kubernetes/eventWatchHelper.go

@@ -0,0 +1,45 @@
+package kubernetes
+
+import (
+	"context"
+	"regexp"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/watch"
+)
+
+type EventWatchHelper struct {
+	kubernetesProvider *Provider
+	NameRegexFilter *regexp.Regexp
+}
+
+func NewEventWatchHelper(kubernetesProvider *Provider, NameRegexFilter *regexp.Regexp) *EventWatchHelper {
+	return &EventWatchHelper{
+		kubernetesProvider: kubernetesProvider,
+		NameRegexFilter: NameRegexFilter,
+	}
+}
+
+// Implements the EventFilterer Interface
+func (wh *EventWatchHelper) Filter(wEvent *WatchEvent) (bool, error) {
+	event, err := wEvent.ToEvent()
+	if err != nil {
+		return false, nil
+	}
+
+	if !wh.NameRegexFilter.MatchString(event.Name) {
+		return false, nil
+	}
+
+	return true, nil
+}
+
+// Implements the WatchCreator Interface
+func (wh *EventWatchHelper) NewWatcher(ctx context.Context, namespace string) (watch.Interface, error) {
+	watcher, err := wh.kubernetesProvider.clientSet.EventsV1().Events(namespace).Watch(ctx, metav1.ListOptions{Watch: true})
+	if err != nil {
+		return nil, err
+	}
+
+	return watcher, nil
+}

shared/kubernetes/mizuTapperSyncer.go

@@ -70,7 +70,7 @@ func CreateAndStartMizuTapperSyncer(ctx context.Context, kubernetesProvider *Pro
 
 func (tapperSyncer *MizuTapperSyncer) watchPodsForTapping() {
 	podWatchHelper := NewPodWatchHelper(tapperSyncer.kubernetesProvider, &tapperSyncer.config.PodFilterRegex)
-	added, modified, removed, errorChan := FilteredWatch(tapperSyncer.context, podWatchHelper, tapperSyncer.config.TargetNamespaces, podWatchHelper)
+	eventChan, errorChan := FilteredWatch(tapperSyncer.context, podWatchHelper, tapperSyncer.config.TargetNamespaces, podWatchHelper)
 
 	restartTappers := func() {
 		err, changeFound := tapperSyncer.updateCurrentlyTappedPods()
@@ -96,9 +96,9 @@ func (tapperSyncer *MizuTapperSyncer) watchPodsForTapping() {
 
 	for {
 		select {
-		case wEvent, ok := <-added:
+		case wEvent, ok := <-eventChan:
 			if !ok {
-				added = nil
+				eventChan = nil
 				continue
 			}
 
@@ -109,44 +109,28 @@ func (tapperSyncer *MizuTapperSyncer) watchPodsForTapping() {
 			}
 
 
-			logger.Log.Debugf("Added matching pod %s, ns: %s", pod.Name, pod.Namespace)
-			restartTappersDebouncer.SetOn()
-		case wEvent, ok := <-removed:
-			if !ok {
-				removed = nil
-				continue
-			}
-
-			pod, err := wEvent.ToPod()
-			if err != nil {
-				tapperSyncer.handleErrorInWatchLoop(err, restartTappersDebouncer)
-				continue
-			}
-
-			logger.Log.Debugf("Removed matching pod %s, ns: %s", pod.Name, pod.Namespace)
-			restartTappersDebouncer.SetOn()
-		case wEvent, ok := <-modified:
-			if !ok {
-				modified = nil
-				continue
-			}
-
-			pod, err := wEvent.ToPod()
-			if err != nil {
-				tapperSyncer.handleErrorInWatchLoop(err, restartTappersDebouncer)
-				continue
-			}
-
-
-			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.
-			// After filtering for IPs, on a normal pod restart this includes the following events:
-			// - Pod deletion
-			// - Pod reaches start state
-			// - Pod reaches ready state
-			// Ready/unready transitions might also trigger this event.
-			if pod.Status.PodIP != "" {
+			switch wEvent.Type {
+			case EventAdded:
+				logger.Log.Debugf("Added matching pod %s, ns: %s", pod.Name, pod.Namespace)
 				restartTappersDebouncer.SetOn()
+			case EventDeleted:
+				logger.Log.Debugf("Removed matching pod %s, ns: %s", pod.Name, pod.Namespace)
+				restartTappersDebouncer.SetOn()
+			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.
+				// After filtering for IPs, on a normal pod restart this includes the following events:
+				// - Pod deletion
+				// - Pod reaches start state
+				// - Pod reaches ready state
+				// Ready/unready transitions might also trigger this event.
+				if pod.Status.PodIP != "" {
+					restartTappersDebouncer.SetOn()
+				}
+			case EventBookmark:
+				break
+			case EventError:
+				break
 			}
 		case err, ok := <-errorChan:
 			if !ok {

shared/kubernetes/podWatchHelper.go

@@ -21,13 +21,13 @@ func NewPodWatchHelper(kubernetesProvider *Provider, NameRegexFilter *regexp.Reg
 }
 
 // Implements the EventFilterer Interface
-func (pwh *PodWatchHelper) Filter(wEvent *WatchEvent) (bool, error) {
+func (wh *PodWatchHelper) Filter(wEvent *WatchEvent) (bool, error) {
 	pod, err := wEvent.ToPod()
 	if err != nil {
 		return false, nil
 	}
 
-	if !pwh.NameRegexFilter.MatchString(pod.Name) {
+	if !wh.NameRegexFilter.MatchString(pod.Name) {
 		return false, nil
 	}
 
@@ -35,8 +35,8 @@ func (pwh *PodWatchHelper) Filter(wEvent *WatchEvent) (bool, error) {
 }
 
 // Implements the WatchCreator Interface
-func (pwh *PodWatchHelper) NewWatcher(ctx context.Context, namespace string) (watch.Interface, error) {
-	watcher, err := pwh.kubernetesProvider.clientSet.CoreV1().Pods(namespace).Watch(ctx, metav1.ListOptions{Watch: true})
+func (wh *PodWatchHelper) NewWatcher(ctx context.Context, namespace string) (watch.Interface, error) {
+	watcher, err := wh.kubernetesProvider.clientSet.CoreV1().Pods(namespace).Watch(ctx, metav1.ListOptions{Watch: true})
 	if err != nil {
 		return nil, err
 	}

shared/kubernetes/provider.go

@@ -6,12 +6,16 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
+	"net/url"
+	"path/filepath"
+	"regexp"
+
 	"github.com/op/go-logging"
 	"github.com/up9inc/mizu/shared"
 	"github.com/up9inc/mizu/shared/logger"
 	"github.com/up9inc/mizu/shared/semver"
 	"github.com/up9inc/mizu/tap/api"
-	"io"
 	v1 "k8s.io/api/apps/v1"
 	core "k8s.io/api/core/v1"
 	rbac "k8s.io/api/rbac/v1"
@@ -32,9 +36,6 @@ import (
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/clientcmd"
 	watchtools "k8s.io/client-go/tools/watch"
-	"net/url"
-	"path/filepath"
-	"regexp"
 )
 
 type Provider struct {
@@ -485,6 +486,11 @@ func (provider *Provider) CreateDaemonsetRBAC(ctx context.Context, namespace str
 				Resources: []string{"daemonsets"},
 				Verbs:     []string{"patch", "get", "list", "create", "delete"},
 			},
+			{
+				APIGroups: []string{"events.k8s.io"},
+				Resources: []string{"events"},
+				Verbs:     []string{"list", "watch"},
+			},
 		},
 	}
 	roleBinding := &rbac.RoleBinding{

shared/kubernetes/watch.go

@@ -9,7 +9,6 @@ import (
 	"sync"
 	"time"
 
-	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/watch"
 )
 
@@ -21,10 +20,8 @@ type WatchCreator interface {
 	NewWatcher(ctx context.Context, namespace string) (watch.Interface, error)
 }
 
-func FilteredWatch(ctx context.Context, watcherCreator WatchCreator, targetNamespaces []string, filterer EventFilterer) (chan *WatchEvent, chan *WatchEvent, chan *WatchEvent, chan error) {
-	addedChan := make(chan *WatchEvent)
-	modifiedChan := make(chan *WatchEvent)
-	removedChan := make(chan *WatchEvent)
+func FilteredWatch(ctx context.Context, watcherCreator WatchCreator, targetNamespaces []string, filterer EventFilterer) (<-chan *WatchEvent, <-chan error) {
+	eventChan := make(chan *WatchEvent)
 	errorChan := make(chan error)
 
 	var wg sync.WaitGroup
@@ -39,11 +36,11 @@ func FilteredWatch(ctx context.Context, watcherCreator WatchCreator, targetNames
 			for {
 				watcher, err := watcherCreator.NewWatcher(ctx, targetNamespace)
 				if err != nil {
-					errorChan <- fmt.Errorf("error in k8 watch: %v", err)
+					errorChan <- fmt.Errorf("error in k8s watch: %v", err)
 					break
 				}
 
-				err = startWatchLoop(ctx, watcher, filterer, addedChan, modifiedChan, removedChan) // blocking
+				err = startWatchLoop(ctx, watcher, filterer, eventChan) // blocking
 				watcher.Stop()
 
 				select {
@@ -54,7 +51,7 @@ func FilteredWatch(ctx context.Context, watcherCreator WatchCreator, targetNames
 				}
 
 				if err != nil {
-					errorChan <- fmt.Errorf("error in k8 watch: %v", err)
+					errorChan <- fmt.Errorf("error in k8s watch: %v", err)
 					break
 				} else {
 					if !watchRestartDebouncer.IsOn() {
@@ -74,16 +71,14 @@ func FilteredWatch(ctx context.Context, watcherCreator WatchCreator, targetNames
 	go func() {
 		<-ctx.Done()
 		wg.Wait()
-		close(addedChan)
-		close(modifiedChan)
-		close(removedChan)
+		close(eventChan)
 		close(errorChan)
 	}()
 
-	return addedChan, modifiedChan, removedChan, errorChan
+	return eventChan, errorChan
 }
 
-func startWatchLoop(ctx context.Context, watcher watch.Interface, filterer EventFilterer, addedChan chan *WatchEvent, modifiedChan chan *WatchEvent, removedChan chan *WatchEvent) error {
+func startWatchLoop(ctx context.Context, watcher watch.Interface, filterer EventFilterer, eventChan chan<- *WatchEvent) error {
 	resultChan := watcher.ResultChan()
 	for {
 		select {
@@ -95,7 +90,7 @@ func startWatchLoop(ctx context.Context, watcher watch.Interface, filterer Event
 			wEvent := WatchEvent(e)
 
 			if wEvent.Type == watch.Error {
-				return apierrors.FromObject(wEvent.Object)
+				return wEvent.ToError()
 			}
 
 			if pass, err := filterer.Filter(&wEvent); err != nil {
@@ -104,14 +99,7 @@ func startWatchLoop(ctx context.Context, watcher watch.Interface, filterer Event
 				continue
 			}
 
-			switch wEvent.Type {
-			case watch.Added:
-				addedChan <- &wEvent
-			case watch.Modified:
-				modifiedChan <- &wEvent
-			case watch.Deleted:
-				removedChan <- &wEvent
-			}
+			eventChan <- &wEvent
 		case <-ctx.Done():
 			return nil
 		}

shared/kubernetes/watchEvent.go

@@ -2,17 +2,51 @@ package kubernetes
 
 import (
 	"fmt"
+	"reflect"
+
 	corev1 "k8s.io/api/core/v1"
+	eventsv1 "k8s.io/api/events/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/watch"
 )
 
+const (
+	EventAdded    watch.EventType = watch.Added
+	EventModified watch.EventType = watch.Modified
+	EventDeleted  watch.EventType = watch.Deleted
+	EventBookmark watch.EventType = watch.Bookmark
+	EventError    watch.EventType = watch.Error
+)
+
+type InvalidObjectType struct {
+	RequestedType reflect.Type
+}
+
+// Implements the error interface
+func (iot *InvalidObjectType) Error() string {
+	return fmt.Sprintf("Cannot convert event to type %s", iot.RequestedType)
+}
+
 type WatchEvent watch.Event
 
 func (we *WatchEvent) ToPod() (*corev1.Pod, error) {
 	pod, ok := we.Object.(*corev1.Pod)
 	if !ok {
-		return nil, fmt.Errorf("Invalid object type on pod event stream")
+		return nil, &InvalidObjectType{RequestedType: reflect.TypeOf(pod)}
 	}
 
 	return pod, nil
 }
+
+func (we *WatchEvent) ToEvent() (*eventsv1.Event, error) {
+	event, ok := we.Object.(*eventsv1.Event)
+	if !ok {
+		return nil, &InvalidObjectType{RequestedType: reflect.TypeOf(event)}
+	}
+
+	return event, nil
+}
+
+func (we *WatchEvent) ToError() error {
+	return apierrors.FromObject(we.Object)
+}

shared/logger/logger.go

@@ -9,7 +9,7 @@ import (
 var Log = logging.MustGetLogger("mizu")
 
 var format = logging.MustStringFormatter(
-	`%{time:2006-01-02T15:04:05.999Z-07:00} %{level:-5s} ▶ %{message} ▶ %{pid} %{shortfile} %{shortfunc}`,
+	`[%{time:2006-01-02T15:04:05.000-0700}] %{level:-5s} ▶ %{message} ▶ [%{pid} %{shortfile} %{shortfunc}]`,
 )
 
 func InitLogger(logPath string) {

tap/api/api.go

@@ -99,7 +99,7 @@ type Dissector interface {
 	Dissect(b *bufio.Reader, isClient bool, tcpID *TcpID, counterPair *CounterPair, superTimer *SuperTimer, superIdentifier *SuperIdentifier, emitter Emitter, options *TrafficFilteringOptions) error
 	Analyze(item *OutputChannelItem, resolvedSource string, resolvedDestination string) *MizuEntry
 	Summarize(entry *MizuEntry) *BaseEntryDetails
-	Represent(pIn Protocol, request map[string]interface{}, response map[string]interface{}) (pOut Protocol, object []byte, bodySize int64, err error)
+	Represent(request map[string]interface{}, response map[string]interface{}) (object []byte, bodySize int64, err error)
 	Macros() map[string]string
 }
 

tap/extensions/amqp/main.go

@@ -325,8 +325,7 @@ func (d dissecting) Summarize(entry *api.MizuEntry) *api.BaseEntryDetails {
 	}
 }
 
-func (d dissecting) Represent(protoIn api.Protocol, request map[string]interface{}, response map[string]interface{}) (protoOut api.Protocol, object []byte, bodySize int64, err error) {
-	protoOut = protocol
+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)
 	var repRequest []interface{}
@@ -363,7 +362,7 @@ func (d dissecting) Represent(protoIn api.Protocol, request map[string]interface
 
 func (d dissecting) Macros() map[string]string {
 	return map[string]string{
-		`amqp`: fmt.Sprintf(`proto.abbr == "%s"`, protocol.Abbreviation),
+		`amqp`: fmt.Sprintf(`proto.name == "%s"`, protocol.Name),
 	}
 }
 

tap/extensions/http/handlers.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"io/ioutil"
 	"net/http"
+	"strings"
 
 	"github.com/up9inc/mizu/tap/api"
 )
@@ -23,8 +24,8 @@ func filterAndEmit(item *api.OutputChannelItem, emitter api.Emitter, options *ap
 	emitter.Emit(item)
 }
 
-func handleHTTP2Stream(grpcAssembler *GrpcAssembler, tcpID *api.TcpID, superTimer *api.SuperTimer, emitter api.Emitter, options *api.TrafficFilteringOptions) error {
-	streamID, messageHTTP1, err := grpcAssembler.readMessage()
+func handleHTTP2Stream(http2Assembler *Http2Assembler, tcpID *api.TcpID, superTimer *api.SuperTimer, emitter api.Emitter, options *api.TrafficFilteringOptions) error {
+	streamID, messageHTTP1, isGrpc, err := http2Assembler.readMessage()
 	if err != nil {
 		return err
 	}
@@ -34,12 +35,13 @@ func handleHTTP2Stream(grpcAssembler *GrpcAssembler, tcpID *api.TcpID, superTime
 	switch messageHTTP1 := messageHTTP1.(type) {
 	case http.Request:
 		ident := fmt.Sprintf(
-			"%s->%s %s->%s %d",
+			"%s->%s %s->%s %d %s",
 			tcpID.SrcIP,
 			tcpID.DstIP,
 			tcpID.SrcPort,
 			tcpID.DstPort,
 			streamID,
+			"HTTP2",
 		)
 		item = reqResMatcher.registerRequest(ident, &messageHTTP1, superTimer.CaptureTime)
 		if item != nil {
@@ -53,12 +55,13 @@ func handleHTTP2Stream(grpcAssembler *GrpcAssembler, tcpID *api.TcpID, superTime
 		}
 	case http.Response:
 		ident := fmt.Sprintf(
-			"%s->%s %s->%s %d",
+			"%s->%s %s->%s %d %s",
 			tcpID.DstIP,
 			tcpID.SrcIP,
 			tcpID.DstPort,
 			tcpID.SrcPort,
 			streamID,
+			"HTTP2",
 		)
 		item = reqResMatcher.registerResponse(ident, &messageHTTP1, superTimer.CaptureTime)
 		if item != nil {
@@ -73,30 +76,41 @@ func handleHTTP2Stream(grpcAssembler *GrpcAssembler, tcpID *api.TcpID, superTime
 	}
 
 	if item != nil {
-		item.Protocol = http2Protocol
+		if isGrpc {
+			item.Protocol = grpcProtocol
+		} else {
+			item.Protocol = http2Protocol
+		}
 		filterAndEmit(item, emitter, options)
 	}
 
 	return nil
 }
 
-func handleHTTP1ClientStream(b *bufio.Reader, tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, emitter api.Emitter, options *api.TrafficFilteringOptions) error {
-	req, err := http.ReadRequest(b)
+func handleHTTP1ClientStream(b *bufio.Reader, tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, emitter api.Emitter, options *api.TrafficFilteringOptions) (switchingProtocolsHTTP2 bool, req *http.Request, err error) {
+	req, err = http.ReadRequest(b)
 	if err != nil {
-		return err
+		return
 	}
 	counterPair.Request++
 
-	body, err := ioutil.ReadAll(req.Body)
+	// Check HTTP2 upgrade - HTTP2 Over Cleartext (H2C)
+	if strings.Contains(strings.ToLower(req.Header.Get("Connection")), "upgrade") && strings.ToLower(req.Header.Get("Upgrade")) == "h2c" {
+		switchingProtocolsHTTP2 = true
+	}
+
+	var body []byte
+	body, err = ioutil.ReadAll(req.Body)
 	req.Body = io.NopCloser(bytes.NewBuffer(body)) // rewind
 
 	ident := fmt.Sprintf(
-		"%s->%s %s->%s %d",
+		"%s->%s %s->%s %d %s",
 		tcpID.SrcIP,
 		tcpID.DstIP,
 		tcpID.SrcPort,
 		tcpID.DstPort,
 		counterPair.Request,
+		"HTTP1",
 	)
 	item := reqResMatcher.registerRequest(ident, req, superTimer.CaptureTime)
 	if item != nil {
@@ -109,26 +123,34 @@ func handleHTTP1ClientStream(b *bufio.Reader, tcpID *api.TcpID, counterPair *api
 		}
 		filterAndEmit(item, emitter, options)
 	}
-	return nil
+	return
 }
 
-func handleHTTP1ServerStream(b *bufio.Reader, tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, emitter api.Emitter, options *api.TrafficFilteringOptions) error {
-	res, err := http.ReadResponse(b, nil)
+func handleHTTP1ServerStream(b *bufio.Reader, tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, emitter api.Emitter, options *api.TrafficFilteringOptions) (switchingProtocolsHTTP2 bool, err error) {
+	var res *http.Response
+	res, err = http.ReadResponse(b, nil)
 	if err != nil {
-		return err
+		return
 	}
 	counterPair.Response++
 
-	body, err := ioutil.ReadAll(res.Body)
+	// Check HTTP2 upgrade - HTTP2 Over Cleartext (H2C)
+	if res.StatusCode == 101 && strings.Contains(strings.ToLower(res.Header.Get("Connection")), "upgrade") && strings.ToLower(res.Header.Get("Upgrade")) == "h2c" {
+		switchingProtocolsHTTP2 = true
+	}
+
+	var body []byte
+	body, err = ioutil.ReadAll(res.Body)
 	res.Body = io.NopCloser(bytes.NewBuffer(body)) // rewind
 
 	ident := fmt.Sprintf(
-		"%s->%s %s->%s %d",
+		"%s->%s %s->%s %d %s",
 		tcpID.DstIP,
 		tcpID.SrcIP,
 		tcpID.DstPort,
 		tcpID.SrcPort,
 		counterPair.Response,
+		"HTTP1",
 	)
 	item := reqResMatcher.registerResponse(ident, res, superTimer.CaptureTime)
 	if item != nil {
@@ -141,5 +163,5 @@ func handleHTTP1ServerStream(b *bufio.Reader, tcpID *api.TcpID, counterPair *api
 		}
 		filterAndEmit(item, emitter, options)
 	}
-	return nil
+	return
 }

tap/extensions/http/http2_assembler.go

@@ -10,6 +10,7 @@ import (
 	"math"
 	"net/http"
 	"net/url"
+	"strconv"
 	"strings"
 
 	"golang.org/x/net/http2"
@@ -27,6 +28,26 @@ const protoMinorHTTP2 = 0
 
 var maxHTTP2DataLen = 1 * 1024 * 1024 // 1MB
 
+var grpcStatusCodes = []string{
+	"OK",
+	"CANCELLED",
+	"UNKNOWN",
+	"INVALID_ARGUMENT",
+	"DEADLINE_EXCEEDED",
+	"NOT_FOUND",
+	"ALREADY_EXISTS",
+	"PERMISSION_DENIED",
+	"RESOURCE_EXHAUSTED",
+	"FAILED_PRECONDITION",
+	"ABORTED",
+	"OUT_OF_RANGE",
+	"UNIMPLEMENTED",
+	"INTERNAL",
+	"UNAVAILABLE",
+	"DATA_LOSS",
+	"UNAUTHENTICATED",
+}
+
 type messageFragment struct {
 	headers []hpack.HeaderField
 	data    []byte
@@ -71,37 +92,38 @@ func (fbs *fragmentsByStream) pop(streamID uint32) ([]hpack.HeaderField, []byte)
 	return headers, data
 }
 
-func createGrpcAssembler(b *bufio.Reader) *GrpcAssembler {
+func createHTTP2Assembler(b *bufio.Reader) *Http2Assembler {
 	var framerOutput bytes.Buffer
 	framer := http2.NewFramer(&framerOutput, b)
 	framer.ReadMetaHeaders = hpack.NewDecoder(initialHeaderTableSize, nil)
-	return &GrpcAssembler{
+	return &Http2Assembler{
 		fragmentsByStream: make(fragmentsByStream),
 		framer:            framer,
 	}
 }
 
-type GrpcAssembler struct {
+type Http2Assembler struct {
 	fragmentsByStream fragmentsByStream
 	framer            *http2.Framer
 }
 
-func (ga *GrpcAssembler) readMessage() (uint32, interface{}, error) {
+func (ga *Http2Assembler) readMessage() (streamID uint32, messageHTTP1 interface{}, isGrpc bool, err error) {
 	// Exactly one Framer is used for each half connection.
 	// (Instead of creating a new Framer for each ReadFrame operation)
 	// This is needed in order to decompress the headers,
 	// because the compression context is updated with each requests/response.
 	frame, err := ga.framer.ReadFrame()
 	if err != nil {
-		return 0, nil, err
+		return
 	}
 
-	streamID := frame.Header().StreamID
+	streamID = frame.Header().StreamID
 
 	ga.fragmentsByStream.appendFrame(streamID, frame)
 
 	if !(ga.isStreamEnd(frame)) {
-		return 0, nil, nil
+		streamID = 0
+		return
 	}
 
 	headers, data := ga.fragmentsByStream.pop(streamID)
@@ -115,13 +137,29 @@ func (ga *GrpcAssembler) readMessage() (uint32, interface{}, error) {
 	dataString := base64.StdEncoding.EncodeToString(data)
 
 	// Use http1 types only because they are expected in http_matcher.
-	// TODO: Create an interface that will be used by http_matcher:registerRequest and http_matcher:registerRequest
-	//       to accept both HTTP/1.x and HTTP/2 requests and responses
-	var messageHTTP1 interface{}
-	if _, ok := headersHTTP1[":method"]; ok {
+	method := headersHTTP1.Get(":method")
+	status := headersHTTP1.Get(":status")
+
+	// gRPC detection
+	grpcStatus := headersHTTP1.Get("Grpc-Status")
+	if grpcStatus != "" {
+		isGrpc = true
+		status = grpcStatus
+	}
+
+	if strings.Contains(headersHTTP1.Get("Content-Type"), "application/grpc") {
+		isGrpc = true
+		grpcPath := headersHTTP1.Get(":path")
+		pathSegments := strings.Split(grpcPath, "/")
+		if len(pathSegments) > 0 {
+			method = pathSegments[len(pathSegments)-1]
+		}
+	}
+
+	if method != "" {
 		messageHTTP1 = http.Request{
 			URL:           &url.URL{},
-			Method:        "POST",
+			Method:        method,
 			Header:        headersHTTP1,
 			Proto:         protoHTTP2,
 			ProtoMajor:    protoMajorHTTP2,
@@ -129,8 +167,16 @@ func (ga *GrpcAssembler) readMessage() (uint32, interface{}, error) {
 			Body:          io.NopCloser(strings.NewReader(dataString)),
 			ContentLength: int64(len(dataString)),
 		}
-	} else if _, ok := headersHTTP1[":status"]; ok {
+	} else if status != "" {
+		var statusCode int
+
+		statusCode, err = strconv.Atoi(status)
+		if err != nil {
+			return
+		}
+
 		messageHTTP1 = http.Response{
+			StatusCode:    statusCode,
 			Header:        headersHTTP1,
 			Proto:         protoHTTP2,
 			ProtoMajor:    protoMajorHTTP2,
@@ -139,13 +185,14 @@ func (ga *GrpcAssembler) readMessage() (uint32, interface{}, error) {
 			ContentLength: int64(len(dataString)),
 		}
 	} else {
-		return 0, nil, errors.New("failed to assemble stream: neither a request nor a message")
+		err = errors.New("failed to assemble stream: neither a request nor a message")
+		return
 	}
 
-	return streamID, messageHTTP1, nil
+	return
 }
 
-func (ga *GrpcAssembler) isStreamEnd(frame http2.Frame) bool {
+func (ga *Http2Assembler) isStreamEnd(frame http2.Frame) bool {
 	switch frame := frame.(type) {
 	case *http2.MetaHeadersFrame:
 		if frame.StreamEnded() {

tap/extensions/http/main.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"io"
 	"log"
+	"net/http"
 	"net/url"
 	"time"
 
@@ -23,21 +24,35 @@ var protocol api.Protocol = api.Protocol{
 	ForegroundColor: "#ffffff",
 	FontSize:        12,
 	ReferenceLink:   "https://datatracker.ietf.org/doc/html/rfc2616",
-	Ports:           []string{"80", "8080", "50051"},
+	Ports:           []string{"80", "443", "8080"},
 	Priority:        0,
 }
 
 var http2Protocol api.Protocol = api.Protocol{
 	Name:            "http",
-	LongName:        "Hypertext Transfer Protocol Version 2 (HTTP/2) (gRPC)",
+	LongName:        "Hypertext Transfer Protocol Version 2 (HTTP/2)",
 	Abbreviation:    "HTTP/2",
-	Macro:           "grpc",
+	Macro:           "http2",
 	Version:         "2.0",
 	BackgroundColor: "#244c5a",
 	ForegroundColor: "#ffffff",
 	FontSize:        11,
 	ReferenceLink:   "https://datatracker.ietf.org/doc/html/rfc7540",
-	Ports:           []string{"80", "8080"},
+	Ports:           []string{"80", "443", "8080"},
+	Priority:        0,
+}
+
+var grpcProtocol api.Protocol = api.Protocol{
+	Name:            "http",
+	LongName:        "Hypertext Transfer Protocol Version 2 (HTTP/2) [ gRPC over HTTP/2 ]",
+	Abbreviation:    "gRPC",
+	Macro:           "grpc",
+	Version:         "2.0",
+	BackgroundColor: "#244c5a",
+	ForegroundColor: "#ffffff",
+	FontSize:        11,
+	ReferenceLink:   "https://grpc.github.io/grpc/core/md_doc_statuscodes.html",
+	Ports:           []string{"80", "443", "8080", "50051"},
 	Priority:        0,
 }
 
@@ -64,20 +79,28 @@ 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 grpcAssembler *GrpcAssembler
+	var http2Assembler *Http2Assembler
 	if isHTTP2 {
 		prepareHTTP2Connection(b, isClient)
-		grpcAssembler = createGrpcAssembler(b)
+		http2Assembler = createHTTP2Assembler(b)
 	}
 
 	dissected := false
+	switchingProtocolsHTTP2 := false
 	for {
+		if switchingProtocolsHTTP2 {
+			switchingProtocolsHTTP2 = false
+			isHTTP2, err = checkIsHTTP2Connection(b, isClient)
+			prepareHTTP2Connection(b, isClient)
+			http2Assembler = createHTTP2Assembler(b)
+		}
+
 		if superIdentifier.Protocol != nil && superIdentifier.Protocol != &protocol {
 			return errors.New("Identified by another protocol")
 		}
 
 		if isHTTP2 {
-			err = handleHTTP2Stream(grpcAssembler, tcpID, superTimer, emitter, options)
+			err = handleHTTP2Stream(http2Assembler, tcpID, superTimer, emitter, options)
 			if err == io.EOF || err == io.ErrUnexpectedEOF {
 				break
 			} else if err != nil {
@@ -85,15 +108,39 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
 			}
 			dissected = true
 		} else if isClient {
-			err = handleHTTP1ClientStream(b, tcpID, counterPair, superTimer, emitter, options)
+			var req *http.Request
+			switchingProtocolsHTTP2, req, err = handleHTTP1ClientStream(b, tcpID, counterPair, superTimer, emitter, options)
 			if err == io.EOF || err == io.ErrUnexpectedEOF {
 				break
 			} else if err != nil {
 				continue
 			}
 			dissected = true
+
+			// In case of an HTTP2 upgrade, duplicate the HTTP1 request into HTTP2 with stream ID 1
+			if switchingProtocolsHTTP2 {
+				ident := fmt.Sprintf(
+					"%s->%s %s->%s 1 %s",
+					tcpID.SrcIP,
+					tcpID.DstIP,
+					tcpID.SrcPort,
+					tcpID.DstPort,
+					"HTTP2",
+				)
+				item := reqResMatcher.registerRequest(ident, req, superTimer.CaptureTime)
+				if item != nil {
+					item.ConnectionInfo = &api.ConnectionInfo{
+						ClientIP:   tcpID.SrcIP,
+						ClientPort: tcpID.SrcPort,
+						ServerIP:   tcpID.DstIP,
+						ServerPort: tcpID.DstPort,
+						IsOutgoing: true,
+					}
+					filterAndEmit(item, emitter, options)
+				}
+			}
 		} else {
-			err = handleHTTP1ServerStream(b, tcpID, counterPair, superTimer, emitter, options)
+			switchingProtocolsHTTP2, err = handleHTTP1ServerStream(b, tcpID, counterPair, superTimer, emitter, options)
 			if err == io.EOF || err == io.ErrUnexpectedEOF {
 				break
 			} else if err != nil {
@@ -110,23 +157,16 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
 	return nil
 }
 
-func SetHostname(address, newHostname string) string {
-	replacedUrl, err := url.Parse(address)
-	if err != nil {
-		return address
-	}
-	replacedUrl.Host = newHostname
-	return replacedUrl.String()
-}
-
 func (d dissecting) Analyze(item *api.OutputChannelItem, resolvedSource string, resolvedDestination string) *api.MizuEntry {
-	var host, scheme, authority, path, service string
+	var host, authority, path, service string
 
 	request := item.Pair.Request.Payload.(map[string]interface{})
 	response := item.Pair.Response.Payload.(map[string]interface{})
 	reqDetails := request["details"].(map[string]interface{})
 	resDetails := response["details"].(map[string]interface{})
 
+	isRequestUpgradedH2C := false
+
 	for _, header := range reqDetails["headers"].([]interface{}) {
 		h := header.(map[string]interface{})
 		if h["name"] == "Host" {
@@ -135,22 +175,25 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, resolvedSource string,
 		if h["name"] == ":authority" {
 			authority = h["value"].(string)
 		}
-		if h["name"] == ":scheme" {
-			scheme = h["value"].(string)
-		}
 		if h["name"] == ":path" {
 			path = h["value"].(string)
 		}
+
+		if h["name"] == "Upgrade" {
+			if h["value"].(string) == "h2c" {
+				isRequestUpgradedH2C = true
+			}
+		}
 	}
 
 	if resDetails["bodySize"].(float64) < 0 {
 		resDetails["bodySize"] = 0
 	}
 
-	if item.Protocol.Version == "2.0" {
-		service = fmt.Sprintf("%s://%s", scheme, authority)
+	if item.Protocol.Version == "2.0" && !isRequestUpgradedH2C {
+		service = authority
 	} else {
-		service = fmt.Sprintf("http://%s", host)
+		service = host
 		u, err := url.Parse(reqDetails["url"].(string))
 		if err != nil {
 			path = reqDetails["url"].(string)
@@ -160,6 +203,7 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, resolvedSource string,
 	}
 
 	request["url"] = reqDetails["url"].(string)
+	reqDetails["targetUri"] = reqDetails["url"]
 	reqDetails["path"] = path
 	reqDetails["summary"] = path
 
@@ -178,9 +222,20 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, resolvedSource string,
 	reqDetails["queryString"] = mapSliceRebuildAsMap(reqDetails["_queryString"].([]interface{}))
 
 	if resolvedDestination != "" {
-		service = SetHostname(service, resolvedDestination)
+		service = resolvedDestination
 	} else if resolvedSource != "" {
-		service = SetHostname(service, resolvedSource)
+		service = resolvedSource
+	}
+
+	method := reqDetails["method"].(string)
+	statusCode := int(resDetails["status"].(float64))
+	if item.Protocol.Abbreviation == "gRPC" {
+		resDetails["statusText"] = grpcStatusCodes[statusCode]
+	}
+
+	if item.Protocol.Version == "2.0" && !isRequestUpgradedH2C {
+		reqDetails["url"] = path
+		request["url"] = path
 	}
 
 	elapsedTime := item.Pair.Response.CaptureTime.Sub(item.Pair.Request.CaptureTime).Round(time.Millisecond).Milliseconds()
@@ -188,10 +243,8 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, resolvedSource string,
 		elapsedTime = 0
 	}
 	httpPair, _ := json.Marshal(item.Pair)
-	_protocol := protocol
-	_protocol.Version = item.Protocol.Version
 	return &api.MizuEntry{
-		Protocol: _protocol,
+		Protocol: item.Protocol,
 		Source: &api.TCP{
 			Name: resolvedSource,
 			IP:   item.ConnectionInfo.ClientIP,
@@ -206,8 +259,8 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, resolvedSource string,
 		Request:             reqDetails,
 		Response:            resDetails,
 		Url:                 fmt.Sprintf("%s%s", service, path),
-		Method:              reqDetails["method"].(string),
-		Status:              int(resDetails["status"].(float64)),
+		Method:              method,
+		Status:              statusCode,
 		RequestSenderIp:     item.ConnectionInfo.ClientIP,
 		Service:             service,
 		Timestamp:           item.Timestamp,
@@ -226,15 +279,9 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, resolvedSource string,
 }
 
 func (d dissecting) Summarize(entry *api.MizuEntry) *api.BaseEntryDetails {
-	var p api.Protocol
-	if entry.Protocol.Version == "2.0" {
-		p = http2Protocol
-	} else {
-		p = protocol
-	}
 	return &api.BaseEntryDetails{
 		Id:              entry.Id,
-		Protocol:        p,
+		Protocol:        entry.Protocol,
 		Url:             entry.Url,
 		RequestSenderIp: entry.RequestSenderIp,
 		Service:         entry.Service,
@@ -264,9 +311,9 @@ func representRequest(request map[string]interface{}) (repRequest []interface{})
 			Selector: `request.method`,
 		},
 		{
-			Name:     "URL",
-			Value:    request["url"].(string),
-			Selector: `request.url`,
+			Name:     "Target URI",
+			Value:    request["targetUri"].(string),
+			Selector: `request.targetUri`,
 		},
 		{
 			Name:     "Path",
@@ -408,12 +455,7 @@ func representResponse(response map[string]interface{}) (repResponse []interface
 	return
 }
 
-func (d dissecting) Represent(protoIn api.Protocol, request map[string]interface{}, response map[string]interface{}) (protoOut api.Protocol, object []byte, bodySize int64, err error) {
-	if protoIn.Version == "2.0" {
-		protoOut = http2Protocol
-	} else {
-		protoOut = protocol
-	}
+func (d dissecting) Represent(request map[string]interface{}, response map[string]interface{}) (object []byte, bodySize int64, err error) {
 	representation := make(map[string]interface{}, 0)
 	repRequest := representRequest(request)
 	repResponse, bodySize := representResponse(response)
@@ -425,9 +467,9 @@ func (d dissecting) Represent(protoIn api.Protocol, request map[string]interface
 
 func (d dissecting) Macros() map[string]string {
 	return map[string]string{
-		`http`:  fmt.Sprintf(`proto.abbr == "%s" and proto.version == "%s"`, protocol.Abbreviation, protocol.Version),
-		`grpc`:  fmt.Sprintf(`proto.abbr == "%s" and proto.version == "%s"`, protocol.Abbreviation, http2Protocol.Version),
-		`http2`: fmt.Sprintf(`proto.abbr == "%s" and proto.version == "%s"`, protocol.Abbreviation, http2Protocol.Version),
+		`http`:  fmt.Sprintf(`proto.name == "%s" and proto.version == "%s"`, protocol.Name, protocol.Version),
+		`http2`: fmt.Sprintf(`proto.name == "%s" and proto.version == "%s"`, protocol.Name, http2Protocol.Version),
+		`grpc`:  fmt.Sprintf(`proto.name == "%s" and proto.version == "%s" and proto.macro == "%s"`, protocol.Name, grpcProtocol.Version, grpcProtocol.Macro),
 	}
 }
 

tap/extensions/http/matcher.go

@@ -92,6 +92,6 @@ func splitIdent(ident string) []string {
 }
 
 func genKey(split []string) string {
-	key := fmt.Sprintf("%s:%s->%s:%s,%s", split[0], split[2], split[1], split[3], split[4])
+	key := fmt.Sprintf("%s:%s->%s:%s,%s%s", split[0], split[2], split[1], split[3], split[4], split[5])
 	return key
 }

tap/extensions/kafka/main.go

@@ -210,8 +210,7 @@ func (d dissecting) Summarize(entry *api.MizuEntry) *api.BaseEntryDetails {
 	}
 }
 
-func (d dissecting) Represent(protoIn api.Protocol, request map[string]interface{}, response map[string]interface{}) (protoOut api.Protocol, object []byte, bodySize int64, err error) {
-	protoOut = _protocol
+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)
 
@@ -258,7 +257,7 @@ func (d dissecting) Represent(protoIn api.Protocol, request map[string]interface
 
 func (d dissecting) Macros() map[string]string {
 	return map[string]string{
-		`kafka`: fmt.Sprintf(`proto.abbr == "%s"`, _protocol.Abbreviation),
+		`kafka`: fmt.Sprintf(`proto.name == "%s"`, _protocol.Name),
 	}
 }
 

tap/extensions/redis/main.go

@@ -146,8 +146,7 @@ func (d dissecting) Summarize(entry *api.MizuEntry) *api.BaseEntryDetails {
 	}
 }
 
-func (d dissecting) Represent(protoIn api.Protocol, request map[string]interface{}, response map[string]interface{}) (protoOut api.Protocol, object []byte, bodySize int64, err error) {
-	protoOut = protocol
+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)
 	repRequest := representGeneric(request, `request.`)
@@ -160,7 +159,7 @@ func (d dissecting) Represent(protoIn api.Protocol, request map[string]interface
 
 func (d dissecting) Macros() map[string]string {
 	return map[string]string{
-		`redis`: fmt.Sprintf(`proto.abbr == "%s"`, protocol.Abbreviation),
+		`redis`: fmt.Sprintf(`proto.name == "%s"`, protocol.Name),
 	}
 }
 

tap/source/tcp_packet_source.go

@@ -121,29 +121,27 @@ func (source *tcpPacketSource) readPackets(ipdefrag bool, packets chan<- TcpPack
 		}
 
 		// defrag the IPv4 packet if required
-		if !ipdefrag {
-			ip4Layer := packet.Layer(layers.LayerTypeIPv4)
-			if ip4Layer == nil {
-				continue
-			}
-			ip4 := ip4Layer.(*layers.IPv4)
-			l := ip4.Length
-			newip4, err := source.defragger.DefragIPv4(ip4)
-			if err != nil {
-				logger.Log.Fatal("Error while de-fragmenting", err)
-			} else if newip4 == nil {
-				logger.Log.Debugf("Fragment...")
-				continue // packet fragment, we don't have whole packet yet.
-			}
-			if newip4.Length != l {
-				diagnose.InternalStats.Ipdefrag++
-				logger.Log.Debugf("Decoding re-assembled packet: %s", newip4.NextLayerType())
-				pb, ok := packet.(gopacket.PacketBuilder)
-				if !ok {
-					logger.Log.Panic("Not a PacketBuilder")
+		if ipdefrag {
+			if ip4Layer := packet.Layer(layers.LayerTypeIPv4); ip4Layer != nil {
+				ip4 := ip4Layer.(*layers.IPv4)
+				l := ip4.Length
+				newip4, err := source.defragger.DefragIPv4(ip4)
+				if err != nil {
+					logger.Log.Fatal("Error while de-fragmenting", err)
+				} else if newip4 == nil {
+					logger.Log.Debugf("Fragment...")
+					continue // packet fragment, we don't have whole packet yet.
+				}
+				if newip4.Length != l {
+					diagnose.InternalStats.Ipdefrag++
+					logger.Log.Debugf("Decoding re-assembled packet: %s", newip4.NextLayerType())
+					pb, ok := packet.(gopacket.PacketBuilder)
+					if !ok {
+						logger.Log.Panic("Not a PacketBuilder")
+					}
+					nextDecoder := newip4.NextLayerType()
+					_ = nextDecoder.Decode(newip4.Payload, pb)
 				}
-				nextDecoder := newip4.NextLayerType()
-				_ = nextDecoder.Decode(newip4.Payload, pb)
 			}
 		}
 

tap/tcp_assembler.go

@@ -16,6 +16,8 @@ import (
 	"github.com/up9inc/mizu/tap/source"
 )
 
+const PACKETS_SEEN_LOG_THRESHOLD = 1000
+
 type tcpAssembler struct {
 	*reassembly.Assembler
 	streamPool     *reassembly.StreamPool
@@ -63,7 +65,11 @@ func (a *tcpAssembler) processPackets(dumpPacket bool, packets <-chan source.Tcp
 
 	for packetInfo := range packets {
 		packetsCount := diagnose.AppStats.IncPacketsCount()
-		logger.Log.Debugf("PACKET #%d", packetsCount)
+		
+		if packetsCount % PACKETS_SEEN_LOG_THRESHOLD == 0 {
+			logger.Log.Debugf("Packets seen: #%d", packetsCount)
+		}
+		
 		packet := packetInfo.Packet
 		data := packet.Data()
 		diagnose.AppStats.UpdateProcessedBytes(uint64(len(data)))
@@ -85,7 +91,7 @@ func (a *tcpAssembler) processPackets(dumpPacket bool, packets <-chan source.Tcp
 				CaptureInfo: packet.Metadata().CaptureInfo,
 			}
 			diagnose.InternalStats.Totalsz += len(tcp.Payload)
-			logger.Log.Debugf("%s : %v -> %s : %v", packet.NetworkLayer().NetworkFlow().Src(), tcp.SrcPort, packet.NetworkLayer().NetworkFlow().Dst(), tcp.DstPort)
+			logger.Log.Debugf("%s:%v -> %s:%v", packet.NetworkLayer().NetworkFlow().Src(), tcp.SrcPort, packet.NetworkLayer().NetworkFlow().Dst(), tcp.DstPort)
 			a.assemblerMutex.Lock()
 			a.AssembleWithContext(packet.NetworkLayer().NetworkFlow(), tcp, &c)
 			a.assemblerMutex.Unlock()

tap/tester/.gitignore

@@ -1 +0,0 @@
-tester

tap/tester/README.md

@@ -1,12 +0,0 @@
-
-This tester used to launch passive-tapper locally without Docker or Kuberenetes environment.
-
-Its good for testing purposes.
-
-# How to run
-
-From the `tap` folder run:
-`./tester/launch.sh`
-
-The tester gets the same arguments the passive_tapper gets, run with `--help` to get a complete list of options.
-`./tester/launch.sh --help`

tap/tester/launch.sh

@@ -1,10 +0,0 @@
-#!/bin/bash
-
-set -e
-
-echo "Building extensions..."
-pushd .. && ./devops/build_extensions.sh && popd
-
-go build -o tester tester/tester.go
-
-sudo ./tester/tester "$@"

tap/tester/tester.go

@@ -1,114 +0,0 @@
-package main
-
-import (
-	"bufio"
-	"io/ioutil"
-	"os"
-	"path"
-	"plugin"
-	"sort"
-	"strings"
-
-	"github.com/op/go-logging"
-
-	"github.com/go-errors/errors"
-	"github.com/up9inc/mizu/shared/logger"
-	"github.com/up9inc/mizu/tap"
-	tapApi "github.com/up9inc/mizu/tap/api"
-)
-
-func loadExtensions() ([]*tapApi.Extension, error) {
-	extensionsDir := "./extensions"
-	files, err := ioutil.ReadDir(extensionsDir)
-
-	if err != nil {
-		return nil, errors.Wrap(err, 0)
-	}
-
-	extensions := make([]*tapApi.Extension, 0)
-	for _, file := range files {
-		filename := file.Name()
-
-		if !strings.HasSuffix(filename, ".so") {
-			continue
-		}
-
-		logger.Log.Infof("Loading extension: %s", filename)
-
-		extension := &tapApi.Extension{
-			Path: path.Join(extensionsDir, filename),
-		}
-
-		plug, err := plugin.Open(extension.Path)
-
-		if err != nil {
-			return nil, errors.Wrap(err, 0)
-		}
-
-		extension.Plug = plug
-		symDissector, err := plug.Lookup("Dissector")
-
-		if err != nil {
-			return nil, errors.Wrap(err, 0)
-		}
-
-		dissector, ok := symDissector.(tapApi.Dissector)
-
-		if !ok {
-			return nil, errors.Errorf("Symbol Dissector type error: %v %T", file, symDissector)
-		}
-
-		dissector.Register(extension)
-		extension.Dissector = dissector
-		extensions = append(extensions, extension)
-	}
-
-	sort.Slice(extensions, func(i, j int) bool {
-		return extensions[i].Protocol.Priority < extensions[j].Protocol.Priority
-	})
-
-	for _, extension := range extensions {
-		logger.Log.Infof("Extension Properties: %+v", extension)
-	}
-
-	return extensions, nil
-}
-
-func internalRun() error {
-	logger.InitLoggerStderrOnly(logging.DEBUG)
-
-	opts := tap.TapOpts{
-		HostMode: false,
-	}
-
-	outputItems := make(chan *tapApi.OutputChannelItem, 1000)
-	extenssions, err := loadExtensions()
-
-	if err != nil {
-		return err
-	}
-
-	tapOpts := tapApi.TrafficFilteringOptions{}
-
-	tap.StartPassiveTapper(&opts, outputItems, extenssions, &tapOpts)
-
-	logger.Log.Infof("Tapping, press enter to exit...")
-	reader := bufio.NewReader(os.Stdin)
-	reader.ReadLine()
-	return nil
-}
-
-func main() {
-	err := internalRun()
-
-	if err != nil {
-		switch err := err.(type) {
-		case *errors.Error:
-			logger.Log.Errorf("Error: %v", err.ErrorStack())
-		default:
-			logger.Log.Errorf("Error: %v", err)
-		}
-
-		os.Exit(1)
-	}
-}

ui/src/components/EntryDetailed/EntrySections.tsx

@@ -130,7 +130,7 @@ export const EntryBodySection: React.FC<EntryBodySectionProps> = ({
             <table>
                 <tbody>
                     <EntryViewLine label={'Mime type'} value={contentType} updateQuery={updateQuery} selector={selector} overrideQueryValue={`r".*"`}/>
-                    <EntryViewLine label={'Encoding'} value={encoding} updateQuery={updateQuery} selector={selector} overrideQueryValue={`r".*"`}/>
+                    {encoding && <EntryViewLine label={'Encoding'} value={encoding} updateQuery={updateQuery} selector={selector} overrideQueryValue={`r".*"`}/>}
                 </tbody>
             </table>
 

ui/src/components/UI/style/misc.module.sass

@@ -5,7 +5,6 @@
     border: solid 1px $secondary-font-color
     margin-left: 4px
     padding: 2px 5px
-    text-transform: uppercase
     font-family: "Source Sans Pro", sans-serif
     font-size: 11px
     font-weight: bold