Merge pull request #89 from up9inc/tap_grooming

Mizu tap analyze grooming

api/go.mod

@@ -11,16 +11,12 @@ require (
 	github.com/go-playground/universal-translator v0.17.0
 	github.com/go-playground/validator/v10 v10.5.0
 	github.com/gofiber/fiber/v2 v2.8.0
-	github.com/google/gopacket v1.1.19
 	github.com/google/martian v2.1.0+incompatible
 	github.com/gorilla/websocket v1.4.2
 	github.com/leodido/go-urn v1.2.1 // indirect
-	github.com/orcaman/concurrent-map v0.0.0-20210106121528-16402b402231
-	github.com/patrickmn/go-cache v2.1.0+incompatible
 	github.com/up9inc/mizu/shared v0.0.0
 	github.com/up9inc/mizu/tap v0.0.0
 	go.mongodb.org/mongo-driver v1.5.1
-	golang.org/x/net v0.0.0-20210421230115-4e50805a0758
 	gorm.io/driver/sqlite v1.1.4
 	gorm.io/gorm v1.21.8
 	k8s.io/api v0.21.0
@@ -29,4 +25,5 @@ require (
 )
 
 replace github.com/up9inc/mizu/shared v0.0.0 => ../shared
+
 replace github.com/up9inc/mizu/tap v0.0.0 => ../tap

api/go.sum

@@ -251,7 +251,6 @@ github.com/onsi/gomega v0.0.0-20170829124025-dcabb60a477c/go.mod h1:C1qb7wdrVGGV
 github.com/onsi/gomega v1.7.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
 github.com/orcaman/concurrent-map v0.0.0-20210106121528-16402b402231 h1:fa50YL1pzKW+1SsBnJDOHppJN9stOEwS+CRWyUtyYGU=
 github.com/orcaman/concurrent-map v0.0.0-20210106121528-16402b402231/go.mod h1:Lu3tH6HLW3feq74c2GC+jIMS/K2CFcDWnWD9XkenwhI=
-github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
 github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
 github.com/pelletier/go-toml v1.7.0/go.mod h1:vwGMzjaWMwyfHwgIBhI2YUM4fB6nL6lVAvS1LBMMhTE=
 github.com/peterbourgon/diskv v2.0.1+incompatible/go.mod h1:uqqh8zWWbv1HBMNONnaR/tNboyR3/BZd58JJSHlUSCU=

api/main.go

@@ -35,12 +35,10 @@ func main() {
 
 	if *standalone {
 		harOutputChannel, outboundLinkOutputChannel := tap.StartPassiveTapper(tapOpts)
-		filteredHarChannel0 := make(chan *tap.OutputChannelItem)
-		filteredHarChannel1 := make(chan *tap.OutputChannelItem)
+		filteredHarChannel := make(chan *tap.OutputChannelItem)
 
-		go filterServices(harOutputChannel, filteredHarChannel0, getTrafficFilteringOptions())
-		go filterHarHeaders(filteredHarChannel0, filteredHarChannel1, getTrafficFilteringOptions())
-		go api.StartReadingEntries(filteredHarChannel1, nil)
+		go filterHarItems(harOutputChannel, filteredHarChannel, getTrafficFilteringOptions())
+		go api.StartReadingEntries(filteredHarChannel, nil)
 		go api.StartReadingOutbound(outboundLinkOutputChannel)
 
 		hostApi(nil)
@@ -66,12 +64,10 @@ func main() {
 		go api.StartReadingOutbound(outboundLinkOutputChannel)
 	} else if *aggregator {
 		socketHarOutChannel := make(chan *tap.OutputChannelItem, 1000)
-		filteredHarChannel0 := make(chan *tap.OutputChannelItem)
-		filteredHarChannel1 := make(chan *tap.OutputChannelItem)
+		filteredHarChannel := make(chan *tap.OutputChannelItem)
 
-		go filterServices(socketHarOutChannel, filteredHarChannel0, getTrafficFilteringOptions())
-		go filterHarHeaders(filteredHarChannel0, filteredHarChannel1, getTrafficFilteringOptions())
-		go api.StartReadingEntries(filteredHarChannel1, nil)
+		go filterHarItems(socketHarOutChannel, filteredHarChannel, getTrafficFilteringOptions())
+		go api.StartReadingEntries(filteredHarChannel, nil)
 
 		hostApi(socketHarOutChannel)
 	}
@@ -129,19 +125,14 @@ func getTrafficFilteringOptions() *shared.TrafficFilteringOptions {
 	return &filteringOptions
 }
 
-func filterServices(inChannel <- chan *tap.OutputChannelItem, outChannel chan *tap.OutputChannelItem, filterOptions *shared.TrafficFilteringOptions) {
+func filterHarItems(inChannel <- chan *tap.OutputChannelItem, outChannel chan *tap.OutputChannelItem, filterOptions *shared.TrafficFilteringOptions) {
 	for message := range inChannel {
-		if api.CheckIsServiceIP(message.ConnectionInfo.ServerIP) {
+		if message.ConnectionInfo.IsOutgoing && api.CheckIsServiceIP(message.ConnectionInfo.ServerIP) {
 			continue
 		}
 
-		outChannel <- message
-	}
-}
-
-func filterHarHeaders(inChannel <- chan *tap.OutputChannelItem, outChannel chan *tap.OutputChannelItem, filterOptions *shared.TrafficFilteringOptions) {
-	for message := range inChannel {
 		sensitiveDataFiltering.FilterSensitiveInfoFromHarRequest(message, filterOptions)
+
 		outChannel <- message
 	}
 }

api/pkg/controllers/entries_controller.go

@@ -1,14 +1,20 @@
 package controllers
 
 import (
+	"bytes"
+	"compress/zlib"
 	"encoding/json"
 	"fmt"
 	"github.com/gofiber/fiber/v2"
 	"github.com/google/martian/har"
+	"io/ioutil"
+	"log"
 	"mizuserver/pkg/database"
 	"mizuserver/pkg/models"
 	"mizuserver/pkg/utils"
 	"mizuserver/pkg/validation"
+	"net/http"
+	"net/url"
 	"time"
 )
 
@@ -138,9 +144,72 @@ func GetHARs(c *fiber.Ctx) error {
 	return c.Status(fiber.StatusOK).SendStream(buffer)
 }
 
+func uploadEntriesImpl(token string, model string, envPrefix string) {
+	sleepTime := time.Second * 10
+
+	var timestampFrom int64 = 0
+
+	for {
+		timestampTo := time.Now().UnixNano() / int64(time.Millisecond)
+		fmt.Printf("Getting entries from %v, to %v\n", timestampFrom, timestampTo)
+		entriesArray := getEntriesFromDb(timestampFrom, timestampTo)
+
+		if len(entriesArray) > 0 {
+			fmt.Printf("About to upload %v entries\n", len(entriesArray))
+
+			body, jMarshalErr := json.Marshal(entriesArray)
+			if jMarshalErr != nil {
+				log.Fatal(jMarshalErr)
+			}
+
+			var in bytes.Buffer
+			w := zlib.NewWriter(&in)
+			_, _ = w.Write(body)
+			_ = w.Close()
+			reqBody := ioutil.NopCloser(bytes.NewReader(in.Bytes()))
+
+			postUrl, _ := url.Parse(fmt.Sprintf("https://traffic.%s/dumpTrafficBulk/%s", envPrefix, model))
+			fmt.Println(postUrl)
+			req := &http.Request{
+				Method: http.MethodPost,
+				URL:    postUrl,
+				Header: map[string][]string{
+					"Content-Encoding": {"deflate"},
+					"Content-Type":     {"application/octet-stream"},
+					"Guest-Auth":       {token},
+				},
+				Body: reqBody,
+			}
+			_, postErr := http.DefaultClient.Do(req)
+			if postErr != nil {
+				log.Fatal(postErr)
+			}
+			fmt.Printf("Finish uploading %v entries to %s\n", len(entriesArray), postUrl)
+
+		} else {
+			fmt.Println("Nothing to upload")
+		}
+
+		fmt.Printf("Sleeping for %v...\n", sleepTime)
+		time.Sleep(sleepTime)
+		timestampFrom = timestampTo
+	}
+}
+
+func UploadEntries(c *fiber.Ctx) error {
+	entriesFilter := &models.UploadEntriesRequestBody{}
+	if err := c.QueryParser(entriesFilter); err != nil {
+		return c.Status(fiber.StatusBadRequest).JSON(err)
+	}
+	if err := validation.Validate(entriesFilter); err != nil {
+		return c.Status(fiber.StatusBadRequest).JSON(err)
+	}
+	go uploadEntriesImpl(entriesFilter.Token, entriesFilter.Model, entriesFilter.Dest)
+	return c.Status(fiber.StatusOK).SendString("OK")
+}
+
 func GetFullEntries(c *fiber.Ctx) error {
 	entriesFilter := &models.HarFetchRequestBody{}
-	order := OrderDesc
 	if err := c.QueryParser(entriesFilter); err != nil {
 		return c.Status(fiber.StatusBadRequest).JSON(err)
 	}
@@ -162,6 +231,12 @@ func GetFullEntries(c *fiber.Ctx) error {
 		timestampTo = entriesFilter.To
 	}
 
+	entriesArray := getEntriesFromDb(timestampFrom, timestampTo)
+	return c.Status(fiber.StatusOK).JSON(entriesArray)
+}
+
+func getEntriesFromDb(timestampFrom int64, timestampTo int64) []har.Entry {
+	order := OrderDesc
 	var entries []models.MizuEntry
 	database.GetEntriesTable().
 		Where(fmt.Sprintf("timestamp BETWEEN %v AND %v", timestampFrom, timestampTo)).
@@ -179,7 +254,7 @@ func GetFullEntries(c *fiber.Ctx) error {
 		_ = json.Unmarshal([]byte(entryData.Entry), &harEntry)
 		entriesArray = append(entriesArray, harEntry)
 	}
-	return c.Status(fiber.StatusOK).JSON(entriesArray)
+	return entriesArray
 }
 
 func GetEntry(c *fiber.Ctx) error {

api/pkg/models/models.go

@@ -49,6 +49,12 @@ type EntriesFilter struct {
 	Timestamp int64  `query:"timestamp" validate:"required,min=1"`
 }
 
+type UploadEntriesRequestBody struct {
+	Token string `query:"token"`
+	Model string `query:"model"`
+	Dest  string `query:"dest"`
+}
+
 type HarFetchRequestBody struct {
 	From int64 `query:"from"`
 	To   int64 `query:"to"`

api/pkg/routes/public_routes.go

@@ -12,7 +12,7 @@ func EntriesRoutes(fiberApp *fiber.App) {
 	routeGroup.Get("/entries", controllers.GetEntries)        // get entries (base/thin entries)
 	routeGroup.Get("/entries/:entryId", controllers.GetEntry) // get single (full) entry
 	routeGroup.Get("/exportEntries", controllers.GetFullEntries)
-
+	routeGroup.Get("/uploadEntries", controllers.UploadEntries)
 
 	routeGroup.Get("/har", controllers.GetHARs)
 

cli/cmd/tap.go

@@ -3,8 +3,10 @@ package cmd
 import (
 	"errors"
 	"fmt"
-	"github.com/up9inc/mizu/cli/mizu"
 	"regexp"
+	"strings"
+
+	"github.com/up9inc/mizu/cli/mizu"
 
 	"github.com/spf13/cobra"
 )
@@ -13,21 +15,23 @@ type MizuTapOptions struct {
 	GuiPort                uint16
 	Namespace              string
 	AllNamespaces          bool
+	Analyze                bool
+	AnalyzeDestination     string
 	KubeConfigPath         string
 	MizuImage              string
 	MizuPodPort            uint16
 	PlainTextFilterRegexes []string
-	Direction              string
+	TapOutgoing            bool
 }
 
-
 var mizuTapOptions = &MizuTapOptions{}
+var direction string
 
 var tapCmd = &cobra.Command{
 	Use:   "tap [POD REGEX]",
 	Short: "Record ingoing traffic of a kubernetes pod",
 	Long: `Record the ingoing traffic of a kubernetes pod.
- Supported protocols are HTTP and gRPC.`,
+Supported protocols are HTTP and gRPC.`,
 	RunE: func(cmd *cobra.Command, args []string) error {
 		if len(args) == 0 {
 			return errors.New("POD REGEX argument is required")
@@ -40,8 +44,13 @@ var tapCmd = &cobra.Command{
 			return errors.New(fmt.Sprintf("%s is not a valid regex %s", args[0], err))
 		}
 
-		if mizuTapOptions.Direction != "in" && mizuTapOptions.Direction != "any" {
-			return errors.New(fmt.Sprintf("%s is not a valid value for flag --direction. Acceptable values are in/any.", mizuTapOptions.Direction))
+		directionLowerCase := strings.ToLower(direction)
+		if directionLowerCase == "any" {
+			mizuTapOptions.TapOutgoing = true
+		} else if directionLowerCase == "in" {
+			mizuTapOptions.TapOutgoing = false
+		} else {
+			return errors.New(fmt.Sprintf("%s is not a valid value for flag --direction. Acceptable values are in/any.", direction))
 		}
 
 		RunMizuTap(regex, mizuTapOptions)
@@ -54,10 +63,12 @@ func init() {
 
 	tapCmd.Flags().Uint16VarP(&mizuTapOptions.GuiPort, "gui-port", "p", 8899, "Provide a custom port for the web interface webserver")
 	tapCmd.Flags().StringVarP(&mizuTapOptions.Namespace, "namespace", "n", "", "Namespace selector")
+	tapCmd.Flags().BoolVar(&mizuTapOptions.Analyze, "analyze", false, "Uploads traffic to UP9 cloud for further analysis")
+	tapCmd.Flags().StringVar(&mizuTapOptions.AnalyzeDestination, "dest", "up9.app", "Destination environment")
 	tapCmd.Flags().BoolVarP(&mizuTapOptions.AllNamespaces, "all-namespaces", "A", false, "Tap all namespaces")
 	tapCmd.Flags().StringVarP(&mizuTapOptions.KubeConfigPath, "kube-config", "k", "", "Path to kube-config file")
 	tapCmd.Flags().StringVarP(&mizuTapOptions.MizuImage, "mizu-image", "", fmt.Sprintf("gcr.io/up9-docker-hub/mizu/%s:latest", mizu.Branch), "Custom image for mizu collector")
 	tapCmd.Flags().Uint16VarP(&mizuTapOptions.MizuPodPort, "mizu-port", "", 8899, "Port which mizu cli will attempt to forward from the mizu collector pod")
 	tapCmd.Flags().StringArrayVarP(&mizuTapOptions.PlainTextFilterRegexes, "regex-masking", "r", nil, "List of regex expressions that are used to filter matching values from text/plain http bodies")
-	tapCmd.Flags().StringVarP(&mizuTapOptions.Direction, "direction", "", "in", "Record traffic that goes in this direction (relative to the tapped pod): in/any")
+	tapCmd.Flags().StringVarP(&direction, "direction", "", "in", "Record traffic that goes in this direction (relative to the tapped pod): in/any")
 }

cli/cmd/tapRunner.go

@@ -2,14 +2,17 @@ package cmd
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
-	"github.com/up9inc/mizu/shared"
+	"net/http"
 	"os"
 	"os/signal"
 	"regexp"
 	"syscall"
 	"time"
 
+	"github.com/up9inc/mizu/shared"
+
 	core "k8s.io/api/core/v1"
 
 	"github.com/up9inc/mizu/cli/debounce"
@@ -45,6 +48,22 @@ func RunMizuTap(podRegexQuery *regexp.Regexp, tappingOptions *MizuTapOptions) {
 		currentlyTappedPods = matchingPods
 	}
 
+	var namespacesStr string
+	if targetNamespace != mizu.K8sAllNamespaces {
+		namespacesStr = fmt.Sprintf("namespace \"%s\"", targetNamespace)
+	} else {
+		namespacesStr = "all namespaces"
+	}
+	fmt.Printf("Tapping pods in %s\n", namespacesStr)
+
+	if len(currentlyTappedPods) == 0 {
+		var suggestionStr string
+		if targetNamespace != mizu.K8sAllNamespaces {
+			suggestionStr = "\nSelect a different namespace with -n or tap all namespaces with -A"
+		}
+		fmt.Printf("Did not find any pods matching the regex argument%s\n", suggestionStr)
+	}
+
 	nodeToTappedPodIPMap, err := getNodeHostToTappedPodIpsMap(currentlyTappedPods)
 	if err != nil {
 		return
@@ -62,12 +81,37 @@ func RunMizuTap(podRegexQuery *regexp.Regexp, tappingOptions *MizuTapOptions) {
 	waitForFinish(ctx, cancel)
 }
 
+type GuestToken struct {
+	Token string `json:"token"`
+	Model string `json:"model"`
+}
+
+func getGuestToken(url string, target *GuestToken) error {
+	resp, err := http.Get(url)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	return json.NewDecoder(resp.Body).Decode(target)
+}
+
+func CreateAnonymousToken(envPrefix string) (*GuestToken, error) {
+	tokenUrl := fmt.Sprintf("https://trcc.%v/anonymous/token", envPrefix)
+	token := &GuestToken{}
+	if err := getGuestToken(tokenUrl, token); err != nil {
+		fmt.Println(err)
+		return nil, err
+	}
+	return token, nil
+}
+
 func createMizuResources(ctx context.Context, kubernetesProvider *kubernetes.Provider, nodeToTappedPodIPMap map[string][]string, tappingOptions *MizuTapOptions, mizuApiFilteringOptions *shared.TrafficFilteringOptions) error {
 	if err := createMizuAggregator(ctx, kubernetesProvider, tappingOptions, mizuApiFilteringOptions); err != nil {
 		return err
 	}
 
-	if err := createMizuTappers(ctx, kubernetesProvider, nodeToTappedPodIPMap, tappingOptions); err != nil {
+	if err := updateMizuTappers(ctx, kubernetesProvider, nodeToTappedPodIPMap, tappingOptions); err != nil {
 		return err
 	}
 
@@ -111,20 +155,27 @@ func getMizuApiFilteringOptions(tappingOptions *MizuTapOptions) (*shared.Traffic
 	return &shared.TrafficFilteringOptions{PlainTextMaskingRegexes: compiledRegexSlice}, nil
 }
 
-func createMizuTappers(ctx context.Context, kubernetesProvider *kubernetes.Provider, nodeToTappedPodIPMap map[string][]string, tappingOptions *MizuTapOptions) error {
-	if err := kubernetesProvider.ApplyMizuTapperDaemonSet(
-		ctx,
-		mizu.ResourcesNamespace,
-		mizu.TapperDaemonSetName,
-		tappingOptions.MizuImage,
-		mizu.TapperPodName,
-		fmt.Sprintf("%s.%s.svc.cluster.local", aggregatorService.Name, aggregatorService.Namespace),
-		nodeToTappedPodIPMap,
-		mizuServiceAccountExists,
-		tappingOptions.Direction,
-	); err != nil {
-		fmt.Printf("Error creating mizu tapper daemonset: %v\n", err)
-		return err
+func updateMizuTappers(ctx context.Context, kubernetesProvider *kubernetes.Provider, nodeToTappedPodIPMap map[string][]string, tappingOptions *MizuTapOptions) error {
+	if len(nodeToTappedPodIPMap) > 0 {
+		if err := kubernetesProvider.ApplyMizuTapperDaemonSet(
+			ctx,
+			mizu.ResourcesNamespace,
+			mizu.TapperDaemonSetName,
+			tappingOptions.MizuImage,
+			mizu.TapperPodName,
+			fmt.Sprintf("%s.%s.svc.cluster.local", aggregatorService.Name, aggregatorService.Namespace),
+			nodeToTappedPodIPMap,
+			mizuServiceAccountExists,
+			tappingOptions.TapOutgoing,
+		); err != nil {
+			fmt.Printf("Error creating mizu tapper daemonset: %v\n", err)
+			return err
+		}
+	} else {
+		if err := kubernetesProvider.RemoveDaemonSet(ctx, mizu.ResourcesNamespace, mizu.TapperDaemonSetName); err != nil {
+			fmt.Printf("Error deleting mizu tapper daemonset: %v\n", err)
+			return err
+		}
 	}
 
 	return nil
@@ -164,7 +215,7 @@ func watchPodsForTapping(ctx context.Context, kubernetesProvider *kubernetes.Pro
 			cancel()
 		}
 
-		if err := createMizuTappers(ctx, kubernetesProvider, nodeToTappedPodIPMap, tappingOptions); err != nil {
+		if err := updateMizuTappers(ctx, kubernetesProvider, nodeToTappedPodIPMap, tappingOptions); err != nil {
 			fmt.Printf("Error updating daemonset: %s (%v,%+v)\n", err, err, err)
 			cancel()
 		}
@@ -217,12 +268,22 @@ func portForwardApiPod(ctx context.Context, kubernetesProvider *kubernetes.Provi
 		case modifiedPod := <-modified:
 			if modifiedPod.Status.Phase == "Running" && !isPodReady {
 				isPodReady = true
-				var err error
-				portForward, err = kubernetes.NewPortForward(kubernetesProvider, mizu.ResourcesNamespace, mizu.AggregatorPodName, tappingOptions.GuiPort, tappingOptions.MizuPodPort, cancel)
-				fmt.Printf("Web interface is now available at http://localhost:%d\n", tappingOptions.GuiPort)
-				if err != nil {
-					fmt.Printf("error forwarding port to pod %s\n", err)
+				var portForwardCreateError error
+				if portForward, portForwardCreateError = kubernetes.NewPortForward(kubernetesProvider, mizu.ResourcesNamespace, mizu.AggregatorPodName, tappingOptions.GuiPort, tappingOptions.MizuPodPort, cancel); portForwardCreateError != nil {
+					fmt.Printf("error forwarding port to pod %s\n", portForwardCreateError)
 					cancel()
+				} else {
+					fmt.Printf("Web interface is now available at http://localhost:%d\n", tappingOptions.GuiPort)
+
+					if tappingOptions.Analyze {
+						token, _ := CreateAnonymousToken(tappingOptions.AnalyzeDestination)
+						if _, err := http.Get(fmt.Sprintf("http://localhost:%d/api/uploadEntries?token=%s&model=%s&dest=%s", tappingOptions.GuiPort, token.Token, token.Model, tappingOptions.AnalyzeDestination)); err != nil {
+							fmt.Println(err)
+						} else {
+							fmt.Println("Staring to upload and analyze the data, it may take a few minutes")
+							fmt.Println("https://" + tappingOptions.AnalyzeDestination + "/share/" + token.Token)
+						}
+					}
 				}
 			}
 

cli/kubernetes/provider.go

@@ -102,7 +102,6 @@ func (provider *Provider) CreateMizuAggregatorPod(ctx context.Context, namespace
 			},
 			DNSPolicy:                     core.DNSClusterFirstWithHostNet,
 			TerminationGracePeriodSeconds: new(int64),
-			// Affinity: TODO: define node selector for all relevant nodes for this mizu instance
 		},
 	}
 	//define the service account only when it exists to prevent pod crash
@@ -215,18 +214,94 @@ func (provider *Provider) CreateMizuRBAC(ctx context.Context, namespace string,
 }
 
 func (provider *Provider) RemovePod(ctx context.Context, namespace string, podName string) error {
+	if isFound, err := provider.CheckPodExists(ctx, namespace, podName);
+	err != nil {
+		return err
+	} else if !isFound {
+		return nil
+	}
+
 	return provider.clientSet.CoreV1().Pods(namespace).Delete(ctx, podName, metav1.DeleteOptions{})
 }
 
 func (provider *Provider) RemoveService(ctx context.Context, namespace string, serviceName string) error {
+	if isFound, err := provider.CheckServiceExists(ctx, namespace, serviceName);
+	err != nil {
+		return err
+	} else if !isFound {
+		return nil
+	}
+
 	return provider.clientSet.CoreV1().Services(namespace).Delete(ctx, serviceName, metav1.DeleteOptions{})
 }
 
 func (provider *Provider) RemoveDaemonSet(ctx context.Context, namespace string, daemonSetName string) error {
+	if isFound, err := provider.CheckDaemonSetExists(ctx, namespace, daemonSetName);
+	err != nil {
+		return err
+	} else if !isFound {
+		return nil
+	}
+
 	return provider.clientSet.AppsV1().DaemonSets(namespace).Delete(ctx, daemonSetName, metav1.DeleteOptions{})
 }
 
-func (provider *Provider) ApplyMizuTapperDaemonSet(ctx context.Context, namespace string, daemonSetName string, podImage string, tapperPodName string, aggregatorPodIp string, nodeToTappedPodIPMap map[string][]string, linkServiceAccount bool, direction string) error {
+func (provider *Provider) CheckPodExists(ctx context.Context, namespace string, name string) (bool, error) {
+	listOptions := metav1.ListOptions{
+		FieldSelector: fmt.Sprintf("metadata.name=%s", name),
+		Limit: 1,
+	}
+	resourceList, err := provider.clientSet.CoreV1().Pods(namespace).List(ctx, listOptions)
+	if err != nil {
+		return false, err
+	}
+
+	if len(resourceList.Items) > 0 {
+		return true, nil
+	}
+
+	return false, nil
+}
+
+func (provider *Provider) CheckServiceExists(ctx context.Context, namespace string, name string) (bool, error) {
+	listOptions := metav1.ListOptions{
+		FieldSelector: fmt.Sprintf("metadata.name=%s", name),
+		Limit: 1,
+	}
+	resourceList, err := provider.clientSet.CoreV1().Services(namespace).List(ctx, listOptions)
+	if err != nil {
+		return false, err
+	}
+
+	if len(resourceList.Items) > 0 {
+		return true, nil
+	}
+
+	return false, nil
+}
+
+func (provider *Provider) CheckDaemonSetExists(ctx context.Context, namespace string, name string) (bool, error) {
+	listOptions := metav1.ListOptions{
+		FieldSelector: fmt.Sprintf("metadata.name=%s", name),
+		Limit: 1,
+	}
+	resourceList, err := provider.clientSet.AppsV1().DaemonSets(namespace).List(ctx, listOptions)
+	if err != nil {
+		return false, err
+	}
+
+	if len(resourceList.Items) > 0 {
+		return true, nil
+	}
+
+	return false, nil
+}
+
+func (provider *Provider) ApplyMizuTapperDaemonSet(ctx context.Context, namespace string, daemonSetName string, podImage string, tapperPodName string, aggregatorPodIp string, nodeToTappedPodIPMap map[string][]string, linkServiceAccount bool, tapOutgoing bool) error {
+	if len(nodeToTappedPodIPMap) == 0 {
+		return fmt.Errorf("Daemon set %s must tap at least 1 pod", daemonSetName)
+	}
+
 	nodeToTappedPodIPMapJsonStr, err := json.Marshal(nodeToTappedPodIPMap)
 	if err != nil {
 		return err
@@ -239,7 +314,7 @@ func (provider *Provider) ApplyMizuTapperDaemonSet(ctx context.Context, namespac
 		"--hardump",
 		"--aggregator-address", fmt.Sprintf("ws://%s/wsTapper", aggregatorPodIp),
 	}
-	if direction == "any" {
+	if tapOutgoing {
 		mizuCmd = append(mizuCmd, "--anydirection")
 	}