TRA_4623 - methods view on timeline stats with coordination to the pie stats (#1175)

* Add select protocol → when selected, the view will be on commands of that exact protocol

* CR fixes

* added const instead of free string

* remove redundant sass file

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) {
-	extensionsMap = ref
+func InitMaps(extensions map[string]*tapApi.Extension, protocols map[string]*tapApi.Protocol) {
+	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 {

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,
-		Capture:  item.Capture,
+		ProtocolId: fmt.Sprintf("%s/%s/%s", protocol.Name, protocol.Version, protocol.Abbreviation),
+		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,
-		Capture:  item.Capture,
+		ProtocolId: fmt.Sprintf("%s/%s/%s", item.Protocol.Name, item.Protocol.Version, item.Protocol.Abbreviation),
+		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]),
-		`http2`: fmt.Sprintf(`proto.name == "%s" and proto.version == "%s"`, http11protocol.Name, http2Protocol.Version),
-		`grpc`:  fmt.Sprintf(`proto.name == "%s" and proto.version == "%s" and proto.macro == "%s"`, http11protocol.Name, grpcProtocol.Version, grpcProtocol.Macro),
-		`gql`:   fmt.Sprintf(`proto.name == "%s" and proto.macro == "%s"`, graphQL1Protocol.Name, graphQL1Protocol.Macro),
+		`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(`protocol == "%s/%s/%s"`, http2Protocol.Name, http2Protocol.Version, http2Protocol.Abbreviation),
+		`grpc`:  fmt.Sprintf(`protocol == "%s/%s/%s"`, grpcProtocol.Name, grpcProtocol.Version, grpcProtocol.Abbreviation),
+		`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")`,
-		"http2": `proto.name == "http" and proto.version == "2.0"`,
-		"grpc":  `proto.name == "http" and proto.version == "2.0" and proto.macro == "grpc"`,
-		"gql":   `proto.name == "http" and proto.macro == "gql"`,
+		"http":  `protocol == "http/1.0/HTTP" or protocol == "http/1.1/HTTP"`,
+		"http2": `protocol == "http/2.0/HTTP/2"`,
+		"grpc":  `protocol == "http/2.0/gRPC"`,
+		"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,
-		Capture:  item.Capture,
+		ProtocolId: fmt.Sprintf("%s/%s/%s", _protocol.Name, _protocol.Version, _protocol.Abbreviation),
+		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,
-		Capture:  item.Capture,
+		ProtocolId: fmt.Sprintf("%s/%s/%s", protocol.Name, protocol.Version, protocol.Abbreviation),
+		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/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/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,70 +1,79 @@
 import styles from "./TimelineBarChart.module.sass";
-import { StatsMode } from "../TrafficStatsModal"
-import React, { useCallback, useEffect, useMemo, useState } from "react";
+import { ALL_PROTOCOLS, StatsMode } from "../TrafficStatsModal"
+import React, { useEffect, useMemo, useState } from "react";
 import {
     BarChart,
     Bar,
     XAxis,
     YAxis,
     Tooltip,
-    Legend
 } from "recharts";
 import { Utils } from "../../../../helpers/Utils";
 
 interface TimelineBarChartProps {
     timeLineBarChartMode: string;
     data: any;
+    selectedProtocol: string;
 }
 
-export const TimelineBarChart: React.FC<TimelineBarChartProps> = ({ timeLineBarChartMode, data }) => {
+export const TimelineBarChart: React.FC<TimelineBarChartProps> = ({ timeLineBarChartMode, data, selectedProtocol }) => {
     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(),
-        ];
-    }, [])
+    const [commandStats, setCommandStats] = useState(null);
+    const [commandNames, setcommandNames] = useState(null);
 
     useEffect(() => {
         if (!data) return;
         const protocolsBarsData = [];
         const prtcNames = [];
         data.forEach(protocolObj => {
-            let obj: { [k: string]: any } = {};
-            obj.timestamp = getHoursAndMinutes(protocolObj.timestamp);
+            let newProtocolbj: { [k: string]: any } = {};
+            newProtocolbj.timestamp = Utils.getHoursAndMinutes(protocolObj.timestamp);
             protocolObj.protocols.forEach(protocol => {
-                obj[`${protocol.name}`] = protocol[StatsMode[timeLineBarChartMode]];
+                newProtocolbj[`${protocol.name}`] = protocol[StatsMode[timeLineBarChartMode]];
                 prtcNames.push({ name: protocol.name, color: protocol.color });
             })
-            protocolsBarsData.push(obj);
+            protocolsBarsData.push(newProtocolbj);
         })
-        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} />
-    }), [protocolsNamesAndColors])
+    useEffect(() => {
+        if (selectedProtocol === ALL_PROTOCOLS) {
+            setCommandStats(null);
+            setcommandNames(null);
+            return;
+        }
+        const commandsNames = [];
+        const protocolsCommands = [];
+        data.forEach(protocolObj => {
+            let newCommandlbj: { [k: string]: any } = {};
+            newCommandlbj.timestamp = Utils.getHoursAndMinutes(protocolObj.timestamp);
+            protocolObj.protocols.find(protocol => protocol.name === selectedProtocol)?.methods.forEach(command => {
+                newCommandlbj[`${command.name}`] = command[StatsMode[timeLineBarChartMode]]
+                if (commandsNames.indexOf(command.name) === -1)
+                    commandsNames.push(command.name);
+            })
+            protocolsCommands.push(newCommandlbj);
+        })
+        protocolsCommands.sort((a, b) => a.timestamp < b.timestamp ? -1 : 1);
+        setcommandNames(commandsNames);
+        setCommandStats(protocolsCommands);
+    }, [data, timeLineBarChartMode, selectedProtocol])
+
+    const bars = useMemo(() => (commandNames || protocolsNamesAndColors).map((entry) => {
+        return <Bar key={entry.name || entry} dataKey={entry.name || entry} stackId="a" fill={entry.color || Utils.stringToColor(entry)} />
+    }), [protocolsNamesAndColors, commandNames])
 
     return (
         <div className={styles.barChartContainer}>
-            <BarChart
+            {protocolStats.length > 0 && <BarChart
                 width={730}
                 height={250}
-                data={protocolStats}
+                data={commandStats || protocolStats}
                 margin={{
                     top: 20,
                     right: 30,
@@ -75,9 +84,8 @@ export const TimelineBarChart: React.FC<TimelineBarChartProps> = ({ timeLineBarC
                 <XAxis dataKey="timestamp" />
                 <YAxis tickFormatter={(value) => timeLineBarChartMode === "VOLUME" ? Utils.humanFileSize(value) : value} />
                 <Tooltip formatter={(value) => timeLineBarChartMode === "VOLUME" ? Utils.humanFileSize(value) : value + " Requests"} />
-                <Legend />
                 {bars}
-            </BarChart>
+            </BarChart>}
         </div>
     );
 }

ui-common/src/components/modals/TrafficStatsModal/TrafficPieChart/TrafficPieChart.module.sass

@@ -1,16 +0,0 @@
-.breadCrumbsContainer
-  margin-top: 15px
-  height: 15px
-
-  .breadCrumbs
-    color: #494677
-    text-align: left
-
-    .clickableTag
-      margin-right: 5px
-      border-bottom: 1px black solid
-      cursor: pointer
-
-    .nonClickableTag
-      margin-left: 5px
-      font-weight: 600

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

@@ -1,21 +1,18 @@
-import React, {useEffect, useMemo, useState} from "react";
-import styles from "./TrafficPieChart.module.sass";
-import {Cell, Legend, Pie, PieChart, Tooltip} from "recharts";
-import {Utils} from "../../../../helpers/Utils";
-import {StatsMode as PieChartMode} from "../TrafficStatsModal"
-
-const COLORS = ['#e6194b', '#3cb44b', '#ffe119', '#4363d8', '#f58231', '#911eb4', '#46f0f0', '#f032e6', '#bcf60c', '#fabebe', '#008080', '#e6beff', '#9a6324', '#fffac8', '#800000', '#aaffc3', '#808000', '#ffd8b1', '#000075', '#808080', '#ffffff', '#000000'];
+import React, { useEffect, useMemo, useState } from "react";
+import { Cell, Legend, Pie, PieChart, Tooltip } from "recharts";
+import { Utils } from "../../../../helpers/Utils";
+import { ALL_PROTOCOLS, StatsMode as PieChartMode } from "../TrafficStatsModal"
 
 const RADIAN = Math.PI / 180;
 const renderCustomizedLabel = ({
-                                 cx,
-                                 cy,
-                                 midAngle,
-                                 innerRadius,
-                                 outerRadius,
-                                 percent,
-                                 index
-                               }: any) => {
+  cx,
+  cy,
+  midAngle,
+  innerRadius,
+  outerRadius,
+  percent,
+  index
+}: any) => {
   const radius = innerRadius + (outerRadius - innerRadius) * 0.5;
   const x = cx + radius * Math.cos(-midAngle * RADIAN);
   const y = cy + radius * Math.sin(-midAngle * RADIAN);
@@ -38,13 +35,13 @@ const renderCustomizedLabel = ({
 interface TrafficPieChartProps {
   pieChartMode: string;
   data: any;
+  selectedProtocol: string;
 }
 
-export const TrafficPieChart: React.FC<TrafficPieChartProps> = ({pieChartMode , data}) => {
+export const TrafficPieChart: React.FC<TrafficPieChartProps> = ({ pieChartMode, data, selectedProtocol }) => {
 
   const [protocolsStats, setProtocolsStats] = useState([]);
   const [commandStats, setCommandStats] = useState(null);
-  const [selectedProtocol, setSelectedProtocol] = useState(null as string);
 
   useEffect(() => {
     if (!data) return;
@@ -59,11 +56,11 @@ export const TrafficPieChart: React.FC<TrafficPieChartProps> = ({pieChartMode ,
   }, [data, pieChartMode])
 
   useEffect(() => {
-    if (!selectedProtocol) {
+    if (selectedProtocol === ALL_PROTOCOLS) {
       setCommandStats(null);
       return;
     }
-    const commandsPieData = data.find(protocol => protocol.name === selectedProtocol).methods.map(command => {
+    const commandsPieData = data.find(protocol => protocol.name === selectedProtocol)?.methods.map(command => {
       return {
         name: command.name,
         value: command[PieChartMode[pieChartMode]]
@@ -75,18 +72,18 @@ export const TrafficPieChart: React.FC<TrafficPieChartProps> = ({pieChartMode ,
   const pieLegend = useMemo(() => {
     if (!data) return;
     let legend;
-    if (!selectedProtocol) {
-      legend = data.map(protocol => <div style={{marginBottom: 5, display: "flex"}}>
-        <div style={{height: 15, width: 30, background: protocol?.color}}/>
-        <span style={{marginLeft: 5}}>
+    if (selectedProtocol === ALL_PROTOCOLS) {
+      legend = data.map(protocol => <div style={{ marginBottom: 5, display: "flex" }}>
+        <div style={{ height: 15, width: 30, background: protocol?.color }} />
+        <span style={{ marginLeft: 5 }}>
           {protocol.name}
         </span>
       </div>)
     } else {
-      legend = data.find(protocol => protocol.name === selectedProtocol).methods.map((method, index) => <div
-        style={{marginBottom: 5, display: "flex"}}>
-        <div style={{height: 15, width: 30, background: COLORS[index % COLORS.length]}}/>
-        <span style={{marginLeft: 5}}>
+      legend = data.find(protocol => protocol.name === selectedProtocol)?.methods.map((method) => <div
+        style={{ marginBottom: 5, display: "flex" }}>
+        <div style={{ height: 15, width: 30, background: Utils.stringToColor(method.name)}} />
+        <span style={{ marginLeft: 5 }}>
           {method.name}
         </span>
       </div>)
@@ -96,15 +93,7 @@ export const TrafficPieChart: React.FC<TrafficPieChartProps> = ({pieChartMode ,
 
   return (
     <div>
-      <div className={styles.breadCrumbsContainer}>
-        {selectedProtocol && <div className={styles.breadCrumbs}>
-          <span className={styles.clickableTag} onClick={() => setSelectedProtocol(null)}>protocols</span>
-          <span>/</span>
-          <span className={styles.nonClickableTag}>{selectedProtocol}</span>
-        </div>}
-      </div>
-
-      {protocolsStats?.length > 0 && <div style={{width: "100%", display: "flex", justifyContent: "center"}}>
+      {protocolsStats?.length > 0 && <div style={{ width: "100%", display: "flex", justifyContent: "center" }}>
         <PieChart width={300} height={300}>
           <Pie
             data={commandStats || protocolsStats}
@@ -114,14 +103,13 @@ export const TrafficPieChart: React.FC<TrafficPieChartProps> = ({pieChartMode ,
             labelLine={false}
             label={renderCustomizedLabel}
             outerRadius={125}
-            fill="#8884d8"
-            onClick={(section) => !commandStats && setSelectedProtocol(section.name)}>
+            fill="#8884d8">
             {(commandStats || protocolsStats).map((entry, index) => (
-              <Cell key={`cell-${index}`} fill={entry.color || COLORS[index % COLORS.length]}/>)
+              <Cell key={`cell-${index}`} fill={entry.color || Utils.stringToColor(entry.name)} />)
             )}
           </Pie>
-          <Legend wrapperStyle={{position: "absolute", width: "auto", height: "auto", right: -150, top: 0}} content={pieLegend}/>
-          <Tooltip formatter={(value) => pieChartMode === "VOLUME" ? Utils.humanFileSize(value) : value + " Requests"}/>
+          <Legend wrapperStyle={{ position: "absolute", width: "auto", height: "auto", right: -150, top: 0 }} content={pieLegend} />
+          <Tooltip formatter={(value) => pieChartMode === "VOLUME" ? Utils.humanFileSize(value) : value + " Requests"} />
         </PieChart>
       </div>}
     </div>

ui-common/src/components/modals/TrafficStatsModal/TrafficStatsModal.module.sass

@@ -1,3 +1,6 @@
+.headlineContainer
+  display: flex
+
 .title
   color: #494677
   font-family: Source Sans Pro,Lucida Grande,Tahoma,sans-serif
@@ -13,6 +16,11 @@
   padding: 30px
   text-align: center
 
+.selectContainer
+  display: flex
+  justify-content: space-evenly
+  margin-bottom: 4%
+
 .select
   border: none
   border-bottom: 1px black solid

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

@@ -1,10 +1,12 @@
-import React, { useEffect, useState } from "react";
-import { Backdrop, Box, Fade, Modal } from "@mui/material";
+import React, { useCallback, useEffect, useState } from "react";
+import { Backdrop, Box, Button, debounce, Fade, Modal } from "@mui/material";
 import styles from "./TrafficStatsModal.module.sass";
 import closeIcon from "assets/close.svg";
 import { TrafficPieChart } from "./TrafficPieChart/TrafficPieChart";
 import { TimelineBarChart } from "./TimelineBarChart/TimelineBarChart";
 import spinnerImg from "assets/spinner.svg";
+import refreshIcon from "assets/refresh.svg";
+import { useCommonStyles } from "../../../helpers/commonStyle";
 
 const modalStyle = {
   position: 'absolute',
@@ -32,15 +34,20 @@ interface TrafficStatsModalProps {
   getTimelineStatsDataApi: () => Promise<any>
 }
 
+export const PROTOCOLS = ["ALL PROTOCOLS","gRPC", "REDIS", "HTTP", "GQL", "AMQP", "KFAKA"];
+export const ALL_PROTOCOLS = PROTOCOLS[0];
+
 export const TrafficStatsModal: React.FC<TrafficStatsModalProps> = ({ isOpen, onClose, getPieStatsDataApi, getTimelineStatsDataApi }) => {
 
   const modes = Object.keys(StatsMode).filter(x => !(parseInt(x) >= 0));
   const [statsMode, setStatsMode] = useState(modes[0]);
+  const [selectedProtocol, setSelectedProtocol] = useState("ALL PROTOCOLS");
   const [pieStatsData, setPieStatsData] = useState(null);
   const [timelineStatsData, setTimelineStatsData] = useState(null);
   const [isLoading, setIsLoading] = useState(false);
+  const commonClasses = useCommonStyles();
 
-  useEffect(() => {
+  const getTrafficStats = useCallback(async () => {
     if (isOpen && getPieStatsDataApi) {
       (async () => {
         try {
@@ -58,6 +65,14 @@ export const TrafficStatsModal: React.FC<TrafficStatsModalProps> = ({ isOpen, on
     }
   }, [isOpen, getPieStatsDataApi, getTimelineStatsDataApi, setPieStatsData, setTimelineStatsData])
 
+  useEffect(() => {
+    getTrafficStats();
+  }, [getTrafficStats])
+
+  const refreshStats = debounce(() => {
+    getTrafficStats();
+  }, 500);
+
   return (
     <Modal
       aria-labelledby="transition-modal-title"
@@ -72,21 +87,40 @@ export const TrafficStatsModal: React.FC<TrafficStatsModalProps> = ({ isOpen, on
           <div className={styles.closeIcon}>
             <img src={closeIcon} alt="close" onClick={() => onClose()} style={{ cursor: "pointer", userSelect: "none" }} />
           </div>
-          <div className={styles.title}>Traffic Statistics</div>
+          <div className={styles.headlineContainer}>
+            <div className={styles.title}>Traffic Statistics</div>
+            <Button style={{ marginLeft: "2%", textTransform: 'unset' }}
+              startIcon={<img src={refreshIcon} className="custom" alt="refresh"></img>}
+              size="medium"
+              variant="contained"
+              className={commonClasses.outlinedButton + " " + commonClasses.imagedButton}
+              onClick={refreshStats}
+            >
+              Refresh
+            </Button>
+          </div>
           <div className={styles.mainContainer}>
-            <div>
-              <span style={{ marginRight: 15 }}>Breakdown By</span>
-              <select className={styles.select} value={statsMode} onChange={(e) => setStatsMode(e.target.value)}>
-                {modes.map(mode => <option key={mode} value={mode}>{mode}</option>)}
-              </select>
+            <div className={styles.selectContainer}>
+              <div>
+                <span style={{ marginRight: 15 }}>Breakdown By</span>
+                <select className={styles.select} value={statsMode} onChange={(e) => setStatsMode(e.target.value)}>
+                  {modes.map(mode => <option key={mode} value={mode}>{mode}</option>)}
+                </select>
+              </div>
+              <div>
+                <span style={{ marginRight: 15 }}>Protocol</span>
+                <select className={styles.select} value={selectedProtocol} onChange={(e) => setSelectedProtocol(e.target.value)}>
+                  {PROTOCOLS.map(protocol => <option key={protocol} value={protocol}>{protocol}</option>)}
+                </select>
+              </div>
             </div>
             <div>
               {isLoading ? <div style={{ textAlign: "center", marginTop: 20 }}>
                 <img alt="spinner" src={spinnerImg} style={{ height: 50 }} />
-              </div> : 
+              </div> :
                 <div>
-                  <TrafficPieChart pieChartMode={statsMode} data={pieStatsData} />
-                  <TimelineBarChart timeLineBarChartMode={statsMode} data={timelineStatsData} />
+                  <TrafficPieChart pieChartMode={statsMode} data={pieStatsData} selectedProtocol={selectedProtocol}/>
+                  <TimelineBarChart timeLineBarChartMode={statsMode} data={timelineStatsData} selectedProtocol={selectedProtocol}/>
                 </div>}
             </div>
           </div>

ui-common/src/components/modals/TrafficStatsModal/assets/refresh.svg

@@ -0,0 +1,3 @@
+<svg width="26" height="26" viewBox="0 0 26 26" fill="none" xmlns="http://www.w3.org/2000/svg">
+<path d="M10.8337 11.9167H7.69308L7.69416 11.907C7.83561 11.2143 8.11247 10.5564 8.50883 9.97105C9.09865 9.10202 9.92598 8.42097 10.8922 8.00913C11.2193 7.87046 11.5606 7.7643 11.9083 7.69388C12.6297 7.54762 13.3731 7.54762 14.0945 7.69388C15.1312 7.90631 16.0825 8.41908 16.8299 9.1683L18.3639 7.63863C17.6725 6.94707 16.8546 6.39501 15.9546 6.01255C15.4956 5.81823 15.0184 5.67016 14.53 5.57055C13.5223 5.36581 12.4838 5.36581 11.4761 5.57055C10.9873 5.67057 10.5098 5.819 10.0504 6.01363C8.69682 6.58791 7.53808 7.54123 6.71374 8.7588C6.15895 9.5798 5.77099 10.5019 5.57191 11.4725C5.54158 11.6188 5.52533 11.7683 5.50366 11.9167H2.16699L6.50033 16.25L10.8337 11.9167ZM15.167 14.0834H18.3076L18.3065 14.092C18.0234 15.4806 17.205 16.7019 16.0282 17.4915C15.443 17.8882 14.7851 18.1651 14.0923 18.3062C13.3713 18.4525 12.6283 18.4525 11.9072 18.3062C11.2146 18.1648 10.5567 17.8879 9.97133 17.4915C9.68383 17.2971 9.41541 17.0758 9.16966 16.8307L7.63783 18.3625C8.32954 19.0539 9.14791 19.6056 10.0482 19.9875C10.5076 20.1825 10.9875 20.331 11.4728 20.4295C12.4801 20.6344 13.5184 20.6344 14.5257 20.4295C16.4676 20.0265 18.1757 18.8819 19.2869 17.2391C19.8412 16.4187 20.2288 15.4974 20.4277 14.5275C20.4569 14.3813 20.4742 14.2318 20.4959 14.0834H23.8337L19.5003 9.75005L15.167 14.0834Z" fill="#205CF5"/>
+</svg>

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 lineNumbersInString = (code:string): number => code.split("\n").length;
+  static isIpAddress = (address: string): boolean => IP_ADDRESS_REGEX.test(address)
+  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,33 @@ 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);
+  }
+
+  static stringToColor = (str) => {
+  	let colors = ["#e51c23", "#e91e63", "#9c27b0", "#673ab7", "#3f51b5", "#5677fc", "#03a9f4", "#00bcd4", "#009688", "#259b24", "#8bc34a", "#afb42b", "#ff9800", "#ff5722", "#795548", "#607d8b"]
+	
+    let hash = 0;
+	if (str.length === 0) return hash;
+    for (let i = 0; i < str.length; i++) {
+        hash = str.charCodeAt(i) + ((hash << 5) - hash);
+        hash = hash & hash;
+    }
+    hash = ((hash % colors.length) + colors.length) % colors.length;
+    return colors[hash];
+}
+
 }