Show the source and destination IP in the entry feed (#485)

Currently shared/kubernetes/watch.go:FilteredWatch only watches pods.
This PR makes it reusable for other types of resources.
This is done in preparation for watching k8s events.

agent/pkg/up9/main.go

@@ -227,6 +227,10 @@ func syncEntriesImpl(token string, model string, envPrefix string, uploadInterva
 		connection.Close()
 	}()
 
+	lastTimeSynced := time.Time{}
+
+	batch := make([]har.Entry, 0)
+
 	handleDataChannel := func(wg *sync.WaitGroup, connection *basenine.Connection, data chan []byte) {
 		defer wg.Done()
 		for {
@@ -239,7 +243,6 @@ func syncEntriesImpl(token string, model string, envPrefix string, uploadInterva
 			var dataMap map[string]interface{}
 			err = json.Unmarshal(dataBytes, &dataMap)
 
-			result := make([]har.Entry, 0)
 			var entry tapApi.MizuEntry
 			if err := json.Unmarshal([]byte(dataBytes), &entry); err != nil {
 				continue
@@ -261,14 +264,22 @@ func syncEntriesImpl(token string, model string, envPrefix string, uploadInterva
 				continue
 			}
 
-			result = append(result, *harEntry)
+			batch = append(batch, *harEntry)
 
-			body, jMarshalErr := json.Marshal(result)
+			now := time.Now()
+			if lastTimeSynced.Add(time.Duration(uploadIntervalSec) * time.Second).After(now) {
+				continue
+			}
+			lastTimeSynced = now
+
+			body, jMarshalErr := json.Marshal(batch)
+			batchSize := len(batch)
 			if jMarshalErr != nil {
 				analyzeInformation.Reset()
 				logger.Log.Infof("Stopping sync entries")
 				logger.Log.Fatal(jMarshalErr)
 			}
+			batch = make([]har.Entry, 0)
 
 			var in bytes.Buffer
 			w := zlib.NewWriter(&in)
@@ -293,7 +304,7 @@ func syncEntriesImpl(token string, model string, envPrefix string, uploadInterva
 				logger.Log.Info("Stopping sync entries")
 				logger.Log.Fatal(postErr)
 			}
-			analyzeInformation.SentCount += 1
+			analyzeInformation.SentCount += batchSize
 
 			if analyzeInformation.SentCount%SentCountLogInterval == 0 {
 				logger.Log.Infof("Uploaded %v entries until now", analyzeInformation.SentCount)

cli/cmd/common.go

@@ -4,9 +4,15 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"github.com/up9inc/mizu/cli/apiserver"
+	"github.com/up9inc/mizu/cli/mizu"
+	"github.com/up9inc/mizu/cli/mizu/fsUtils"
+	"github.com/up9inc/mizu/cli/telemetry"
 	"os"
 	"os/signal"
+	"path"
 	"syscall"
+	"time"
 
 	"github.com/up9inc/mizu/cli/config"
 	"github.com/up9inc/mizu/cli/config/configStructs"
@@ -64,3 +70,42 @@ func handleKubernetesProviderError(err error) {
 		logger.Log.Error(err)
 	}
 }
+
+func finishMizuExecution(kubernetesProvider *kubernetes.Provider, apiProvider *apiserver.Provider) {
+	telemetry.ReportAPICalls(apiProvider)
+	removalCtx, cancel := context.WithTimeout(context.Background(), cleanupTimeout)
+	defer cancel()
+	dumpLogsIfNeeded(removalCtx, kubernetesProvider)
+	cleanUpMizuResources(removalCtx, cancel, kubernetesProvider)
+}
+
+func dumpLogsIfNeeded(ctx context.Context, kubernetesProvider *kubernetes.Provider) {
+	if !config.Config.DumpLogs {
+		return
+	}
+	mizuDir := mizu.GetMizuFolderPath()
+	filePath := path.Join(mizuDir, fmt.Sprintf("mizu_logs_%s.zip", time.Now().Format("2006_01_02__15_04_05")))
+	if err := fsUtils.DumpLogs(ctx, kubernetesProvider, filePath); err != nil {
+		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

@@ -5,27 +5,24 @@ import (
 	"errors"
 	"fmt"
 	"io/ioutil"
-	"path"
 	"regexp"
 	"strings"
 	"time"
 
-	"github.com/up9inc/mizu/cli/cmd/goUtils"
+	"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/getkin/kin-openapi/openapi3"
 	"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"
-	"gopkg.in/yaml.v3"
-	core "k8s.io/api/core/v1"
-
 	"github.com/up9inc/mizu/cli/mizu"
 	"github.com/up9inc/mizu/cli/mizu/fsUtils"
-	"github.com/up9inc/mizu/cli/telemetry"
 	"github.com/up9inc/mizu/cli/uiUtils"
 	"github.com/up9inc/mizu/shared"
 	"github.com/up9inc/mizu/shared/kubernetes"
@@ -119,6 +116,9 @@ func RunMizuTap() {
 	logger.Log.Infof("Tapping pods in %s", namespacesStr)
 
 	if config.Config.Tap.DryRun {
+		if err := printTappedPodsPreview(ctx, kubernetesProvider, targetNamespaces); err != nil {
+			logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Error listing pods: %v", errormessage.FormatError(err)))
+		}
 		return
 	}
 
@@ -134,7 +134,7 @@ func RunMizuTap() {
 		return
 	}
 	if config.Config.Tap.DaemonMode {
-		if err := handleDaemonModePostCreation(cancel, kubernetesProvider); err != nil {
+		if err := handleDaemonModePostCreation(ctx, cancel, kubernetesProvider, targetNamespaces); err != nil {
 			defer finishMizuExecution(kubernetesProvider, apiProvider)
 			cancel()
 		} else {
@@ -156,28 +156,38 @@ func RunMizuTap() {
 	}
 }
 
-func handleDaemonModePostCreation(cancel context.CancelFunc, kubernetesProvider *kubernetes.Provider) error {
+func handleDaemonModePostCreation(ctx context.Context, cancel context.CancelFunc, kubernetesProvider *kubernetes.Provider, namespaces []string) error {
+	if err := printTappedPodsPreview(ctx, kubernetesProvider, namespaces); err != nil {
+		return err
+	}
+
 	apiProvider := apiserver.NewProvider(GetApiServerUrl(), 90, 1*time.Second)
 
 	if err := waitForDaemonModeToBeReady(cancel, kubernetesProvider, apiProvider); err != nil {
 		return err
 	}
-	if err := printDaemonModeTappedPods(apiProvider); err != nil {
-		return err
-	}
 
 	return nil
 }
 
-func printDaemonModeTappedPods(apiProvider *apiserver.Provider) error {
-	if healthStatus, err := apiProvider.GetHealthStatus(); err != nil {
+/*
+this function is a bit problematic as it might be detached from the actual pods the mizu api server will tap.
+The alternative would be to wait for api server to be ready and then query it for the pods it listens to, this has
+the arguably worse drawback of taking a relatively very long time before the user sees which pods are targeted, if any.
+*/
+func printTappedPodsPreview(ctx context.Context, kubernetesProvider *kubernetes.Provider, namespaces []string) error {
+	if matchingPods, err := kubernetesProvider.ListAllRunningPodsMatchingRegex(ctx, config.Config.Tap.PodRegex(), namespaces); err != nil {
 		return err
 	} else {
-		for _, tappedPod := range healthStatus.TapStatus.Pods {
+		if len(matchingPods) == 0 {
+			printNoPodsFoundSuggestion(namespaces)
+		}
+		logger.Log.Info("Pods that match the provided criteria at this instant:")
+		for _, tappedPod := range matchingPods {
 			logger.Log.Infof(uiUtils.Green, fmt.Sprintf("+%s", tappedPod.Name))
 		}
+		return nil
 	}
-	return nil
 }
 
 func waitForDaemonModeToBeReady(cancel context.CancelFunc, kubernetesProvider *kubernetes.Provider, apiProvider *apiserver.Provider) error {
@@ -215,11 +225,7 @@ func startTapperSyncer(ctx context.Context, cancel context.CancelFunc, provider
 	}
 
 	if len(tapperSyncer.CurrentlyTappedPods) == 0 {
-		var suggestionStr string
-		if !shared.Contains(targetNamespaces, kubernetes.K8sAllNamespaces) {
-			suggestionStr = ". Select a different namespace with -n or tap all namespaces with -A"
-		}
-		logger.Log.Warningf(uiUtils.Warning, fmt.Sprintf("Did not find any pods matching the regex argument%s", suggestionStr))
+		printNoPodsFoundSuggestion(targetNamespaces)
 	}
 
 	go func() {
@@ -252,6 +258,14 @@ func startTapperSyncer(ctx context.Context, cancel context.CancelFunc, provider
 	return nil
 }
 
+func printNoPodsFoundSuggestion(targetNamespaces []string) {
+	var suggestionStr string
+	if !shared.Contains(targetNamespaces, kubernetes.K8sAllNamespaces) {
+		suggestionStr = ". You can also try selecting a different namespace with -n or tap all namespaces with -A"
+	}
+	logger.Log.Warningf(uiUtils.Warning, fmt.Sprintf("Did not find any currently running pods that match the regex argument, mizu will automatically tap matching pods if any are created later%s", suggestionStr))
+}
+
 func getErrorDisplayTextForK8sTapManagerError(err kubernetes.K8sTapManagerError) string {
 	switch err.TapManagerReason {
 	case kubernetes.TapManagerPodListError:
@@ -359,20 +373,9 @@ func createMizuApiServerPod(ctx context.Context, kubernetesProvider *kubernetes.
 }
 
 func createMizuApiServerDeployment(ctx context.Context, kubernetesProvider *kubernetes.Provider, opts *kubernetes.ApiServerOptions) error {
-	isDefaultStorageClassAvailable, err := kubernetesProvider.IsDefaultStorageProviderAvailable(ctx)
 	volumeClaimCreated := false
-	if err != nil {
-		return err
-	}
-	if isDefaultStorageClassAvailable {
-		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 will mean that mizu's data will be lost on pod restart")
-			logger.Log.Debugf("error creating persistent volume claim: %v", err)
-		} else {
-			volumeClaimCreated = true
-		}
-	} else {
-		logger.Log.Warningf(uiUtils.Yellow, "Could not find default volume provider in this cluster, this will mean that mizu's data will be lost on pod restart")
+	if !config.Config.Tap.NoPersistentVolumeClaim {
+		volumeClaimCreated = TryToCreatePersistentVolumeClaim(ctx, kubernetesProvider)
 	}
 
 	pod, err := kubernetesProvider.GetMizuApiServerPodObject(opts, volumeClaimCreated, kubernetes.PersistentVolumeClaimName)
@@ -387,6 +390,26 @@ func createMizuApiServerDeployment(ctx context.Context, kubernetesProvider *kube
 	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
 
@@ -421,45 +444,6 @@ func getSyncEntriesConfig() *shared.SyncEntriesConfig {
 	}
 }
 
-func finishMizuExecution(kubernetesProvider *kubernetes.Provider, apiProvider *apiserver.Provider) {
-	telemetry.ReportAPICalls(apiProvider)
-	removalCtx, cancel := context.WithTimeout(context.Background(), cleanupTimeout)
-	defer cancel()
-	dumpLogsIfNeeded(removalCtx, kubernetesProvider)
-	cleanUpMizuResources(removalCtx, cancel, kubernetesProvider)
-}
-
-func dumpLogsIfNeeded(ctx context.Context, kubernetesProvider *kubernetes.Provider) {
-	if !config.Config.DumpLogs {
-		return
-	}
-	mizuDir := mizu.GetMizuFolderPath()
-	filePath := path.Join(mizuDir, fmt.Sprintf("mizu_logs_%s.zip", time.Now().Format("2006_01_02__15_04_05")))
-	if err := fsUtils.DumpLogs(ctx, kubernetesProvider, filePath); err != nil {
-		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)
-	}
-}
-
 func cleanUpRestrictedMode(ctx context.Context, kubernetesProvider *kubernetes.Provider) []string {
 	leftoverResources := make([]string, 0)
 
@@ -570,7 +554,8 @@ func waitUntilNamespaceDeleted(ctx context.Context, cancel context.CancelFunc, k
 
 func watchApiServerPod(ctx context.Context, kubernetesProvider *kubernetes.Provider, cancel context.CancelFunc) {
 	podExactRegex := regexp.MustCompile(fmt.Sprintf("^%s$", kubernetes.ApiServerPodName))
-	added, modified, removed, errorChan := kubernetes.FilteredWatch(ctx, kubernetesProvider, []string{config.Config.MizuResourcesNamespace}, podExactRegex)
+	podWatchHelper := kubernetes.NewPodWatchHelper(kubernetesProvider, podExactRegex)
+	added, modified, removed, errorChan := kubernetes.FilteredWatch(ctx, podWatchHelper, []string{config.Config.MizuResourcesNamespace}, podWatchHelper)
 	isPodReady := false
 	timeAfter := time.After(25 * time.Second)
 	for {
@@ -591,12 +576,19 @@ func watchApiServerPod(ctx context.Context, kubernetesProvider *kubernetes.Provi
 			logger.Log.Infof("%s removed", kubernetes.ApiServerPodName)
 			cancel()
 			return
-		case modifiedPod, ok := <-modified:
+		case wEvent, ok := <-modified:
 			if !ok {
 				modified = nil
 				continue
 			}
 
+			modifiedPod, err := wEvent.ToPod()
+			if err != nil {
+				logger.Log.Errorf(uiUtils.Error, err)
+				cancel()
+				continue
+			}
+
 			logger.Log.Debugf("Watching API Server pod loop, modified: %v", modifiedPod.Status.Phase)
 
 			if modifiedPod.Status.Phase == core.PodPending {
@@ -657,34 +649,57 @@ func watchApiServerPod(ctx context.Context, kubernetesProvider *kubernetes.Provi
 
 func watchTapperPod(ctx context.Context, kubernetesProvider *kubernetes.Provider, cancel context.CancelFunc) {
 	podExactRegex := regexp.MustCompile(fmt.Sprintf("^%s.*", kubernetes.TapperDaemonSetName))
-	added, modified, removed, errorChan := kubernetes.FilteredWatch(ctx, kubernetesProvider, []string{config.Config.MizuResourcesNamespace}, podExactRegex)
+	podWatchHelper := kubernetes.NewPodWatchHelper(kubernetesProvider, podExactRegex)
+	added, modified, removed, errorChan := kubernetes.FilteredWatch(ctx, podWatchHelper, []string{config.Config.MizuResourcesNamespace}, podWatchHelper)
 	var prevPodPhase core.PodPhase
 	for {
 		select {
-		case addedPod, ok := <-added:
+		case wEvent, ok := <-added:
 			if !ok {
 				added = nil
 				continue
 			}
 
+			addedPod, err := wEvent.ToPod()
+			if err != nil {
+				logger.Log.Errorf(uiUtils.Error, err)
+				cancel()
+				continue
+			}
+
 			logger.Log.Debugf("Tapper is created [%s]", addedPod.Name)
-		case removedPod, ok := <-removed:
+		case wEvent, ok := <-removed:
 			if !ok {
 				removed = nil
 				continue
 			}
 
+			removedPod, err := wEvent.ToPod()
+			if err != nil {
+				logger.Log.Errorf(uiUtils.Error, err)
+				cancel()
+				continue
+			}
+
+
 			logger.Log.Debugf("Tapper is removed [%s]", removedPod.Name)
-		case modifiedPod, ok := <-modified:
+		case wEvent, ok := <-modified:
 			if !ok {
 				modified = nil
 				continue
 			}
 
+			modifiedPod, err := wEvent.ToPod()
+			if err != nil {
+				logger.Log.Errorf(uiUtils.Error, err)
+				cancel()
+				continue
+			}
+
 			if modifiedPod.Status.Phase == core.PodPending && modifiedPod.Status.Conditions[0].Type == core.PodScheduled && modifiedPod.Status.Conditions[0].Status != core.ConditionTrue {
 				logger.Log.Infof(uiUtils.Red, fmt.Sprintf("Wasn't able to deploy the tapper %s. Reason: \"%s\"", modifiedPod.Name, modifiedPod.Status.Conditions[0].Message))
 				cancel()
-				break
+				continue
 			}
 
 			podStatus := modifiedPod.Status

cli/config/configStructs/tapConfig.go

@@ -3,6 +3,8 @@ package configStructs
 import (
 	"errors"
 	"fmt"
+	"github.com/up9inc/mizu/cli/uiUtils"
+	"github.com/up9inc/mizu/shared/logger"
 	"regexp"
 
 	"github.com/up9inc/mizu/shared"
@@ -26,25 +28,26 @@ const (
 )
 
 type TapConfig struct {
-	UploadIntervalSec      int              `yaml:"upload-interval" default:"10"`
-	PodRegexStr            string           `yaml:"regex" default:".*"`
-	GuiPort                uint16           `yaml:"gui-port" default:"8899"`
-	ProxyHost              string           `yaml:"proxy-host" default:"127.0.0.1"`
-	Namespaces             []string         `yaml:"namespaces"`
-	Analysis               bool             `yaml:"analysis" default:"false"`
-	AllNamespaces          bool             `yaml:"all-namespaces" default:"false"`
-	PlainTextFilterRegexes []string         `yaml:"regex-masking"`
-	IgnoredUserAgents      []string         `yaml:"ignored-user-agents"`
-	DisableRedaction       bool             `yaml:"no-redact" default:"false"`
-	HumanMaxEntriesDBSize  string           `yaml:"max-entries-db-size" default:"200MB"`
-	DryRun                 bool             `yaml:"dry-run" default:"false"`
-	Workspace              string           `yaml:"workspace"`
-	EnforcePolicyFile      string           `yaml:"traffic-validation-file"`
-	ContractFile           string           `yaml:"contract"`
-	AskUploadConfirmation  bool             `yaml:"ask-upload-confirmation" default:"true"`
-	ApiServerResources     shared.Resources `yaml:"api-server-resources"`
-	TapperResources        shared.Resources `yaml:"tapper-resources"`
-	DaemonMode             bool             `yaml:"daemon" default:"false"`
+	UploadIntervalSec       int              `yaml:"upload-interval" default:"10"`
+	PodRegexStr             string           `yaml:"regex" default:".*"`
+	GuiPort                 uint16           `yaml:"gui-port" default:"8899"`
+	ProxyHost               string           `yaml:"proxy-host" default:"127.0.0.1"`
+	Namespaces              []string         `yaml:"namespaces"`
+	Analysis                bool             `yaml:"analysis" default:"false"`
+	AllNamespaces           bool             `yaml:"all-namespaces" default:"false"`
+	PlainTextFilterRegexes  []string         `yaml:"regex-masking"`
+	IgnoredUserAgents       []string         `yaml:"ignored-user-agents"`
+	DisableRedaction        bool             `yaml:"no-redact" default:"false"`
+	HumanMaxEntriesDBSize   string           `yaml:"max-entries-db-size" default:"200MB"`
+	DryRun                  bool             `yaml:"dry-run" default:"false"`
+	Workspace               string           `yaml:"workspace"`
+	EnforcePolicyFile       string           `yaml:"traffic-validation-file"`
+	ContractFile            string           `yaml:"contract"`
+	AskUploadConfirmation   bool             `yaml:"ask-upload-confirmation" default:"true"`
+	ApiServerResources      shared.Resources `yaml:"api-server-resources"`
+	TapperResources         shared.Resources `yaml:"tapper-resources"`
+	DaemonMode              bool             `yaml:"daemon" default:"false"`
+	NoPersistentVolumeClaim bool             `yaml:"no-persistent-volume-claim" default:"false"`
 }
 
 func (config *TapConfig) PodRegex() *regexp.Regexp {
@@ -79,5 +82,9 @@ func (config *TapConfig) Validate() error {
 		return errors.New(fmt.Sprintf("Can't run with both --%s and --%s flags", AnalysisTapName, WorkspaceTapName))
 	}
 
+	if config.NoPersistentVolumeClaim && !config.DaemonMode {
+		logger.Log.Warningf(uiUtils.Warning, fmt.Sprintf("the --set tap.no-persistent-volume-claim=true flag has no effect without the --%s flag, the claim will not be created anyway.", DaemonModeTapName))
+	}
+
 	return nil
 }

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

@@ -7,15 +7,15 @@ rules:
   - apiGroups: [""]
     resources: ["pods"]
     verbs: ["get", "list", "watch", "delete"]
-  - apiGroups: [ "" ]
+  - apiGroups: [ "apps" ]
     resources: [ "deployments" ]
-    verbs: [ "create", "delete" ]
+    verbs: [ "get", "create", "delete" ]
   - apiGroups: [""]
     resources: ["services"]
     verbs: ["get", "list", "watch", "create", "delete"]
   - apiGroups: ["apps"]
     resources: ["daemonsets"]
-    verbs: ["create", "patch", "delete"]
+    verbs: ["get", "create", "patch", "delete", "list"]
   - apiGroups: [""]
     resources: ["namespaces"]
     verbs: ["get", "list", "watch", "create", "delete"]

examples/roles/permissions-ns-daemon.yaml

@@ -8,7 +8,7 @@ rules:
 - apiGroups: [""]
   resources: ["pods"]
   verbs: ["get", "list", "watch", "delete"]
-- apiGroups: [ "" ]
+- apiGroups: [ "apps" ]
   resources: [ "deployments" ]
   verbs: [ "get", "create", "delete" ]
 - apiGroups: [""]
@@ -16,7 +16,7 @@ rules:
   verbs: ["get", "list", "watch", "create", "delete"]
 - apiGroups: ["apps"]
   resources: ["daemonsets"]
-  verbs: ["get", "create", "patch", "delete"]
+  verbs: ["get", "create", "patch", "delete", "list"]
 - apiGroups: [""]
   resources: ["services/proxy"]
   verbs: ["get"]
@@ -32,7 +32,7 @@ rules:
 - apiGroups: ["rbac.authorization.k8s.io"]
   resources: ["rolebindings"]
   verbs: ["get", "create", "delete"]
-- apiGroups: ["apps", "extensions"]
+- apiGroups: ["apps", "extensions", ""]
   resources: ["pods"]
   verbs: ["get", "list", "watch"]
 - apiGroups: ["apps", "extensions"]

shared/kubernetes/mizuTapperSyncer.go

@@ -68,7 +68,8 @@ func CreateAndStartMizuTapperSyncer(ctx context.Context, kubernetesProvider *Pro
 }
 
 func (tapperSyncer *MizuTapperSyncer) watchPodsForTapping() {
-	added, modified, removed, errorChan := FilteredWatch(tapperSyncer.context, tapperSyncer.kubernetesProvider, tapperSyncer.config.TargetNamespaces, &tapperSyncer.config.PodFilterRegex)
+	podWatchHelper := NewPodWatchHelper(tapperSyncer.kubernetesProvider, &tapperSyncer.config.PodFilterRegex)
+	added, modified, removed, errorChan := FilteredWatch(tapperSyncer.context, podWatchHelper, tapperSyncer.config.TargetNamespaces, podWatchHelper)
 
 	restartTappers := func() {
 		err, changeFound := tapperSyncer.updateCurrentlyTappedPods()
@@ -94,28 +95,48 @@ func (tapperSyncer *MizuTapperSyncer) watchPodsForTapping() {
 
 	for {
 		select {
-		case pod, ok := <-added:
+		case wEvent, ok := <-added:
 			if !ok {
 				added = nil
 				continue
 			}
 
+			pod, err := wEvent.ToPod()
+			if err != nil {
+				tapperSyncer.handleErrorInWatchLoop(err, restartTappersDebouncer)
+				continue
+			}
+
+
 			logger.Log.Debugf("Added matching pod %s, ns: %s", pod.Name, pod.Namespace)
 			restartTappersDebouncer.SetOn()
-		case pod, ok := <-removed:
+		case wEvent, ok := <-removed:
 			if !ok {
 				removed = nil
 				continue
 			}
 
+			pod, err := wEvent.ToPod()
+			if err != nil {
+				tapperSyncer.handleErrorInWatchLoop(err, restartTappersDebouncer)
+				continue
+			}
+
 			logger.Log.Debugf("Removed matching pod %s, ns: %s", pod.Name, pod.Namespace)
 			restartTappersDebouncer.SetOn()
-		case pod, ok := <-modified:
+		case wEvent, ok := <-modified:
 			if !ok {
 				modified = nil
 				continue
 			}
 
+			pod, err := wEvent.ToPod()
+			if err != nil {
+				tapperSyncer.handleErrorInWatchLoop(err, restartTappersDebouncer)
+				continue
+			}
+
+
 			logger.Log.Debugf("Modified matching pod %s, ns: %s, phase: %s, ip: %s", pod.Name, pod.Namespace, pod.Status.Phase, pod.Status.PodIP)
 			// Act only if the modified pod has already obtained an IP address.
 			// After filtering for IPs, on a normal pod restart this includes the following events:
@@ -132,12 +153,8 @@ func (tapperSyncer *MizuTapperSyncer) watchPodsForTapping() {
 				continue
 			}
 
-			logger.Log.Debugf("Watching pods loop, got error %v, stopping `restart tappers debouncer`", err)
-			restartTappersDebouncer.Cancel()
-			tapperSyncer.ErrorOut <- K8sTapManagerError{
-				OriginalError:    err,
-				TapManagerReason: TapManagerPodWatchError,
-			}
+			tapperSyncer.handleErrorInWatchLoop(err, restartTappersDebouncer)
+			continue
 
 		case <-tapperSyncer.context.Done():
 			logger.Log.Debugf("Watching pods loop, context done, stopping `restart tappers debouncer`")
@@ -148,6 +165,15 @@ func (tapperSyncer *MizuTapperSyncer) watchPodsForTapping() {
 	}
 }
 
+func (tapperSyncer *MizuTapperSyncer) handleErrorInWatchLoop(err error, restartTappersDebouncer *debounce.Debouncer) {
+	logger.Log.Debugf("Watching pods loop, got error %v, stopping `restart tappers debouncer`", err)
+	restartTappersDebouncer.Cancel()
+	tapperSyncer.ErrorOut <- K8sTapManagerError{
+		OriginalError:    err,
+		TapManagerReason: TapManagerPodWatchError,
+	}
+}
+
 func (tapperSyncer *MizuTapperSyncer) updateCurrentlyTappedPods() (err error, changesFound bool) {
 	if matchingPods, err := tapperSyncer.kubernetesProvider.ListAllRunningPodsMatchingRegex(tapperSyncer.context, &tapperSyncer.config.PodFilterRegex, tapperSyncer.config.TargetNamespaces); err != nil {
 		return err, false

shared/kubernetes/podWatchHelper.go

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

shared/kubernetes/provider.go

@@ -153,14 +153,6 @@ func (provider *Provider) WaitUtilNamespaceDeleted(ctx context.Context, name str
 	return err
 }
 
-func (provider *Provider) GetPodWatcher(ctx context.Context, namespace string) watch.Interface {
-	watcher, err := provider.clientSet.CoreV1().Pods(namespace).Watch(ctx, metav1.ListOptions{Watch: true})
-	if err != nil {
-		panic(err.Error())
-	}
-	return watcher
-}
-
 func (provider *Provider) CreateNamespace(ctx context.Context, name string) (*core.Namespace, error) {
 	namespaceSpec := &core.Namespace{
 		ObjectMeta: metav1.ObjectMeta{
@@ -579,6 +571,11 @@ func (provider *Provider) RemoveDaemonSet(ctx context.Context, namespace string,
 	return provider.handleRemovalError(err)
 }
 
+func (provider *Provider) RemovePersistentVolumeClaim(ctx context.Context, namespace string, volumeClaimName string) error {
+	err := provider.clientSet.CoreV1().PersistentVolumeClaims(namespace).Delete(ctx, volumeClaimName, metav1.DeleteOptions{})
+	return provider.handleRemovalError(err)
+}
+
 func (provider *Provider) handleRemovalError(err error) error {
 	// Ignore NotFound - There is nothing to delete.
 	// Ignore Forbidden - Assume that a user could not have created the resource in the first place.
@@ -859,10 +856,6 @@ func (provider *Provider) CreatePersistentVolumeClaim(ctx context.Context, names
 	return provider.clientSet.CoreV1().PersistentVolumeClaims(namespace).Create(ctx, volumeClaim, metav1.CreateOptions{})
 }
 
-func (provider *Provider) RemovePersistentVolumeClaim(ctx context.Context, namespace string, volumeClaimName string) error {
-	return provider.clientSet.CoreV1().PersistentVolumeClaims(namespace).Delete(ctx, volumeClaimName, metav1.DeleteOptions{})
-}
-
 func getClientSet(config *restclient.Config) (*kubernetes.Clientset, error) {
 	clientSet, err := kubernetes.NewForConfig(config)
 	if err != nil {

shared/kubernetes/watch.go

@@ -6,19 +6,25 @@ import (
 	"fmt"
 	"github.com/up9inc/mizu/shared/debounce"
 	"github.com/up9inc/mizu/shared/logger"
-	"regexp"
 	"sync"
 	"time"
 
-	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/watch"
 )
 
-func FilteredWatch(ctx context.Context, kubernetesProvider *Provider, targetNamespaces []string, podFilter *regexp.Regexp) (chan *corev1.Pod, chan *corev1.Pod, chan *corev1.Pod, chan error) {
-	addedChan := make(chan *corev1.Pod)
-	modifiedChan := make(chan *corev1.Pod)
-	removedChan := make(chan *corev1.Pod)
+type EventFilterer interface {
+	Filter(*WatchEvent) (bool, error)
+}
+
+type WatchCreator interface {
+	NewWatcher(ctx context.Context, namespace string) (watch.Interface, error)
+}
+
+func FilteredWatch(ctx context.Context, watcherCreator WatchCreator, targetNamespaces []string, filterer EventFilterer) (chan *WatchEvent, chan *WatchEvent, chan *WatchEvent, chan error) {
+	addedChan := make(chan *WatchEvent)
+	modifiedChan := make(chan *WatchEvent)
+	removedChan := make(chan *WatchEvent)
 	errorChan := make(chan error)
 
 	var wg sync.WaitGroup
@@ -31,8 +37,13 @@ func FilteredWatch(ctx context.Context, kubernetesProvider *Provider, targetName
 			watchRestartDebouncer := debounce.NewDebouncer(1 * time.Minute, func() {})
 
 			for {
-				watcher := kubernetesProvider.GetPodWatcher(ctx, targetNamespace)
-				err := startWatchLoop(ctx, watcher, podFilter, addedChan, modifiedChan, removedChan) // blocking
+				watcher, err := watcherCreator.NewWatcher(ctx, targetNamespace)
+				if err != nil {
+					errorChan <- fmt.Errorf("error in k8 watch: %v", err)
+					break
+				}
+
+				err = startWatchLoop(ctx, watcher, filterer, addedChan, modifiedChan, removedChan) // blocking
 				watcher.Stop()
 
 				select {
@@ -72,7 +83,7 @@ func FilteredWatch(ctx context.Context, kubernetesProvider *Provider, targetName
 	return addedChan, modifiedChan, removedChan, errorChan
 }
 
-func startWatchLoop(ctx context.Context, watcher watch.Interface, podFilter *regexp.Regexp, addedChan chan *corev1.Pod, modifiedChan chan *corev1.Pod, removedChan chan *corev1.Pod) error {
+func startWatchLoop(ctx context.Context, watcher watch.Interface, filterer EventFilterer, addedChan chan *WatchEvent, modifiedChan chan *WatchEvent, removedChan chan *WatchEvent) error {
 	resultChan := watcher.ResultChan()
 	for {
 		select {
@@ -81,26 +92,25 @@ func startWatchLoop(ctx context.Context, watcher watch.Interface, podFilter *reg
 				return nil
 			}
 
-			if e.Type == watch.Error {
-				return apierrors.FromObject(e.Object)
+			wEvent := WatchEvent(e)
+
+			if wEvent.Type == watch.Error {
+				return apierrors.FromObject(wEvent.Object)
 			}
 
-			pod, ok := e.Object.(*corev1.Pod)
-			if !ok {
+			if pass, err := filterer.Filter(&wEvent); err != nil {
+				return err
+			} else if !pass {
 				continue
 			}
 
-			if !podFilter.MatchString(pod.Name) {
-				continue
-			}
-
-			switch e.Type {
+			switch wEvent.Type {
 			case watch.Added:
-				addedChan <- pod
+				addedChan <- &wEvent
 			case watch.Modified:
-				modifiedChan <- pod
+				modifiedChan <- &wEvent
 			case watch.Deleted:
-				removedChan <- pod
+				removedChan <- &wEvent
 			}
 		case <-ctx.Done():
 			return nil

shared/kubernetes/watchEvent.go

@@ -0,0 +1,18 @@
+package kubernetes
+
+import (
+	"fmt"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/watch"
+)
+
+type WatchEvent watch.Event
+
+func (we *WatchEvent) ToPod() (*corev1.Pod, error) {
+	pod, ok := we.Object.(*corev1.Pod)
+	if !ok {
+		return nil, fmt.Errorf("Invalid object type on pod event stream")
+	}
+
+	return pod, nil
+}

ui/src/components/EntryDetailed/EntrySections.tsx

@@ -15,7 +15,7 @@ interface EntryViewLineProps {
 }
 
 const EntryViewLine: React.FC<EntryViewLineProps> = ({label, value, updateQuery, selector, overrideQueryValue}) => {
-    return (label && value && <tr className={styles.dataLine}>
+    return (label && <tr className={styles.dataLine}>
                 <td
                     className={`queryable ${styles.dataKey}`}
                     onClick={() => {

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

@@ -84,7 +84,14 @@
   padding: 4px
   padding-left: 12px
 
-.port
+.tcpInfo
   font-size: 12px
   color: $secondary-font-color
-  margin: 5px
+  margin-top: 5px
+  margin-bottom: 5px
+
+.port
+  margin-right: 5px
+
+.ip
+  margin-left: 5px

ui/src/components/EntryListItem/EntryListItem.tsx

@@ -171,7 +171,17 @@ export const EntryItem: React.FC<EntryProps> = ({entry, setFocusedEntryId, style
             }
             <div className={styles.separatorRight}>
                 <span
-                    className={`queryable ${styles.port}`}
+                    className={`queryable ${styles.tcpInfo} ${styles.ip}`}
+                    title="Source IP"
+                    onClick={() => {
+                        updateQuery(`src.ip == "${entry.sourceIp}"`)
+                    }}
+                >
+                    {entry.sourceIp}
+                </span>
+                <span className={`${styles.tcpInfo}`}>:</span>
+                <span
+                    className={`queryable ${styles.tcpInfo} ${styles.port}`}
                     title="Source Port"
                     onClick={() => {
                         updateQuery(`src.port == "${entry.sourcePort}"`)
@@ -199,7 +209,17 @@ export const EntryItem: React.FC<EntryProps> = ({entry, setFocusedEntryId, style
                     />
                 }
                 <span
-                    className={`queryable ${styles.port}`}
+                    className={`queryable ${styles.tcpInfo} ${styles.ip}`}
+                    title="Destination IP"
+                    onClick={() => {
+                        updateQuery(`dst.ip == "${entry.destinationIp}"`)
+                    }}
+                >
+                    {entry.destinationIp}
+                </span>
+                <span className={`${styles.tcpInfo}`}>:</span>
+                <span
+                    className={`queryable ${styles.tcpInfo} ${styles.port}`}
                     title="Destination Port"
                     onClick={() => {
                         updateQuery(`dst.port == "${entry.destinationPort}"`)

ui/src/components/TrafficPage.tsx

@@ -101,6 +101,7 @@ export const TrafficPage: React.FC<TrafficPageProps> = ({setAnalyzeStatus, onTLS
     const listEntry = useRef(null);
 
     const openWebSocket = (query) => {
+        setFocusedEntryId(null);
         setEntries([]);
         setEntriesBuffer([]);
         ws.current = new WebSocket(MizuWebsocketURL);

ui/src/components/UI/FancyTextDisplay.tsx

@@ -17,7 +17,7 @@ interface Props {
 const FancyTextDisplay: React.FC<Props> = ({text, className, isPossibleToCopy = true, applyTextEllipsis = true, flipped = false, useTooltip= false, displayIconOnMouseOver = false, buttonOnly = false}) => {
     const [showCopiedNotification, setCopied] = useState(false);
     const [showTooltip, setShowTooltip] = useState(false);
-    const displayText = text || '';
+    text = String(text);
 
     const onCopy = () => {
         setCopied(true)
@@ -33,12 +33,12 @@ const FancyTextDisplay: React.FC<Props> = ({text, className, isPossibleToCopy =
         return () => clearTimeout(timer);
     }, [showCopiedNotification]);
 
-    const textElement = <span className={'FancyTextDisplay-Text'}>{displayText}</span>;
+    const textElement = <span className={'FancyTextDisplay-Text'}>{text}</span>;
 
-    const copyButton = isPossibleToCopy && displayText ? <CopyToClipboard text={displayText} onCopy={onCopy}>
+    const copyButton = isPossibleToCopy && text ? <CopyToClipboard text={text} onCopy={onCopy}>
                     <span
                         className={`FancyTextDisplay-Icon`}
-                        title={`Copy "${displayText}" value to clipboard`}
+                        title={`Copy "${text}" value to clipboard`}
                     >
                         <img src={duplicateImg} alt="Duplicate full value"/>
                         {showCopiedNotification && <span className={'FancyTextDisplay-CopyNotifier'}>Copied</span>}
@@ -48,14 +48,14 @@ const FancyTextDisplay: React.FC<Props> = ({text, className, isPossibleToCopy =
     return (
         <p
             className={`FancyTextDisplay-Container ${className ? className : ''} ${displayIconOnMouseOver ? 'displayIconOnMouseOver ' : ''} ${applyTextEllipsis ? ' FancyTextDisplay-ContainerEllipsis' : ''}`}
-            title={displayText.toString()}
+            title={text}
             onMouseOver={ e => setShowTooltip(true)}
             onMouseLeave={ e => setShowTooltip(false)}
         >
             {!buttonOnly && flipped && textElement}
             {copyButton}
             {!buttonOnly && !flipped && textElement}
-            {useTooltip && showTooltip && <span className={'FancyTextDisplay-CopyNotifier'}>{displayText}</span>}
+            {useTooltip && showTooltip && <span className={'FancyTextDisplay-CopyNotifier'}>{text}</span>}
         </p>
     );
 };