Refactoring (#84)

* Only use one channel for filtering HARs.

* Only check if dest is service ip if message is outgoing.

* Parse direction flag on input.

* Renamed filterHarHeaders -> filterHarItems.

* Fixed compilation bugs.

api/main.go

@@ -35,12 +35,10 @@ func main() {
 
 	if *standalone {
 		harOutputChannel, outboundLinkOutputChannel := tap.StartPassiveTapper(tapOpts)
-		filteredHarChannel0 := make(chan *tap.OutputChannelItem)
+		filteredHarChannel := make(chan *tap.OutputChannelItem)
-		filteredHarChannel1 := make(chan *tap.OutputChannelItem)
 
-		go filterServices(harOutputChannel, filteredHarChannel0, getTrafficFilteringOptions())
+		go filterHarItems(harOutputChannel, filteredHarChannel, getTrafficFilteringOptions())
-		go filterHarHeaders(filteredHarChannel0, filteredHarChannel1, getTrafficFilteringOptions())
+		go api.StartReadingEntries(filteredHarChannel, nil)
-		go api.StartReadingEntries(filteredHarChannel1, 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)
+		filteredHarChannel := make(chan *tap.OutputChannelItem)
-		filteredHarChannel1 := make(chan *tap.OutputChannelItem)
 
-		go filterServices(socketHarOutChannel, filteredHarChannel0, getTrafficFilteringOptions())
+		go filterHarItems(socketHarOutChannel, filteredHarChannel, getTrafficFilteringOptions())
-		go filterHarHeaders(filteredHarChannel0, filteredHarChannel1, getTrafficFilteringOptions())
+		go api.StartReadingEntries(filteredHarChannel, nil)
-		go api.StartReadingEntries(filteredHarChannel1, 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
 	}
 }

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"
 )
@@ -17,11 +19,12 @@ type MizuTapOptions struct {
 	MizuImage              string
 	MizuPodPort            uint16
 	PlainTextFilterRegexes []string
-	Direction              string
+	TapOutgoing            bool
 }
 
 
 var mizuTapOptions = &MizuTapOptions{}
+var direction string
 
 var tapCmd = &cobra.Command{
 	Use:   "tap [POD REGEX]",
@@ -40,8 +43,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" {
+		directionLowerCase := strings.ToLower(direction)
-			return errors.New(fmt.Sprintf("%s is not a valid value for flag --direction. Acceptable values are in/any.", mizuTapOptions.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)
@@ -59,5 +67,5 @@ func init() {
 	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

@@ -121,7 +121,7 @@ func createMizuTappers(ctx context.Context, kubernetesProvider *kubernetes.Provi
 		fmt.Sprintf("%s.%s.svc.cluster.local", aggregatorService.Name, aggregatorService.Namespace),
 		nodeToTappedPodIPMap,
 		mizuServiceAccountExists,
-		tappingOptions.Direction,
+		tappingOptions.TapOutgoing,
 	); err != nil {
 		fmt.Printf("Error creating mizu tapper daemonset: %v\n", err)
 		return err

cli/kubernetes/provider.go

@@ -226,7 +226,7 @@ func (provider *Provider) RemoveDaemonSet(ctx context.Context, namespace string,
 	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) ApplyMizuTapperDaemonSet(ctx context.Context, namespace string, daemonSetName string, podImage string, tapperPodName string, aggregatorPodIp string, nodeToTappedPodIPMap map[string][]string, linkServiceAccount bool, tapOutgoing bool) error {
 	nodeToTappedPodIPMapJsonStr, err := json.Marshal(nodeToTappedPodIPMap)
 	if err != nil {
 		return err
@@ -239,7 +239,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")
 	}