TRA-4065 support inflight tap target update (#556)

* WIP

* WIP

* Update main.go

* Update main.go and passive_tapper.go

* Update passive_tapper.go

* Update passive_tapper.go

* Update passive_tapper.go

* Update passive_tapper.go

agent/main.go

@@ -55,7 +55,7 @@ var extensionsMap map[string]*tapApi.Extension // global
 var startTime int64
 
 const (
-	socketConnectionRetries    = 10
+	socketConnectionRetries    = 30
 	socketConnectionRetryDelay = time.Second * 2
 	socketHandshakeTimeout     = time.Second * 2
 )
@@ -425,12 +425,32 @@ func dialSocketWithRetry(socketAddress string, retryAmount int, retryDelay time.
 				time.Sleep(retryDelay)
 			}
 		} else {
+			go handleIncomingMessageAsTapper(socketConnection)
 			return socketConnection, nil
 		}
 	}
 	return nil, lastErr
 }
 
+func handleIncomingMessageAsTapper(socketConnection *websocket.Conn) {
+	for {
+		if _, message, err := socketConnection.ReadMessage(); err != nil {
+			logger.Log.Errorf("error reading message from socket connection, err: %s, (%v,%+v)", err, err, err)
+			if errors.Is(err, syscall.EPIPE) {
+				// socket has disconnected, we can safely stop this goroutine
+				return
+			}
+		} else {
+			var tapConfigMessage *shared.WebSocketTapConfigMessage
+			if err := json.Unmarshal(message, &tapConfigMessage); err != nil {
+				logger.Log.Errorf("received unknown message from socket connection: %s, err: %s, (%v,%+v)", string(message), err, err, err)
+			} else {
+				tap.UpdateTapTargets(tapConfigMessage.TapTargets)
+			}
+		}
+	}
+}
+
 func startMizuTapperSyncer(ctx context.Context, provider *kubernetes.Provider) (*kubernetes.MizuTapperSyncer, error) {
 	tapperSyncer, err := kubernetes.CreateAndStartMizuTapperSyncer(ctx, provider, kubernetes.TapperSyncerConfig{
 		TargetNamespaces:         config.Config.TargetNamespaces,

cli/cmd/cleanRunner.go

@@ -1,6 +1,9 @@
 package cmd
 
-import "github.com/up9inc/mizu/cli/apiserver"
+import (
+	"github.com/up9inc/mizu/cli/apiserver"
+	"github.com/up9inc/mizu/cli/config"
+)
 
 func performCleanCommand() {
 	kubernetesProvider, err := getKubernetesProviderForCli()
@@ -8,5 +11,5 @@ func performCleanCommand() {
 		return
 	}
 
-	finishMizuExecution(kubernetesProvider, apiserver.NewProvider(GetApiServerUrl(), apiserver.DefaultRetries, apiserver.DefaultTimeout))
+	finishMizuExecution(kubernetesProvider, apiserver.NewProvider(GetApiServerUrl(), apiserver.DefaultRetries, apiserver.DefaultTimeout), config.Config.IsNsRestrictedMode(), config.Config.MizuResourcesNamespace)
 }

cli/cmd/common.go

@@ -7,11 +7,9 @@ import (
 	"github.com/up9inc/mizu/cli/apiserver"
 	"github.com/up9inc/mizu/cli/mizu"
 	"github.com/up9inc/mizu/cli/mizu/fsUtils"
+	"github.com/up9inc/mizu/cli/resources"
 	"github.com/up9inc/mizu/cli/telemetry"
-	"os"
-	"os/signal"
 	"path"
-	"syscall"
 	"time"
 
 	"github.com/up9inc/mizu/cli/config"
@@ -37,22 +35,6 @@ func startProxyReportErrorIfAny(kubernetesProvider *kubernetes.Provider, cancel
 	logger.Log.Debugf("proxy ended")
 }
 
-func waitForFinish(ctx context.Context, cancel context.CancelFunc) {
-	logger.Log.Debugf("waiting for finish...")
-	sigChan := make(chan os.Signal, 1)
-	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
-
-	// block until ctx cancel is called or termination signal is received
-	select {
-	case <-ctx.Done():
-		logger.Log.Debugf("ctx done")
-		break
-	case <-sigChan:
-		logger.Log.Debugf("Got termination signal, canceling execution...")
-		cancel()
-	}
-}
-
 func getKubernetesProviderForCli() (*kubernetes.Provider, error) {
 	kubernetesProvider, err := kubernetes.NewProvider(config.Config.KubeConfigPath())
 	if err != nil {
@@ -71,12 +53,12 @@ func handleKubernetesProviderError(err error) {
 	}
 }
 
-func finishMizuExecution(kubernetesProvider *kubernetes.Provider, apiProvider *apiserver.Provider) {
+func finishMizuExecution(kubernetesProvider *kubernetes.Provider, apiProvider *apiserver.Provider, isNsRestrictedMode bool, mizuResourcesNamespace string) {
 	telemetry.ReportAPICalls(apiProvider)
 	removalCtx, cancel := context.WithTimeout(context.Background(), cleanupTimeout)
 	defer cancel()
 	dumpLogsIfNeeded(removalCtx, kubernetesProvider)
-	cleanUpMizuResources(removalCtx, cancel, kubernetesProvider)
+	resources.CleanUpMizuResources(removalCtx, cancel, kubernetesProvider, isNsRestrictedMode, mizuResourcesNamespace)
 }
 
 func dumpLogsIfNeeded(ctx context.Context, kubernetesProvider *kubernetes.Provider) {
@@ -89,23 +71,3 @@ func dumpLogsIfNeeded(ctx context.Context, kubernetesProvider *kubernetes.Provid
 		logger.Log.Errorf("Failed dump logs %v", err)
 	}
 }
-
-func cleanUpMizuResources(ctx context.Context, cancel context.CancelFunc, kubernetesProvider *kubernetes.Provider) {
-	logger.Log.Infof("\nRemoving mizu resources")
-
-	var leftoverResources []string
-
-	if config.Config.IsNsRestrictedMode() {
-		leftoverResources = cleanUpRestrictedMode(ctx, kubernetesProvider)
-	} else {
-		leftoverResources = cleanUpNonRestrictedMode(ctx, cancel, kubernetesProvider)
-	}
-
-	if len(leftoverResources) > 0 {
-		errMsg := fmt.Sprintf("Failed to remove the following resources, for more info check logs at %s:", fsUtils.GetLogFilePath())
-		for _, resource := range leftoverResources {
-			errMsg += "\n- " + resource
-		}
-		logger.Log.Errorf(uiUtils.Error, errMsg)
-	}
-}

cli/cmd/tapRunner.go

@@ -4,26 +4,24 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"github.com/up9inc/mizu/cli/resources"
+	"github.com/up9inc/mizu/cli/utils"
 	"io/ioutil"
 	"regexp"
 	"strings"
 	"time"
 
-	"k8s.io/apimachinery/pkg/util/intstr"
-
 	"github.com/getkin/kin-openapi/openapi3"
 	"gopkg.in/yaml.v3"
 	core "k8s.io/api/core/v1"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/util/wait"
 
 	"github.com/up9inc/mizu/cli/apiserver"
 	"github.com/up9inc/mizu/cli/cmd/goUtils"
 	"github.com/up9inc/mizu/cli/config"
 	"github.com/up9inc/mizu/cli/config/configStructs"
 	"github.com/up9inc/mizu/cli/errormessage"
-	"github.com/up9inc/mizu/cli/mizu"
 	"github.com/up9inc/mizu/cli/mizu/fsUtils"
 	"github.com/up9inc/mizu/cli/uiUtils"
 	"github.com/up9inc/mizu/shared"
@@ -37,9 +35,6 @@ const cleanupTimeout = time.Minute
 type tapState struct {
 	startTime        time.Time
 	targetNamespaces []string
-
-	apiServerService         *core.Service
-	tapperSyncer             *kubernetes.MizuTapperSyncer
 	mizuServiceAccountExists bool
 }
 
@@ -131,7 +126,7 @@ func RunMizuTap() {
 	}
 
 	logger.Log.Infof("Waiting for Mizu Agent to start...")
-	if err := createMizuResources(ctx, cancel, kubernetesProvider, serializedValidationRules, serializedContract, serializedMizuConfig); err != nil {
+	if state.mizuServiceAccountExists, err = resources.CreateMizuResources(ctx, cancel, kubernetesProvider, serializedValidationRules, serializedContract, serializedMizuConfig, config.Config.IsNsRestrictedMode(), config.Config.MizuResourcesNamespace, config.Config.Tap.DaemonMode, config.Config.AgentImage, getSyncEntriesConfig(), config.Config.Tap.MaxEntriesDBSizeBytes(), config.Config.Tap.ApiServerResources, config.Config.ImagePullPolicy(), config.Config.LogLevel(), config.Config.Tap.NoPersistentVolumeClaim); err != nil {
 		logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Error creating resources: %v", errormessage.FormatError(err)))
 
 		var statusError *k8serrors.StatusError
@@ -143,24 +138,24 @@ func RunMizuTap() {
 		return
 	}
 	if config.Config.Tap.DaemonMode {
-		if err := handleDaemonModePostCreation(ctx, cancel, kubernetesProvider, state.targetNamespaces); err != nil {
-			defer finishMizuExecution(kubernetesProvider, apiProvider)
+		if err := handleDaemonModePostCreation(cancel, kubernetesProvider); err != nil {
+			defer finishMizuExecution(kubernetesProvider, apiProvider, config.Config.IsNsRestrictedMode(), config.Config.MizuResourcesNamespace)
 			cancel()
 		} else {
 			logger.Log.Infof(uiUtils.Magenta, "Mizu is now running in daemon mode, run `mizu view` to connect to the mizu daemon instance")
 		}
 	} else {
-		defer finishMizuExecution(kubernetesProvider, apiProvider)
+		defer finishMizuExecution(kubernetesProvider, apiProvider, config.Config.IsNsRestrictedMode(), config.Config.MizuResourcesNamespace)
 
 		go goUtils.HandleExcWrapper(watchApiServerEvents, ctx, kubernetesProvider, cancel)
 		go goUtils.HandleExcWrapper(watchApiServerPod, ctx, kubernetesProvider, cancel)
 
 		// block until exit signal or error
-		waitForFinish(ctx, cancel)
+		utils.WaitForFinish(ctx, cancel)
 	}
 }
 
-func handleDaemonModePostCreation(ctx context.Context, cancel context.CancelFunc, kubernetesProvider *kubernetes.Provider, namespaces []string) error {
+func handleDaemonModePostCreation(cancel context.CancelFunc, kubernetesProvider *kubernetes.Provider) error {
 	apiProvider := apiserver.NewProvider(GetApiServerUrl(), 90, 1*time.Second)
 
 	if err := waitForDaemonModeToBeReady(cancel, kubernetesProvider, apiProvider); err != nil {
@@ -252,8 +247,6 @@ func startTapperSyncer(ctx context.Context, cancel context.CancelFunc, provider
 		}
 	}()
 
-	state.tapperSyncer = tapperSyncer
-
 	return nil
 }
 
@@ -287,137 +280,6 @@ func readValidationRules(file string) (string, error) {
 	return string(newContent), nil
 }
 
-func createMizuResources(ctx context.Context, cancel context.CancelFunc, kubernetesProvider *kubernetes.Provider, serializedValidationRules string, serializedContract string, serializedMizuConfig string) error {
-	if !config.Config.IsNsRestrictedMode() {
-		if err := createMizuNamespace(ctx, kubernetesProvider); err != nil {
-			return err
-		}
-	}
-
-	if err := createMizuConfigmap(ctx, kubernetesProvider, serializedValidationRules, serializedContract, serializedMizuConfig); err != nil {
-		logger.Log.Warningf(uiUtils.Warning, fmt.Sprintf("Failed to create resources required for policy validation. Mizu will not validate policy rules. error: %v", errormessage.FormatError(err)))
-	}
-
-	var err error
-	state.mizuServiceAccountExists, err = createRBACIfNecessary(ctx, kubernetesProvider)
-	if err != nil {
-		if !config.Config.Tap.DaemonMode {
-			logger.Log.Warningf(uiUtils.Warning, fmt.Sprintf("Failed to ensure the resources required for IP resolving. Mizu will not resolve target IPs to names. error: %v", errormessage.FormatError(err)))
-		}
-	}
-
-	var serviceAccountName string
-	if state.mizuServiceAccountExists {
-		serviceAccountName = kubernetes.ServiceAccountName
-	} else {
-		serviceAccountName = ""
-	}
-
-	opts := &kubernetes.ApiServerOptions{
-		Namespace:             config.Config.MizuResourcesNamespace,
-		PodName:               kubernetes.ApiServerPodName,
-		PodImage:              config.Config.AgentImage,
-		ServiceAccountName:    serviceAccountName,
-		IsNamespaceRestricted: config.Config.IsNsRestrictedMode(),
-		SyncEntriesConfig:     getSyncEntriesConfig(),
-		MaxEntriesDBSizeBytes: config.Config.Tap.MaxEntriesDBSizeBytes(),
-		Resources:             config.Config.Tap.ApiServerResources,
-		ImagePullPolicy:       config.Config.ImagePullPolicy(),
-		LogLevel:              config.Config.LogLevel(),
-	}
-
-	if config.Config.Tap.DaemonMode {
-		if !state.mizuServiceAccountExists {
-			defer cleanUpMizuResources(ctx, cancel, kubernetesProvider)
-			logger.Log.Fatalf(uiUtils.Red, fmt.Sprintf("Failed to ensure the resources required for mizu to run in daemon mode. cannot proceed. error: %v", errormessage.FormatError(err)))
-		}
-		if err := createMizuApiServerDeployment(ctx, kubernetesProvider, opts); err != nil {
-			return err
-		}
-	} else {
-		if err := createMizuApiServerPod(ctx, kubernetesProvider, opts); err != nil {
-			return err
-		}
-	}
-
-	state.apiServerService, err = kubernetesProvider.CreateService(ctx, config.Config.MizuResourcesNamespace, kubernetes.ApiServerPodName, kubernetes.ApiServerPodName)
-	if err != nil {
-		return err
-	}
-	logger.Log.Debugf("Successfully created service: %s", kubernetes.ApiServerPodName)
-
-	return nil
-}
-
-func createMizuConfigmap(ctx context.Context, kubernetesProvider *kubernetes.Provider, serializedValidationRules string, serializedContract string, serializedMizuConfig string) error {
-	err := kubernetesProvider.CreateConfigMap(ctx, config.Config.MizuResourcesNamespace, kubernetes.ConfigMapName, serializedValidationRules, serializedContract, serializedMizuConfig)
-	return err
-}
-
-func createMizuNamespace(ctx context.Context, kubernetesProvider *kubernetes.Provider) error {
-	_, err := kubernetesProvider.CreateNamespace(ctx, config.Config.MizuResourcesNamespace)
-	return err
-}
-
-func createMizuApiServerPod(ctx context.Context, kubernetesProvider *kubernetes.Provider, opts *kubernetes.ApiServerOptions) error {
-	pod, err := kubernetesProvider.GetMizuApiServerPodObject(opts, false, "")
-	if err != nil {
-		return err
-	}
-	if _, err = kubernetesProvider.CreatePod(ctx, config.Config.MizuResourcesNamespace, pod); err != nil {
-		return err
-	}
-	logger.Log.Debugf("Successfully created API server pod: %s", kubernetes.ApiServerPodName)
-	return nil
-}
-
-func createMizuApiServerDeployment(ctx context.Context, kubernetesProvider *kubernetes.Provider, opts *kubernetes.ApiServerOptions) error {
-	volumeClaimCreated := false
-	if !config.Config.Tap.NoPersistentVolumeClaim {
-		volumeClaimCreated = TryToCreatePersistentVolumeClaim(ctx, kubernetesProvider)
-	}
-
-	pod, err := kubernetesProvider.GetMizuApiServerPodObject(opts, volumeClaimCreated, kubernetes.PersistentVolumeClaimName)
-	if err != nil {
-		return err
-	}
-	pod.Spec.Containers[0].LivenessProbe = &core.Probe{
-		Handler: core.Handler{
-			HTTPGet: &core.HTTPGetAction{
-				Path: "/echo",
-				Port: intstr.FromInt(shared.DefaultApiServerPort),
-			},
-		},
-		InitialDelaySeconds: 1,
-		PeriodSeconds:       10,
-	}
-	if _, err = kubernetesProvider.CreateDeployment(ctx, config.Config.MizuResourcesNamespace, opts.PodName, pod); err != nil {
-		return err
-	}
-	logger.Log.Debugf("Successfully created API server deployment: %s", kubernetes.ApiServerPodName)
-	return nil
-}
-
-func TryToCreatePersistentVolumeClaim(ctx context.Context, kubernetesProvider *kubernetes.Provider) bool {
-	isDefaultStorageClassAvailable, err := kubernetesProvider.IsDefaultStorageProviderAvailable(ctx)
-	if err != nil {
-		logger.Log.Warningf(uiUtils.Yellow, "An error occured when checking if a default storage provider exists in this cluster, this means mizu data will be lost on mizu-api-server pod restart")
-		logger.Log.Debugf("error checking if default storage class exists: %v", err)
-		return false
-	} else if !isDefaultStorageClassAvailable {
-		logger.Log.Warningf(uiUtils.Yellow, "Could not find default storage provider in this cluster, this means mizu data will be lost on mizu-api-server pod restart")
-		return false
-	}
-
-	if _, err = kubernetesProvider.CreatePersistentVolumeClaim(ctx, config.Config.MizuResourcesNamespace, kubernetes.PersistentVolumeClaimName, config.Config.Tap.MaxEntriesDBSizeBytes()+mizu.DaemonModePersistentVolumeSizeBufferBytes); err != nil {
-		logger.Log.Warningf(uiUtils.Yellow, "An error has occured while creating a persistent volume claim for mizu, this means mizu data will be lost on mizu-api-server pod restart")
-		logger.Log.Debugf("error creating persistent volume claim: %v", err)
-		return false
-	}
-
-	return true
-}
-
 func getMizuApiFilteringOptions() (*api.TrafficFilteringOptions, error) {
 	var compiledRegexSlice []*api.SerializableRegexp
 
@@ -452,114 +314,6 @@ func getSyncEntriesConfig() *shared.SyncEntriesConfig {
 	}
 }
 
-func cleanUpRestrictedMode(ctx context.Context, kubernetesProvider *kubernetes.Provider) []string {
-	leftoverResources := make([]string, 0)
-
-	if err := kubernetesProvider.RemoveService(ctx, config.Config.MizuResourcesNamespace, kubernetes.ApiServerPodName); err != nil {
-		resourceDesc := fmt.Sprintf("Service %s in namespace %s", kubernetes.ApiServerPodName, config.Config.MizuResourcesNamespace)
-		handleDeletionError(err, resourceDesc, &leftoverResources)
-	}
-
-	if err := kubernetesProvider.RemoveDaemonSet(ctx, config.Config.MizuResourcesNamespace, kubernetes.TapperDaemonSetName); err != nil {
-		resourceDesc := fmt.Sprintf("DaemonSet %s in namespace %s", kubernetes.TapperDaemonSetName, config.Config.MizuResourcesNamespace)
-		handleDeletionError(err, resourceDesc, &leftoverResources)
-	}
-
-	if err := kubernetesProvider.RemoveConfigMap(ctx, config.Config.MizuResourcesNamespace, kubernetes.ConfigMapName); err != nil {
-		resourceDesc := fmt.Sprintf("ConfigMap %s in namespace %s", kubernetes.ConfigMapName, config.Config.MizuResourcesNamespace)
-		handleDeletionError(err, resourceDesc, &leftoverResources)
-	}
-
-	if err := kubernetesProvider.RemoveServicAccount(ctx, config.Config.MizuResourcesNamespace, kubernetes.ServiceAccountName); err != nil {
-		resourceDesc := fmt.Sprintf("Service Account %s in namespace %s", kubernetes.ServiceAccountName, config.Config.MizuResourcesNamespace)
-		handleDeletionError(err, resourceDesc, &leftoverResources)
-	}
-
-	if err := kubernetesProvider.RemoveRole(ctx, config.Config.MizuResourcesNamespace, kubernetes.RoleName); err != nil {
-		resourceDesc := fmt.Sprintf("Role %s in namespace %s", kubernetes.RoleName, config.Config.MizuResourcesNamespace)
-		handleDeletionError(err, resourceDesc, &leftoverResources)
-	}
-
-	if err := kubernetesProvider.RemovePod(ctx, config.Config.MizuResourcesNamespace, kubernetes.ApiServerPodName); err != nil {
-		resourceDesc := fmt.Sprintf("Pod %s in namespace %s", kubernetes.ApiServerPodName, config.Config.MizuResourcesNamespace)
-		handleDeletionError(err, resourceDesc, &leftoverResources)
-	}
-
-	//daemon mode resources
-	if err := kubernetesProvider.RemoveRoleBinding(ctx, config.Config.MizuResourcesNamespace, kubernetes.RoleBindingName); err != nil {
-		resourceDesc := fmt.Sprintf("RoleBinding %s in namespace %s", kubernetes.RoleBindingName, config.Config.MizuResourcesNamespace)
-		handleDeletionError(err, resourceDesc, &leftoverResources)
-	}
-
-	if err := kubernetesProvider.RemoveDeployment(ctx, config.Config.MizuResourcesNamespace, kubernetes.ApiServerPodName); err != nil {
-		resourceDesc := fmt.Sprintf("Deployment %s in namespace %s", kubernetes.ApiServerPodName, config.Config.MizuResourcesNamespace)
-		handleDeletionError(err, resourceDesc, &leftoverResources)
-	}
-
-	if err := kubernetesProvider.RemovePersistentVolumeClaim(ctx, config.Config.MizuResourcesNamespace, kubernetes.PersistentVolumeClaimName); err != nil {
-		resourceDesc := fmt.Sprintf("PersistentVolumeClaim %s in namespace %s", kubernetes.PersistentVolumeClaimName, config.Config.MizuResourcesNamespace)
-		handleDeletionError(err, resourceDesc, &leftoverResources)
-	}
-
-	if err := kubernetesProvider.RemoveRole(ctx, config.Config.MizuResourcesNamespace, kubernetes.DaemonRoleName); err != nil {
-		resourceDesc := fmt.Sprintf("Role %s in namespace %s", kubernetes.DaemonRoleName, config.Config.MizuResourcesNamespace)
-		handleDeletionError(err, resourceDesc, &leftoverResources)
-	}
-
-	if err := kubernetesProvider.RemoveRoleBinding(ctx, config.Config.MizuResourcesNamespace, kubernetes.DaemonRoleBindingName); err != nil {
-		resourceDesc := fmt.Sprintf("RoleBinding %s in namespace %s", kubernetes.DaemonRoleBindingName, config.Config.MizuResourcesNamespace)
-		handleDeletionError(err, resourceDesc, &leftoverResources)
-	}
-
-	return leftoverResources
-}
-
-func cleanUpNonRestrictedMode(ctx context.Context, cancel context.CancelFunc, kubernetesProvider *kubernetes.Provider) []string {
-	leftoverResources := make([]string, 0)
-
-	if err := kubernetesProvider.RemoveNamespace(ctx, config.Config.MizuResourcesNamespace); err != nil {
-		resourceDesc := fmt.Sprintf("Namespace %s", config.Config.MizuResourcesNamespace)
-		handleDeletionError(err, resourceDesc, &leftoverResources)
-	} else {
-		defer waitUntilNamespaceDeleted(ctx, cancel, kubernetesProvider)
-	}
-
-	if err := kubernetesProvider.RemoveClusterRole(ctx, kubernetes.ClusterRoleName); err != nil {
-		resourceDesc := fmt.Sprintf("ClusterRole %s", kubernetes.ClusterRoleName)
-		handleDeletionError(err, resourceDesc, &leftoverResources)
-	}
-
-	if err := kubernetesProvider.RemoveClusterRoleBinding(ctx, kubernetes.ClusterRoleBindingName); err != nil {
-		resourceDesc := fmt.Sprintf("ClusterRoleBinding %s", kubernetes.ClusterRoleBindingName)
-		handleDeletionError(err, resourceDesc, &leftoverResources)
-	}
-
-	return leftoverResources
-}
-
-func handleDeletionError(err error, resourceDesc string, leftoverResources *[]string) {
-	logger.Log.Debugf("Error removing %s: %v", resourceDesc, errormessage.FormatError(err))
-	*leftoverResources = append(*leftoverResources, resourceDesc)
-}
-
-func waitUntilNamespaceDeleted(ctx context.Context, cancel context.CancelFunc, kubernetesProvider *kubernetes.Provider) {
-	// Call cancel if a terminating signal was received. Allows user to skip the wait.
-	go func() {
-		waitForFinish(ctx, cancel)
-	}()
-
-	if err := kubernetesProvider.WaitUtilNamespaceDeleted(ctx, config.Config.MizuResourcesNamespace); err != nil {
-		switch {
-		case ctx.Err() == context.Canceled:
-			logger.Log.Debugf("Do nothing. User interrupted the wait")
-		case err == wait.ErrWaitTimeout:
-			logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Timeout while removing Namespace %s", config.Config.MizuResourcesNamespace))
-		default:
-			logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Error while waiting for Namespace %s to be deleted: %v", config.Config.MizuResourcesNamespace, errormessage.FormatError(err)))
-		}
-	}
-}
-
 func watchApiServerPod(ctx context.Context, kubernetesProvider *kubernetes.Provider, cancel context.CancelFunc) {
 	podExactRegex := regexp.MustCompile(fmt.Sprintf("^%s$", kubernetes.ApiServerPodName))
 	podWatchHelper := kubernetes.NewPodWatchHelper(kubernetesProvider, podExactRegex)
@@ -705,21 +459,3 @@ func getNamespaces(kubernetesProvider *kubernetes.Provider) []string {
 		return []string{currentNamespace}
 	}
 }
-
-func createRBACIfNecessary(ctx context.Context, kubernetesProvider *kubernetes.Provider) (bool, error) {
-	if !config.Config.IsNsRestrictedMode() {
-		if err := kubernetesProvider.CreateMizuRBAC(ctx, config.Config.MizuResourcesNamespace, kubernetes.ServiceAccountName, kubernetes.ClusterRoleName, kubernetes.ClusterRoleBindingName, mizu.RBACVersion); err != nil {
-			return false, err
-		}
-	} else {
-		if err := kubernetesProvider.CreateMizuRBACNamespaceRestricted(ctx, config.Config.MizuResourcesNamespace, kubernetes.ServiceAccountName, kubernetes.RoleName, kubernetes.RoleBindingName, mizu.RBACVersion); err != nil {
-			return false, err
-		}
-	}
-	if config.Config.Tap.DaemonMode {
-		if err := kubernetesProvider.CreateDaemonsetRBAC(ctx, config.Config.MizuResourcesNamespace, kubernetes.ServiceAccountName, kubernetes.DaemonRoleName, kubernetes.DaemonRoleBindingName, mizu.RBACVersion); err != nil {
-			return false, err
-		}
-	}
-	return true, nil
-}

cli/cmd/viewRunner.go

@@ -3,6 +3,7 @@ package cmd
 import (
 	"context"
 	"fmt"
+	"github.com/up9inc/mizu/cli/utils"
 	"net/http"
 
 	"github.com/up9inc/mizu/cli/apiserver"
@@ -71,5 +72,5 @@ func runMizuView() {
 		return
 	}
 
-	waitForFinish(ctx, cancel)
+	utils.WaitForFinish(ctx, cancel)
 }

cli/resources/cleanResources.go

@@ -0,0 +1,141 @@
+package resources
+
+import (
+	"context"
+	"fmt"
+	"github.com/up9inc/mizu/cli/errormessage"
+	"github.com/up9inc/mizu/cli/mizu/fsUtils"
+	"github.com/up9inc/mizu/cli/uiUtils"
+	"github.com/up9inc/mizu/cli/utils"
+	"github.com/up9inc/mizu/shared/kubernetes"
+	"github.com/up9inc/mizu/shared/logger"
+	"k8s.io/apimachinery/pkg/util/wait"
+)
+
+func CleanUpMizuResources(ctx context.Context, cancel context.CancelFunc, kubernetesProvider *kubernetes.Provider, isNsRestrictedMode bool, mizuResourcesNamespace string) {
+	logger.Log.Infof("\nRemoving mizu resources")
+
+	var leftoverResources []string
+
+	if isNsRestrictedMode {
+		leftoverResources = cleanUpRestrictedMode(ctx, kubernetesProvider, mizuResourcesNamespace)
+	} else {
+		leftoverResources = cleanUpNonRestrictedMode(ctx, cancel, kubernetesProvider, mizuResourcesNamespace)
+	}
+
+	if len(leftoverResources) > 0 {
+		errMsg := fmt.Sprintf("Failed to remove the following resources, for more info check logs at %s:", fsUtils.GetLogFilePath())
+		for _, resource := range leftoverResources {
+			errMsg += "\n- " + resource
+		}
+		logger.Log.Errorf(uiUtils.Error, errMsg)
+	}
+}
+
+func cleanUpNonRestrictedMode(ctx context.Context, cancel context.CancelFunc, kubernetesProvider *kubernetes.Provider, mizuResourcesNamespace string) []string {
+	leftoverResources := make([]string, 0)
+
+	if err := kubernetesProvider.RemoveNamespace(ctx, mizuResourcesNamespace); err != nil {
+		resourceDesc := fmt.Sprintf("Namespace %s", mizuResourcesNamespace)
+		handleDeletionError(err, resourceDesc, &leftoverResources)
+	} else {
+		defer waitUntilNamespaceDeleted(ctx, cancel, kubernetesProvider, mizuResourcesNamespace)
+	}
+
+	if err := kubernetesProvider.RemoveClusterRole(ctx, kubernetes.ClusterRoleName); err != nil {
+		resourceDesc := fmt.Sprintf("ClusterRole %s", kubernetes.ClusterRoleName)
+		handleDeletionError(err, resourceDesc, &leftoverResources)
+	}
+
+	if err := kubernetesProvider.RemoveClusterRoleBinding(ctx, kubernetes.ClusterRoleBindingName); err != nil {
+		resourceDesc := fmt.Sprintf("ClusterRoleBinding %s", kubernetes.ClusterRoleBindingName)
+		handleDeletionError(err, resourceDesc, &leftoverResources)
+	}
+
+	return leftoverResources
+}
+
+func waitUntilNamespaceDeleted(ctx context.Context, cancel context.CancelFunc, kubernetesProvider *kubernetes.Provider, mizuResourcesNamespace string) {
+	// Call cancel if a terminating signal was received. Allows user to skip the wait.
+	go func() {
+		utils.WaitForFinish(ctx, cancel)
+	}()
+
+	if err := kubernetesProvider.WaitUtilNamespaceDeleted(ctx, mizuResourcesNamespace); err != nil {
+		switch {
+		case ctx.Err() == context.Canceled:
+			logger.Log.Debugf("Do nothing. User interrupted the wait")
+		case err == wait.ErrWaitTimeout:
+			logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Timeout while removing Namespace %s", mizuResourcesNamespace))
+		default:
+			logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Error while waiting for Namespace %s to be deleted: %v", mizuResourcesNamespace, errormessage.FormatError(err)))
+		}
+	}
+}
+
+func cleanUpRestrictedMode(ctx context.Context, kubernetesProvider *kubernetes.Provider, mizuResourcesNamespace string) []string {
+	leftoverResources := make([]string, 0)
+
+	if err := kubernetesProvider.RemoveService(ctx, mizuResourcesNamespace, kubernetes.ApiServerPodName); err != nil {
+		resourceDesc := fmt.Sprintf("Service %s in namespace %s", kubernetes.ApiServerPodName, mizuResourcesNamespace)
+		handleDeletionError(err, resourceDesc, &leftoverResources)
+	}
+
+	if err := kubernetesProvider.RemoveDaemonSet(ctx, mizuResourcesNamespace, kubernetes.TapperDaemonSetName); err != nil {
+		resourceDesc := fmt.Sprintf("DaemonSet %s in namespace %s", kubernetes.TapperDaemonSetName, mizuResourcesNamespace)
+		handleDeletionError(err, resourceDesc, &leftoverResources)
+	}
+
+	if err := kubernetesProvider.RemoveConfigMap(ctx, mizuResourcesNamespace, kubernetes.ConfigMapName); err != nil {
+		resourceDesc := fmt.Sprintf("ConfigMap %s in namespace %s", kubernetes.ConfigMapName, mizuResourcesNamespace)
+		handleDeletionError(err, resourceDesc, &leftoverResources)
+	}
+
+	if err := kubernetesProvider.RemoveServicAccount(ctx, mizuResourcesNamespace, kubernetes.ServiceAccountName); err != nil {
+		resourceDesc := fmt.Sprintf("Service Account %s in namespace %s", kubernetes.ServiceAccountName, mizuResourcesNamespace)
+		handleDeletionError(err, resourceDesc, &leftoverResources)
+	}
+
+	if err := kubernetesProvider.RemoveRole(ctx, mizuResourcesNamespace, kubernetes.RoleName); err != nil {
+		resourceDesc := fmt.Sprintf("Role %s in namespace %s", kubernetes.RoleName, mizuResourcesNamespace)
+		handleDeletionError(err, resourceDesc, &leftoverResources)
+	}
+
+	if err := kubernetesProvider.RemovePod(ctx, mizuResourcesNamespace, kubernetes.ApiServerPodName); err != nil {
+		resourceDesc := fmt.Sprintf("Pod %s in namespace %s", kubernetes.ApiServerPodName, mizuResourcesNamespace)
+		handleDeletionError(err, resourceDesc, &leftoverResources)
+	}
+
+	//install mode resources
+	if err := kubernetesProvider.RemoveRoleBinding(ctx, mizuResourcesNamespace, kubernetes.RoleBindingName); err != nil {
+		resourceDesc := fmt.Sprintf("RoleBinding %s in namespace %s", kubernetes.RoleBindingName, mizuResourcesNamespace)
+		handleDeletionError(err, resourceDesc, &leftoverResources)
+	}
+
+	if err := kubernetesProvider.RemoveDeployment(ctx, mizuResourcesNamespace, kubernetes.ApiServerPodName); err != nil {
+		resourceDesc := fmt.Sprintf("Deployment %s in namespace %s", kubernetes.ApiServerPodName, mizuResourcesNamespace)
+		handleDeletionError(err, resourceDesc, &leftoverResources)
+	}
+
+	if err := kubernetesProvider.RemovePersistentVolumeClaim(ctx, mizuResourcesNamespace, kubernetes.PersistentVolumeClaimName); err != nil {
+		resourceDesc := fmt.Sprintf("PersistentVolumeClaim %s in namespace %s", kubernetes.PersistentVolumeClaimName, mizuResourcesNamespace)
+		handleDeletionError(err, resourceDesc, &leftoverResources)
+	}
+
+	if err := kubernetesProvider.RemoveRole(ctx, mizuResourcesNamespace, kubernetes.DaemonRoleName); err != nil {
+		resourceDesc := fmt.Sprintf("Role %s in namespace %s", kubernetes.DaemonRoleName, mizuResourcesNamespace)
+		handleDeletionError(err, resourceDesc, &leftoverResources)
+	}
+
+	if err := kubernetesProvider.RemoveRoleBinding(ctx, mizuResourcesNamespace, kubernetes.DaemonRoleBindingName); err != nil {
+		resourceDesc := fmt.Sprintf("RoleBinding %s in namespace %s", kubernetes.DaemonRoleBindingName, mizuResourcesNamespace)
+		handleDeletionError(err, resourceDesc, &leftoverResources)
+	}
+
+	return leftoverResources
+}
+
+func handleDeletionError(err error, resourceDesc string, leftoverResources *[]string) {
+	logger.Log.Debugf("Error removing %s: %v", resourceDesc, errormessage.FormatError(err))
+	*leftoverResources = append(*leftoverResources, resourceDesc)
+}

cli/resources/createResources.go

@@ -0,0 +1,163 @@
+package resources
+
+import (
+	"context"
+	"fmt"
+	"github.com/op/go-logging"
+	"github.com/up9inc/mizu/cli/errormessage"
+	"github.com/up9inc/mizu/cli/mizu"
+	"github.com/up9inc/mizu/cli/uiUtils"
+	"github.com/up9inc/mizu/shared"
+	"github.com/up9inc/mizu/shared/kubernetes"
+	"github.com/up9inc/mizu/shared/logger"
+	core "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
+)
+
+func CreateMizuResources(ctx context.Context, cancel context.CancelFunc, kubernetesProvider *kubernetes.Provider, serializedValidationRules string, serializedContract string, serializedMizuConfig string, isNsRestrictedMode bool, mizuResourcesNamespace string, isInstallMode bool, agentImage string, syncEntriesConfig *shared.SyncEntriesConfig, maxEntriesDBSizeBytes int64, apiServerResources shared.Resources, imagePullPolicy core.PullPolicy, logLevel logging.Level, noPersistentVolumeClaim bool) (bool, error) {
+	if !isNsRestrictedMode {
+		if err := createMizuNamespace(ctx, kubernetesProvider, mizuResourcesNamespace); err != nil {
+			return false, err
+		}
+	}
+
+	if err := createMizuConfigmap(ctx, kubernetesProvider, serializedValidationRules, serializedContract, serializedMizuConfig, mizuResourcesNamespace); err != nil {
+		logger.Log.Warningf(uiUtils.Warning, fmt.Sprintf("Failed to create resources required for policy validation. Mizu will not validate policy rules. error: %v", errormessage.FormatError(err)))
+	}
+
+	mizuServiceAccountExists, err := createRBACIfNecessary(ctx, kubernetesProvider, isNsRestrictedMode, mizuResourcesNamespace, isInstallMode)
+	if err != nil {
+		if !isInstallMode {
+			logger.Log.Warningf(uiUtils.Warning, fmt.Sprintf("Failed to ensure the resources required for IP resolving. Mizu will not resolve target IPs to names. error: %v", errormessage.FormatError(err)))
+		}
+	}
+
+	var serviceAccountName string
+	if mizuServiceAccountExists {
+		serviceAccountName = kubernetes.ServiceAccountName
+	} else {
+		serviceAccountName = ""
+	}
+
+	opts := &kubernetes.ApiServerOptions{
+		Namespace:             mizuResourcesNamespace,
+		PodName:               kubernetes.ApiServerPodName,
+		PodImage:              agentImage,
+		ServiceAccountName:    serviceAccountName,
+		IsNamespaceRestricted: isNsRestrictedMode,
+		SyncEntriesConfig:     syncEntriesConfig,
+		MaxEntriesDBSizeBytes: maxEntriesDBSizeBytes,
+		Resources:             apiServerResources,
+		ImagePullPolicy:       imagePullPolicy,
+		LogLevel:              logLevel,
+	}
+
+	if isInstallMode {
+		if !mizuServiceAccountExists {
+			defer CleanUpMizuResources(ctx, cancel, kubernetesProvider, isNsRestrictedMode, mizuResourcesNamespace)
+			logger.Log.Fatalf(uiUtils.Red, fmt.Sprintf("Failed to ensure the resources required for mizu to run in daemon mode. cannot proceed. error: %v", errormessage.FormatError(err)))
+		}
+		if err := createMizuApiServerDeployment(ctx, kubernetesProvider, opts, noPersistentVolumeClaim); err != nil {
+			return mizuServiceAccountExists, err
+		}
+	} else {
+		if err := createMizuApiServerPod(ctx, kubernetesProvider, opts); err != nil {
+			return mizuServiceAccountExists, err
+		}
+	}
+
+	_, err = kubernetesProvider.CreateService(ctx, mizuResourcesNamespace, kubernetes.ApiServerPodName, kubernetes.ApiServerPodName)
+	if err != nil {
+		return mizuServiceAccountExists, err
+	}
+	logger.Log.Debugf("Successfully created service: %s", kubernetes.ApiServerPodName)
+
+	return mizuServiceAccountExists, nil
+}
+
+func createMizuNamespace(ctx context.Context, kubernetesProvider *kubernetes.Provider, mizuResourcesNamespace string) error {
+	_, err := kubernetesProvider.CreateNamespace(ctx, mizuResourcesNamespace)
+	return err
+}
+
+func createMizuConfigmap(ctx context.Context, kubernetesProvider *kubernetes.Provider, serializedValidationRules string, serializedContract string, serializedMizuConfig string, mizuResourcesNamespace string) error {
+	err := kubernetesProvider.CreateConfigMap(ctx, mizuResourcesNamespace, kubernetes.ConfigMapName, serializedValidationRules, serializedContract, serializedMizuConfig)
+	return err
+}
+
+func createRBACIfNecessary(ctx context.Context, kubernetesProvider *kubernetes.Provider, isNsRestrictedMode bool, mizuResourcesNamespace string, isInstallMode bool) (bool, error) {
+	if !isNsRestrictedMode {
+		if err := kubernetesProvider.CreateMizuRBAC(ctx, mizuResourcesNamespace, kubernetes.ServiceAccountName, kubernetes.ClusterRoleName, kubernetes.ClusterRoleBindingName, mizu.RBACVersion); err != nil {
+			return false, err
+		}
+	} else {
+		if err := kubernetesProvider.CreateMizuRBACNamespaceRestricted(ctx, mizuResourcesNamespace, kubernetes.ServiceAccountName, kubernetes.RoleName, kubernetes.RoleBindingName, mizu.RBACVersion); err != nil {
+			return false, err
+		}
+	}
+	if isInstallMode {
+		if err := kubernetesProvider.CreateDaemonsetRBAC(ctx, mizuResourcesNamespace, kubernetes.ServiceAccountName, kubernetes.DaemonRoleName, kubernetes.DaemonRoleBindingName, mizu.RBACVersion); err != nil {
+			return false, err
+		}
+	}
+	return true, nil
+}
+
+func createMizuApiServerDeployment(ctx context.Context, kubernetesProvider *kubernetes.Provider, opts *kubernetes.ApiServerOptions, noPersistentVolumeClaim bool) error {
+	volumeClaimCreated := false
+	if !noPersistentVolumeClaim {
+		volumeClaimCreated = tryToCreatePersistentVolumeClaim(ctx, kubernetesProvider, opts)
+	}
+
+	pod, err := kubernetesProvider.GetMizuApiServerPodObject(opts, volumeClaimCreated, kubernetes.PersistentVolumeClaimName)
+	if err != nil {
+		return err
+	}
+	pod.Spec.Containers[0].LivenessProbe = &core.Probe{
+		Handler: core.Handler{
+			HTTPGet: &core.HTTPGetAction{
+				Path: "/echo",
+				Port: intstr.FromInt(shared.DefaultApiServerPort),
+			},
+		},
+		InitialDelaySeconds: 1,
+		PeriodSeconds:       10,
+	}
+	if _, err = kubernetesProvider.CreateDeployment(ctx, opts.Namespace, opts.PodName, pod); err != nil {
+		return err
+	}
+	logger.Log.Debugf("Successfully created API server deployment: %s", kubernetes.ApiServerPodName)
+	return nil
+}
+
+func tryToCreatePersistentVolumeClaim(ctx context.Context, kubernetesProvider *kubernetes.Provider, opts *kubernetes.ApiServerOptions) bool {
+	isDefaultStorageClassAvailable, err := kubernetesProvider.IsDefaultStorageProviderAvailable(ctx)
+	if err != nil {
+		logger.Log.Warningf(uiUtils.Yellow, "An error occured when checking if a default storage provider exists in this cluster, this means mizu data will be lost on mizu-api-server pod restart")
+		logger.Log.Debugf("error checking if default storage class exists: %v", err)
+		return false
+	} else if !isDefaultStorageClassAvailable {
+		logger.Log.Warningf(uiUtils.Yellow, "Could not find default storage provider in this cluster, this means mizu data will be lost on mizu-api-server pod restart")
+		return false
+	}
+
+	if _, err = kubernetesProvider.CreatePersistentVolumeClaim(ctx, opts.Namespace, kubernetes.PersistentVolumeClaimName, opts.MaxEntriesDBSizeBytes + mizu.DaemonModePersistentVolumeSizeBufferBytes); err != nil {
+		logger.Log.Warningf(uiUtils.Yellow, "An error has occured while creating a persistent volume claim for mizu, this means mizu data will be lost on mizu-api-server pod restart")
+		logger.Log.Debugf("error creating persistent volume claim: %v", err)
+		return false
+	}
+
+	return true
+}
+
+func createMizuApiServerPod(ctx context.Context, kubernetesProvider *kubernetes.Provider, opts *kubernetes.ApiServerOptions) error {
+	pod, err := kubernetesProvider.GetMizuApiServerPodObject(opts, false, "")
+	if err != nil {
+		return err
+	}
+	if _, err = kubernetesProvider.CreatePod(ctx, opts.Namespace, pod); err != nil {
+		return err
+	}
+	logger.Log.Debugf("Successfully created API server pod: %s", kubernetes.ApiServerPodName)
+	return nil
+}

cli/utils/waitUtils.go

@@ -0,0 +1,25 @@
+package utils
+
+import (
+	"context"
+	"github.com/up9inc/mizu/shared/logger"
+	"os"
+	"os/signal"
+	"syscall"
+)
+
+func WaitForFinish(ctx context.Context, cancel context.CancelFunc) {
+	logger.Log.Debugf("waiting for finish...")
+	sigChan := make(chan os.Signal, 1)
+	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
+
+	// block until ctx cancel is called or termination signal is received
+	select {
+	case <-ctx.Done():
+		logger.Log.Debugf("ctx done")
+		break
+	case <-sigChan:
+		logger.Log.Debugf("Got termination signal, canceling execution...")
+		cancel()
+	}
+}

shared/models.go

@@ -1,11 +1,13 @@
 package shared
 
 import (
+	"io/ioutil"
+	"strings"
+
 	"github.com/op/go-logging"
 	"github.com/up9inc/mizu/shared/logger"
 	"github.com/up9inc/mizu/tap/api"
-	"io/ioutil"
-	"strings"
+	v1 "k8s.io/api/core/v1"
 
 	"gopkg.in/yaml.v3"
 )
@@ -21,6 +23,7 @@ const (
 	WebSocketMessageTypeToast         WebSocketMessageType = "toast"
 	WebSocketMessageTypeQueryMetadata WebSocketMessageType = "queryMetadata"
 	WebSocketMessageTypeStartTime     WebSocketMessageType = "startTime"
+	WebSocketMessageTypeTapConfig     WebSocketMessageType = "tapConfig"
 )
 
 type Resources struct {
@@ -67,6 +70,11 @@ type WebSocketStatusMessage struct {
 	TappingStatus []TappedPodStatus `json:"tappingStatus"`
 }
 
+type WebSocketTapConfigMessage struct {
+	*WebSocketMessageMetadata
+	TapTargets []v1.Pod `json:"pods"`
+}
+
 type TapperStatus struct {
 	TapperName string `json:"tapperName"`
 	NodeName   string `json:"nodeName"`

tap/passive_tapper.go

@@ -11,6 +11,7 @@ package tap
 import (
 	"encoding/json"
 	"flag"
+	"fmt"
 	"os"
 	"runtime"
 	"strings"
@@ -60,8 +61,11 @@ type TapOpts struct {
 	FilterAuthorities []v1.Pod
 }
 
-var extensions []*api.Extension                   // global
-var filteringOptions *api.TrafficFilteringOptions // global
+var extensions []*api.Extension                     // global
+var filteringOptions *api.TrafficFilteringOptions   // global
+var tapTargets []v1.Pod                             // global
+var packetSourceManager *source.PacketSourceManager // global
+var mainPacketInputChan chan source.TcpPacketInfo   // global
 
 func inArrayInt(arr []int, valueToCheck int) bool {
 	for _, value := range arr {
@@ -86,7 +90,9 @@ func StartPassiveTapper(opts *TapOpts, outputItems chan *api.OutputChannelItem,
 	filteringOptions = options
 
 	if opts.FilterAuthorities == nil {
-		opts.FilterAuthorities = []v1.Pod{}
+		tapTargets = []v1.Pod{}
+	} else {
+		tapTargets = opts.FilterAuthorities
 	}
 
 	if GetMemoryProfilingEnabled() {
@@ -96,6 +102,23 @@ func StartPassiveTapper(opts *TapOpts, outputItems chan *api.OutputChannelItem,
 	go startPassiveTapper(opts, outputItems)
 }
 
+func UpdateTapTargets(newTapTargets []v1.Pod) {
+	tapTargets = newTapTargets
+	if err := initializePacketSources(); err != nil {
+		logger.Log.Fatal(err)
+	}
+	printNewTapTargets()
+}
+
+func printNewTapTargets() {
+	printStr := ""
+	for _, tapTarget := range tapTargets {
+		printStr += fmt.Sprintf("%s (%s), ", tapTarget.Status.PodIP, tapTarget.Name)
+	}
+	printStr = strings.TrimRight(printStr, ", ")
+	logger.Log.Infof("Now tapping: %s", printStr)
+}
+
 func printPeriodicStats(cleaner *Cleaner) {
 	statsPeriod := time.Second * time.Duration(*statsevery)
 	ticker := time.NewTicker(statsPeriod)
@@ -136,7 +159,11 @@ func printPeriodicStats(cleaner *Cleaner) {
 	}
 }
 
-func initializePacketSources(opts *TapOpts) (*source.PacketSourceManager, error) {
+func initializePacketSources() error {
+	if packetSourceManager != nil {
+		packetSourceManager.Close()
+	}
+
 	var bpffilter string
 	if len(flag.Args()) > 0 {
 		bpffilter = strings.Join(flag.Args(), " ")
@@ -151,7 +178,13 @@ func initializePacketSources(opts *TapOpts) (*source.PacketSourceManager, error)
 		BpfFilter:   bpffilter,
 	}
 
-	return source.NewPacketSourceManager(*procfs, *pids, *fname, *iface, *istio, opts.FilterAuthorities, behaviour)
+	var err error
+	if packetSourceManager, err = source.NewPacketSourceManager(*procfs, *pids, *fname, *iface, *istio, tapTargets, behaviour); err != nil {
+		return err
+	} else {
+		packetSourceManager.ReadPackets(!*nodefrag, mainPacketInputChan)
+		return nil
+	}
 }
 
 func startPassiveTapper(opts *TapOpts, outputItems chan *api.OutputChannelItem) {
@@ -161,25 +194,15 @@ func startPassiveTapper(opts *TapOpts, outputItems chan *api.OutputChannelItem)
 	diagnose.InitializeErrorsMap(*debug, *verbose, *quiet)
 	diagnose.InitializeTapperInternalStats()
 
-	sources, err := initializePacketSources(opts)
-
-	if err != nil {
+	if err := initializePacketSources(); err != nil {
 		logger.Log.Fatal(err)
 	}
 
-	defer sources.Close()
-
-	if err != nil {
-		logger.Log.Fatal(err)
-	}
-
-	packets := make(chan source.TcpPacketInfo)
+	mainPacketInputChan = make(chan source.TcpPacketInfo)
 	assembler := NewTcpAssembler(outputItems, streamsMap, opts)
 
 	diagnose.AppStats.SetStartTime(time.Now())
 
-	sources.ReadPackets(!*nodefrag, packets)
-
 	staleConnectionTimeout := time.Second * time.Duration(*staleTimeoutSeconds)
 	cleaner := Cleaner{
 		assembler:         assembler.Assembler,
@@ -191,7 +214,7 @@ func startPassiveTapper(opts *TapOpts, outputItems chan *api.OutputChannelItem)
 
 	go printPeriodicStats(&cleaner)
 
-	assembler.processPackets(*hexdumppkt, packets)
+	assembler.processPackets(*hexdumppkt, mainPacketInputChan)
 
 	if diagnose.TapErrors.OutputLevel >= 2 {
 		assembler.dumpStreamPool()

tap/tcp_assembler.go

@@ -65,11 +65,11 @@ func (a *tcpAssembler) processPackets(dumpPacket bool, packets <-chan source.Tcp
 
 	for packetInfo := range packets {
 		packetsCount := diagnose.AppStats.IncPacketsCount()
-		
-		if packetsCount % PACKETS_SEEN_LOG_THRESHOLD == 0 {
+
+		if packetsCount%PACKETS_SEEN_LOG_THRESHOLD == 0 {
 			logger.Log.Debugf("Packets seen: #%d", packetsCount)
 		}
-		
+
 		packet := packetInfo.Packet
 		data := packet.Data()
 		diagnose.AppStats.UpdateProcessedBytes(uint64(len(data)))
@@ -91,7 +91,6 @@ func (a *tcpAssembler) processPackets(dumpPacket bool, packets <-chan source.Tcp
 				CaptureInfo: packet.Metadata().CaptureInfo,
 			}
 			diagnose.InternalStats.Totalsz += len(tcp.Payload)
-			logger.Log.Debugf("%s:%v -> %s:%v", packet.NetworkLayer().NetworkFlow().Src(), tcp.SrcPort, packet.NetworkLayer().NetworkFlow().Dst(), tcp.DstPort)
 			a.assemblerMutex.Lock()
 			a.AssembleWithContext(packet.NetworkLayer().NetworkFlow(), tcp, &c)
 			a.assemblerMutex.Unlock()

tap/tcp_stream.go

@@ -2,7 +2,6 @@ package tap
 
 import (
 	"encoding/binary"
-	"fmt"
 	"sync"
 
 	"github.com/google/gopacket"
@@ -75,7 +74,7 @@ func (t *tcpStream) Accept(tcp *layers.TCP, ci gopacket.CaptureInfo, dir reassem
 }
 
 func (t *tcpStream) ReassembledSG(sg reassembly.ScatterGather, ac reassembly.AssemblerContext) {
-	dir, start, end, skip := sg.Info()
+	dir, _, _, skip := sg.Info()
 	length, saved := sg.Lengths()
 	// update stats
 	sgStats := sg.Stats()
@@ -103,13 +102,6 @@ func (t *tcpStream) ReassembledSG(sg reassembly.ScatterGather, ac reassembly.Ass
 	diagnose.InternalStats.OverlapBytes += sgStats.OverlapBytes
 	diagnose.InternalStats.OverlapPackets += sgStats.OverlapPackets
 
-	var ident string
-	if dir == reassembly.TCPDirClientToServer {
-		ident = fmt.Sprintf("%v %v(%s): ", t.net, t.transport, dir)
-	} else {
-		ident = fmt.Sprintf("%v %v(%s): ", t.net.Reverse(), t.transport.Reverse(), dir)
-	}
-	diagnose.TapErrors.Debug("%s: SG reassembled packet with %d bytes (start:%v,end:%v,skip:%d,saved:%d,nb:%d,%d,overlap:%d,%d)", ident, length, start, end, skip, saved, sgStats.Packets, sgStats.Chunks, sgStats.OverlapBytes, sgStats.OverlapPackets)
 	if skip == -1 && *allowmissinginit {
 		// this is allowed
 	} else if skip != 0 {
@@ -174,7 +166,6 @@ func (t *tcpStream) ReassembledSG(sg reassembly.ScatterGather, ac reassembly.Ass
 }
 
 func (t *tcpStream) ReassemblyComplete(ac reassembly.AssemblerContext) bool {
-	diagnose.TapErrors.Debug("%s: Connection closed", t.ident)
 	if t.isTapTarget && !t.isClosed {
 		t.Close()
 	}

tap/tcp_stream_factory.go

@@ -54,7 +54,6 @@ func NewTcpStreamFactory(emitter api.Emitter, streamsMap *tcpStreamMap, opts *Ta
 }
 
 func (factory *tcpStreamFactory) New(net, transport gopacket.Flow, tcp *layers.TCP, ac reassembly.AssemblerContext) reassembly.Stream {
-	logger.Log.Debugf("* NEW: %s %s", net, transport)
 	fsmOptions := reassembly.TCPSimpleFSMOptions{
 		SupportMissingEstablishment: *allowmissinginit,
 	}
@@ -152,22 +151,17 @@ func inArrayPod(pods []v1.Pod, address string) bool {
 
 func (factory *tcpStreamFactory) getStreamProps(srcIP string, srcPort string, dstIP string, dstPort string) *streamProps {
 	if factory.opts.HostMode {
-		if inArrayPod(factory.opts.FilterAuthorities, fmt.Sprintf("%s:%s", dstIP, dstPort)) {
-			logger.Log.Debugf("getStreamProps %s", fmt.Sprintf("+ host1 %s:%s", dstIP, dstPort))
+		if inArrayPod(tapTargets, fmt.Sprintf("%s:%s", dstIP, dstPort)) {
 			return &streamProps{isTapTarget: true, isOutgoing: false}
-		} else if inArrayPod(factory.opts.FilterAuthorities, dstIP) {
-			logger.Log.Debugf("getStreamProps %s", fmt.Sprintf("+ host2 %s", dstIP))
+		} else if inArrayPod(tapTargets, dstIP) {
 			return &streamProps{isTapTarget: true, isOutgoing: false}
-		} else if inArrayPod(factory.opts.FilterAuthorities, fmt.Sprintf("%s:%s", srcIP, srcPort)) {
-			logger.Log.Debugf("getStreamProps %s", fmt.Sprintf("+ host3 %s:%s", srcIP, srcPort))
+		} else if inArrayPod(tapTargets, fmt.Sprintf("%s:%s", srcIP, srcPort)) {
 			return &streamProps{isTapTarget: true, isOutgoing: true}
-		} else if inArrayPod(factory.opts.FilterAuthorities, srcIP) {
-			logger.Log.Debugf("getStreamProps %s", fmt.Sprintf("+ host4 %s", srcIP))
+		} else if inArrayPod(tapTargets, srcIP) {
 			return &streamProps{isTapTarget: true, isOutgoing: true}
 		}
 		return &streamProps{isTapTarget: false, isOutgoing: false}
 	} else {
-		logger.Log.Debugf("getStreamProps %s", fmt.Sprintf("+ notHost3 %s:%s -> %s:%s", srcIP, srcPort, dstIP, dstPort))
 		return &streamProps{isTapTarget: true}
 	}
 }

ui/package-lock.json

@@ -7747,9 +7747,9 @@
       "integrity": "sha512-l9sfDFsuqtOqKDsQdqrMRk0U85RZc0RtOR9yPI7mRVOa4FsR/BVnZ0shmQRM96Ji99kYZP/7hn1cedc1+ApsTQ=="
     },
     "highlight.js": {
-      "version": "10.7.2",
-      "resolved": "https://registry.npmjs.org/highlight.js/-/highlight.js-10.7.2.tgz",
-      "integrity": "sha512-oFLl873u4usRM9K63j4ME9u3etNF0PLiJhSQ8rdfuL51Wn3zkD6drf9ZW0dOzjnZI22YYG24z30JcmfCZjMgYg=="
+      "version": "11.3.1",
+      "resolved": "https://registry.npmjs.org/highlight.js/-/highlight.js-11.3.1.tgz",
+      "integrity": "sha512-PUhCRnPjLtiLHZAQ5A/Dt5F8cWZeMyj9KRsACsWT+OD6OP0x6dp5OmT5jdx0JgEyPxPZZIPQpRN2TciUT7occw=="
     },
     "hmac-drbg": {
       "version": "1.0.1",
@@ -10234,6 +10234,11 @@
       "resolved": "https://registry.npmjs.org/jsesc/-/jsesc-2.5.2.tgz",
       "integrity": "sha512-OYu7XEzjkCQ3C5Ps3QIZsQfNpqoJyZZA99wd9aWd05NCtC5pWOkShK2mkL6HXQR6/Cy2lbNdPlZBpuQHXE63gA=="
     },
+    "json-beautify": {
+      "version": "1.1.1",
+      "resolved": "https://registry.npmjs.org/json-beautify/-/json-beautify-1.1.1.tgz",
+      "integrity": "sha512-17j+Hk2lado0xqKtUcyAjK0AtoHnPSIgktWRsEXgdFQFG9UnaGw6CHa0J7xsvulxRpFl6CrkDFHght1p5ZJc4A=="
+    },
     "json-parse-better-errors": {
       "version": "1.0.2",
       "resolved": "https://registry.npmjs.org/json-parse-better-errors/-/json-parse-better-errors-1.0.2.tgz",
@@ -10612,6 +10617,13 @@
       "requires": {
         "fault": "^1.0.0",
         "highlight.js": "~10.7.0"
+      },
+      "dependencies": {
+        "highlight.js": {
+          "version": "10.7.3",
+          "resolved": "https://registry.npmjs.org/highlight.js/-/highlight.js-10.7.3.tgz",
+          "integrity": "sha512-tzcUFauisWKNHaRkN4Wjl/ZA07gENAjFl3J/c480dprkGTg5EQstgaNFqBfUqCq54kZRIEcreTsAgF/m2quD7A=="
+        }
       }
     },
     "lru-cache": {
@@ -13577,6 +13589,34 @@
       "resolved": "https://registry.npmjs.org/react-is/-/react-is-16.13.1.tgz",
       "integrity": "sha512-24e6ynE2H+OKt4kqsOvNd8kBpV65zoxbA4BVsEOB3ARVWQki/DHzaUoC5KuON/BiccDaCCTZBuOcfZs70kR8bQ=="
     },
+    "react-lowlight": {
+      "version": "3.0.0",
+      "resolved": "https://registry.npmjs.org/react-lowlight/-/react-lowlight-3.0.0.tgz",
+      "integrity": "sha512-s0+T81PsCbUZYd/0XrplGc6kQEUdiwLKI0G6umJP1ViqRoZRCvSuHvXOy20Usd2ywDKWLuVETQgBDPeNQhPNZg==",
+      "requires": {
+        "lowlight": "^2.4.1"
+      },
+      "dependencies": {
+        "fault": {
+          "version": "2.0.1",
+          "resolved": "https://registry.npmjs.org/fault/-/fault-2.0.1.tgz",
+          "integrity": "sha512-WtySTkS4OKev5JtpHXnib4Gxiurzh5NCGvWrFaZ34m6JehfTUhKZvn9njTfw48t6JumVQOmrKqpmGcdwxnhqBQ==",
+          "requires": {
+            "format": "^0.2.0"
+          }
+        },
+        "lowlight": {
+          "version": "2.4.1",
+          "resolved": "https://registry.npmjs.org/lowlight/-/lowlight-2.4.1.tgz",
+          "integrity": "sha512-mQkAG0zGQ9lcYecEft+hl9uV1fD6HpURA83/TYrsxKvb8xX2mfyB+aaV/A/aWmhhEcWVzr9Cc+l/fvUYfEUumw==",
+          "requires": {
+            "@types/hast": "^2.0.0",
+            "fault": "^2.0.0",
+            "highlight.js": "~11.3.0"
+          }
+        }
+      }
+    },
     "react-refresh": {
       "version": "0.8.3",
       "resolved": "https://registry.npmjs.org/react-refresh/-/react-refresh-0.8.3.tgz",
@@ -13663,6 +13703,13 @@
         "lowlight": "^1.17.0",
         "prismjs": "^1.22.0",
         "refractor": "^3.2.0"
+      },
+      "dependencies": {
+        "highlight.js": {
+          "version": "10.7.3",
+          "resolved": "https://registry.npmjs.org/highlight.js/-/highlight.js-10.7.3.tgz",
+          "integrity": "sha512-tzcUFauisWKNHaRkN4Wjl/ZA07gENAjFl3J/c480dprkGTg5EQstgaNFqBfUqCq54kZRIEcreTsAgF/m2quD7A=="
+        }
       }
     },
     "react-toastify": {
@@ -18149,11 +18196,24 @@
       "resolved": "https://registry.npmjs.org/ws/-/ws-7.4.5.tgz",
       "integrity": "sha512-xzyu3hFvomRfXKH8vOFMU3OguG6oOvhXMo3xsGy3xWExqaM2dxBbVxuD99O7m3ZUFMvvscsZDqxfgMaRr/Nr1g=="
     },
+    "xml-formatter": {
+      "version": "2.6.0",
+      "resolved": "https://registry.npmjs.org/xml-formatter/-/xml-formatter-2.6.0.tgz",
+      "integrity": "sha512-+bQeoiE5W3CJdDCHTlveYSWFfQWnYB3uHGeRJ6LlEsL5kT++mWy9iN1cMeEDfBbgOnXO2DNUbmQ6elkR/mCcjg==",
+      "requires": {
+        "xml-parser-xo": "^3.2.0"
+      }
+    },
     "xml-name-validator": {
       "version": "3.0.0",
       "resolved": "https://registry.npmjs.org/xml-name-validator/-/xml-name-validator-3.0.0.tgz",
       "integrity": "sha512-A5CUptxDsvxKJEU3yO6DuWBSJz/qizqzJKOMIfUJHETbBw/sFaDxgd6fxm1ewUaM0jZ444Fc5vC5ROYurg/4Pw=="
     },
+    "xml-parser-xo": {
+      "version": "3.2.0",
+      "resolved": "https://registry.npmjs.org/xml-parser-xo/-/xml-parser-xo-3.2.0.tgz",
+      "integrity": "sha512-8LRU6cq+d7mVsoDaMhnkkt3CTtAs4153p49fRo+HIB3I1FD1o5CeXRjRH29sQevIfVJIcPjKSsPU/+Ujhq09Rg=="
+    },
     "xmlchars": {
       "version": "2.2.0",
       "resolved": "https://registry.npmjs.org/xmlchars/-/xmlchars-2.2.0.tgz",

ui/package.json

@@ -15,6 +15,8 @@
     "@types/react-dom": "^17.0.3",
     "@uiw/react-textarea-code-editor": "^1.4.12",
     "axios": "^0.21.1",
+    "highlight.js": "^11.3.1",
+    "json-beautify": "^1.1.1",
     "jsonpath": "^1.1.1",
     "moment": "^2.29.1",
     "node-sass": "^5.0.0",
@@ -23,12 +25,14 @@
     "react": "^17.0.2",
     "react-copy-to-clipboard": "^5.0.3",
     "react-dom": "^17.0.2",
+    "react-lowlight": "^3.0.0",
     "react-scripts": "4.0.3",
     "react-scrollable-feed-virtualized": "^1.4.9",
     "react-syntax-highlighter": "^15.4.3",
     "react-toastify": "^8.0.3",
     "typescript": "^4.2.4",
-    "web-vitals": "^1.1.1"
+    "web-vitals": "^1.1.1",
+    "xml-formatter": "^2.6.0"
   },
   "scripts": {
     "start": "react-scripts start",

ui/src/components/EntryDetailed/EntrySections.tsx

@@ -6,6 +6,8 @@ import FancyTextDisplay from "../UI/FancyTextDisplay";
 import Queryable from "../UI/Queryable";
 import Checkbox from "../UI/Checkbox";
 import ProtobufDecoder from "protobuf-decoder";
+import {default as jsonBeautify} from "json-beautify";
+import {default as xmlBeautify} from "xml-formatter";
 
 interface EntryViewLineProps {
     label: string;
@@ -121,23 +123,41 @@ export const EntryBodySection: React.FC<EntryBodySectionProps> = ({
     contentType,
     selector,
 }) => {
-    const MAXIMUM_BYTES_TO_HIGHLIGHT = 10000; // The maximum of chars to highlight in body, in case the response can be megabytes
-    const supportedLanguages = [['html', 'html'], ['json', 'json'], ['application/grpc', 'json']]; // [[indicator, languageToUse],...]
-    const jsonLikeFormats = ['json'];
+    const MAXIMUM_BYTES_TO_FORMAT = 1000000; // The maximum of chars to highlight in body, in case the response can be megabytes
+    const jsonLikeFormats = ['json', 'yaml', 'yml'];
+    const xmlLikeFormats = ['xml', 'html'];
     const protobufFormats = ['application/grpc'];
-    const [isWrapped, setIsWrapped] = useState(false);
+    const supportedFormats = jsonLikeFormats.concat(xmlLikeFormats, protobufFormats);
 
-    const formatTextBody = (body): string => {
-        const chunk = body.slice(0, MAXIMUM_BYTES_TO_HIGHLIGHT);
-        const bodyBuf = encoding === 'base64' ? atob(chunk) : chunk;
+    const [isPretty, setIsPretty] = useState(true);
+    const [showLineNumbers, setShowLineNumbers] = useState(true);
+    const [decodeBase64, setDecodeBase64] = useState(true);
+
+    const isBase64Encoding = encoding === 'base64';
+    const supportsPrettying = supportedFormats.some(format => contentType?.indexOf(format) > -1);
+
+    const formatTextBody = (body: any): string => {
+        if (!decodeBase64) return body;
+
+        const chunk = body.slice(0, MAXIMUM_BYTES_TO_FORMAT);
+        const bodyBuf = isBase64Encoding ? atob(chunk) : chunk;
+
+        if (!isPretty) return bodyBuf;
 
         try {
             if (jsonLikeFormats.some(format => contentType?.indexOf(format) > -1)) {
-                return JSON.stringify(JSON.parse(bodyBuf), null, 2);
+                return jsonBeautify(JSON.parse(bodyBuf), null, 2, 80);
+            }  else if (xmlLikeFormats.some(format => contentType?.indexOf(format) > -1)) {
+                return xmlBeautify(bodyBuf, {
+                    indentation: '  ',
+                    filter: (node) => node.type !== 'Comment',
+                    collapseContent: true,
+                    lineSeparator: '\n'
+                });
             } else if (protobufFormats.some(format => contentType?.indexOf(format) > -1)) {
                 // Replace all non printable characters (ASCII)
                 const protobufDecoder = new ProtobufDecoder(bodyBuf, true);
-                return JSON.stringify(protobufDecoder.decode().toSimple(), null, 2);
+                return jsonBeautify(protobufDecoder.decode().toSimple(), null, 2, 80);
             }
         } catch (error) {
             console.error(error);
@@ -145,13 +165,6 @@ export const EntryBodySection: React.FC<EntryBodySectionProps> = ({
         return bodyBuf;
     }
 
-    const getLanguage = (mimetype) => {
-        const chunk = content?.slice(0, 100);
-        if (chunk.indexOf('html') > 0 || chunk.indexOf('HTML') > 0) return supportedLanguages[0][1];
-        const language = supportedLanguages.find(el => (mimetype + contentType).indexOf(el[0]) > -1);
-        return language ? language[1] : 'default';
-    }
-
     return <React.Fragment>
         {content && content?.length > 0 && <EntrySectionContainer
                                                 title='Body'
@@ -159,24 +172,26 @@ export const EntryBodySection: React.FC<EntryBodySectionProps> = ({
                                                 query={`${selector} == r".*"`}
                                                 updateQuery={updateQuery}
                                             >
-            <table>
-                <tbody>
-                    <EntryViewLine label={'Mime type'} value={contentType} useTooltip={false}/>
-                    {encoding && <EntryViewLine label={'Encoding'} value={encoding} useTooltip={false}/>}
-                </tbody>
-            </table>
+            <div style={{display: 'flex', alignItems: 'center', alignContent: 'center', margin: "5px 0"}}>
+                {supportsPrettying && <div style={{paddingTop: 3}}>
+                    <Checkbox checked={isPretty} onToggle={() => {setIsPretty(!isPretty)}}/>
+                </div>}
+                {supportsPrettying && <span style={{marginLeft: '.2rem'}}>Pretty</span>}
 
-            <div style={{display: 'flex', alignItems: 'center', alignContent: 'center', margin: "5px 0"}} onClick={() => setIsWrapped(!isWrapped)}>
-                <div style={{paddingTop: 3}}>
-                    <Checkbox checked={isWrapped} onToggle={() => {}}/>
+                <div style={{paddingTop: 3, paddingLeft: supportsPrettying ? 20 : 0}}>
+                    <Checkbox checked={showLineNumbers} onToggle={() => {setShowLineNumbers(!showLineNumbers)}}/>
                 </div>
-                <span style={{marginLeft: '.5rem'}}>Wrap text</span>
+                <span style={{marginLeft: '.2rem'}}>Line numbers</span>
+
+                {isBase64Encoding && <div style={{paddingTop: 3, paddingLeft: 20}}>
+                    <Checkbox checked={decodeBase64} onToggle={() => {setDecodeBase64(!decodeBase64)}}/>
+                </div>}
+                {isBase64Encoding && <span style={{marginLeft: '.2rem'}}>Decode Base64</span>}
             </div>
 
             <SyntaxHighlighter
-                isWrapped={isWrapped}
                 code={formatTextBody(content)}
-                language={content?.mimeType ? getLanguage(content.mimeType) : 'default'}
+                showLineNumbers={showLineNumbers}
             />
         </EntrySectionContainer>}
     </React.Fragment>
@@ -334,7 +349,6 @@ export const EntryContractSection: React.FC<EntryContractSectionProps> = ({color
         </EntrySectionContainer>}
         {contractContent && <EntrySectionContainer title="Contract" color={color}>
             <SyntaxHighlighter
-                isWrapped={false}
                 code={contractContent}
                 language={"yaml"}
             />

ui/src/components/EntryListItem/EntryListItem.module.sass

@@ -68,6 +68,7 @@
   flex-direction: column
   overflow: hidden
   padding-right: 10px
+  padding-top: 4px
   flex-grow: 1
 
 .separatorRight

ui/src/components/Filters.tsx

@@ -167,7 +167,6 @@ export const QueryForm: React.FC<QueryFormProps> = ({query, setQuery, background
                                 This is a simple query that matches to HTTP packets with request path "/catalogue":
                             </Typography>
                             <SyntaxHighlighter
-                                isWrapped={false}
                                 showLineNumbers={false}
                                 code={`http and request.path == "/catalogue"`}
                                 language="python"
@@ -176,7 +175,6 @@ export const QueryForm: React.FC<QueryFormProps> = ({query, setQuery, background
                                 The same query can be negated for HTTP path and written like this:
                             </Typography>
                             <SyntaxHighlighter
-                                isWrapped={false}
                                 showLineNumbers={false}
                                 code={`http and request.path != "/catalogue"`}
                                 language="python"
@@ -185,7 +183,6 @@ export const QueryForm: React.FC<QueryFormProps> = ({query, setQuery, background
                                 The syntax supports regular expressions. Here is a query that matches the HTTP requests that send JSON to a server:
                             </Typography>
                             <SyntaxHighlighter
-                                isWrapped={false}
                                 showLineNumbers={false}
                                 code={`http and request.headers["Accept"] == r"application/json.*"`}
                                 language="python"
@@ -194,7 +191,6 @@ export const QueryForm: React.FC<QueryFormProps> = ({query, setQuery, background
                                 Here is another query that matches HTTP responses with status code 4xx:
                             </Typography>
                             <SyntaxHighlighter
-                                isWrapped={false}
                                 showLineNumbers={false}
                                 code={`http and response.status == r"4.*"`}
                                 language="python"
@@ -203,7 +199,6 @@ export const QueryForm: React.FC<QueryFormProps> = ({query, setQuery, background
                                 The same exact query can be as integer comparison:
                             </Typography>
                             <SyntaxHighlighter
-                                isWrapped={false}
                                 showLineNumbers={false}
                                 code={`http and response.status >= 400`}
                                 language="python"
@@ -212,7 +207,6 @@ export const QueryForm: React.FC<QueryFormProps> = ({query, setQuery, background
                                 The results can be queried based on their timestamps:
                             </Typography>
                             <SyntaxHighlighter
-                                isWrapped={false}
                                 showLineNumbers={false}
                                 code={`timestamp < datetime("10/28/2021, 9:13:02.905 PM")`}
                                 language="python"
@@ -224,7 +218,6 @@ export const QueryForm: React.FC<QueryFormProps> = ({query, setQuery, background
                                 Since Mizu supports various protocols like gRPC, AMQP, Kafka and Redis. It's possible to write complex queries that match multiple protocols like this:
                             </Typography>
                             <SyntaxHighlighter
-                                isWrapped={false}
                                 showLineNumbers={false}
                                 code={`(http and request.method == "PUT") or (amqp and request.queue.startsWith("test"))\n or (kafka and response.payload.errorCode == 2) or (redis and request.key == "example")\n or (grpc and request.headers[":path"] == r".*foo.*")`}
                                 language="python"
@@ -242,7 +235,6 @@ export const QueryForm: React.FC<QueryFormProps> = ({query, setQuery, background
                                 Such that; clicking this icon in left-pane, would append the query below:
                             </Typography>
                             <SyntaxHighlighter
-                                isWrapped={false}
                                 showLineNumbers={false}
                                 code={`and dst.name == "carts.sock-shop"`}
                                 language="python"
@@ -260,7 +252,6 @@ export const QueryForm: React.FC<QueryFormProps> = ({query, setQuery, background
                                 A query that compares one selector to another is also a valid query:
                             </Typography>
                             <SyntaxHighlighter
-                                isWrapped={false}
                                 showLineNumbers={false}
                                 code={`http and (request.query["x"] == response.headers["y"]\n or response.content.text.contains(request.query["x"]))`}
                                 language="python"
@@ -276,7 +267,6 @@ export const QueryForm: React.FC<QueryFormProps> = ({query, setQuery, background
                                 true if the given selector's value starts with the string:
                             </Typography>
                             <SyntaxHighlighter
-                                isWrapped={false}
                                 showLineNumbers={false}
                                 code={`request.path.startsWith("something")`}
                                 language="python"
@@ -285,7 +275,6 @@ export const QueryForm: React.FC<QueryFormProps> = ({query, setQuery, background
                                 true if the given selector's value ends with the string:
                             </Typography>
                             <SyntaxHighlighter
-                                isWrapped={false}
                                 showLineNumbers={false}
                                 code={`request.path.endsWith("something")`}
                                 language="python"
@@ -294,7 +283,6 @@ export const QueryForm: React.FC<QueryFormProps> = ({query, setQuery, background
                                 true if the given selector's value contains the string:
                             </Typography>
                             <SyntaxHighlighter
-                                isWrapped={false}
                                 showLineNumbers={false}
                                 code={`request.path.contains("something")`}
                                 language="python"
@@ -303,7 +291,6 @@ export const QueryForm: React.FC<QueryFormProps> = ({query, setQuery, background
                                 returns the UNIX timestamp which is the equivalent of the time that's provided by the string. Invalid input evaluates to false:
                             </Typography>
                             <SyntaxHighlighter
-                                isWrapped={false}
                                 showLineNumbers={false}
                                 code={`timestamp >= datetime("10/19/2021, 6:29:02.593 PM")`}
                                 language="python"
@@ -312,7 +299,6 @@ export const QueryForm: React.FC<QueryFormProps> = ({query, setQuery, background
                                 limits the number of records that are streamed back as a result of a query. Always evaluates to true:
                             </Typography>
                             <SyntaxHighlighter
-                                isWrapped={false}
                                 showLineNumbers={false}
                                 code={`and limit(100)`}
                                 language="python"

ui/src/components/UI/SyntaxHighlighter/highlighterStyle.ts

@@ -1,152 +0,0 @@
-export const highlighterStyle = {
-    "code[class*=\"language-\"]": {
-        "color": "#494677",
-        "fontFamily": "Inconsolata, Monaco, Consolas, 'Courier New', Courier, monospace",
-        "direction": "ltr",
-        "textAlign": "left",
-        "whiteSpace": "pre",
-        "wordSpacing": "normal",
-        "wordBreak": "normal",
-        "lineHeight": "1.5",
-        "MozTabSize": "4",
-        "OTabSize": "4",
-        "tabSize": "4",
-        "padding": "1rem",
-        "WebkitHyphetokenns": "none",
-        "MozHyphens": "none",
-        "msHyphens": "none",
-        "hyphens": "none"
-    },
-    "pre[class*=\"language-\"]": {
-        "color": "#494677",
-        "fontFamily": "Inconsolata, Monaco, Consolas, 'Courier New', Courier, monospace",
-        "direction": "ltr",
-        "textAlign": "left",
-        "whiteSpace": "pre",
-        "wordSpacing": "normal",
-        "wordBreak": "normal",
-        "lineHeight": "1.2",
-        "MozTabSize": "4",
-        "OTabSize": "4",
-        "tabSize": "4",
-        "WebkitHyphens": "none",
-        "MozHyphens": "none",
-        "msHyphens": "none",
-        "hyphens": "none",
-        "padding": "0",
-        "margin": ".5em 0",
-        "overflow": "auto",
-        "borderRadius": "0.3em",
-        "background": "#F7F9FC"
-    },
-    ":not(pre) > code[class*=\"language-\"]": {
-        "background": "#F7F9FC",
-        "padding": ".1em",
-        "borderRadius": ".3em"
-    },
-    "comment": {
-        "color": "#5d6aa0"
-    },
-    "prolog": {
-        "color": "#494677"
-    },
-    "doctype": {
-        "color": "#494677"
-    },
-    "cdata": {
-        "color": "#494677"
-    },
-    "punctuation": {
-        "color": "#494677"
-    },
-    ".namespace": {
-        "Opacity": ".7"
-    },
-    "property": {
-        "color": "#627ef7"
-    },
-    "keyword": {
-        "color": "#627ef7"
-    },
-    "tag": {
-        "color": "#627ef7"
-    },
-    "class-name": {
-        "color": "#3eb545",
-        "textDecoration": "underline"
-    },
-    "boolean": {
-        "color": "#3eb545"
-    },
-    "constant": {
-        "color": "#3eb545"
-    },
-    "symbol": {
-        "color": "#ff3a30"
-    },
-    "deleted": {
-        "color": "#ff3a30"
-    },
-    "number": {
-        "color": "#ff16f7"
-    },
-    "selector": {
-        "color": "rgb(9,224,19)"
-    },
-    "attr-name": {
-        "color": "rgb(9,224,19)"
-    },
-    "string": {
-        "color": "rgb(9,224,19)"
-    },
-    "char": {
-        "color": "rgb(9,224,19)"
-    },
-    "builtin": {
-        "color": "rgb(9,224,19)"
-    },
-    "inserted": {
-        "color": "rgb(9,224,19)"
-    },
-    "variable": {
-        "color": "#C6C5FE"
-    },
-    "operator": {
-        "color": "#A1A1A1"
-    },
-    "entity": {
-        "color": "#fdab2b",
-        "cursor": "help"
-    },
-    "url": {
-        "color": "#96CBFE"
-    },
-    ".language-css .token.string": {
-        "color": "#87C38A"
-    },
-    ".style .token.string": {
-        "color": "#87C38A"
-    },
-    "atrule": {
-        "color": "#fdab2b"
-    },
-    "attr-value": {
-        "color": "#f8c575"
-    },
-    "function": {
-        "color": "#fdab2b"
-    },
-    "regex": {
-        "color": "#fab248"
-    },
-    "important": {
-        "color": "#fd971f",
-        "fontWeight": "bold"
-    },
-    "bold": {
-        "fontWeight": "bold"
-    },
-    "italic": {
-        "fontStyle": "italic"
-    }
-};

ui/src/components/UI/SyntaxHighlighter/index.scss

@@ -26,12 +26,24 @@
     }
 }
 
-.wrapped{
-    pre {
-        code {
-            &:last-child {
-                white-space: pre-wrap!important
-            }
-        }
-    }
-}
+code.hljs {
+    white-space: pre-wrap;
+}
+
+code.hljs:before {
+    counter-reset: listing;
+}
+
+code.hljs .hljs-marker-line {
+    counter-increment: listing;
+}
+
+code.hljs .hljs-marker-line:before {
+    content: counter(listing) "  ";
+    display: inline-block;
+    width: 3rem;
+    padding-left: auto;
+    margin-left: auto;
+    text-align: right;
+    opacity: .5;
+}

ui/src/components/UI/SyntaxHighlighter/index.tsx

@@ -1,30 +1,47 @@
 import React from 'react';
-import {Prism as SyntaxHighlighterContainer} from 'react-syntax-highlighter';
-import {highlighterStyle} from './highlighterStyle'
+import Lowlight from 'react-lowlight'
+import 'highlight.js/styles/atom-one-light.css'
 import './index.scss';
 
+import xml from 'highlight.js/lib/languages/xml'
+import json from 'highlight.js/lib/languages/json'
+import protobuf from 'highlight.js/lib/languages/protobuf'
+import javascript from 'highlight.js/lib/languages/javascript'
+import actionscript from 'highlight.js/lib/languages/actionscript'
+import wasm from 'highlight.js/lib/languages/wasm'
+import handlebars from 'highlight.js/lib/languages/handlebars'
+import yaml from 'highlight.js/lib/languages/yaml'
+import python from 'highlight.js/lib/languages/python'
+
+Lowlight.registerLanguage('python', python);
+Lowlight.registerLanguage('xml', xml);
+Lowlight.registerLanguage('json', json);
+Lowlight.registerLanguage('yaml', yaml);
+Lowlight.registerLanguage('protobuf', protobuf);
+Lowlight.registerLanguage('javascript', javascript);
+Lowlight.registerLanguage('actionscript', actionscript);
+Lowlight.registerLanguage('wasm', wasm);
+Lowlight.registerLanguage('handlebars', handlebars);
+
 interface Props {
     code: string;
-    style?: any;
     showLineNumbers?: boolean;
-    className?: string;
     language?: string;
-    isWrapped?: boolean;
 }
 
 export const SyntaxHighlighter: React.FC<Props> = ({
-                                                code,
-                                                style = highlighterStyle,
-                                                showLineNumbers = true,
-                                                className,
-                                                language = 'python',
-                                                isWrapped = false,
-                                            }) => {
-    return <div className={`highlighterContainer ${className ? className : ''} ${isWrapped ? 'wrapped' : ''}`}>
-        <SyntaxHighlighterContainer language={language} style={style} showLineNumbers={showLineNumbers}>
-            {code ?? ""}
-        </SyntaxHighlighterContainer>
-    </div>;
+        code,
+        showLineNumbers = false,
+        language = null
+    }) => {
+    const markers = showLineNumbers ? code.split("\n").map((item, i) => {
+        return {
+            line: i + 1,
+            className: 'hljs-marker-line'
+        }
+    }) : [];
+
+    return <div style={{fontSize: ".75rem"}}><Lowlight language={language ? language : ""} value={code} markers={markers}/></div>;
 };
 
 export default SyntaxHighlighter;