Fix replay to be like the test (#1173)

Represent should get entry that Marshaled and UnMarshaled

Dockerfile

@@ -54,14 +54,14 @@ ENV BPF_TARGET=arm64 BPF_CFLAGS="-O2 -g -D__TARGET_ARCH_arm64"
 
 
 ### Builder image for x86-64 to AArch64 cross-compilation
-FROM up9inc/linux-arm64-musl-go-libpcap-capstone-bpf AS builder-from-amd64-to-arm64v8
+FROM up9inc/linux-arm64-musl-go-libpcap-capstone-bpf:capstone-5.0-rc2 AS builder-from-amd64-to-arm64v8
 ENV CGO_ENABLED=1 GOOS=linux
 ENV GOARCH=arm64 CGO_CFLAGS="-I/work/libpcap -I/work/capstone/include"
 ENV BPF_TARGET=arm64 BPF_CFLAGS="-O2 -g -D__TARGET_ARCH_arm64 -I/usr/xcc/aarch64-linux-musl-cross/aarch64-linux-musl/include/"
 
 
 ### Builder image for AArch64 to x86-64 cross-compilation
-FROM up9inc/linux-x86_64-musl-go-libpcap-capstone-bpf AS builder-from-arm64v8-to-amd64
+FROM up9inc/linux-x86_64-musl-go-libpcap-capstone-bpf:capstone-5.0-rc2 AS builder-from-arm64v8-to-amd64
 ENV CGO_ENABLED=1 GOOS=linux
 ENV GOARCH=amd64 CGO_CFLAGS="-I/libpcap -I/capstone/include"
 ENV BPF_TARGET=amd64 BPF_CFLAGS="-O2 -g -D__TARGET_ARCH_x86  -I/usr/local/musl/x86_64-unknown-linux-musl/include/"

agent/go.mod

@@ -83,7 +83,7 @@ require (
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/klauspost/compress v1.14.2 // indirect
-	github.com/knightsc/gapstone v0.0.0-20211014144438-5e0e64002a6e // indirect
+	github.com/knightsc/gapstone v0.0.0-20191231144527-6fa5afaf11a9 // indirect
 	github.com/leodido/go-urn v1.2.1 // indirect
 	github.com/liggitt/tabwriter v0.0.0-20181228230101-89fcab3d43de // indirect
 	github.com/mailru/easyjson v0.7.7 // indirect

agent/go.sum

@@ -453,8 +453,8 @@ github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+o
 github.com/klauspost/compress v1.9.8/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
 github.com/klauspost/compress v1.14.2 h1:S0OHlFk/Gbon/yauFJ4FfJJF5V0fc5HbBTJazi28pRw=
 github.com/klauspost/compress v1.14.2/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
-github.com/knightsc/gapstone v0.0.0-20211014144438-5e0e64002a6e h1:6J5obSn9umEThiYzWzndcPOZR0Qj/sVCZpH6V1G7yNE=
+github.com/knightsc/gapstone v0.0.0-20191231144527-6fa5afaf11a9 h1:1KszOoXSFt0aRQ6wxxcKm7QKgfLPI0TWO47UcY/f+vA=
-github.com/knightsc/gapstone v0.0.0-20211014144438-5e0e64002a6e/go.mod h1:1K5hEzsMBLTPdRJKEHqBFJ8Zt2VRqDhomcQ11KH0WW4=
+github.com/knightsc/gapstone v0.0.0-20191231144527-6fa5afaf11a9/go.mod h1:1K5hEzsMBLTPdRJKEHqBFJ8Zt2VRqDhomcQ11KH0WW4=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/kr/fs v0.1.0/go.mod h1:FFnZGqtBN9Gxj7eW1uZ42v5BccTP0vu6NEaFoC2HwRg=

agent/main.go

@@ -131,6 +131,7 @@ func hostApi(socketHarOutputChannel chan<- *tapApi.OutputChannelItem) *gin.Engin
 	routes.MetadataRoutes(ginApp)
 	routes.StatusRoutes(ginApp)
 	routes.DbRoutes(ginApp)
+	routes.ReplayRoutes(ginApp)
 
 	return ginApp
 }
@@ -155,7 +156,7 @@ func runInTapperMode() {
 
 	hostMode := os.Getenv(shared.HostModeEnvVar) == "1"
 	tapOpts := &tap.TapOpts{
-		HostMode:     hostMode,
+		HostMode: hostMode,
 	}
 
 	filteredOutputItemsChannel := make(chan *tapApi.OutputChannelItem)

agent/pkg/api/entry_streamer_socket_connector.go

@@ -22,7 +22,16 @@ func (e *DefaultEntryStreamerSocketConnector) SendEntry(socketId int, entry *tap
 	if params.EnableFullEntries {
 		message, _ = models.CreateFullEntryWebSocketMessage(entry)
 	} else {
-		extension := extensionsMap[entry.Protocol.Name]
+		protocol, ok := protocolsMap[entry.ProtocolId]
+		if !ok {
+			return fmt.Errorf("protocol not found, protocol: %v", protocol)
+		}
+
+		extension, ok := extensionsMap[protocol.Name]
+		if !ok {
+			return fmt.Errorf("extension not found, extension: %v", protocol.Name)
+		}
+
 		base := extension.Dissector.Summarize(entry)
 		message, _ = models.CreateBaseEntryWebSocketMessage(base)
 	}

agent/pkg/api/main.go

@@ -134,7 +134,7 @@ func startReadingChannel(outputItems <-chan *tapApi.OutputChannelItem, extension
 		serviceMapGenerator.NewTCPEntry(mizuEntry.Source, mizuEntry.Destination, &item.Protocol)
 
 		oasGenerator := dependency.GetInstance(dependency.OasGeneratorDependency).(oas.OasGeneratorSink)
-		oasGenerator.HandleEntry(mizuEntry)
+		oasGenerator.HandleEntry(mizuEntry, &item.Protocol)
 	}
 }
 

agent/pkg/api/socket_routes.go

@@ -14,10 +14,14 @@ import (
 	tapApi "github.com/up9inc/mizu/tap/api"
 )
 
-var extensionsMap map[string]*tapApi.Extension // global
+var (
+	extensionsMap map[string]*tapApi.Extension // global
+	protocolsMap  map[string]*tapApi.Protocol  //global
+)
 
-func InitExtensionsMap(ref map[string]*tapApi.Extension) {
+func InitMaps(extensions map[string]*tapApi.Extension, protocols map[string]*tapApi.Protocol) {
-	extensionsMap = ref
+	extensionsMap = extensions
+	protocolsMap = protocols
 }
 
 type EventHandlers interface {

agent/pkg/app/main.go

@@ -11,8 +11,8 @@ import (
 	"github.com/up9inc/mizu/agent/pkg/api"
 	"github.com/up9inc/mizu/agent/pkg/utils"
 	"github.com/up9inc/mizu/logger"
-	"github.com/up9inc/mizu/tap/dbgctl"
 	tapApi "github.com/up9inc/mizu/tap/api"
+	"github.com/up9inc/mizu/tap/dbgctl"
 	amqpExt "github.com/up9inc/mizu/tap/extensions/amqp"
 	httpExt "github.com/up9inc/mizu/tap/extensions/http"
 	kafkaExt "github.com/up9inc/mizu/tap/extensions/kafka"
@@ -22,11 +22,13 @@ import (
 var (
 	Extensions    []*tapApi.Extension          // global
 	ExtensionsMap map[string]*tapApi.Extension // global
+	ProtocolsMap  map[string]*tapApi.Protocol  //global
 )
 
 func LoadExtensions() {
 	Extensions = make([]*tapApi.Extension, 0)
 	ExtensionsMap = make(map[string]*tapApi.Extension)
+	ProtocolsMap = make(map[string]*tapApi.Protocol)
 
 	extensionHttp := &tapApi.Extension{}
 	dissectorHttp := httpExt.NewDissector()
@@ -34,6 +36,10 @@ func LoadExtensions() {
 	extensionHttp.Dissector = dissectorHttp
 	Extensions = append(Extensions, extensionHttp)
 	ExtensionsMap[extensionHttp.Protocol.Name] = extensionHttp
+	protocolsHttp := dissectorHttp.GetProtocols()
+	for k, v := range protocolsHttp {
+		ProtocolsMap[k] = v
+	}
 
 	if !dbgctl.MizuTapperDisableNonHttpExtensions {
 		extensionAmqp := &tapApi.Extension{}
@@ -42,6 +48,10 @@ func LoadExtensions() {
 		extensionAmqp.Dissector = dissectorAmqp
 		Extensions = append(Extensions, extensionAmqp)
 		ExtensionsMap[extensionAmqp.Protocol.Name] = extensionAmqp
+		protocolsAmqp := dissectorAmqp.GetProtocols()
+		for k, v := range protocolsAmqp {
+			ProtocolsMap[k] = v
+		}
 
 		extensionKafka := &tapApi.Extension{}
 		dissectorKafka := kafkaExt.NewDissector()
@@ -49,6 +59,10 @@ func LoadExtensions() {
 		extensionKafka.Dissector = dissectorKafka
 		Extensions = append(Extensions, extensionKafka)
 		ExtensionsMap[extensionKafka.Protocol.Name] = extensionKafka
+		protocolsKafka := dissectorKafka.GetProtocols()
+		for k, v := range protocolsKafka {
+			ProtocolsMap[k] = v
+		}
 
 		extensionRedis := &tapApi.Extension{}
 		dissectorRedis := redisExt.NewDissector()
@@ -56,13 +70,17 @@ func LoadExtensions() {
 		extensionRedis.Dissector = dissectorRedis
 		Extensions = append(Extensions, extensionRedis)
 		ExtensionsMap[extensionRedis.Protocol.Name] = extensionRedis
+		protocolsRedis := dissectorRedis.GetProtocols()
+		for k, v := range protocolsRedis {
+			ProtocolsMap[k] = v
+		}
 	}
 
 	sort.Slice(Extensions, func(i, j int) bool {
 		return Extensions[i].Protocol.Priority < Extensions[j].Protocol.Priority
 	})
 
-	api.InitExtensionsMap(ExtensionsMap)
+	api.InitMaps(ExtensionsMap, ProtocolsMap)
 }
 
 func ConfigureBasenineServer(host string, port string, dbSize int64, logLevel logging.Level, insertionFilter string) {

agent/pkg/controllers/replay_controller.go

@@ -0,0 +1,34 @@
+package controllers
+
+import (
+	"net/http"
+	"time"
+
+	"github.com/gin-gonic/gin"
+	"github.com/up9inc/mizu/agent/pkg/replay"
+	"github.com/up9inc/mizu/agent/pkg/validation"
+	"github.com/up9inc/mizu/logger"
+)
+
+const (
+	replayTimeout = 10 * time.Second
+)
+
+func ReplayRequest(c *gin.Context) {
+	logger.Log.Debug("Starting replay")
+	replayDetails := &replay.Details{}
+	if err := c.Bind(replayDetails); err != nil {
+		c.JSON(http.StatusBadRequest, err)
+		return
+	}
+
+	logger.Log.Debugf("Validating replay, %v", replayDetails)
+	if err := validation.Validate(replayDetails); err != nil {
+		c.JSON(http.StatusBadRequest, err)
+		return
+	}
+
+	logger.Log.Debug("Executing replay, %v", replayDetails)
+	result := replay.ExecuteRequest(replayDetails, replayTimeout)
+	c.JSON(http.StatusOK, result)
+}

agent/pkg/entries/entries_provider.go

@@ -3,6 +3,7 @@ package entries
 import (
 	"encoding/json"
 	"errors"
+	"fmt"
 	"time"
 
 	basenine "github.com/up9inc/basenine/client/go"
@@ -38,11 +39,20 @@ func (e *BasenineEntriesProvider) GetEntries(entriesRequest *models.EntriesReque
 			return nil, nil, err
 		}
 
-		extension := app.ExtensionsMap[entry.Protocol.Name]
+		protocol, ok := app.ProtocolsMap[entry.ProtocolId]
+		if !ok {
+			return nil, nil, fmt.Errorf("protocol not found, protocol: %v", protocol)
+		}
+
+		extension, ok := app.ExtensionsMap[protocol.Name]
+		if !ok {
+			return nil, nil, fmt.Errorf("extension not found, extension: %v", protocol.Name)
+		}
+
 		base := extension.Dissector.Summarize(entry)
 
 		dataSlice = append(dataSlice, &tapApi.EntryWrapper{
-			Protocol: entry.Protocol,
+			Protocol: *protocol,
 			Data:     entry,
 			Base:     base,
 		})
@@ -68,7 +78,16 @@ func (e *BasenineEntriesProvider) GetEntry(singleEntryRequest *models.SingleEntr
 		return nil, errors.New(string(bytes))
 	}
 
-	extension := app.ExtensionsMap[entry.Protocol.Name]
+	protocol, ok := app.ProtocolsMap[entry.ProtocolId]
+	if !ok {
+		return nil, fmt.Errorf("protocol not found, protocol: %v", protocol)
+	}
+
+	extension, ok := app.ExtensionsMap[protocol.Name]
+	if !ok {
+		return nil, fmt.Errorf("extension not found, extension: %v", protocol.Name)
+	}
+
 	base := extension.Dissector.Summarize(entry)
 	var representation []byte
 	representation, err = extension.Dissector.Represent(entry.Request, entry.Response)
@@ -78,7 +97,7 @@ func (e *BasenineEntriesProvider) GetEntry(singleEntryRequest *models.SingleEntr
 
 	var rules []map[string]interface{}
 	var isRulesEnabled bool
-	if entry.Protocol.Name == "http" {
+	if protocol.Name == "http" {
 		harEntry, _ := har.NewEntry(entry.Request, entry.Response, entry.StartTime, entry.ElapsedTime)
 		_, rulesMatched, _isRulesEnabled := models.RunValidationRulesState(*harEntry, entry.Destination.Name)
 		isRulesEnabled = _isRulesEnabled
@@ -89,7 +108,7 @@ func (e *BasenineEntriesProvider) GetEntry(singleEntryRequest *models.SingleEntr
 	}
 
 	return &tapApi.EntryWrapper{
-		Protocol:       entry.Protocol,
+		Protocol:       *protocol,
 		Representation: string(representation),
 		Data:           entry,
 		Base:           base,

agent/pkg/oas/oas_generator.go

@@ -6,9 +6,8 @@ import (
 	"sync"
 
 	"github.com/up9inc/mizu/agent/pkg/har"
-	"github.com/up9inc/mizu/tap/api"
-
 	"github.com/up9inc/mizu/logger"
+	"github.com/up9inc/mizu/tap/api"
 )
 
 var (
@@ -17,7 +16,7 @@ var (
 )
 
 type OasGeneratorSink interface {
-	HandleEntry(mizuEntry *api.Entry)
+	HandleEntry(mizuEntry *api.Entry, protocol *api.Protocol)
 }
 
 type OasGenerator interface {
@@ -59,12 +58,12 @@ func (g *defaultOasGenerator) IsStarted() bool {
 	return g.started
 }
 
-func (g *defaultOasGenerator) HandleEntry(mizuEntry *api.Entry) {
+func (g *defaultOasGenerator) HandleEntry(mizuEntry *api.Entry, protocol *api.Protocol) {
 	if !g.started {
 		return
 	}
 
-	if mizuEntry.Protocol.Name == "http" {
+	if protocol.Name == "http" {
 		dest := mizuEntry.Destination.Name
 		if dest == "" {
 			logger.Log.Debugf("OAS: Unresolved entry %d", mizuEntry.Id)
@@ -86,7 +85,7 @@ func (g *defaultOasGenerator) HandleEntry(mizuEntry *api.Entry) {
 
 		g.handleHARWithSource(entryWSource)
 	} else {
-		logger.Log.Debugf("OAS: Unsupported protocol in entry %d: %s", mizuEntry.Id, mizuEntry.Protocol.Name)
+		logger.Log.Debugf("OAS: Unsupported protocol in entry %d: %s", mizuEntry.Id, protocol.Name)
 	}
 }
 

agent/pkg/providers/tappedPods/tapped_pods_provider.go

@@ -60,7 +60,7 @@ func GetTappedPodsStatus() []shared.TappedPodStatus {
 
 func SetNodeToTappedPodMap(nodeToTappedPodsMap shared.NodeToPodsMap) {
 	summary := nodeToTappedPodsMap.Summary()
-	logger.Log.Infof("Setting node to tapped pods map to %v", summary)
+	logger.Log.Debugf("Setting node to tapped pods map to %v", summary)
 
 	nodeHostToTappedPodsMap = nodeToTappedPodsMap
 }

agent/pkg/replay/replay.go

@@ -0,0 +1,186 @@
+package replay
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/up9inc/mizu/agent/pkg/app"
+	tapApi "github.com/up9inc/mizu/tap/api"
+	mizuhttp "github.com/up9inc/mizu/tap/extensions/http"
+)
+
+var (
+	inProcessRequestsLocker = sync.Mutex{}
+	inProcessRequests       = 0
+)
+
+const maxParallelAction = 5
+
+type Details struct {
+	Method  string            `json:"method"`
+	Url     string            `json:"url"`
+	Body    string            `json:"body"`
+	Headers map[string]string `json:"headers"`
+}
+
+type Response struct {
+	Success      bool        `json:"status"`
+	Data         interface{} `json:"data"`
+	ErrorMessage string      `json:"errorMessage"`
+}
+
+func incrementCounter() bool {
+	result := false
+	inProcessRequestsLocker.Lock()
+	if inProcessRequests < maxParallelAction {
+		inProcessRequests++
+		result = true
+	}
+	inProcessRequestsLocker.Unlock()
+	return result
+}
+
+func decrementCounter() {
+	inProcessRequestsLocker.Lock()
+	inProcessRequests--
+	inProcessRequestsLocker.Unlock()
+}
+
+func getEntryFromRequestResponse(extension *tapApi.Extension, request *http.Request, response *http.Response) *tapApi.Entry {
+	captureTime := time.Now()
+
+	itemTmp := tapApi.OutputChannelItem{
+		Protocol: *extension.Protocol,
+		ConnectionInfo: &tapApi.ConnectionInfo{
+			ClientIP:   "",
+			ClientPort: "1",
+			ServerIP:   "",
+			ServerPort: "1",
+			IsOutgoing: false,
+		},
+		Capture:   "",
+		Timestamp: time.Now().UnixMilli(),
+		Pair: &tapApi.RequestResponsePair{
+			Request: tapApi.GenericMessage{
+				IsRequest:   true,
+				CaptureTime: captureTime,
+				CaptureSize: 0,
+				Payload: &mizuhttp.HTTPPayload{
+					Type: mizuhttp.TypeHttpRequest,
+					Data: request,
+				},
+			},
+			Response: tapApi.GenericMessage{
+				IsRequest:   false,
+				CaptureTime: captureTime,
+				CaptureSize: 0,
+				Payload: &mizuhttp.HTTPPayload{
+					Type: mizuhttp.TypeHttpResponse,
+					Data: response,
+				},
+			},
+		},
+	}
+
+	// Analyze is expecting an item that's marshalled and unmarshalled
+	itemMarshalled, err := json.Marshal(itemTmp)
+	if err != nil {
+		return nil
+	}
+	var finalItem *tapApi.OutputChannelItem
+	if err := json.Unmarshal(itemMarshalled, &finalItem); err != nil {
+		return nil
+	}
+
+	return extension.Dissector.Analyze(finalItem, "", "", "")
+}
+
+func ExecuteRequest(replayData *Details, timeout time.Duration) *Response {
+	if incrementCounter() {
+		defer decrementCounter()
+
+		client := &http.Client{
+			Timeout: timeout,
+		}
+
+		request, err := http.NewRequest(strings.ToUpper(replayData.Method), replayData.Url, bytes.NewBufferString(replayData.Body))
+		if err != nil {
+			return &Response{
+				Success:      false,
+				Data:         nil,
+				ErrorMessage: err.Error(),
+			}
+		}
+
+		for headerKey, headerValue := range replayData.Headers {
+			request.Header.Add(headerKey, headerValue)
+		}
+		request.Header.Add("x-mizu", uuid.New().String())
+		response, requestErr := client.Do(request)
+
+		if requestErr != nil {
+			return &Response{
+				Success:      false,
+				Data:         nil,
+				ErrorMessage: requestErr.Error(),
+			}
+		}
+
+		extension := app.ExtensionsMap["http"] // # TODO: maybe pass the extension to the function so it can be tested
+		entry := getEntryFromRequestResponse(extension, request, response)
+		base := extension.Dissector.Summarize(entry)
+		var representation []byte
+
+		// Represent is expecting an entry that's marshalled and unmarshalled
+		entryMarshalled, err := json.Marshal(entry)
+		if err != nil {
+			return &Response{
+				Success:      false,
+				Data:         nil,
+				ErrorMessage: err.Error(),
+			}
+		}
+		var entryUnmarshalled *tapApi.Entry
+		if err := json.Unmarshal(entryMarshalled, &entryUnmarshalled); err != nil {
+			return &Response{
+				Success:      false,
+				Data:         nil,
+				ErrorMessage: err.Error(),
+			}
+		}
+
+		representation, err = extension.Dissector.Represent(entryUnmarshalled.Request, entryUnmarshalled.Response)
+		if err != nil {
+			return &Response{
+				Success:      false,
+				Data:         nil,
+				ErrorMessage: err.Error(),
+			}
+		}
+
+		return &Response{
+			Success: true,
+			Data: &tapApi.EntryWrapper{
+				Protocol:       *extension.Protocol,
+				Representation: string(representation),
+				Data:           entryUnmarshalled,
+				Base:           base,
+				Rules:          nil,
+				IsRulesEnabled: false,
+			},
+			ErrorMessage: "",
+		}
+	} else {
+		return &Response{
+			Success:      false,
+			Data:         nil,
+			ErrorMessage: fmt.Sprintf("reached threshold of %d requests", maxParallelAction),
+		}
+	}
+}

agent/pkg/replay/replay_internal_test.go

@@ -0,0 +1,108 @@
+package replay
+
+import (
+	"bytes"
+	"fmt"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+
+	"encoding/json"
+
+	"github.com/google/uuid"
+	tapApi "github.com/up9inc/mizu/tap/api"
+	mizuhttp "github.com/up9inc/mizu/tap/extensions/http"
+)
+
+func TestValid(t *testing.T) {
+	client := &http.Client{
+		Timeout: 10 * time.Second,
+	}
+
+	tests := map[string]*Details{
+		"40x": {
+			Method:  "GET",
+			Url:     "http://httpbin.org/status/404",
+			Body:    "",
+			Headers: map[string]string{},
+		},
+		"20x": {
+			Method:  "GET",
+			Url:     "http://httpbin.org/status/200",
+			Body:    "",
+			Headers: map[string]string{},
+		},
+		"50x": {
+			Method:  "GET",
+			Url:     "http://httpbin.org/status/500",
+			Body:    "",
+			Headers: map[string]string{},
+		},
+		// TODO: this should be fixes, currently not working because of header name with ":"
+		//":path-header": {
+		//	Method: "GET",
+		//	Url:    "http://httpbin.org/get",
+		//	Body:   "",
+		//	Headers: map[string]string{
+		//		":path": "/get",
+		//	},
+		// },
+	}
+
+	for testCaseName, replayData := range tests {
+		t.Run(fmt.Sprintf("%+v", testCaseName), func(t *testing.T) {
+			request, err := http.NewRequest(strings.ToUpper(replayData.Method), replayData.Url, bytes.NewBufferString(replayData.Body))
+			if err != nil {
+				t.Errorf("Error executing request")
+			}
+
+			for headerKey, headerValue := range replayData.Headers {
+				request.Header.Add(headerKey, headerValue)
+			}
+			request.Header.Add("x-mizu", uuid.New().String())
+			response, requestErr := client.Do(request)
+
+			if requestErr != nil {
+				t.Errorf("failed: %v, ", requestErr)
+			}
+
+			extensionHttp := &tapApi.Extension{}
+			dissectorHttp := mizuhttp.NewDissector()
+			dissectorHttp.Register(extensionHttp)
+			extensionHttp.Dissector = dissectorHttp
+			extension := extensionHttp
+
+			entry := getEntryFromRequestResponse(extension, request, response)
+			base := extension.Dissector.Summarize(entry)
+
+			// Represent is expecting an entry that's marshalled and unmarshalled
+			entryMarshalled, err := json.Marshal(entry)
+			if err != nil {
+				t.Errorf("failed marshaling entry: %v, ", err)
+			}
+			var entryUnmarshalled *tapApi.Entry
+			if err := json.Unmarshal(entryMarshalled, &entryUnmarshalled); err != nil {
+				t.Errorf("failed unmarshaling entry: %v, ", err)
+			}
+
+			var representation []byte
+			representation, err = extension.Dissector.Represent(entryUnmarshalled.Request, entryUnmarshalled.Response)
+			if err != nil {
+				t.Errorf("failed: %v, ", err)
+			}
+
+			result := &tapApi.EntryWrapper{
+				Protocol:       *extension.Protocol,
+				Representation: string(representation),
+				Data:           entry,
+				Base:           base,
+				Rules:          nil,
+				IsRulesEnabled: false,
+			}
+			t.Logf("%+v", result)
+			//data, _ := json.MarshalIndent(result, "", "  ")
+			//t.Logf("%+v", string(data))
+		})
+	}
+}

agent/pkg/routes/replay_routes.go

@@ -0,0 +1,13 @@
+package routes
+
+import (
+	"github.com/gin-gonic/gin"
+	"github.com/up9inc/mizu/agent/pkg/controllers"
+)
+
+// ReplayRoutes defines the group of replay routes.
+func ReplayRoutes(app *gin.Engine) {
+	routeGroup := app.Group("/replay")
+
+	routeGroup.POST("/", controllers.ReplayRequest)
+}

cli/cmd/tap.go

@@ -57,4 +57,5 @@ func init() {
 	tapCmd.Flags().Bool(configStructs.ServiceMeshName, defaultTapConfig.ServiceMesh, "Record decrypted traffic if the cluster is configured with a service mesh and with mtls")
 	tapCmd.Flags().Bool(configStructs.TlsName, defaultTapConfig.Tls, "Record tls traffic")
 	tapCmd.Flags().Bool(configStructs.ProfilerName, defaultTapConfig.Profiler, "Run pprof server")
+	tapCmd.Flags().Int(configStructs.MaxLiveStreamsName, defaultTapConfig.MaxLiveStreams, "Maximum live tcp streams to handle concurrently")
 }

cli/cmd/tapRunner.go

@@ -176,6 +176,7 @@ func startTapperSyncer(ctx context.Context, cancel context.CancelFunc, provider
 		MizuServiceAccountExists: state.mizuServiceAccountExists,
 		ServiceMesh:              config.Config.Tap.ServiceMesh,
 		Tls:                      config.Config.Tap.Tls,
+		MaxLiveStreams:           config.Config.Tap.MaxLiveStreams,
 	}, startTime)
 
 	if err != nil {

cli/config/configStructs/tapConfig.go

@@ -27,6 +27,7 @@ const (
 	ServiceMeshName               = "service-mesh"
 	TlsName                       = "tls"
 	ProfilerName                  = "profiler"
+	MaxLiveStreamsName            = "max-live-streams"
 )
 
 type TapConfig struct {
@@ -47,6 +48,7 @@ type TapConfig struct {
 	ServiceMesh            bool             `yaml:"service-mesh" default:"false"`
 	Tls                    bool             `yaml:"tls" default:"false"`
 	Profiler               bool             `yaml:"profiler" default:"false"`
+	MaxLiveStreams         int              `yaml:"max-live-streams" default:"500"`
 }
 
 func (config *TapConfig) PodRegex() *regexp.Regexp {

devops/install-capstone.sh

@@ -5,8 +5,8 @@ if (( $EUID != 0 )); then
     SUDO='sudo'
 fi
 
-curl https://github.com/capstone-engine/capstone/archive/4.0.2.tar.gz -Lo ./capstone.tar.gz \
+curl https://github.com/capstone-engine/capstone/releases/download/5.0-rc2/capstone-5.0-rc2.tar.xz -Lo ./capstone.tar.xz \
- && tar -xzf capstone.tar.gz && mv ./capstone-* ./capstone \
+ && tar -xf capstone.tar.xz && mv ./capstone-* ./capstone \
  && cd capstone \
  && CAPSTONE_ARCHS="aarch64 x86" ./make.sh \
  && $SUDO ./make.sh install

devops/linux-arm64-musl-go-libpcap-capstone-bpf/Dockerfile

@@ -20,8 +20,8 @@ RUN ./configure --host=arm && make \
 WORKDIR /work
 
 # Build and install Capstone from source
-RUN curl https://github.com/capstone-engine/capstone/archive/4.0.2.tar.gz -Lo ./capstone.tar.gz \
+RUN curl https://github.com/capstone-engine/capstone/releases/download/5.0-rc2/capstone-5.0-rc2.tar.xz -Lo ./capstone.tar.xz \
- && tar -xzf capstone.tar.gz && mv ./capstone-* ./capstone
+ && tar -xf capstone.tar.xz && mv ./capstone-* ./capstone
 WORKDIR /work/capstone
 RUN CAPSTONE_ARCHS="aarch64" CAPSTONE_STATIC=yes ./make.sh \
 && cp /work/capstone/libcapstone.a /usr/xcc/aarch64-linux-musl-cross/lib/gcc/aarch64-linux-musl/*/

devops/linux-arm64-musl-go-libpcap-capstone-bpf/build-push.sh

@@ -1,4 +1,5 @@
 #!/bin/bash
 set -e
 
-docker build . -t up9inc/linux-arm64-musl-go-libpcap-capstone-bpf && docker push up9inc/linux-arm64-musl-go-libpcap-capstone-bpf
+# Build it on x86_64
+docker build . -t up9inc/linux-arm64-musl-go-libpcap-capstone-bpf:capstone-5.0-rc2 && docker push up9inc/linux-arm64-musl-go-libpcap-capstone-bpf:capstone-5.0-rc2

devops/linux-x86_64-musl-go-libpcap-capstone-bpf/Dockerfile

@@ -44,8 +44,8 @@ RUN ./configure --host=x86_64 && make \
 WORKDIR /
 
 # Build and install Capstone from source
-RUN curl https://github.com/capstone-engine/capstone/archive/4.0.2.tar.gz -Lo ./capstone.tar.gz \
+RUN curl https://github.com/capstone-engine/capstone/releases/download/5.0-rc2/capstone-5.0-rc2.tar.xz -Lo ./capstone.tar.xz \
- && tar -xzf capstone.tar.gz && mv ./capstone-* ./capstone
+ && tar -xf capstone.tar.xz && mv ./capstone-* ./capstone
 WORKDIR /capstone
 RUN CAPSTONE_ARCHS="x86" CAPSTONE_STATIC=yes ./make.sh \
  && cp /capstone/libcapstone.a /usr/local/musl/lib/gcc/x86_64-unknown-linux-musl/*/

devops/linux-x86_64-musl-go-libpcap-capstone-bpf/build-push.sh

@@ -1,4 +1,5 @@
 #!/bin/bash
 set -e
 
-docker build . -t up9inc/linux-x86_64-musl-go-libpcap-capstone-bpf && docker push up9inc/linux-x86_64-musl-go-libpcap-capstone-bpf
+# Build it on arm64
+docker build . -t up9inc/linux-x86_64-musl-go-libpcap-capstone-bpf:capstone-5.0-rc2 && docker push up9inc/linux-x86_64-musl-go-libpcap-capstone-bpf:capstone-5.0-rc2

shared/kubernetes/mizuTapperSyncer.go

@@ -48,6 +48,7 @@ type TapperSyncerConfig struct {
 	MizuServiceAccountExists bool
 	ServiceMesh              bool
 	Tls                      bool
+	MaxLiveStreams           int
 }
 
 func CreateAndStartMizuTapperSyncer(ctx context.Context, kubernetesProvider *Provider, config TapperSyncerConfig, startTime time.Time) (*MizuTapperSyncer, error) {
@@ -337,7 +338,8 @@ func (tapperSyncer *MizuTapperSyncer) updateMizuTappers() error {
 			tapperSyncer.config.MizuApiFilteringOptions,
 			tapperSyncer.config.LogLevel,
 			tapperSyncer.config.ServiceMesh,
-			tapperSyncer.config.Tls); err != nil {
+			tapperSyncer.config.Tls,
+			tapperSyncer.config.MaxLiveStreams); err != nil {
 			return err
 		}
 

shared/kubernetes/provider.go

@@ -10,6 +10,7 @@ import (
 	"net/url"
 	"path/filepath"
 	"regexp"
+	"strconv"
 
 	"github.com/op/go-logging"
 	"github.com/up9inc/mizu/logger"
@@ -382,11 +383,11 @@ func (provider *Provider) GetMizuApiServerPodObject(opts *ApiServerOptions, moun
 			Tolerations: []core.Toleration{
 				{
 					Operator: core.TolerationOpExists,
-					Effect: core.TaintEffectNoExecute,
+					Effect:   core.TaintEffectNoExecute,
 				},
 				{
 					Operator: core.TolerationOpExists,
-					Effect: core.TaintEffectNoSchedule,
+					Effect:   core.TaintEffectNoSchedule,
 				},
 			},
 		},
@@ -711,7 +712,7 @@ func (provider *Provider) CreateConfigMap(ctx context.Context, namespace string,
 	return nil
 }
 
-func (provider *Provider) ApplyMizuTapperDaemonSet(ctx context.Context, namespace string, daemonSetName string, podImage string, tapperPodName string, apiServerPodIp string, nodeNames []string, serviceAccountName string, resources shared.Resources, imagePullPolicy core.PullPolicy, mizuApiFilteringOptions api.TrafficFilteringOptions, logLevel logging.Level, serviceMesh bool, tls bool) error {
+func (provider *Provider) ApplyMizuTapperDaemonSet(ctx context.Context, namespace string, daemonSetName string, podImage string, tapperPodName string, apiServerPodIp string, nodeNames []string, serviceAccountName string, resources shared.Resources, imagePullPolicy core.PullPolicy, mizuApiFilteringOptions api.TrafficFilteringOptions, logLevel logging.Level, serviceMesh bool, tls bool, maxLiveStreams int) error {
 	logger.Log.Debugf("Applying %d tapper daemon sets, ns: %s, daemonSetName: %s, podImage: %s, tapperPodName: %s", len(nodeNames), namespace, daemonSetName, podImage, tapperPodName)
 
 	if len(nodeNames) == 0 {
@@ -729,6 +730,7 @@ func (provider *Provider) ApplyMizuTapperDaemonSet(ctx context.Context, namespac
 		"--tap",
 		"--api-server-address", fmt.Sprintf("ws://%s/wsTapper", apiServerPodIp),
 		"--nodefrag",
+		"--max-live-streams", strconv.Itoa(maxLiveStreams),
 	}
 
 	if serviceMesh {

shared/models.go

@@ -55,7 +55,6 @@ type WebSocketMessageMetadata struct {
 	MessageType WebSocketMessageType `json:"messageType,omitempty"`
 }
 
-
 type WebSocketStatusMessage struct {
 	*WebSocketMessageMetadata
 	TappingStatus []TappedPodStatus `json:"tappingStatus"`

tap/api/api.go

@@ -91,7 +91,6 @@ type OutputChannelItem struct {
 	Timestamp      int64
 	ConnectionInfo *ConnectionInfo
 	Pair           *RequestResponsePair
-	Summary        *BaseEntry
 	Namespace      string
 }
 
@@ -116,6 +115,7 @@ func (p *ReadProgress) Reset() {
 
 type Dissector interface {
 	Register(*Extension)
+	GetProtocols() map[string]*Protocol
 	Ping()
 	Dissect(b *bufio.Reader, reader TcpReader, options *TrafficFilteringOptions) error
 	Analyze(item *OutputChannelItem, resolvedSource string, resolvedDestination string, namespace string) *Entry
@@ -151,7 +151,7 @@ func (e *Emitting) Emit(item *OutputChannelItem) {
 
 type Entry struct {
 	Id           string                 `json:"id"`
-	Protocol     Protocol               `json:"proto"`
+	ProtocolId   string                 `json:"protocol"`
 	Capture      Capture                `json:"capture"`
 	Source       *TCP                   `json:"src"`
 	Destination  *TCP                   `json:"dst"`

tap/extensions/amqp/Makefile

@@ -13,4 +13,4 @@ test-pull-bin:
 
 test-pull-expect:
 	@mkdir -p expect
-	@[ "${skipexpect}" ] && echo "Skipping downloading expected JSONs" || gsutil  -o 'GSUtil:parallel_process_count=5' -o 'GSUtil:parallel_thread_count=5'  -m cp -r gs://static.up9.io/mizu/test-pcap/expect11/amqp/\* expect
+	@[ "${skipexpect}" ] && echo "Skipping downloading expected JSONs" || gsutil  -o 'GSUtil:parallel_process_count=5' -o 'GSUtil:parallel_thread_count=5'  -m cp -r gs://static.up9.io/mizu/test-pcap/expect13/amqp/\* expect

tap/extensions/amqp/main.go

@@ -26,12 +26,20 @@ var protocol = api.Protocol{
 	Priority:        1,
 }
 
+var protocolsMap = map[string]*api.Protocol{
+	fmt.Sprintf("%s/%s/%s", protocol.Name, protocol.Version, protocol.Abbreviation): &protocol,
+}
+
 type dissecting string
 
 func (d dissecting) Register(extension *api.Extension) {
 	extension.Protocol = &protocol
 }
 
+func (d dissecting) GetProtocols() map[string]*api.Protocol {
+	return protocolsMap
+}
+
 func (d dissecting) Ping() {
 	log.Printf("pong %s", protocol.Name)
 }
@@ -214,8 +222,8 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, resolvedSource string,
 
 	reqDetails["method"] = request["method"]
 	return &api.Entry{
-		Protocol: protocol,
+		ProtocolId: fmt.Sprintf("%s/%s/%s", protocol.Name, protocol.Version, protocol.Abbreviation),
-		Capture:  item.Capture,
+		Capture:    item.Capture,
 		Source: &api.TCP{
 			Name: resolvedSource,
 			IP:   item.ConnectionInfo.ClientIP,
@@ -277,7 +285,7 @@ func (d dissecting) Summarize(entry *api.Entry) *api.BaseEntry {
 
 	return &api.BaseEntry{
 		Id:           entry.Id,
-		Protocol:     entry.Protocol,
+		Protocol:     *protocolsMap[entry.ProtocolId],
 		Capture:      entry.Capture,
 		Summary:      summary,
 		SummaryQuery: summaryQuery,
@@ -322,7 +330,7 @@ func (d dissecting) Represent(request map[string]interface{}, response map[strin
 
 func (d dissecting) Macros() map[string]string {
 	return map[string]string{
-		`amqp`: fmt.Sprintf(`proto.name == "%s"`, protocol.Name),
+		`amqp`: fmt.Sprintf(`protocol == "%s/%s/%s"`, protocol.Name, protocol.Version, protocol.Abbreviation),
 	}
 }
 

tap/extensions/amqp/main_test.go

@@ -44,7 +44,7 @@ func TestRegister(t *testing.T) {
 
 func TestMacros(t *testing.T) {
 	expectedMacros := map[string]string{
-		"amqp": `proto.name == "amqp"`,
+		"amqp": `protocol == "amqp/0-9-1/AMQP"`,
 	}
 	dissector := NewDissector()
 	macros := dissector.Macros()

tap/extensions/http/Makefile

@@ -13,4 +13,4 @@ test-pull-bin:
 
 test-pull-expect:
 	@mkdir -p expect
-	@[ "${skipexpect}" ] && echo "Skipping downloading expected JSONs" || gsutil  -o 'GSUtil:parallel_process_count=5' -o 'GSUtil:parallel_thread_count=5'  -m cp -r gs://static.up9.io/mizu/test-pcap/expect12/http/\* expect
+	@[ "${skipexpect}" ] && echo "Skipping downloading expected JSONs" || gsutil  -o 'GSUtil:parallel_process_count=5' -o 'GSUtil:parallel_thread_count=5'  -m cp -r gs://static.up9.io/mizu/test-pcap/expect13/http/\* expect

tap/extensions/http/main.go

@@ -98,6 +98,15 @@ var graphQL2Protocol = api.Protocol{
 	Priority:        0,
 }
 
+var protocolsMap = map[string]*api.Protocol{
+	fmt.Sprintf("%s/%s/%s", http10protocol.Name, http10protocol.Version, http10protocol.Abbreviation):       &http10protocol,
+	fmt.Sprintf("%s/%s/%s", http11protocol.Name, http11protocol.Version, http11protocol.Abbreviation):       &http11protocol,
+	fmt.Sprintf("%s/%s/%s", http2Protocol.Name, http2Protocol.Version, http2Protocol.Abbreviation):          &http2Protocol,
+	fmt.Sprintf("%s/%s/%s", grpcProtocol.Name, grpcProtocol.Version, grpcProtocol.Abbreviation):             &grpcProtocol,
+	fmt.Sprintf("%s/%s/%s", graphQL1Protocol.Name, graphQL1Protocol.Version, graphQL1Protocol.Abbreviation): &graphQL1Protocol,
+	fmt.Sprintf("%s/%s/%s", graphQL2Protocol.Name, graphQL2Protocol.Version, graphQL2Protocol.Abbreviation): &graphQL2Protocol,
+}
+
 const (
 	TypeHttpRequest = iota
 	TypeHttpResponse
@@ -109,6 +118,10 @@ func (d dissecting) Register(extension *api.Extension) {
 	extension.Protocol = &http11protocol
 }
 
+func (d dissecting) GetProtocols() map[string]*api.Protocol {
+	return protocolsMap
+}
+
 func (d dissecting) Ping() {
 	log.Printf("pong %s", http11protocol.Name)
 }
@@ -281,8 +294,8 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, resolvedSource string,
 	}
 
 	return &api.Entry{
-		Protocol: item.Protocol,
+		ProtocolId: fmt.Sprintf("%s/%s/%s", item.Protocol.Name, item.Protocol.Version, item.Protocol.Abbreviation),
-		Capture:  item.Capture,
+		Capture:    item.Capture,
 		Source: &api.TCP{
 			Name: resolvedSource,
 			IP:   item.ConnectionInfo.ClientIP,
@@ -315,7 +328,7 @@ func (d dissecting) Summarize(entry *api.Entry) *api.BaseEntry {
 
 	return &api.BaseEntry{
 		Id:           entry.Id,
-		Protocol:     entry.Protocol,
+		Protocol:     *protocolsMap[entry.ProtocolId],
 		Capture:      entry.Capture,
 		Summary:      summary,
 		SummaryQuery: summaryQuery,
@@ -503,10 +516,10 @@ func (d dissecting) Represent(request map[string]interface{}, response map[strin
 
 func (d dissecting) Macros() map[string]string {
 	return map[string]string{
-		`http`:  fmt.Sprintf(`proto.name == "%s" and proto.version.startsWith("%c")`, http11protocol.Name, http11protocol.Version[0]),
+		`http`:  fmt.Sprintf(`protocol == "%s/%s/%s" or protocol == "%s/%s/%s"`, http10protocol.Name, http10protocol.Version, http10protocol.Abbreviation, http11protocol.Name, http11protocol.Version, http11protocol.Abbreviation),
-		`http2`: fmt.Sprintf(`proto.name == "%s" and proto.version == "%s"`, http11protocol.Name, http2Protocol.Version),
+		`http2`: fmt.Sprintf(`protocol == "%s/%s/%s"`, http2Protocol.Name, http2Protocol.Version, http2Protocol.Abbreviation),
-		`grpc`:  fmt.Sprintf(`proto.name == "%s" and proto.version == "%s" and proto.macro == "%s"`, http11protocol.Name, grpcProtocol.Version, grpcProtocol.Macro),
+		`grpc`:  fmt.Sprintf(`protocol == "%s/%s/%s"`, grpcProtocol.Name, grpcProtocol.Version, grpcProtocol.Abbreviation),
-		`gql`:   fmt.Sprintf(`proto.name == "%s" and proto.macro == "%s"`, graphQL1Protocol.Name, graphQL1Protocol.Macro),
+		`gql`:   fmt.Sprintf(`protocol == "%s/%s/%s" or protocol == "%s/%s/%s"`, graphQL1Protocol.Name, graphQL1Protocol.Version, graphQL1Protocol.Abbreviation, graphQL2Protocol.Name, graphQL2Protocol.Version, graphQL2Protocol.Abbreviation),
 	}
 }
 

tap/extensions/http/main_test.go

@@ -44,10 +44,10 @@ func TestRegister(t *testing.T) {
 
 func TestMacros(t *testing.T) {
 	expectedMacros := map[string]string{
-		"http":  `proto.name == "http" and proto.version.startsWith("1")`,
+		"http":  `protocol == "http/1.0/HTTP" or protocol == "http/1.1/HTTP"`,
-		"http2": `proto.name == "http" and proto.version == "2.0"`,
+		"http2": `protocol == "http/2.0/HTTP/2"`,
-		"grpc":  `proto.name == "http" and proto.version == "2.0" and proto.macro == "grpc"`,
+		"grpc":  `protocol == "http/2.0/gRPC"`,
-		"gql":   `proto.name == "http" and proto.macro == "gql"`,
+		"gql":   `protocol == "http/1.1/GQL" or protocol == "http/2.0/GQL"`,
 	}
 	dissector := NewDissector()
 	macros := dissector.Macros()

tap/extensions/kafka/Makefile

@@ -13,4 +13,4 @@ test-pull-bin:
 
 test-pull-expect:
 	@mkdir -p expect
-	@[ "${skipexpect}" ] && echo "Skipping downloading expected JSONs" || gsutil  -o 'GSUtil:parallel_process_count=5' -o 'GSUtil:parallel_thread_count=5'  -m cp -r gs://static.up9.io/mizu/test-pcap/expect11/kafka/\* expect
+	@[ "${skipexpect}" ] && echo "Skipping downloading expected JSONs" || gsutil  -o 'GSUtil:parallel_process_count=5' -o 'GSUtil:parallel_thread_count=5'  -m cp -r gs://static.up9.io/mizu/test-pcap/expect13/kafka/\* expect

tap/extensions/kafka/main.go

@@ -24,12 +24,20 @@ var _protocol = api.Protocol{
 	Priority:        2,
 }
 
+var protocolsMap = map[string]*api.Protocol{
+	fmt.Sprintf("%s/%s/%s", _protocol.Name, _protocol.Version, _protocol.Abbreviation): &_protocol,
+}
+
 type dissecting string
 
 func (d dissecting) Register(extension *api.Extension) {
 	extension.Protocol = &_protocol
 }
 
+func (d dissecting) GetProtocols() map[string]*api.Protocol {
+	return protocolsMap
+}
+
 func (d dissecting) Ping() {
 	log.Printf("pong %s", _protocol.Name)
 }
@@ -62,8 +70,8 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, resolvedSource string,
 		elapsedTime = 0
 	}
 	return &api.Entry{
-		Protocol: _protocol,
+		ProtocolId: fmt.Sprintf("%s/%s/%s", _protocol.Name, _protocol.Version, _protocol.Abbreviation),
-		Capture:  item.Capture,
+		Capture:    item.Capture,
 		Source: &api.TCP{
 			Name: resolvedSource,
 			IP:   item.ConnectionInfo.ClientIP,
@@ -187,7 +195,7 @@ func (d dissecting) Summarize(entry *api.Entry) *api.BaseEntry {
 
 	return &api.BaseEntry{
 		Id:           entry.Id,
-		Protocol:     entry.Protocol,
+		Protocol:     *protocolsMap[entry.ProtocolId],
 		Capture:      entry.Capture,
 		Summary:      summary,
 		SummaryQuery: summaryQuery,
@@ -243,7 +251,7 @@ func (d dissecting) Represent(request map[string]interface{}, response map[strin
 
 func (d dissecting) Macros() map[string]string {
 	return map[string]string{
-		`kafka`: fmt.Sprintf(`proto.name == "%s"`, _protocol.Name),
+		`kafka`: fmt.Sprintf(`protocol == "%s/%s/%s"`, _protocol.Name, _protocol.Version, _protocol.Abbreviation),
 	}
 }
 

tap/extensions/kafka/main_test.go

@@ -44,7 +44,7 @@ func TestRegister(t *testing.T) {
 
 func TestMacros(t *testing.T) {
 	expectedMacros := map[string]string{
-		"kafka": `proto.name == "kafka"`,
+		"kafka": `protocol == "kafka/12/KAFKA"`,
 	}
 	dissector := NewDissector()
 	macros := dissector.Macros()

tap/extensions/redis/Makefile

@@ -13,4 +13,4 @@ test-pull-bin:
 
 test-pull-expect:
 	@mkdir -p expect
-	@[ "${skipexpect}" ] && echo "Skipping downloading expected JSONs" || gsutil  -o 'GSUtil:parallel_process_count=5' -o 'GSUtil:parallel_thread_count=5'  -m cp -r gs://static.up9.io/mizu/test-pcap/expect11/redis/\* expect
+	@[ "${skipexpect}" ] && echo "Skipping downloading expected JSONs" || gsutil  -o 'GSUtil:parallel_process_count=5' -o 'GSUtil:parallel_thread_count=5'  -m cp -r gs://static.up9.io/mizu/test-pcap/expect13/redis/\* expect

tap/extensions/redis/main.go

@@ -24,12 +24,20 @@ var protocol = api.Protocol{
 	Priority:        3,
 }
 
+var protocolsMap = map[string]*api.Protocol{
+	fmt.Sprintf("%s/%s/%s", protocol.Name, protocol.Version, protocol.Abbreviation): &protocol,
+}
+
 type dissecting string
 
 func (d dissecting) Register(extension *api.Extension) {
 	extension.Protocol = &protocol
 }
 
+func (d dissecting) GetProtocols() map[string]*api.Protocol {
+	return protocolsMap
+}
+
 func (d dissecting) Ping() {
 	log.Printf("pong %s", protocol.Name)
 }
@@ -70,8 +78,8 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, resolvedSource string,
 		elapsedTime = 0
 	}
 	return &api.Entry{
-		Protocol: protocol,
+		ProtocolId: fmt.Sprintf("%s/%s/%s", protocol.Name, protocol.Version, protocol.Abbreviation),
-		Capture:  item.Capture,
+		Capture:    item.Capture,
 		Source: &api.TCP{
 			Name: resolvedSource,
 			IP:   item.ConnectionInfo.ClientIP,
@@ -115,7 +123,7 @@ func (d dissecting) Summarize(entry *api.Entry) *api.BaseEntry {
 
 	return &api.BaseEntry{
 		Id:           entry.Id,
-		Protocol:     entry.Protocol,
+		Protocol:     *protocolsMap[entry.ProtocolId],
 		Capture:      entry.Capture,
 		Summary:      summary,
 		SummaryQuery: summaryQuery,
@@ -144,7 +152,7 @@ func (d dissecting) Represent(request map[string]interface{}, response map[strin
 
 func (d dissecting) Macros() map[string]string {
 	return map[string]string{
-		`redis`: fmt.Sprintf(`proto.name == "%s"`, protocol.Name),
+		`redis`: fmt.Sprintf(`protocol == "%s/%s/%s"`, protocol.Name, protocol.Version, protocol.Abbreviation),
 	}
 }
 

tap/extensions/redis/main_test.go

@@ -45,7 +45,7 @@ func TestRegister(t *testing.T) {
 
 func TestMacros(t *testing.T) {
 	expectedMacros := map[string]string{
-		"redis": `proto.name == "redis"`,
+		"redis": `protocol == "redis/3.x/REDIS"`,
 	}
 	dissector := NewDissector()
 	macros := dissector.Macros()

tap/go.mod

@@ -8,7 +8,7 @@ require (
 	github.com/go-errors/errors v1.4.2
 	github.com/google/gopacket v1.1.19
 	github.com/hashicorp/golang-lru v0.5.4
-	github.com/knightsc/gapstone v0.0.0-20211014144438-5e0e64002a6e
+	github.com/knightsc/gapstone v0.0.0-20191231144527-6fa5afaf11a9
 	github.com/shirou/gopsutil v3.21.11+incompatible
 	github.com/struCoder/pidusage v0.2.1
 	github.com/up9inc/mizu/logger v0.0.0

tap/go.sum

@@ -81,8 +81,8 @@ github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnr
 github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
-github.com/knightsc/gapstone v0.0.0-20211014144438-5e0e64002a6e h1:6J5obSn9umEThiYzWzndcPOZR0Qj/sVCZpH6V1G7yNE=
+github.com/knightsc/gapstone v0.0.0-20191231144527-6fa5afaf11a9 h1:1KszOoXSFt0aRQ6wxxcKm7QKgfLPI0TWO47UcY/f+vA=
-github.com/knightsc/gapstone v0.0.0-20211014144438-5e0e64002a6e/go.mod h1:1K5hEzsMBLTPdRJKEHqBFJ8Zt2VRqDhomcQ11KH0WW4=
+github.com/knightsc/gapstone v0.0.0-20191231144527-6fa5afaf11a9/go.mod h1:1K5hEzsMBLTPdRJKEHqBFJ8Zt2VRqDhomcQ11KH0WW4=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pretty v0.2.0/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
 github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0=

tap/tcp_assembler.go

@@ -85,7 +85,7 @@ func NewTcpAssembler(outputItems chan *api.OutputChannelItem, streamsMap api.Tcp
 
 	maxBufferedPagesTotal := GetMaxBufferedPagesPerConnection()
 	maxBufferedPagesPerConnection := GetMaxBufferedPagesTotal()
-	logger.Log.Infof("Assembler options: maxBufferedPagesTotal=%d, maxBufferedPagesPerConnection=%d, opts=%v",
+	logger.Log.Infof("Assembler options: maxBufferedPagesTotal=%d, maxBufferedPagesPerConnection=%d, opts=%+v",
 		maxBufferedPagesTotal, maxBufferedPagesPerConnection, opts)
 	a.Assembler.AssemblerOptions.MaxBufferedPagesTotal = maxBufferedPagesTotal
 	a.Assembler.AssemblerOptions.MaxBufferedPagesPerConnection = maxBufferedPagesPerConnection

tap/tlstapper/bpf/go_uprobes.c

@@ -60,7 +60,18 @@ Capstone Engine: https://www.capstone-engine.org/
 
 static __always_inline __u32 go_crypto_tls_get_fd_from_tcp_conn(struct pt_regs *ctx) {
     struct go_interface conn;
-    long err = bpf_probe_read(&conn, sizeof(conn), (void*)GO_ABI_INTERNAL_PT_REGS_R1(ctx));
+    long err;
+    __u64 addr;
+#if defined(bpf_target_arm64)
+    err = bpf_probe_read(&addr, sizeof(addr), (void*)GO_ABI_INTERNAL_PT_REGS_SP(ctx)+0x8);
+    if (err != 0) {
+        return invalid_fd;
+    }
+#else
+    addr = GO_ABI_INTERNAL_PT_REGS_R1(ctx);
+#endif
+
+    err = bpf_probe_read(&conn, sizeof(conn), (void*)addr);
     if (err != 0) {
         return invalid_fd;
     }
@@ -88,14 +99,23 @@ static __always_inline void go_crypto_tls_uprobe(struct pt_regs *ctx, struct bpf
     }
 
     struct ssl_info info = new_ssl_info();
+    long err;
 
+#if defined(bpf_target_arm64)
+    err = bpf_probe_read(&info.buffer_len, sizeof(__u32), (void*)GO_ABI_INTERNAL_PT_REGS_SP(ctx)+0x18);
+    if (err != 0) {
+        log_error(ctx, LOG_ERROR_READING_BYTES_COUNT, pid_tgid, err, ORIGIN_SSL_UPROBE_CODE);
+        return;
+    }
+#else
     info.buffer_len = GO_ABI_INTERNAL_PT_REGS_R2(ctx);
+#endif
     info.buffer = (void*)GO_ABI_INTERNAL_PT_REGS_R4(ctx);
     info.fd = go_crypto_tls_get_fd_from_tcp_conn(ctx);
 
     // GO_ABI_INTERNAL_PT_REGS_GP is Goroutine address
     __u64 pid_fp = pid << 32 | GO_ABI_INTERNAL_PT_REGS_GP(ctx);
-    long err = bpf_map_update_elem(go_context, &pid_fp, &info, BPF_ANY);
+    err = bpf_map_update_elem(go_context, &pid_fp, &info, BPF_ANY);
 
     if (err != 0) {
         log_error(ctx, LOG_ERROR_PUTTING_SSL_CONTEXT, pid_tgid, err, 0l);
@@ -130,7 +150,15 @@ static __always_inline void go_crypto_tls_ex_uprobe(struct pt_regs *ctx, struct
 
     // In case of read, the length is determined on return
     if (flags == FLAGS_IS_READ_BIT) {
+#if defined(bpf_target_arm64)
+        // On ARM64 we look at a general-purpose register as an indicator of error return
+        if (GO_ABI_INTERNAL_PT_REGS_R6(ctx) == 0x10) {
+            return;
+        }
+        info.buffer_len = GO_ABI_INTERNAL_PT_REGS_R7(ctx); // n in return n, nil
+#else
         info.buffer_len = GO_ABI_INTERNAL_PT_REGS_R1(ctx); // n in return n, nil
+#endif
         // This check achieves ignoring 0 length reads (the reads result with an error)
         if (info.buffer_len <= 0) {
             return;

tap/tlstapper/bpf/include/go_abi_internal.h

@@ -66,11 +66,12 @@ https://go.googlesource.com/go/+/refs/heads/dev.regabi/src/cmd/compile/internal-
 https://github.com/golang/go/blob/go1.17.6/src/cmd/compile/internal/ssa/gen/AMD64Ops.go#L100
 */
 #define GO_ABI_INTERNAL_PT_REGS_R1(x) ((x)->eax)
-#define GO_ABI_INTERNAL_PT_REGS_P2(x) ((x)->ecx)
+#define GO_ABI_INTERNAL_PT_REGS_R2(x) ((x)->ecx)
-#define GO_ABI_INTERNAL_PT_REGS_P3(x) ((x)->edx)
+#define GO_ABI_INTERNAL_PT_REGS_R3(x) ((x)->edx)
-#define GO_ABI_INTERNAL_PT_REGS_P4(x) 0
+#define GO_ABI_INTERNAL_PT_REGS_R4(x) 0
-#define GO_ABI_INTERNAL_PT_REGS_P5(x) 0
+#define GO_ABI_INTERNAL_PT_REGS_R5(x) 0
-#define GO_ABI_INTERNAL_PT_REGS_P6(x) 0
+#define GO_ABI_INTERNAL_PT_REGS_R6(x) 0
+#define GO_ABI_INTERNAL_PT_REGS_R7(x) 0
 #define GO_ABI_INTERNAL_PT_REGS_SP(x) ((x)->esp)
 #define GO_ABI_INTERNAL_PT_REGS_FP(x) ((x)->ebp)
 #define GO_ABI_INTERNAL_PT_REGS_GP(x) ((x)->e14)
@@ -83,6 +84,7 @@ https://github.com/golang/go/blob/go1.17.6/src/cmd/compile/internal/ssa/gen/AMD6
 #define GO_ABI_INTERNAL_PT_REGS_R4(x) ((x)->rbx)
 #define GO_ABI_INTERNAL_PT_REGS_R5(x) ((x)->rbp)
 #define GO_ABI_INTERNAL_PT_REGS_R6(x) ((x)->rsi)
+#define GO_ABI_INTERNAL_PT_REGS_R7(x) ((x)->rdi)
 #define GO_ABI_INTERNAL_PT_REGS_SP(x) ((x)->rsp)
 #define GO_ABI_INTERNAL_PT_REGS_FP(x) ((x)->rbp)
 #define GO_ABI_INTERNAL_PT_REGS_GP(x) ((x)->r14)
@@ -101,7 +103,8 @@ https://github.com/golang/go/blob/go1.17.6/src/cmd/compile/internal/ssa/gen/ARM6
 #define GO_ABI_INTERNAL_PT_REGS_R4(x) ((x)->uregs[3])
 #define GO_ABI_INTERNAL_PT_REGS_R5(x) ((x)->uregs[4])
 #define GO_ABI_INTERNAL_PT_REGS_R6(x) ((x)->uregs[5])
-#define GO_ABI_INTERNAL_PT_REGS_SP(x) ((x)->uregs[14])
+#define GO_ABI_INTERNAL_PT_REGS_R7(x) ((x)->uregs[6])
+#define GO_ABI_INTERNAL_PT_REGS_SP(x) ((x)->uregs[13])
 #define GO_ABI_INTERNAL_PT_REGS_FP(x) ((x)->uregs[29])
 #define GO_ABI_INTERNAL_PT_REGS_GP(x) ((x)->uregs[28])
 
@@ -116,7 +119,8 @@ struct pt_regs;
 #define GO_ABI_INTERNAL_PT_REGS_R4(x) (((PT_REGS_ARM64 *)(x))->regs[3])
 #define GO_ABI_INTERNAL_PT_REGS_R5(x) (((PT_REGS_ARM64 *)(x))->regs[4])
 #define GO_ABI_INTERNAL_PT_REGS_R6(x) (((PT_REGS_ARM64 *)(x))->regs[5])
-#define GO_ABI_INTERNAL_PT_REGS_SP(x) (((PT_REGS_ARM64 *)(x))->regs[30])
+#define GO_ABI_INTERNAL_PT_REGS_R7(x) (((PT_REGS_ARM64 *)(x))->regs[6])
+#define GO_ABI_INTERNAL_PT_REGS_SP(x) (((PT_REGS_ARM64 *)(x))->sp)
 #define GO_ABI_INTERNAL_PT_REGS_FP(x) (((PT_REGS_ARM64 *)(x))->regs[29])
 #define GO_ABI_INTERNAL_PT_REGS_GP(x) (((PT_REGS_ARM64 *)(x))->regs[28])
 
@@ -132,6 +136,7 @@ https://github.com/golang/go/blob/go1.17.6/src/cmd/compile/internal/ssa/gen/PPC6
 #define GO_ABI_INTERNAL_PT_REGS_R4(x) ((x)->gpr[6])
 #define GO_ABI_INTERNAL_PT_REGS_R5(x) ((x)->gpr[7])
 #define GO_ABI_INTERNAL_PT_REGS_R6(x) ((x)->gpr[8])
+#define GO_ABI_INTERNAL_PT_REGS_R7(x) ((x)->gpr[9])
 #define GO_ABI_INTERNAL_PT_REGS_SP(x) ((x)->sp)
 #define GO_ABI_INTERNAL_PT_REGS_FP(x) ((x)->gpr[12])
 #define GO_ABI_INTERNAL_PT_REGS_GP(x) ((x)->gpr[30])

tap/tlstapper/go_offsets.go

@@ -77,7 +77,7 @@ func getOffsets(filePath string) (offsets map[string]*goExtendedOffset, err erro
 	case "arm64":
 		engine, err = gapstone.New(
 			gapstone.CS_ARCH_ARM64,
-			gapstone.CS_MODE_ARM,
+			gapstone.CS_MODE_LITTLE_ENDIAN,
 		)
 	default:
 		err = fmt.Errorf("Unsupported architecture: %v", runtime.GOARCH)
@@ -86,6 +86,16 @@ func getOffsets(filePath string) (offsets map[string]*goExtendedOffset, err erro
 		return
 	}
 
+	engineMajor, engineMinor := engine.Version()
+	logger.Log.Infof(
+		"Disassembling %s with Capstone %d.%d (arch: %d, mode: %d)",
+		filePath,
+		engineMajor,
+		engineMinor,
+		engine.Arch(),
+		engine.Mode(),
+	)
+
 	offsets = make(map[string]*goExtendedOffset)
 	var fd *os.File
 	fd, err = os.Open(filePath)
@@ -133,7 +143,7 @@ func getOffsets(filePath string) (offsets map[string]*goExtendedOffset, err erro
 		extendedOffset := &goExtendedOffset{enter: offset}
 
 		// source: https://gist.github.com/grantseltzer/3efa8ecc5de1fb566e8091533050d608
-		// skip over any symbols that aren't functinons/methods
+		// skip over any symbols that aren't functions/methods
 		if sym.Info != byte(2) && sym.Info != byte(18) {
 			offsets[sym.Name] = extendedOffset
 			continue

tap/tlstapper/tls_tapper.go

@@ -193,7 +193,8 @@ func (t *TlsTapper) tapGoPid(procfs string, pid uint32, namespace string) error
 	hooks := goHooks{}
 
 	if err := hooks.installUprobes(&t.bpfObjects, exePath); err != nil {
-		return err
+		logger.Log.Infof("PID skipped not a Go binary or symbol table is stripped (pid: %v) %v", pid, exePath)
+		return nil // hide the error on purpose, its OK for a process to be not a Go binary or stripped Go binary
 	}
 
 	logger.Log.Infof("Tapping TLS (pid: %v) (Go: %v)", pid, exePath)

ui-common/src/components/modals/TrafficStatsModal/TimelineBarChart/TimelineBarChart.tsx

@@ -1,6 +1,6 @@
 import styles from "./TimelineBarChart.module.sass";
 import { StatsMode } from "../TrafficStatsModal"
-import React, { useCallback, useEffect, useMemo, useState } from "react";
+import React, { useEffect, useMemo, useState } from "react";
 import {
     BarChart,
     Bar,
@@ -20,40 +20,24 @@ export const TimelineBarChart: React.FC<TimelineBarChartProps> = ({ timeLineBarC
     const [protocolStats, setProtocolStats] = useState([]);
     const [protocolsNamesAndColors, setProtocolsNamesAndColors] = useState([]);
 
-    const padTo2Digits = useCallback((num) => {
-        return String(num).padStart(2, '0');
-    }, [])
-
-    const getHoursAndMinutes = useCallback((protocolTimeKey) => {
-        const time = new Date(protocolTimeKey)
-        const hoursAndMinutes = padTo2Digits(time.getHours()) + ':' + padTo2Digits(time.getMinutes());
-        return hoursAndMinutes;
-    }, [padTo2Digits])
-
-    const creatUniqueObjArray = useCallback((objArray) => {
-        return [
-            ...new Map(objArray.map((item) => [item["name"], item])).values(),
-        ];
-    }, [])
-
     useEffect(() => {
         if (!data) return;
         const protocolsBarsData = [];
         const prtcNames = [];
         data.forEach(protocolObj => {
             let obj: { [k: string]: any } = {};
-            obj.timestamp = getHoursAndMinutes(protocolObj.timestamp);
+            obj.timestamp = Utils.getHoursAndMinutes(protocolObj.timestamp);
             protocolObj.protocols.forEach(protocol => {
                 obj[`${protocol.name}`] = protocol[StatsMode[timeLineBarChartMode]];
                 prtcNames.push({ name: protocol.name, color: protocol.color });
             })
             protocolsBarsData.push(obj);
         })
-        const uniqueObjArray = creatUniqueObjArray(prtcNames);
+        const uniqueObjArray = Utils.creatUniqueObjArrayByProp(prtcNames, "name")
         protocolsBarsData.sort((a, b) => a.timestamp < b.timestamp ? -1 : 1);
         setProtocolStats(protocolsBarsData);
         setProtocolsNamesAndColors(uniqueObjArray);
-    }, [data, timeLineBarChartMode, setProtocolStats, setProtocolsNamesAndColors, creatUniqueObjArray, getHoursAndMinutes])
+    }, [data, timeLineBarChartMode])
 
     const bars = useMemo(() => protocolsNamesAndColors.map((protocolToDIsplay) => {
         return <Bar key={protocolToDIsplay.name} dataKey={protocolToDIsplay.name} stackId="a" fill={protocolToDIsplay.color} />
@@ -61,7 +45,7 @@ export const TimelineBarChart: React.FC<TimelineBarChartProps> = ({ timeLineBarC
 
     return (
         <div className={styles.barChartContainer}>
-            <BarChart
+            {protocolStats.length > 0 && <BarChart
                 width={730}
                 height={250}
                 data={protocolStats}
@@ -77,7 +61,7 @@ export const TimelineBarChart: React.FC<TimelineBarChartProps> = ({ timeLineBarC
                 <Tooltip formatter={(value) => timeLineBarChartMode === "VOLUME" ? Utils.humanFileSize(value) : value + " Requests"} />
                 <Legend />
                 {bars}
-            </BarChart>
+            </BarChart>}
         </div>
     );
 }

ui-common/src/helpers/Utils.ts

@@ -2,10 +2,10 @@ const IP_ADDRESS_REGEX = /([0-9]{1,3}.[0-9]{1,3}.[0-9]{1,3}.[0-9]{1,3})(:([0-9]{
 
 
 export class Utils {
-    static isIpAddress = (address: string): boolean => IP_ADDRESS_REGEX.test(address)
+  static isIpAddress = (address: string): boolean => IP_ADDRESS_REGEX.test(address)
-    static lineNumbersInString = (code:string): number => code.split("\n").length;
+  static lineNumbersInString = (code: string): number => code.split("\n").length;
 
-  static humanFileSize(bytes, si=false, dp=1) {
+  static humanFileSize(bytes, si = false, dp = 1) {
     const thresh = si ? 1000 : 1024;
 
     if (Math.abs(bytes) < thresh) {
@@ -16,7 +16,7 @@ export class Utils {
       ? ['kB', 'MB', 'GB', 'TB', 'PB', 'EB', 'ZB', 'YB']
       : ['KiB', 'MiB', 'GiB', 'TiB', 'PiB', 'EiB', 'ZiB', 'YiB'];
     let u = -1;
-    const r = 10**dp;
+    const r = 10 ** dp;
 
     do {
       bytes /= thresh;
@@ -26,4 +26,20 @@ export class Utils {
 
     return bytes.toFixed(dp) + ' ' + units[u];
   }
+
+  static padTo2Digits = (num) => {
+    return String(num).padStart(2, '0');
+  }
+
+  static getHoursAndMinutes = (protocolTimeKey) => {
+    const time = new Date(protocolTimeKey)
+    const hoursAndMinutes = Utils.padTo2Digits(time.getHours()) + ':' + Utils.padTo2Digits(time.getMinutes());
+    return hoursAndMinutes;
+  }
+
+  static creatUniqueObjArrayByProp = (objArray, prop) => {
+    const map = new Map(objArray.map((item) => [item[prop], item])).values()
+    return Array.from(map);
+  }
+
 }