TRA-4060 fix proxying error (#542)

cli/cmd/tapRunner.go

@@ -151,6 +151,7 @@ func RunMizuTap() {
 	} else {
 		defer finishMizuExecution(kubernetesProvider, apiProvider)
 
+		go goUtils.HandleExcWrapper(watchApiServerEvents, ctx, kubernetesProvider, cancel)
 		go goUtils.HandleExcWrapper(watchApiServerPod, ctx, kubernetesProvider, cancel)
 
 		// block until exit signal or error
@@ -564,6 +565,67 @@ 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))
+	podWatchHelper := kubernetes.NewPodWatchHelper(kubernetesProvider, podExactRegex)
+	eventChan, errorChan := kubernetes.FilteredWatch(ctx, podWatchHelper, []string{config.Config.MizuResourcesNamespace}, podWatchHelper)
+	isPodReady := false
+	timeAfter := time.After(25 * time.Second)
+	for {
+		select {
+		case wEvent, ok := <-eventChan:
+			if !ok {
+				eventChan = nil
+				continue
+			}
+
+			switch wEvent.Type {
+			case kubernetes.EventAdded:
+				logger.Log.Debugf("Watching API Server pod loop, added")
+			case kubernetes.EventDeleted:
+				logger.Log.Infof("%s removed", kubernetes.ApiServerPodName)
+				cancel()
+				return
+			case kubernetes.EventModified:
+				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.PodRunning && !isPodReady {
+					isPodReady = true
+					postApiServerStarted(ctx, kubernetesProvider, cancel, err)
+				}
+			case kubernetes.EventBookmark:
+				break
+			case kubernetes.EventError:
+				break
+			}
+		case err, ok := <-errorChan:
+			if !ok {
+				errorChan = nil
+				continue
+			}
+
+			logger.Log.Errorf("[ERROR] Agent creation, watching %v namespace, error: %v", config.Config.MizuResourcesNamespace, err)
+			cancel()
+
+		case <-timeAfter:
+			if !isPodReady {
+				logger.Log.Errorf(uiUtils.Error, "Mizu API server was not ready in time")
+				cancel()
+			}
+		case <-ctx.Done():
+			logger.Log.Debugf("Watching API Server pod loop, ctx done")
+			return
+		}
+	}
+}
+
+func watchApiServerEvents(ctx context.Context, kubernetesProvider *kubernetes.Provider, cancel context.CancelFunc) {
 	podExactRegex := regexp.MustCompile(fmt.Sprintf("^%s", kubernetes.ApiServerPodName))
 	eventWatchHelper := kubernetes.NewEventWatchHelper(kubernetesProvider, podExactRegex, "pod")
 	eventChan, errorChan := kubernetes.FilteredWatch(ctx, eventWatchHelper, []string{config.Config.MizuResourcesNamespace}, eventWatchHelper)
@@ -594,26 +656,7 @@ func watchApiServerPod(ctx context.Context, kubernetesProvider *kubernetes.Provi
 					event.Note))
 
 			switch event.Reason {
-			case "Started":
-				go startProxyReportErrorIfAny(kubernetesProvider, cancel)
-
-				url := GetApiServerUrl()
-				if err := apiProvider.TestConnection(); err != nil {
-					logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Couldn't connect to API server, for more info check logs at %s", fsUtils.GetLogFilePath()))
-					cancel()
-					break
-				}
-				options, _ := getMizuApiFilteringOptions()
-				if err = startTapperSyncer(ctx, cancel, kubernetesProvider, state.targetNamespaces, *options, state.startTime); err != nil {
-					logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Error starting mizu tapper syncer: %v", err))
-					cancel()
-				}
-
-				logger.Log.Infof("Mizu is available at %s", url)
-				if !config.Config.HeadlessMode {
-					uiUtils.OpenBrowser(url)
-				}
-			case "FailedScheduling", "Failed", "Killing":
+			case "FailedScheduling", "Failed":
 				logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Mizu API Server status: %s - %s", event.Reason, event.Note))
 				cancel()
 				break
@@ -632,6 +675,27 @@ func watchApiServerPod(ctx context.Context, kubernetesProvider *kubernetes.Provi
 	}
 }
 
