Handle unexpected socket close and replace the default rlimit(100) filter with leftOff(-1) filter (#508)

* Handle unexpected socket close and replace the default `rlimit(100)` filter with `leftOff(-1)` filter

* Rename `dontClear` parameter to `resetEntriesBuffer` and remove negation

Dockerfile

@@ -42,8 +42,8 @@ RUN go build -ldflags="-s -w \
      -X 'mizuserver/pkg/version.SemVer=${SEM_VER}'" -o mizuagent .
 
 # Download Basenine executable, verify the sha1sum and move it to a directory in $PATH
-ADD https://github.com/up9inc/basenine/releases/download/v0.2.11/basenine_linux_amd64 ./basenine_linux_amd64
-ADD https://github.com/up9inc/basenine/releases/download/v0.2.11/basenine_linux_amd64.sha256 ./basenine_linux_amd64.sha256
+ADD https://github.com/up9inc/basenine/releases/download/v0.2.12/basenine_linux_amd64 ./basenine_linux_amd64
+ADD https://github.com/up9inc/basenine/releases/download/v0.2.12/basenine_linux_amd64.sha256 ./basenine_linux_amd64.sha256
 RUN shasum -a 256 -c basenine_linux_amd64.sha256
 RUN chmod +x ./basenine_linux_amd64
 

agent/go.mod

@@ -16,7 +16,7 @@ require (
 	github.com/op/go-logging v0.0.0-20160315200505-970db520ece7
 	github.com/orcaman/concurrent-map v0.0.0-20210106121528-16402b402231
 	github.com/patrickmn/go-cache v2.1.0+incompatible
-	github.com/up9inc/basenine/client/go v0.0.0-20211121072216-04366911881c
+	github.com/up9inc/basenine/client/go v0.0.0-20211125004153-d0e8aec03fea
 	github.com/up9inc/mizu/shared v0.0.0
 	github.com/up9inc/mizu/tap v0.0.0
 	github.com/up9inc/mizu/tap/api v0.0.0

agent/go.sum

@@ -450,8 +450,8 @@ github.com/ugorji/go v1.1.7 h1:/68gy2h+1mWMrwZFeD1kQialdSzAb432dtpeJ42ovdo=
 github.com/ugorji/go v1.1.7/go.mod h1:kZn38zHttfInRq0xu/PH0az30d+z6vm202qpg1oXVMw=
 github.com/ugorji/go/codec v1.1.7 h1:2SvQaVZ1ouYrrKKwoSk2pzd4A9evlKJb9oTL+OaLUSs=
 github.com/ugorji/go/codec v1.1.7/go.mod h1:Ax+UKWsSmolVDwsd+7N3ZtXu+yMGCf907BLYF3GoBXY=
-github.com/up9inc/basenine/client/go v0.0.0-20211121072216-04366911881c h1:GJsCVhDKjV/k3mNG255VN7hAQ7fxyNgX5T+VJyzoOQ0=
-github.com/up9inc/basenine/client/go v0.0.0-20211121072216-04366911881c/go.mod h1:SvJGPoa/6erhUQV7kvHBwM/0x5LyO6XaG2lUaCaKiUI=
+github.com/up9inc/basenine/client/go v0.0.0-20211125004153-d0e8aec03fea h1:GpJGO2PNTS/S0j2E1yqBi3ST/VXfokzuOrTQyaYgWnA=
+github.com/up9inc/basenine/client/go v0.0.0-20211125004153-d0e8aec03fea/go.mod h1:SvJGPoa/6erhUQV7kvHBwM/0x5LyO6XaG2lUaCaKiUI=
 github.com/vektah/gqlparser v1.1.2/go.mod h1:1ycwN7Ij5njmMkPPAOaRFY4rET2Enx7IkVv3vaXspKw=
 github.com/vishvananda/netns v0.0.0-20210104183010-2eb08e3e575f h1:p4VB7kIXpOQvVn1ZaTIVp+3vuYAXFe3OJEvjbUYJLaA=
 github.com/vishvananda/netns v0.0.0-20210104183010-2eb08e3e575f/go.mod h1:DD4vA1DwXk04H54A1oHXtwZmA0grkVMdPxx/VGLCah0=

agent/pkg/api/socket_routes.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"mizuserver/pkg/models"
 	"net/http"
+	"strconv"
 	"sync"
 	"time"
 
@@ -94,6 +95,8 @@ func websocketHandler(w http.ResponseWriter, r *http.Request, eventHandlers Even
 	startTimeBytes, _ := models.CreateWebsocketStartTimeMessage(startTime)
 	SendToSocket(socketId, startTimeBytes)
 
+	queryRecieved := false
+
 	for {
 		_, msg, err := ws.ReadMessage()
 		if err != nil {
@@ -101,65 +104,75 @@ func websocketHandler(w http.ResponseWriter, r *http.Request, eventHandlers Even
 			break
 		}
 
-		if !isTapper && !isQuerySet {
-			query := string(msg)
-			err = basenine.Validate(shared.BasenineHost, shared.BaseninePort, query)
-			if err != nil {
-				toastBytes, _ := models.CreateWebsocketToastMessage(&models.ToastMessage{
-					Type:      "error",
-					AutoClose: 5000,
-					Text:      fmt.Sprintf("Syntax error: %s", err.Error()),
-				})
-				SendToSocket(socketId, toastBytes)
-				break
-			}
-
-			isQuerySet = true
-
-			handleDataChannel := func(c *basenine.Connection, data chan []byte) {
-				for {
-					bytes := <-data
-
-					if string(bytes) == basenine.CloseChannel {
-						return
-					}
-
-					var dataMap map[string]interface{}
-					err = json.Unmarshal(bytes, &dataMap)
-
-					base := dataMap["base"].(map[string]interface{})
-					base["id"] = uint(dataMap["id"].(float64))
-
-					baseEntryBytes, _ := models.CreateBaseEntryWebSocketMessage(base)
-					SendToSocket(socketId, baseEntryBytes)
+		if !queryRecieved {
+			if !isTapper && !isQuerySet {
+				queryRecieved = true
+				query := string(msg)
+				err = basenine.Validate(shared.BasenineHost, shared.BaseninePort, query)
+				if err != nil {
+					toastBytes, _ := models.CreateWebsocketToastMessage(&models.ToastMessage{
+						Type:      "error",
+						AutoClose: 5000,
+						Text:      fmt.Sprintf("Syntax error: %s", err.Error()),
+					})
+					SendToSocket(socketId, toastBytes)
+					break
 				}
-			}
 
-			handleMetaChannel := func(c *basenine.Connection, meta chan []byte) {
-				for {
-					bytes := <-meta
+				isQuerySet = true
 
-					if string(bytes) == basenine.CloseChannel {
-						return
+				handleDataChannel := func(c *basenine.Connection, data chan []byte) {
+					for {
+						bytes := <-data
+
+						if string(bytes) == basenine.CloseChannel {
+							return
+						}
+
+						var dataMap map[string]interface{}
+						err = json.Unmarshal(bytes, &dataMap)
+
+						base := dataMap["base"].(map[string]interface{})
+						base["id"] = uint(dataMap["id"].(float64))
+
+						baseEntryBytes, _ := models.CreateBaseEntryWebSocketMessage(base)
+						SendToSocket(socketId, baseEntryBytes)
 					}
-
-					var metadata *basenine.Metadata
-					err = json.Unmarshal(bytes, &metadata)
-					if err != nil {
-						logger.Log.Debugf("Error recieving metadata: %v", err.Error())
-					}
-
-					metadataBytes, _ := models.CreateWebsocketQueryMetadataMessage(metadata)
-					SendToSocket(socketId, metadataBytes)
 				}
+
+				handleMetaChannel := func(c *basenine.Connection, meta chan []byte) {
+					for {
+						bytes := <-meta
+
+						if string(bytes) == basenine.CloseChannel {
+							return
+						}
+
+						var metadata *basenine.Metadata
+						err = json.Unmarshal(bytes, &metadata)
+						if err != nil {
+							logger.Log.Debugf("Error recieving metadata: %v", err.Error())
+						}
+
+						metadataBytes, _ := models.CreateWebsocketQueryMetadataMessage(metadata)
+						SendToSocket(socketId, metadataBytes)
+					}
+				}
+
+				go handleDataChannel(connection, data)
+				go handleMetaChannel(connection, meta)
+
+				connection.Query(query, data, meta)
+			} else {
+				eventHandlers.WebSocketMessage(socketId, msg)
 			}
-
-			go handleDataChannel(connection, data)
-			go handleMetaChannel(connection, meta)
-
-			connection.Query(query, data, meta)
 		} else {
-			eventHandlers.WebSocketMessage(socketId, msg)
+			id, err := strconv.Atoi(string(msg))
+			if err != nil {
+				continue
+			}
+			focusEntryBytes, _ := models.CreateWebsocketFocusEntry(id)
+			SendToSocket(socketId, focusEntryBytes)
 		}
 	}
 }

agent/pkg/models/models.go

@@ -57,6 +57,11 @@ type WebSocketStartTimeMessage struct {
 	Data int64 `json:"data"`
 }
 
+type WebSocketFocusEntryMessage struct {
+	*shared.WebSocketMessageMetadata
+	Id int `json:"id"`
+}
+
 func CreateBaseEntryWebSocketMessage(base map[string]interface{}) ([]byte, error) {
 	message := &WebSocketEntryMessage{
 		WebSocketMessageMetadata: &shared.WebSocketMessageMetadata{
@@ -117,6 +122,16 @@ func CreateWebsocketStartTimeMessage(base int64) ([]byte, error) {
 	return json.Marshal(message)
 }
 
+func CreateWebsocketFocusEntry(id int) ([]byte, error) {
+	message := &WebSocketFocusEntryMessage{
+		WebSocketMessageMetadata: &shared.WebSocketMessageMetadata{
+			MessageType: shared.WebSocketMessageFocusEntry,
+		},
+		Id: id,
+	}
+	return json.Marshal(message)
+}
+
 // ExtendedHAR is the top level object of a HAR log.
 type ExtendedHAR struct {
 	Log *ExtendedLog `json:"log"`

debug.Dockerfile

@@ -37,8 +37,8 @@ COPY agent .
 RUN go build -gcflags="all=-N -l" -o mizuagent .
 
 # Download Basenine executable, verify the sha1sum and move it to a directory in $PATH
-ADD https://github.com/up9inc/basenine/releases/download/v0.2.11/basenine_linux_amd64 ./basenine_linux_amd64
-ADD https://github.com/up9inc/basenine/releases/download/v0.2.11/basenine_linux_amd64.sha256 ./basenine_linux_amd64.sha256
+ADD https://github.com/up9inc/basenine/releases/download/v0.2.12/basenine_linux_amd64 ./basenine_linux_amd64
+ADD https://github.com/up9inc/basenine/releases/download/v0.2.12/basenine_linux_amd64.sha256 ./basenine_linux_amd64.sha256
 RUN shasum -a 256 -c basenine_linux_amd64.sha256
 RUN chmod +x ./basenine_linux_amd64
 

shared/models.go

@@ -1,11 +1,12 @@
 package shared
 
 import (
+	"io/ioutil"
+	"strings"
+
 	"github.com/op/go-logging"
 	"github.com/up9inc/mizu/shared/logger"
 	"github.com/up9inc/mizu/tap/api"
-	"io/ioutil"
-	"strings"
 
 	"gopkg.in/yaml.v3"
 )
@@ -21,6 +22,7 @@ const (
 	WebSocketMessageTypeToast         WebSocketMessageType = "toast"
 	WebSocketMessageTypeQueryMetadata WebSocketMessageType = "queryMetadata"
 	WebSocketMessageTypeStartTime     WebSocketMessageType = "startTime"
+	WebSocketMessageFocusEntry        WebSocketMessageType = "focusEntry"
 )
 
 type Resources struct {

ui/src/components/EntryDetailed.tsx

@@ -1,9 +1,8 @@
 import React from "react";
 import EntryViewer from "./EntryDetailed/EntryViewer";
+import {EntryItem} from "./EntryListItem/EntryListItem";
 import {makeStyles} from "@material-ui/core";
 import Protocol from "./UI/Protocol"
-import StatusCode from "./UI/StatusCode";
-import {Summary} from "./UI/Summary";
 
 const useStyles = makeStyles(() => ({
     entryTitle: {
@@ -12,6 +11,7 @@ const useStyles = makeStyles(() => ({
         maxHeight: 46,
         alignItems: 'center',
         marginBottom: 4,
+        marginLeft: 6,
         padding: 2,
         paddingBottom: 0
     },
@@ -64,18 +64,17 @@ const EntryTitle: React.FC<any> = ({protocol, data, bodySize, elapsedTime, updat
 };
 
 const EntrySummary: React.FC<any> = ({data, updateQuery}) => {
-    const classes = useStyles();
+    const entry = data.base;
 
-    const response = data.response;
-
-    return <div className={classes.entrySummary}>
-        {response && "status" in response && <div style={{marginRight: 8}}>
-            <StatusCode statusCode={response.status} updateQuery={updateQuery}/>
-        </div>}
-        <div style={{flexGrow: 1, overflow: 'hidden'}}>
-            <Summary method={data.method} summary={data.summary} updateQuery={updateQuery}/>
-        </div>
-    </div>;
+    return <EntryItem
+        key={entry.id}
+        entry={entry}
+        focusedEntryId={null}
+        setFocusedEntryId={null}
+        style={{}}
+        updateQuery={updateQuery}
+        headingMode={true}
+    />;
 };
 
 export const EntryDetailed: React.FC<EntryDetailedProps> = ({entryData, updateQuery}) => {

ui/src/components/EntryDetailed/EntryViewer.module.sass

@@ -4,6 +4,7 @@
     font-family: "Source Sans Pro", Lucida Grande, Tahoma, sans-serif
     height: calc(100% - 70px)
     width: 100%
+    margin-top: 10px
 
     h3,
     h4

ui/src/components/EntryListItem/EntryListItem.tsx

@@ -1,4 +1,4 @@
-import React, {useState} from "react";
+import React from "react";
 import styles from './EntryListItem.module.sass';
 import StatusCode, {getClassification, StatusCodeClassification} from "../UI/StatusCode";
 import Protocol, {ProtocolInterface} from "../UI/Protocol"
@@ -37,14 +37,16 @@ interface Rules {
 
 interface EntryProps {
     entry: Entry;
+    focusedEntryId: string;
     setFocusedEntryId: (id: string) => void;
     style: object;
     updateQuery: any;
+    headingMode: boolean;
 }
 
-export const EntryItem: React.FC<EntryProps> = ({entry, setFocusedEntryId, style, updateQuery}) => {
+export const EntryItem: React.FC<EntryProps> = ({entry, focusedEntryId, setFocusedEntryId, style, updateQuery, headingMode}) => {
 
-    const [isSelected, setIsSelected] = useState(false);
+    const isSelected = focusedEntryId === entry.id.toString();
 
     const classification = getClassification(entry.statusCode)
     const numberOfRules = entry.rules.numberOfRules
@@ -122,22 +124,22 @@ export const EntryItem: React.FC<EntryProps> = ({entry, setFocusedEntryId, style
             className={`${styles.row}
             ${isSelected && !rule && !contractEnabled ? styles.rowSelected : additionalRulesProperties}`}
             onClick={() => {
-                setIsSelected(!isSelected);
+                if (!setFocusedEntryId) return;
                 setFocusedEntryId(entry.id.toString());
             }}
             style={{
                 border: isSelected ? `1px ${entry.protocol.backgroundColor} solid` : "1px transparent solid",
-                position: "absolute",
+                position: !headingMode ? "absolute" : "unset",
                 top: style['top'],
                 marginTop: style['marginTop'],
-                width: "calc(100% - 25px)",
+                width: !headingMode ? "calc(100% - 25px)" : "calc(100% - 18px)",
             }}
         >
-            <Protocol
+            {!headingMode ? <Protocol
                 protocol={entry.protocol}
                 horizontal={false}
                 updateQuery={updateQuery}
-            />
+            /> : null}
             {((entry.protocol.name === "http" && "statusCode" in entry) || entry.statusCode !== 0) && <div>
                 <StatusCode statusCode={entry.statusCode} updateQuery={updateQuery}/>
             </div>}

ui/src/components/Filters.tsx

@@ -13,7 +13,7 @@ interface FiltersProps {
     setQuery: any
     backgroundColor: string
     ws: any
-    openWebSocket: (query: string) => void;
+    openWebSocket: (query: string, resetEntriesBuffer: boolean) => void;
 }
 
 export const Filters: React.FC<FiltersProps> = ({query, setQuery, backgroundColor, ws, openWebSocket}) => {
@@ -33,7 +33,7 @@ interface QueryFormProps {
     setQuery: any
     backgroundColor: string
     ws: any
-    openWebSocket: (query: string) => void;
+    openWebSocket: (query: string, resetEntriesBuffer: boolean) => void;
 }
 
 const style = {
@@ -63,8 +63,8 @@ export const QueryForm: React.FC<QueryFormProps> = ({query, setQuery, background
     }
 
     const handleSubmit = (e) => {
-        ws.close()
-        openWebSocket(query)
+        ws.close();
+        openWebSocket(query, true);
         e.preventDefault();
     }
 

ui/src/components/TrafficPage.tsx

@@ -66,6 +66,7 @@ export const TrafficPage: React.FC<TrafficPageProps> = ({setAnalyzeStatus, onTLS
 
     const [queriedCurrent, setQueriedCurrent] = useState(0);
     const [queriedTotal, setQueriedTotal] = useState(0);
+    const [leftOff, setLeftOff] = useState(0);
 
     const [startTime, setStartTime] = useState(0);
 
@@ -100,16 +101,30 @@ export const TrafficPage: React.FC<TrafficPageProps> = ({setAnalyzeStatus, onTLS
 
     const listEntry = useRef(null);
 
-    const openWebSocket = (query) => {
-        setFocusedEntryId(null);
-        setEntries([]);
-        setEntriesBuffer([]);
+    const openWebSocket = (query: string, resetEntriesBuffer: boolean) => {
+        if (resetEntriesBuffer) {
+            setFocusedEntryId(null);
+            setEntries([]);
+            setEntriesBuffer([]);
+        } else {
+            setEntriesBuffer(entries);
+        }
         ws.current = new WebSocket(MizuWebsocketURL);
         ws.current.onopen = () => {
-            ws.current.send(query)
             setConnection(ConnectionStatus.Connected);
+            ws.current.send(query);
+        }
+        ws.current.onclose = () => {
+            setConnection(ConnectionStatus.Closed);
+        }
+        ws.current.onerror = (event) => {
+            console.error("WebSocket error:", event);
+            if (query) {
+                openWebSocket(`(${query}) and leftOff(${leftOff})`, false);
+            } else {
+                openWebSocket(`leftOff(${leftOff})`, false);
+            }
         }
-        ws.current.onclose = () => setConnection(ConnectionStatus.Closed);
     }
 
     if (ws.current) {
@@ -119,15 +134,21 @@ export const TrafficPage: React.FC<TrafficPageProps> = ({setAnalyzeStatus, onTLS
             switch (message.messageType) {
                 case "entry":
                     const entry = message.data;
-                    if (!focusedEntryId) setFocusedEntryId(entry.id.toString());
+                    var focusThis = false;
+                    if (!focusedEntryId) {
+                        focusThis = true;
+                        setFocusedEntryId(entry.id.toString());
+                    }
                     setEntriesBuffer([
                         ...entriesBuffer,
                         <EntryItem
                             key={entry.id}
                             entry={entry}
+                            focusedEntryId={focusThis ? entry.id.toString() : focusedEntryId}
                             setFocusedEntryId={setFocusedEntryId}
                             style={{}}
                             updateQuery={updateQuery}
+                            headingMode={false}
                         />
                     ]);
                     break
@@ -155,11 +176,22 @@ export const TrafficPage: React.FC<TrafficPageProps> = ({setAnalyzeStatus, onTLS
                 case "queryMetadata":
                     setQueriedCurrent(message.data.current);
                     setQueriedTotal(message.data.total);
+                    setLeftOff(message.data.leftOff);
                     setEntries(entriesBuffer);
                     break;
                 case "startTime":
                     setStartTime(message.data);
                     break;
+                case "focusEntry":
+                    // To achieve selecting only one entry, render all elements in the buffer
+                    // with the current `focusedEntryId` value.
+                    entriesBuffer.forEach((entry: any, i: number) => {
+                        entriesBuffer[i] = React.cloneElement(entry, {
+                            focusedEntryId: focusedEntryId
+                        });
+                    })
+                    setEntries(entriesBuffer);
+                    break;
                 default:
                     console.error(`unsupported websocket message type, Got: ${message.messageType}`)
             }
@@ -168,7 +200,7 @@ export const TrafficPage: React.FC<TrafficPageProps> = ({setAnalyzeStatus, onTLS
 
     useEffect(() => {
         (async () => {
-            openWebSocket("rlimit(100)");
+            openWebSocket("leftOff(-1)", true);
             try{
                 const tapStatusResponse = await api.tapStatus();
                 setTappingStatus(tapStatusResponse);
@@ -185,21 +217,28 @@ export const TrafficPage: React.FC<TrafficPageProps> = ({setAnalyzeStatus, onTLS
     useEffect(() => {
         if (!focusedEntryId) return;
         setSelectedEntryData(null);
+
+        if (ws.current.readyState === WebSocket.OPEN) {
+            ws.current.send(focusedEntryId);
+        }
+
         (async () => {
             try {
                 const entryData = await api.getEntry(focusedEntryId);
                 setSelectedEntryData(entryData);
             } catch (error) {
-                toast[error.response.data.type](`Entry[${focusedEntryId}]: ${error.response.data.msg}`, {
-                    position: "bottom-right",
-                    theme: "colored",
-                    autoClose: error.response.data.autoClose,
-                    hideProgressBar: false,
-                    closeOnClick: true,
-                    pauseOnHover: true,
-                    draggable: true,
-                    progress: undefined,
-                });
+                if (error.response) {
+                    toast[error.response.data.type](`Entry[${focusedEntryId}]: ${error.response.data.msg}`, {
+                        position: "bottom-right",
+                        theme: "colored",
+                        autoClose: error.response.data.autoClose,
+                        hideProgressBar: false,
+                        closeOnClick: true,
+                        pauseOnHover: true,
+                        draggable: true,
+                        progress: undefined,
+                    });
+                }
                 console.error(error);
             }
         })()
@@ -209,8 +248,11 @@ export const TrafficPage: React.FC<TrafficPageProps> = ({setAnalyzeStatus, onTLS
         if (connection === ConnectionStatus.Connected) {
             ws.current.close();
         } else {
-            openWebSocket(query);
-            setConnection(ConnectionStatus.Connected);
+            if (query) {
+                openWebSocket(`(${query}) and leftOff(${leftOff})`, false);
+            } else {
+                openWebSocket(`leftOff(${leftOff})`, false);
+            }
         }
     }