Migrate from SQLite to Basenine and introduce a new filtering syntax (#279)

* Fix the OOMKilled error by calling `debug.FreeOSMemory` periodically

* Remove `MAX_NUMBER_OF_GOROUTINES` environment variable

* Change the line

* Increase the default value of `TCP_STREAM_CHANNEL_TIMEOUT_MS` to `10000`

* Write the client and integrate to the new real-time database

* Refactor the WebSocket implementaiton for `/ws`

* Adapt the UI to the new filtering system

* Fix the rest of the issues in the UI

* Increase the buffer of the scanner

* Implement accessing single records

* Increase the buffer of another scanner

* Populate `Request` and `Response` fields of `MizuEntry`

* Add syntax highlighting for the query

* Add database to `Dockerfile`

* Fix some issues

* Update the `realtime_dbms` Git module commit hash

* Upgrade Gin version and print the query string

* Revert "Upgrade Gin version and print the query string"

This reverts commit aa09f904ee.

* Use WebSocket's itself to query instead of the query string

* Fix some errors related to conversion to HAR

* Fix the issues caused by the latest merge

* Fix the build error

* Fix PR validation GitHub workflow

* Replace the git submodule with latest Basenine version `0.1.0`

Remove `realtime_client.go` and use the official client library `github.com/up9inc/basenine/client/go` instead.

* Move Basenine host and port constants to `shared` module

* Reliably execute and wait for Basenine to become available

* Upgrade Basenine version

* Properly close WebSocket and data channel

* Fix the issues caused by the recent merge commit

* Clean up the TypeScript code

* Update `.gitignore`

* Limit the database size

* Add `Macros` method signature to `Dissector` interface and set the macros provided by the protocol extensions

* Run `go mod tidy` on `agent`

* Upgrade `github.com/up9inc/basenine/client/go` version

* Implement a mechanism to update the query using click events in the UI and use it for protocol macros

* Update the query on click to timestamps

* Fix some issues in the WebSocket and channel handling

* Update the query on clicks to status code

* Update the query on clicks to method, path and service

* Update the query on clicks to is outgoing, source and destination ports

* Add an API endpoint to validate the query against syntax errors

* Move the query background color state into `TrafficPage`

* Fix the logic in `setQuery`

* Display a toast message in case of a syntax error in the query

* Remove a call to `fmt.Printf`

* Upgrade Basenine version to `0.1.3`

* Fix an issue related to getting `MAX_ENTRIES_DB_BYTES` environment variable

* Have the `path` key in request details, in HTTP

* Rearrange the HTTP headers for the querying

* Do the same thing for `cookies` and `queryString`

* Update the query on click to table elements

Add the selectors for `TABLE` type representations in HTTP extension.

* Update the query on click to `bodySize` and `elapsedTime` in `EntryTitle`

* Add the selectors for `TABLE` type representations in AMQP extension

* Add the selectors for `TABLE` type representations in Kafka extension

* Add the selectors for `TABLE` type representations in Redis extension

* Define a struct in `tap/api.go` for the section representation data

* Add the selectors for `BODY` type representations

* Add `request.path` to the HTTP request details

* Change the summary string's field name from `path` to `summary`

* Introduce `queryable` CSS class for queryable UI elements and underline them on hover

* Instead of `N requests` at the bottom, make it `Displaying N results (queried X/Y)` and live update the values

Upgrade Basenine version to `0.2.0`.

* Verify the sha256sum of Basenine executable inside `Dockerfile`

* Pass the start time to web UI through WebSocket and always show the `EntriesList` footer

* Pipe the `stderr` of Basenine as well

* Fix the layout issues related to `CodeEditor` in the UI

* Use the correct `shasum` command in `Dockerfile`

* Upgrade Basenine version to `0.2.1`

* Limit the height of `CodeEditor` container

* Remove `Paused` enum `ConnectionStatus` in UI

* Fix the issue caused by the recent merge

* Add the filtering guide (cheatsheet)

* Update open cheatsheet button's title

* Update cheatsheet content

* Remove the old SQLite code, adapt the `--analyze` related code to Basenine

* Change the method signature of `NewEntry`

* Change the method signature of `Represent`

* Introduce `HTTPPair` field in `MizuEntry` specific to HTTP

* Remove `Entry`, `EntryId` and `EstimatedSizeBytes` fields from `MizuEntry`

Also remove the `getEstimatedEntrySizeBytes` method.

* Remove `gorm.io/gorm` dependency

* Remove unused `sensitiveDataFiltering` folder

* Increase the left margin of open cheatsheet button

* Add `overflow: auto` to the cheatsheet `Modal`

* Fix `GetEntry` method

* Fix the macro for gRPC

* Fix an interface conversion in case of AMQP

* Fix two more interface conversion errors in AMQP

* Make the `syncEntriesImpl` method blocking

* Fix a grammar mistake in the cheatsheet

* Adapt to the changes in the recent merge commit

* Improve the cheatsheet text

* Always display the timestamp in `en-US`

* Upgrade Basenine version to `0.2.2`

* Fix the order of closing Basenine connections and channels

* Don't close the Basenine channels at all

* Upgrade Basenine version to `0.2.3`

* Set the initial filter to `rlimit(100)`

* Make Basenine persistent

* Upgrade Basenine version to `0.2.4`

* Update `debug.Dockerfile`

* Fix a failing test

* Upgrade Basenine version to `0.2.5`

* Revert "Do not show play icon when disconnected (#428)"

This reverts commit 8af2e562f8.

* Upgrade Basenine version to `0.2.6`

* Make all non-informative things informative

* Make `100` a constant

* Use `===` in JavaScript no matter what

* Remove a forgotten `console.log`

* Add a comment and update the `query` in `syncEntriesImpl`

* Don't call `panic` in `GetEntry`

* Replace `panic` calls in `startBasenineServer` with `logger.Log.Panicf`

* Remove unnecessary `\n` characters in the logs

agent/pkg/up9/main.go

@@ -7,15 +7,16 @@ import (
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
-	"mizuserver/pkg/database"
 	"mizuserver/pkg/utils"
 	"net/http"
 	"net/url"
 	"regexp"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/google/martian/har"
+	basenine "github.com/up9inc/basenine/client/go"
 	"github.com/up9inc/mizu/shared"
 	"github.com/up9inc/mizu/shared/logger"
 	tapApi "github.com/up9inc/mizu/tap/api"
@@ -23,6 +24,7 @@ import (
 
 const (
 	AnalyzeCheckSleepTime = 5 * time.Second
+	SentCountLogInterval  = 100
 )
 
 type GuestToken struct {
@@ -204,44 +206,62 @@ func syncEntriesImpl(token string, model string, envPrefix string, uploadInterva
 	analyzeInformation.AnalyzeDestination = envPrefix
 	analyzeInformation.SentCount = 0
 
-	sleepTime := time.Second * time.Duration(uploadIntervalSec)
+	// "http or grpc" filter indicates that we're only interested in HTTP and gRPC entries
+	query := "http or grpc"
 
-	var timeFrom time.Time
-	protocolFilter := "http"
+	logger.Log.Infof("Getting entries from the database\n")
 
-	for {
-		timeTo := time.Now()
-		logger.Log.Infof("Getting entries from %v, to %v\n", timeFrom.Format(time.RFC3339Nano), timeTo.Format(time.RFC3339Nano))
-		entriesArray := database.GetEntriesFromDb(timeFrom, timeTo, &protocolFilter)
+	var connection *basenine.Connection
+	var err error
+	connection, err = basenine.NewConnection(shared.BasenineHost, shared.BaseninePort)
+	if err != nil {
+		panic(err)
+	}
 
-		if len(entriesArray) > 0 {
-			result := make([]har.Entry, 0)
-			for _, data := range entriesArray {
-				var pair tapApi.RequestResponsePair
-				if err := json.Unmarshal([]byte(data.Entry), &pair); err != nil {
-					continue
-				}
-				harEntry, err := utils.NewEntry(&pair)
-				if err != nil {
-					continue
-				}
-				if data.ResolvedSource != "" {
-					harEntry.Request.Headers = append(harEntry.Request.Headers, har.Header{Name: "x-mizu-source", Value: data.ResolvedSource})
-				}
-				if data.ResolvedDestination != "" {
-					harEntry.Request.Headers = append(harEntry.Request.Headers, har.Header{Name: "x-mizu-destination", Value: data.ResolvedDestination})
-					harEntry.Request.URL = utils.SetHostname(harEntry.Request.URL, data.ResolvedDestination)
-				}
+	data := make(chan []byte)
+	meta := make(chan []byte)
 
-				// go's default marshal behavior is to encode []byte fields to base64, python's default unmarshal behavior is to not decode []byte fields from base64
-				if harEntry.Response.Content.Text, err = base64.StdEncoding.DecodeString(string(harEntry.Response.Content.Text)); err != nil {
-					continue
-				}
+	defer func() {
+		data <- []byte(basenine.CloseChannel)
+		meta <- []byte(basenine.CloseChannel)
+		connection.Close()
+	}()
 
-				result = append(result, *harEntry)
+	handleDataChannel := func(wg *sync.WaitGroup, connection *basenine.Connection, data chan []byte) {
+		defer wg.Done()
+		for {
+			dataBytes := <-data
+
+			if string(dataBytes) == basenine.CloseChannel {
+				return
 			}
 
-			logger.Log.Infof("About to upload %v entries\n", len(result))
+			var dataMap map[string]interface{}
+			err = json.Unmarshal(dataBytes, &dataMap)
+
+			result := make([]har.Entry, 0)
+			var entry tapApi.MizuEntry
+			if err := json.Unmarshal([]byte(dataBytes), &entry); err != nil {
+				continue
+			}
+			harEntry, err := utils.NewEntry(entry.Request, entry.Response, entry.StartTime, entry.ElapsedTime)
+			if err != nil {
+				continue
+			}
+			if entry.ResolvedSource != "" {
+				harEntry.Request.Headers = append(harEntry.Request.Headers, har.Header{Name: "x-mizu-source", Value: entry.ResolvedSource})
+			}
+			if entry.ResolvedDestination != "" {
+				harEntry.Request.Headers = append(harEntry.Request.Headers, har.Header{Name: "x-mizu-destination", Value: entry.ResolvedDestination})
+				harEntry.Request.URL = utils.SetHostname(harEntry.Request.URL, entry.ResolvedDestination)
+			}
+
+			// go's default marshal behavior is to encode []byte fields to base64, python's default unmarshal behavior is to not decode []byte fields from base64
+			if harEntry.Response.Content.Text, err = base64.StdEncoding.DecodeString(string(harEntry.Response.Content.Text)); err != nil {
+				continue
+			}
+
+			result = append(result, *harEntry)
 
 			body, jMarshalErr := json.Marshal(result)
 			if jMarshalErr != nil {
@@ -273,18 +293,33 @@ func syncEntriesImpl(token string, model string, envPrefix string, uploadInterva
 				logger.Log.Info("Stopping sync entries")
 				logger.Log.Fatal(postErr)
 			}
-			analyzeInformation.SentCount += len(entriesArray)
-			logger.Log.Infof("Finish uploading %v entries to %s\n", len(entriesArray), GetTrafficDumpUrl(envPrefix, model))
+			analyzeInformation.SentCount += 1
 
-			logger.Log.Infof("Uploaded %v entries until now", analyzeInformation.SentCount)
-		} else {
-			logger.Log.Infof("Nothing to upload")
+			if analyzeInformation.SentCount%SentCountLogInterval == 0 {
+				logger.Log.Infof("Uploaded %v entries until now", analyzeInformation.SentCount)
+			}
 		}
-
-		logger.Log.Infof("Sleeping for %v...\n", sleepTime)
-		time.Sleep(sleepTime)
-		timeFrom = timeTo
 	}
+
+	handleMetaChannel := func(wg *sync.WaitGroup, connection *basenine.Connection, meta chan []byte) {
+		defer wg.Done()
+		for {
+			metaBytes := <-meta
+
+			if string(metaBytes) == basenine.CloseChannel {
+				return
+			}
+		}
+	}
+
+	var wg sync.WaitGroup
+	go handleDataChannel(&wg, connection, data)
+	go handleMetaChannel(&wg, connection, meta)
+	wg.Add(2)
+
+	connection.Query(query, data, meta)
+
+	wg.Wait()
 }
 
 func UpdateAnalyzeStatus(callback func(data []byte)) {