+func postApiServerStarted(ctx context.Context, kubernetesProvider *kubernetes.Provider, cancel context.CancelFunc, err error) {
+	go startProxyReportErrorIfAny(kubernetesProvider, cancel)
+
+	url := GetApiServerUrl()
+	if err := apiProvider.TestConnection(); err != nil {
+		logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Couldn't connect to API server, for more info check logs at %s", fsUtils.GetLogFilePath()))
+		cancel()
+		return
+	}
+	options, _ := getMizuApiFilteringOptions()
+	if err = startTapperSyncer(ctx, cancel, kubernetesProvider, state.targetNamespaces, *options, state.startTime); err != nil {
+		logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Error starting mizu tapper syncer: %v", err))
+		cancel()
+	}
+
+	logger.Log.Infof("Mizu is available at %s", url)
+	if !config.Config.HeadlessMode {
+		uiUtils.OpenBrowser(url)
+	}
+}
+
 func getNamespaces(kubernetesProvider *kubernetes.Provider) []string {
 	if config.Config.Tap.AllNamespaces {
 		return []string{kubernetes.K8sAllNamespaces}

tap/extensions/amqp/helpers.go

@@ -579,12 +579,18 @@ func representConnectionStart(event map[string]interface{}) []interface{} {
 }
 
 func representConnectionClose(event map[string]interface{}) []interface{} {
+	replyCode := ""
+
+	if event["replyCode"] != nil {
+		replyCode = fmt.Sprintf("%g", event["replyCode"].(float64))
+	}
+
 	rep := make([]interface{}, 0)
 
 	details, _ := json.Marshal([]api.TableData{
 		{
 			Name:     "Reply Code",
-			Value:    fmt.Sprintf("%g", event["replyCode"].(float64)),
+			Value:    replyCode,
 			Selector: `request.replyCode`,
 		},
 		{

ui/src/components/EntriesList.tsx

@@ -66,7 +66,7 @@ export const EntriesList: React.FC<EntriesListProps> = ({entries, setEntries, qu
         }
         setIsLoadingTop(true);
         const data = await api.fetchEntries(leftOffTop, -1, query, 100, 3000);
-        if (!data || !data.meta) {
+        if (!data || data.data === null || data.meta === null) {
             setNoMoreDataTop(true);
             setIsLoadingTop(false);
             return;

ui/src/components/TrafficPage.tsx

@@ -1,4 +1,4 @@
-import React, {useEffect, useRef, useState} from "react";
+import React, {useEffect, useMemo, useRef, useState} from "react";
 import {Filters} from "./Filters";
 import {EntriesList} from "./EntriesList";
 import {makeStyles} from "@material-ui/core";
@@ -12,6 +12,7 @@ import {StatusBar} from "./UI/StatusBar";
 import Api, {MizuWebsocketURL} from "../helpers/api";
 import { ToastContainer, toast } from 'react-toastify';
 import 'react-toastify/dist/ReactToastify.css';
+import debounce from 'lodash/debounce';
 
 const useLayoutStyles = makeStyles(() => ({
     details: {
@@ -72,22 +73,24 @@ export const TrafficPage: React.FC<TrafficPageProps> = ({setAnalyzeStatus, onTLS
 
     const [startTime, setStartTime] = useState(0);
 
-    useEffect(() => {
-        (async function() {
-            if (!query) {
-                setQueryBackgroundColor("#f5f5f5")
-            } else {
-                const data = await api.validateQuery(query);
-                if (!data) {
-                    return;
-                }
-                if (data.valid) {
-                    setQueryBackgroundColor("#d2fad2");
-                } else {
-                    setQueryBackgroundColor("#fad6dc");
-                }
+    const handleQueryChange = useMemo(() => debounce(async (query: string) => {
+        if (!query) {
+            setQueryBackgroundColor("#f5f5f5")
+        } else {
+            const data = await api.validateQuery(query);
+            if (!data) {
+                return;
             }
-        })();
+            if (data.valid) {
+                setQueryBackgroundColor("#d2fad2");
+            } else {
+                setQueryBackgroundColor("#fad6dc");
+            }
+        }
+    }, 500), []) as (query: string) => void;
+
+    useEffect(() => {
+        handleQueryChange(query);
     }, [query]);
 
     useEffect(() => {