Implement accessing single records

agent/pkg/api/main.go

@@ -5,7 +5,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"mizuserver/pkg/database"
 	"mizuserver/pkg/holder"
 	"net/url"
 	"os"
@@ -116,9 +115,8 @@ func startReadingChannel(outputItems <-chan *tapApi.OutputChannelItem, extension
 		mizuEntry := extension.Dissector.Analyze(item, primitive.NewObjectID().Hex(), resolvedSource, resolvedDestionation)
 		baseEntry := extension.Dissector.Summarize(mizuEntry)
 		mizuEntry.EstimatedSizeBytes = getEstimatedEntrySizeBytes(mizuEntry)
-		database.CreateEntry(mizuEntry)
+		mizuEntry.Summary = baseEntry
-		item.Summary = baseEntry
+		Insert(mizuEntry, conn)
-		Insert(item, conn)
 	}
 }
 

agent/pkg/api/realtime_client.go

@@ -47,6 +47,49 @@ func Insert(entry interface{}, conn net.Conn) {
 	conn.Write([]byte("\n"))
 }
 
+func Single(index uint) (entry map[string]interface{}) {
+	conn := Connect("localhost", "8000")
+
+	var wg sync.WaitGroup
+	go readConnectionSingle(&wg, conn, &entry)
+	wg.Add(1)
+
+	conn.SetWriteDeadline(time.Now().Add(1 * time.Second))
+	conn.Write([]byte("/single\n"))
+
+	conn.SetWriteDeadline(time.Now().Add(1 * time.Second))
+	conn.Write([]byte(fmt.Sprintf("%d\n", index)))
+
+	wg.Wait()
+	return
+}
+
+func readConnectionSingle(wg *sync.WaitGroup, conn net.Conn, entry *map[string]interface{}) {
+	defer wg.Done()
+	for {
+		scanner := bufio.NewScanner(conn)
+
+		for {
+			ok := scanner.Scan()
+			text := scanner.Text()
+
+			command := handleCommands(text)
+			if !command {
+				fmt.Printf("\b\b** %s\n> ", text)
+				if err := json.Unmarshal([]byte(text), entry); err != nil {
+					panic(err)
+				}
+				return
+			}
+
+			if !ok {
+				fmt.Println("Reached EOF on server connection.")
+				break
+			}
+		}
+	}
+}
+
 func Query(query string, conn net.Conn, ws *websocket.Conn) {
 	var wg sync.WaitGroup
 	go readConnection(&wg, conn, ws)
@@ -87,7 +130,10 @@ func readConnection(wg *sync.WaitGroup, conn net.Conn, ws *websocket.Conn) {
 					panic(err)
 				}
 
-				baseEntryBytes, _ := models.CreateBaseEntryWebSocketMessage(data["Summary"].(map[string]interface{}))
+				summary := data["summary"].(map[string]interface{})
+				summary["id"] = uint(data["id"].(float64))
+
+				baseEntryBytes, _ := models.CreateBaseEntryWebSocketMessage(summary)
 				ws.WriteMessage(1, baseEntryBytes)
 			}
 

agent/pkg/controllers/entries_controller.go

