Run unit tests when tap/api changes (#1016)

* update test files paths

.github/workflows/test.yml

@@ -56,7 +56,7 @@ jobs:
       
       - name: Check extensions modified files
         id: ext_modified_files
-        run: devops/check_modified_files.sh tap/extensions/
+        run: devops/check_modified_files.sh tap/extensions/ tap/api/
 
       - name: Extensions Test
         if: github.event_name == 'push' || steps.ext_modified_files.outputs.matched == 'true'
@@ -64,4 +64,3 @@ jobs:
 
       - name: Upload coverage to Codecov
         uses: codecov/codecov-action@v2
-

agent/pkg/api/main.go

@@ -128,6 +128,11 @@ BasenineReconnect:
 	for item := range outputItems {
 		extension := extensionsMap[item.Protocol.Name]
 		resolvedSource, resolvedDestionation, namespace := resolveIP(item.ConnectionInfo)
+		
+		if namespace == "" && item.Namespace != tapApi.UNKNOWN_NAMESPACE {
+			namespace = item.Namespace
+		}
+		
 		mizuEntry := extension.Dissector.Analyze(item, resolvedSource, resolvedDestionation, namespace)
 		if extension.Protocol.Name == "http" {
 			if !disableOASValidation {

agent/pkg/providers/status_provider.go

@@ -4,18 +4,13 @@ import (
 	"encoding/json"
 	"fmt"
 	"os"
-	"time"
 
-	"github.com/patrickmn/go-cache"
 	"github.com/up9inc/mizu/agent/pkg/models"
 	"github.com/up9inc/mizu/shared"
 )
 
-const tlsLinkRetainmentTime = time.Minute * 15
-
 var (
-	authStatus     *models.AuthStatus
-	RecentTLSLinks = cache.New(tlsLinkRetainmentTime, tlsLinkRetainmentTime)
+	authStatus *models.AuthStatus
 )
 
 func GetAuthStatus() (*models.AuthStatus, error) {

shared/kubernetes/utils.go

@@ -26,7 +26,8 @@ func GetNodeHostToTappedPodsMap(tappedPods []core.Pod) shared.NodeToPodsMap {
 func getMinimizedPod(fullPod core.Pod) core.Pod {
 	return core.Pod{
 		ObjectMeta: metav1.ObjectMeta{
-			Name: fullPod.Name,
+			Name:      fullPod.Name,
+			Namespace: fullPod.Namespace,
 		},
 		Status: core.PodStatus{
 			PodIP:             fullPod.Status.PodIP,

tap/api/api.go

@@ -18,6 +18,7 @@ import (
 )
 
 const mizuTestEnvVar = "MIZU_TEST"
+const UNKNOWN_NAMESPACE = ""
 
 var UnknownIp net.IP = net.IP{0, 0, 0, 0}
 var UnknownPort uint16 = 0
@@ -92,14 +93,15 @@ type RequestResponsePair struct {
 	Response GenericMessage `json:"response"`
 }
 
-// `Protocol` is modified in the later stages of data propagation. Therefore it's not a pointer.
 type OutputChannelItem struct {
+	// `Protocol` is modified in later stages of data propagation. Therefore, it's not a pointer.
 	Protocol       Protocol
 	Capture        Capture
 	Timestamp      int64
 	ConnectionInfo *ConnectionInfo
 	Pair           *RequestResponsePair
 	Summary        *BaseEntry
+	Namespace      string
 }
 
 type SuperTimer struct {

tap/extensions/amqp/Makefile

@@ -13,4 +13,4 @@ test-pull-bin:
 
 test-pull-expect:
 	@mkdir -p expect
-	@[ "${skipexpect}" ] && echo "Skipping downloading expected JSONs" || gsutil  -o 'GSUtil:parallel_process_count=5' -o 'GSUtil:parallel_thread_count=5'  -m cp -r gs://static.up9.io/mizu/test-pcap/expect7/amqp/\* expect
+	@[ "${skipexpect}" ] && echo "Skipping downloading expected JSONs" || gsutil  -o 'GSUtil:parallel_process_count=5' -o 'GSUtil:parallel_thread_count=5'  -m cp -r gs://static.up9.io/mizu/test-pcap/expect8/amqp/\* expect

tap/extensions/http/Makefile

@@ -13,4 +13,4 @@ test-pull-bin:
 
 test-pull-expect:
 	@mkdir -p expect
-	@[ "${skipexpect}" ] && echo "Skipping downloading expected JSONs" || gsutil  -o 'GSUtil:parallel_process_count=5' -o 'GSUtil:parallel_thread_count=5'  -m cp -r gs://static.up9.io/mizu/test-pcap/expect7/http/\* expect
+	@[ "${skipexpect}" ] && echo "Skipping downloading expected JSONs" || gsutil  -o 'GSUtil:parallel_process_count=5' -o 'GSUtil:parallel_thread_count=5'  -m cp -r gs://static.up9.io/mizu/test-pcap/expect8/http/\* expect

tap/extensions/kafka/Makefile

@@ -13,4 +13,4 @@ test-pull-bin:
 
 test-pull-expect:
 	@mkdir -p expect
-	@[ "${skipexpect}" ] && echo "Skipping downloading expected JSONs" || gsutil  -o 'GSUtil:parallel_process_count=5' -o 'GSUtil:parallel_thread_count=5'  -m cp -r gs://static.up9.io/mizu/test-pcap/expect7/kafka/\* expect
+	@[ "${skipexpect}" ] && echo "Skipping downloading expected JSONs" || gsutil  -o 'GSUtil:parallel_process_count=5' -o 'GSUtil:parallel_thread_count=5'  -m cp -r gs://static.up9.io/mizu/test-pcap/expect8/kafka/\* expect

tap/extensions/redis/Makefile

@@ -13,4 +13,4 @@ test-pull-bin:
 
 test-pull-expect:
 	@mkdir -p expect
-	@[ "${skipexpect}" ] && echo "Skipping downloading expected JSONs" || gsutil  -o 'GSUtil:parallel_process_count=5' -o 'GSUtil:parallel_thread_count=5'  -m cp -r gs://static.up9.io/mizu/test-pcap/expect7/redis/\* expect
+	@[ "${skipexpect}" ] && echo "Skipping downloading expected JSONs" || gsutil  -o 'GSUtil:parallel_process_count=5' -o 'GSUtil:parallel_thread_count=5'  -m cp -r gs://static.up9.io/mizu/test-pcap/expect8/redis/\* expect

tap/tlstapper/tls_emitter.go

@@ -0,0 +1,13 @@
+package tlstapper
+
+import "github.com/up9inc/mizu/tap/api"
+
+type tlsEmitter struct {
+	delegate  api.Emitter
+	namespace string
+}
+
+func (e *tlsEmitter) Emit(item *api.OutputChannelItem) {
+	item.Namespace = e.namespace
+	e.delegate.Emit(item)
+}

tap/tlstapper/tls_poller.go

@@ -5,6 +5,7 @@ import (
 	"bytes"
 	"fmt"
 	"net"
+	"sync"
 
 	"encoding/binary"
 	"encoding/hex"
@@ -19,13 +20,14 @@ import (
 )
 
 type tlsPoller struct {
-	tls           *TlsTapper
-	readers       map[string]*tlsReader
-	closedReaders chan string
-	reqResMatcher api.RequestResponseMatcher
-	chunksReader  *perf.Reader
-	extension     *api.Extension
-	procfs        string
+	tls            *TlsTapper
+	readers        map[string]*tlsReader
+	closedReaders  chan string
+	reqResMatcher  api.RequestResponseMatcher
+	chunksReader   *perf.Reader
+	extension      *api.Extension
+	procfs         string
+	pidToNamespace sync.Map
 }
 
 func newTlsPoller(tls *TlsTapper, extension *api.Extension, procfs string) *tlsPoller {
@@ -151,16 +153,21 @@ func (p *tlsPoller) startNewTlsReader(chunk *tlsChunk, ip net.IP, port uint16, k
 
 	tcpid := p.buildTcpId(chunk, ip, port)
 
-	go dissect(extension, reader, chunk.isRequest(), &tcpid, emitter, options, p.reqResMatcher)
+	tlsEmitter := &tlsEmitter{
+		delegate:  emitter,
+		namespace: p.getNamespace(chunk.Pid),
+	}
+
+	go dissect(extension, reader, chunk.isRequest(), &tcpid, tlsEmitter, options, p.reqResMatcher)
 	return reader
 }
 
 func dissect(extension *api.Extension, reader *tlsReader, isRequest bool, tcpid *api.TcpID,
-	emitter api.Emitter, options *api.TrafficFilteringOptions, reqResMatcher api.RequestResponseMatcher) {
+	tlsEmitter *tlsEmitter, options *api.TrafficFilteringOptions, reqResMatcher api.RequestResponseMatcher) {
 	b := bufio.NewReader(reader)
 
 	err := extension.Dissector.Dissect(b, reader.progress, api.Ebpf, isRequest, tcpid, &api.CounterPair{},
-		&api.SuperTimer{}, &api.SuperIdentifier{}, emitter, options, reqResMatcher)
+		&api.SuperTimer{}, &api.SuperIdentifier{}, tlsEmitter, options, reqResMatcher)
 
 	if err != nil {
 		logger.Log.Warningf("Error dissecting TLS %v - %v", tcpid, err)
@@ -205,6 +212,33 @@ func (p *tlsPoller) buildTcpId(chunk *tlsChunk, ip net.IP, port uint16) api.TcpI
 	}
 }
 
+func (p *tlsPoller) addPid(pid uint32, namespace string) {
+	p.pidToNamespace.Store(pid, namespace)
+}
+
+func (p *tlsPoller) getNamespace(pid uint32) string {
+	namespaceIfc, ok := p.pidToNamespace.Load(pid)
+
+	if !ok {
+		return api.UNKNOWN_NAMESPACE
+	}
+
+	namespace, ok := namespaceIfc.(string)
+
+	if !ok {
+		return api.UNKNOWN_NAMESPACE
+	}
+
+	return namespace
+}
+
+func (p *tlsPoller) clearPids() {
+	p.pidToNamespace.Range(func(key, v interface{}) bool {
+		p.pidToNamespace.Delete(key)
+		return true
+	})
+}
+
 func (p *tlsPoller) logTls(chunk *tlsChunk, ip net.IP, port uint16) {
 	var flagsStr string
 

tap/tlstapper/tls_process_discoverer.go

@@ -24,11 +24,11 @@ func UpdateTapTargets(tls *TlsTapper, pods *[]v1.Pod, procfs string) error {
 	if err != nil {
 		return err
 	}
-	
+
 	tls.ClearPids()
 
-	for _, pid := range containerPids {
-		if err := tls.AddPid(procfs, pid); err != nil {
+	for pid, pod := range containerPids {
+		if err := tls.AddPid(procfs, pid, pod.Namespace); err != nil {
 			LogError(err)
 		}
 	}
@@ -36,8 +36,8 @@ func UpdateTapTargets(tls *TlsTapper, pods *[]v1.Pod, procfs string) error {
 	return nil
 }
 
-func findContainerPids(procfs string, containerIds map[string]bool) ([]uint32, error) {
-	result := make([]uint32, 0)
+func findContainerPids(procfs string, containerIds map[string]v1.Pod) (map[uint32]v1.Pod, error) {
+	result := make(map[uint32]v1.Pod)
 
 	pids, err := ioutil.ReadDir(procfs)
 
@@ -63,7 +63,9 @@ func findContainerPids(procfs string, containerIds map[string]bool) ([]uint32, e
 			continue
 		}
 
-		if _, ok := containerIds[cgroup]; !ok {
+		pod, ok := containerIds[cgroup]
+
+		if !ok {
 			continue
 		}
 
@@ -73,14 +75,14 @@ func findContainerPids(procfs string, containerIds map[string]bool) ([]uint32, e
 			continue
 		}
 
-		result = append(result, uint32(pidNumber))
+		result[uint32(pidNumber)] = pod
 	}
 
 	return result, nil
 }
 
-func buildContainerIdsMap(pods *[]v1.Pod) map[string]bool {
-	result := make(map[string]bool)
+func buildContainerIdsMap(pods *[]v1.Pod) map[string]v1.Pod {
+	result := make(map[string]v1.Pod)
 
 	for _, pod := range *pods {
 		for _, container := range pod.Status.ContainerStatuses {
@@ -91,7 +93,7 @@ func buildContainerIdsMap(pods *[]v1.Pod) map[string]bool {
 				continue
 			}
 
-			result[url.Host] = true
+			result[url.Host] = pod
 		}
 	}
 
@@ -141,14 +143,14 @@ func extractCgroup(lines []string) string {
 //  /kubepods.slice/kubepods-burstable.slice/kubepods-burstable-pod3beae8e0_164d_4689_a087_efd902d8c2ab.slice/docker-<ID>.scope
 //  /kubepods/besteffort/pod7709c1d5-447c-428f-bed9-8ddec35c93f4/<ID>
 //
-// This function extract the <ID> out of the cgroup path, the <ID> should match 
+// This function extract the <ID> out of the cgroup path, the <ID> should match
 //	the "Container ID:" field when running kubectl describe pod <POD>
 //
 func normalizeCgroup(cgrouppath string) string {
 	basename := strings.TrimSpace(path.Base(cgrouppath))
-	
+
 	if strings.Contains(basename, "-") {
-		basename = basename[strings.Index(basename, "-") + 1:]
+		basename = basename[strings.Index(basename, "-")+1:]
 	}
 
 	if strings.Contains(basename, ".") {

tap/tlstapper/tls_tapper.go

@@ -1,11 +1,12 @@
 package tlstapper
 
 import (
+	"sync"
+
 	"github.com/cilium/ebpf/rlimit"
 	"github.com/go-errors/errors"
 	"github.com/up9inc/mizu/shared/logger"
 	"github.com/up9inc/mizu/tap/api"
-	"sync"
 )
 
 const GLOABL_TAP_PID = 0
@@ -58,10 +59,10 @@ func (t *TlsTapper) PollForLogging() {
 }
 
 func (t *TlsTapper) GlobalTap(sslLibrary string) error {
-	return t.tapPid(GLOABL_TAP_PID, sslLibrary)
+	return t.tapPid(GLOABL_TAP_PID, sslLibrary, api.UNKNOWN_NAMESPACE)
 }
 
-func (t *TlsTapper) AddPid(procfs string, pid uint32) error {
+func (t *TlsTapper) AddPid(procfs string, pid uint32, namespace string) error {
 	sslLibrary, err := findSsllib(procfs, pid)
 
 	if err != nil {
@@ -69,7 +70,7 @@ func (t *TlsTapper) AddPid(procfs string, pid uint32) error {
 		return nil // hide the error on purpose, its OK for a process to not use libssl.so
 	}
 
-	return t.tapPid(pid, sslLibrary)
+	return t.tapPid(pid, sslLibrary, namespace)
 }
 
 func (t *TlsTapper) RemovePid(pid uint32) error {
@@ -85,12 +86,13 @@ func (t *TlsTapper) RemovePid(pid uint32) error {
 }
 
 func (t *TlsTapper) ClearPids() {
+	t.poller.clearPids()
 	t.registeredPids.Range(func(key, v interface{}) bool {
 		pid := key.(uint32)
 		if pid == GLOABL_TAP_PID {
 			return true
 		}
-		
+
 		if err := t.RemovePid(pid); err != nil {
 			LogError(err)
 		}
@@ -133,7 +135,7 @@ func setupRLimit() error {
 	return nil
 }
 
-func (t *TlsTapper) tapPid(pid uint32, sslLibrary string) error {
+func (t *TlsTapper) tapPid(pid uint32, sslLibrary string, namespace string) error {
 	logger.Log.Infof("Tapping TLS (pid: %v) (sslLibrary: %v)", pid, sslLibrary)
 
 	newSsl := sslHooks{}
@@ -144,12 +146,14 @@ func (t *TlsTapper) tapPid(pid uint32, sslLibrary string) error {
 
 	t.sslHooksStructs = append(t.sslHooksStructs, newSsl)
 
+	t.poller.addPid(pid, namespace)
+
 	pids := t.bpfObjects.tlsTapperMaps.PidsMap
 
 	if err := pids.Put(pid, uint32(1)); err != nil {
 		return errors.Wrap(err, 0)
 	}
-	
+
 	t.registeredPids.Store(pid, true)
 
 	return nil

ui-common/example/src/api.js

@@ -54,11 +54,6 @@ export default class Api {
         return response.data;
     }
 
-    getRecentTLSLinks = async () => {
-        const response = await client.get("/status/recentTLSLinks");
-        return response.data;
-    }
-
     getOasServices = async () => {
         const response = await client.get("/oas");
         return response.data;
@@ -121,4 +116,4 @@ export function getWebsocketUrl(){
     }
 
     return websocketUrl;
-}
+}

ui-common/src/components/TLSWarning/TLSWarning.sass

@@ -1,12 +0,0 @@
-.httpsDomains
-  display: none
-  margin: 0
-  padding: 0
-  list-style: none
-
-.customWarningStyle
-  &:hover
-    overflow-y: scroll
-    height: 85px
-    .httpsDomains
-      display: block

ui-common/src/components/TLSWarning/TLSWarning.tsx

@@ -1,44 +0,0 @@
-import {Snackbar} from "@material-ui/core";
-import MuiAlert from "@material-ui/lab/Alert";
-import React, {useEffect} from "react";
-import { RecoilState, useRecoilValue } from "recoil";
-import TrafficViewerApiAtom from "../../recoil/TrafficViewerApi/atom";
-import TrafficViewerApi from "../TrafficViewer/TrafficViewerApi";
-import './TLSWarning.sass';
-
-interface TLSWarningProps {
-    showTLSWarning: boolean
-    setShowTLSWarning: (show: boolean) => void
-    addressesWithTLS: Set<string>
-    setAddressesWithTLS: (addresses: Set<string>) => void
-    userDismissedTLSWarning: boolean
-    setUserDismissedTLSWarning: (flag: boolean) => void
-}
-
-export const TLSWarning: React.FC<TLSWarningProps>  = ({showTLSWarning, setShowTLSWarning, addressesWithTLS, setAddressesWithTLS, userDismissedTLSWarning, setUserDismissedTLSWarning}) => {
-
-    const trafficViewerApi = useRecoilValue(TrafficViewerApiAtom as RecoilState<TrafficViewerApi>)
-    useEffect(() => {
-        (async () => {
-            try {
-                const getRecentTLSLinksFunc = trafficViewerApi?.getRecentTLSLinks ? trafficViewerApi?.getRecentTLSLinks : function(){}
-                const recentTLSLinks = await getRecentTLSLinksFunc();
-                if (recentTLSLinks?.length > 0) {
-                    setAddressesWithTLS(new Set(recentTLSLinks));
-                    setShowTLSWarning(true);
-                }
-            } catch (e) {
-                console.error(e);
-            }
-        })();
-    }, [setShowTLSWarning, setAddressesWithTLS,trafficViewerApi]);
-
-    return (<Snackbar open={showTLSWarning && !userDismissedTLSWarning}>
-        <MuiAlert classes={{filledWarning: 'customWarningStyle'}} elevation={6} variant="filled"
-                  onClose={() => setUserDismissedTLSWarning(true)} severity="warning">
-            Mizu is detecting TLS traffic, this type of traffic will not be displayed.
-            {addressesWithTLS.size > 0 &&
-            <ul className="httpsDomains"> {Array.from(addressesWithTLS, address => <li>{address}</li>)} </ul>}
-        </MuiAlert>
-    </Snackbar>);
-}

ui-common/src/components/TrafficViewer/TrafficViewer.tsx

@@ -14,7 +14,6 @@ import {RecoilRoot, RecoilState, useRecoilState, useRecoilValue, useSetRecoilSta
 import entriesAtom from "../../recoil/entries";
 import focusedEntryIdAtom from "../../recoil/focusedEntryId";
 import queryAtom from "../../recoil/query";
-import {TLSWarning} from "../TLSWarning/TLSWarning";
 import trafficViewerApiAtom from "../../recoil/TrafficViewerApi"
 import TrafficViewerApi from "./TrafficViewerApi";
 import {StatusBar} from "../UI/StatusBar";
@@ -77,10 +76,6 @@ export const TrafficViewer: React.FC<TrafficViewerProps> = ({
   const setLeftOffTop = useSetRecoilState(leftOffTopAtom);
   const scrollableRef = useRef(null);
 
-  const [showTLSWarning, setShowTLSWarning] = useState(false);
-  const [userDismissedTLSWarning, setUserDismissedTLSWarning] = useState(false);
-  const [addressesWithTLS, setAddressesWithTLS] = useState(new Set<string>());
-
   const handleQueryChange = useMemo(
     () =>
       debounce(async (query: string) => {
@@ -286,12 +281,6 @@ export const TrafficViewer: React.FC<TrafficViewerProps> = ({
           <EntryDetailed/>
         </div>
       </div>}
-      <TLSWarning showTLSWarning={showTLSWarning}
-                  setShowTLSWarning={setShowTLSWarning}
-                  addressesWithTLS={addressesWithTLS}
-                  setAddressesWithTLS={setAddressesWithTLS}
-                  userDismissedTLSWarning={userDismissedTLSWarning}
-                  setUserDismissedTLSWarning={setUserDismissedTLSWarning}/>
     </div>
   );
 };

ui-common/src/components/TrafficViewer/TrafficViewerApi.ts

@@ -4,7 +4,6 @@ type TrafficViewerApi = {
   analyzeStatus: () => any
   fetchEntries: (leftOff: any, direction: number, query: any, limit: number, timeoutMs: number) => any
   getEntry: (entryId: any, query: string) => any
-  getRecentTLSLinks: () => any,
   webSocket: {
     close: () => void
   }

ui/src/helpers/api.js

@@ -62,11 +62,6 @@ export default class Api {
         return response.data;
     }
 
-    getRecentTLSLinks = async () => {
-        const response = await client.get("/status/recentTLSLinks");
-        return response.data;
-    }
-
     getAuthStatus = async () => {
         const response = await client.get("/status/auth");
         return response.data;