@@ -3,7 +3,7 @@ package controllers
 import (
 	"encoding/json"
 	"fmt"
-	"github.com/google/martian/har"
+	"mizuserver/pkg/api"
 	"mizuserver/pkg/database"
 	"mizuserver/pkg/models"
 	"mizuserver/pkg/providers"
@@ -11,8 +11,11 @@ import (
 	"mizuserver/pkg/utils"
 	"mizuserver/pkg/validation"
 	"net/http"
+	"strconv"
 	"time"
 
+	"github.com/google/martian/har"
+
 	"github.com/gin-gonic/gin"
 	"github.com/romana/rlog"
 
@@ -133,12 +136,41 @@ func GetFullEntries(c *gin.Context) {
 }
 
 func GetEntry(c *gin.Context) {
-	var entryData tapApi.MizuEntry
+	id, _ := strconv.Atoi(c.Param("entryId"))
-	database.GetEntriesTable().
+	fmt.Printf("GetEntry id: %v\n", id)
-		Where(map[string]string{"entryId": c.Param("entryId")}).
+	entry := api.Single(uint(id))
-		First(&entryData)
+	entryData := tapApi.MizuEntry{
+		Protocol: tapApi.Protocol{
+			Name:            entry["proto"].(map[string]interface{})["name"].(string),
+			LongName:        entry["proto"].(map[string]interface{})["longName"].(string),
+			Abbreviation:    entry["proto"].(map[string]interface{})["abbreviation"].(string),
+			Version:         entry["proto"].(map[string]interface{})["version"].(string),
+			BackgroundColor: entry["proto"].(map[string]interface{})["backgroundColor"].(string),
+			ForegroundColor: entry["proto"].(map[string]interface{})["foregroundColor"].(string),
+			FontSize:        int8(entry["proto"].(map[string]interface{})["fontSize"].(float64)),
+			ReferenceLink:   entry["proto"].(map[string]interface{})["referenceLink"].(string),
+			Priority:        uint8(entry["proto"].(map[string]interface{})["priority"].(float64)),
+		},
+		EntryId:         entry["entryId"].(string),
+		Entry:           entry["entry"].(string),
+		Url:             entry["url"].(string),
+		Method:          entry["method"].(string),
+		Status:          int(entry["status"].(float64)),
+		RequestSenderIp: entry["requestSenderIp"].(string),
+		Service:         entry["service"].(string),
+		Timestamp:       int64(entry["timestamp"].(float64)),
+		ElapsedTime:     int64(entry["elapsedTime"].(float64)),
+		Path:            entry["path"].(string),
+		// ResolvedSource:      entry["resolvedSource"].(string),
+		// ResolvedDestination: entry["resolvedDestination"].(string),
+		SourceIp:        entry["sourceIp"].(string),
+		DestinationIp:   entry["destinationIp"].(string),
+		SourcePort:      entry["sourcePort"].(string),
+		DestinationPort: entry["destinationPort"].(string),
+		// IsOutgoing:      entry["isOutgoing"].(bool),
+	}
 
-	extension := extensionsMap[entryData.ProtocolName]
+	extension := extensionsMap[entryData.Protocol.Name]
 	protocol, representation, bodySize, _ := extension.Dissector.Represent(&entryData)
 	c.JSON(http.StatusOK, tapApi.MizuEntryWrapper{
 		Protocol:       protocol,

agent/pkg/database/main.go

@@ -3,11 +3,8 @@ package database
 import (
 	"fmt"
 	"mizuserver/pkg/utils"
-	"time"
 
-	"gorm.io/driver/sqlite"
 	"gorm.io/gorm"
-	"gorm.io/gorm/logger"
 
 	tapApi "github.com/up9inc/mizu/tap/api"
 )
@@ -34,28 +31,18 @@ var (
 )
 
 func init() {
-	DB = initDataBase(DBPath)
-	go StartEnforcingDatabaseSize()
 }
 
 func GetEntriesTable() *gorm.DB {
 	return DB.Table("mizu_entries")
 }
 
-func CreateEntry(entry *tapApi.MizuEntry) {
+// func CreateEntry(entry *tapApi.MizuEntry) {
-	if IsDBLocked {
+// 	if IsDBLocked {
-		return
+// 		return
-	}
+// 	}
-	GetEntriesTable().Create(entry)
+// 	GetEntriesTable().Create(entry)
-}
+// }
-
-func initDataBase(databasePath string) *gorm.DB {
-	temp, _ := gorm.Open(sqlite.Open(databasePath), &gorm.Config{
-		Logger: &utils.TruncatingLogger{LogLevel: logger.Warn, SlowThreshold: 500 * time.Millisecond},
-	})
-	_ = temp.AutoMigrate(&tapApi.MizuEntry{}) // this will ensure table is created
-	return temp
-}
 
 func GetEntriesFromDb(timestampFrom int64, timestampTo int64, protocolName *string) []tapApi.MizuEntry {
 	order := OrderDesc

agent/pkg/database/size_enforcer.go

@@ -1,122 +0,0 @@
-package database
-
-import (
-	"log"
-	"os"
-	"strconv"
-	"time"
-
-	"github.com/fsnotify/fsnotify"
-	"github.com/romana/rlog"
-	"github.com/up9inc/mizu/shared"
-	"github.com/up9inc/mizu/shared/debounce"
-	"github.com/up9inc/mizu/shared/units"
-	tapApi "github.com/up9inc/mizu/tap/api"
-)
-
-const percentageOfMaxSizeBytesToPrune = 15
-const defaultMaxDatabaseSizeBytes int64 = 200 * 1000 * 1000
-
-func StartEnforcingDatabaseSize() {
-	watcher, err := fsnotify.NewWatcher()
-	if err != nil {
-		log.Fatalf("Error creating filesystem watcher for db size enforcement: %v\n", err)
-		return
-	}
-
-	maxEntriesDBByteSize, err := getMaxEntriesDBByteSize()
-	if err != nil {
-		log.Fatalf("Error parsing max db size: %v\n", err)
-		return
-	}
-
-	checkFileSizeDebouncer := debounce.NewDebouncer(5*time.Second, func() {
-		checkFileSize(maxEntriesDBByteSize)
-	})
-
-	go func() {
-		for {
-			select {
-			case event, ok := <-watcher.Events:
-				if !ok {
-					return // closed channel
-				}
-				if event.Op == fsnotify.Write {
-					checkFileSizeDebouncer.SetOn()
-				}
-			case err, ok := <-watcher.Errors:
-				if !ok {
-					return // closed channel
-				}
-				rlog.Errorf("filesystem watcher encountered error:%v", err)
-			}
-		}
-	}()
-
-	err = watcher.Add(DBPath)
-	if err != nil {
-		log.Fatalf("Error adding %s to filesystem watcher for db size enforcement: %v\n", DBPath, err)
-	}
-}
-
-func getMaxEntriesDBByteSize() (int64, error) {
-	maxEntriesDBByteSize := defaultMaxDatabaseSizeBytes
-	var err error
-
-	maxEntriesDBSizeByteSEnvVarValue := os.Getenv(shared.MaxEntriesDBSizeBytesEnvVar)
-	if maxEntriesDBSizeByteSEnvVarValue != "" {
-		maxEntriesDBByteSize, err = strconv.ParseInt(maxEntriesDBSizeByteSEnvVarValue, 10, 64)
-	}
-	return maxEntriesDBByteSize, err
-}
-
-func checkFileSize(maxSizeBytes int64) {
-	fileStat, err := os.Stat(DBPath)
-	if err != nil {
-		rlog.Errorf("Error checking %s file size: %v", DBPath, err)
-	} else {
-		if fileStat.Size() > maxSizeBytes {
-			pruneOldEntries(fileStat.Size())
-		}
-	}
-}
-
-func pruneOldEntries(currentFileSize int64) {
-	// sqlite locks the database while delete or VACUUM are running and sqlite is terrible at handling its own db lock while a lot of inserts are attempted, we prevent a significant bottleneck by handling the db lock ourselves here
-	IsDBLocked = true
-	defer func() { IsDBLocked = false }()
-
-	amountOfBytesToTrim := currentFileSize / (100 / percentageOfMaxSizeBytesToPrune)
-
-	rows, err := GetEntriesTable().Limit(10000).Order("id").Rows()
-	if err != nil {
-		rlog.Errorf("Error getting 10000 first db rows: %v", err)
-		return
-	}
-
-	entryIdsToRemove := make([]uint, 0)
-	bytesToBeRemoved := int64(0)
-	for rows.Next() {
-		if bytesToBeRemoved >= amountOfBytesToTrim {
-			break
-		}
-		var entry tapApi.MizuEntry
-		err = DB.ScanRows(rows, &entry)
-		if err != nil {
-			rlog.Errorf("Error scanning db row: %v", err)
-			continue
-		}
-
-		entryIdsToRemove = append(entryIdsToRemove, entry.ID)
-		bytesToBeRemoved += int64(entry.EstimatedSizeBytes)
-	}
-
-	if len(entryIdsToRemove) > 0 {
-		GetEntriesTable().Where(entryIdsToRemove).Delete(tapApi.MizuEntry{})
-		// VACUUM causes sqlite to shrink the db file after rows have been deleted, the db file will not shrink without this
-		DB.Exec("VACUUM")
-		rlog.Errorf("Removed %d rows and cleared %s", len(entryIdsToRemove), units.BytesToHumanReadable(bytesToBeRemoved))
-	} else {
-		rlog.Error("Found no rows to remove when pruning")
-	}
-}

tap/api/api.go

@@ -102,25 +102,19 @@ func (e *Emitting) Emit(item *OutputChannelItem) {
 }
 
 type MizuEntry struct {
-	ID                      uint `gorm:"primarykey"`
+	Id                  uint              `json:"id"`
-	CreatedAt               time.Time
+	Protocol            Protocol          `json:"proto"`
-	UpdatedAt               time.Time
+	Timestamp           int64             `json:"timestamp"`
-	ProtocolName            string `json:"protocolName" gorm:"column:protocolName"`
+	Request             interface{}       `json:"request"`
-	ProtocolLongName        string `json:"protocolLongName" gorm:"column:protocolLongName"`
+	Response            interface{}       `json:"response"`
-	ProtocolAbbreviation    string `json:"protocolAbbreviation" gorm:"column:protocolVersion"`
+	Summary             *BaseEntryDetails `json:"summary"`
-	ProtocolVersion         string `json:"protocolVersion" gorm:"column:protocolVersion"`
+	Entry               string            `json:"entry,omitempty"`
-	ProtocolBackgroundColor string `json:"protocolBackgroundColor" gorm:"column:protocolBackgroundColor"`
-	ProtocolForegroundColor string `json:"protocolForegroundColor" gorm:"column:protocolForegroundColor"`
-	ProtocolFontSize        int8   `json:"protocolFontSize" gorm:"column:protocolFontSize"`
-	ProtocolReferenceLink   string `json:"protocolReferenceLink" gorm:"column:protocolReferenceLink"`
-	Entry                   string `json:"entry,omitempty" gorm:"column:entry"`
 	EntryId             string            `json:"entryId" gorm:"column:entryId"`
 	Url                 string            `json:"url" gorm:"column:url"`
 	Method              string            `json:"method" gorm:"column:method"`
 	Status              int               `json:"status" gorm:"column:status"`
 	RequestSenderIp     string            `json:"requestSenderIp" gorm:"column:requestSenderIp"`
 	Service             string            `json:"service" gorm:"column:service"`
-	Timestamp               int64  `json:"timestamp" gorm:"column:timestamp"`
 	ElapsedTime         int64             `json:"elapsedTime" gorm:"column:elapsedTime"`
 	Path                string            `json:"path" gorm:"column:path"`
 	ResolvedSource      string            `json:"resolvedSource,omitempty" gorm:"column:resolvedSource"`
@@ -141,7 +135,7 @@ type MizuEntryWrapper struct {
 }
 
 type BaseEntryDetails struct {
-	Id              string          `json:"id,omitempty"`
+	Id              uint            `json:"id"`
 	Protocol        Protocol        `json:"protocol,omitempty"`
 	Url             string          `json:"url,omitempty"`
 	RequestSenderIp string          `json:"requestSenderIp,omitempty"`
@@ -171,17 +165,8 @@ type DataUnmarshaler interface {
 }
 
 func (bed *BaseEntryDetails) UnmarshalData(entry *MizuEntry) error {
-	bed.Protocol = Protocol{
+	bed.Protocol = entry.Protocol
-		Name:            entry.ProtocolName,
+	bed.Id = entry.Id
-		LongName:        entry.ProtocolLongName,
-		Abbreviation:    entry.ProtocolAbbreviation,
-		Version:         entry.ProtocolVersion,
-		BackgroundColor: entry.ProtocolBackgroundColor,
-		ForegroundColor: entry.ProtocolForegroundColor,
-		FontSize:        entry.ProtocolFontSize,
-		ReferenceLink:   entry.ProtocolReferenceLink,
-	}
-	bed.Id = entry.EntryId
 	bed.Url = entry.Url
 	bed.Service = entry.Service
 	bed.Summary = entry.Path

tap/extensions/amqp/main.go

@@ -267,14 +267,7 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, entryId string, resolve
 	request["url"] = summary
 	entryBytes, _ := json.Marshal(item.Pair)
 	return &api.MizuEntry{
-		ProtocolName:            protocol.Name,
+		Protocol:            protocol,
-		ProtocolLongName:        protocol.LongName,
-		ProtocolAbbreviation:    protocol.Abbreviation,
-		ProtocolVersion:         protocol.Version,
-		ProtocolBackgroundColor: protocol.BackgroundColor,
-		ProtocolForegroundColor: protocol.ForegroundColor,
-		ProtocolFontSize:        protocol.FontSize,
-		ProtocolReferenceLink:   protocol.ReferenceLink,
 		EntryId:             entryId,
 		Entry:               string(entryBytes),
 		Url:                 fmt.Sprintf("%s%s", service, summary),
@@ -298,7 +291,7 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, entryId string, resolve
 
 func (d dissecting) Summarize(entry *api.MizuEntry) *api.BaseEntryDetails {
 	return &api.BaseEntryDetails{
-		Id:              entry.EntryId,
+		Id:              entry.Id,
 		Protocol:        protocol,
 		Url:             entry.Url,
 		RequestSenderIp: entry.RequestSenderIp,

tap/extensions/http/main.go

@@ -171,15 +171,10 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, entryId string, resolve
 
 	elapsedTime := item.Pair.Response.CaptureTime.Sub(item.Pair.Request.CaptureTime).Round(time.Millisecond).Milliseconds()
 	entryBytes, _ := json.Marshal(item.Pair)
+	_protocol := protocol
+	_protocol.Version = item.Protocol.Version
 	return &api.MizuEntry{
-		ProtocolName:            protocol.Name,
+		Protocol:            _protocol,
-		ProtocolLongName:        protocol.LongName,
-		ProtocolAbbreviation:    protocol.Abbreviation,
-		ProtocolVersion:         item.Protocol.Version,
-		ProtocolBackgroundColor: protocol.BackgroundColor,
-		ProtocolForegroundColor: protocol.ForegroundColor,
-		ProtocolFontSize:        protocol.FontSize,
-		ProtocolReferenceLink:   protocol.ReferenceLink,
 		EntryId:             entryId,
 		Entry:               string(entryBytes),
 		Url:                 fmt.Sprintf("%s%s", service, path),
@@ -202,13 +197,13 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, entryId string, resolve
 
 func (d dissecting) Summarize(entry *api.MizuEntry) *api.BaseEntryDetails {
 	var p api.Protocol
-	if entry.ProtocolVersion == "2.0" {
+	if entry.Protocol.Version == "2.0" {
 		p = http2Protocol
 	} else {
 		p = protocol
 	}
 	return &api.BaseEntryDetails{
-		Id:              entry.EntryId,
+		Id:              entry.Id,
 		Protocol:        p,
 		Url:             entry.Url,
 		RequestSenderIp: entry.RequestSenderIp,
@@ -377,7 +372,7 @@ func representResponse(response map[string]interface{}) (repResponse []interface
 }
 
 func (d dissecting) Represent(entry *api.MizuEntry) (p api.Protocol, object []byte, bodySize int64, err error) {
-	if entry.ProtocolVersion == "2.0" {
+	if entry.Protocol.Version == "2.0" {
 		p = http2Protocol
 	} else {
 		p = protocol

tap/extensions/kafka/main.go

@@ -142,14 +142,7 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, entryId string, resolve
 	elapsedTime := item.Pair.Response.CaptureTime.Sub(item.Pair.Request.CaptureTime).Round(time.Millisecond).Milliseconds()
 	entryBytes, _ := json.Marshal(item.Pair)
 	return &api.MizuEntry{
-		ProtocolName:            _protocol.Name,
+		Protocol:            _protocol,
-		ProtocolLongName:        _protocol.LongName,
-		ProtocolAbbreviation:    _protocol.Abbreviation,
-		ProtocolVersion:         _protocol.Version,
-		ProtocolBackgroundColor: _protocol.BackgroundColor,
-		ProtocolForegroundColor: _protocol.ForegroundColor,
-		ProtocolFontSize:        _protocol.FontSize,
-		ProtocolReferenceLink:   _protocol.ReferenceLink,
 		EntryId:             entryId,
 		Entry:               string(entryBytes),
 		Url:                 fmt.Sprintf("%s%s", service, summary),
@@ -172,7 +165,7 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, entryId string, resolve
 
 func (d dissecting) Summarize(entry *api.MizuEntry) *api.BaseEntryDetails {
 	return &api.BaseEntryDetails{
-		Id:              entry.EntryId,
+		Id:              entry.Id,
 		Protocol:        _protocol,
 		Url:             entry.Url,
 		RequestSenderIp: entry.RequestSenderIp,

ui/src/components/EntriesList.tsx

@@ -88,7 +88,7 @@ export const EntriesList: React.FC<EntriesListProps> = ({entries, setEntries, fo
                         {filteredEntries.map(entry => <EntryItem key={entry.id}
                                                         entry={entry}
                                                         setFocusedEntryId={setFocusedEntryId}
-                                                        isSelected={focusedEntryId === entry.id}
+                                                        isSelected={focusedEntryId === entry.id.toString()}
                                                         style={{}}/>)}
                     </ScrollableFeedVirtualized>
                     {!connectionOpen && !noMoreDataBottom && <div className={styles.fetchButtonContainer}>

ui/src/components/EntryListItem/EntryListItem.tsx

@@ -15,7 +15,7 @@ interface Entry {
     method?: string,
     summary: string,
     service: string,
-    id: string,
+    id: number,
     statusCode?: number;
     url?: string;
     timestamp: Date;
@@ -100,10 +100,10 @@ export const EntryItem: React.FC<EntryProps> = ({entry, setFocusedEntryId, isSel
     }
     return <>
         <div
-            id={entry.id}
+            id={entry.id.toString()}
             className={`${styles.row}
             ${isSelected ? styles.rowSelected : backgroundColor}`}
-            onClick={() => setFocusedEntryId(entry.id)}
+            onClick={() => setFocusedEntryId(entry.id.toString())}
             style={{
                 border: isSelected ? `1px ${entry.protocol.backgroundColor} solid` : "1px transparent solid",
                 position: "absolute",

ui/src/components/TrafficPage.tsx

@@ -86,7 +86,7 @@ export const TrafficPage: React.FC<TrafficPageProps> = ({setAnalyzeStatus, onTLS
                         setNoMoreDataBottom(false)
                         return;
                     }
-                    if (!focusedEntryId) setFocusedEntryId(entry.id)
+                    if (!focusedEntryId) setFocusedEntryId(entry.id.toString())
                     let newEntries = [...entries];
                     setEntries([...newEntries, entry])
                     if(listEntry.current) {