TRA-4410 Display namespace field in the UI (#974)

* Added log message
* Remove Reset function from OasGenerator interface, use Stop+Start instead
* SetEntriesQuery returns a bool stating whether the query changed

agent/main.go

@@ -199,7 +199,7 @@ func runInHarReaderMode() {
 func enableExpFeatureIfNeeded() {
 	if config.Config.OAS {
 		oasGenerator := dependency.GetInstance(dependency.OasGeneratorDependency).(oas.OasGenerator)
-		oasGenerator.Start()
+		oasGenerator.Start(nil)
 	}
 	if config.Config.ServiceMap {
 		serviceMapGenerator := dependency.GetInstance(dependency.ServiceMapGeneratorDependency).(servicemap.ServiceMap)
@@ -371,7 +371,7 @@ func handleIncomingMessageAsTapper(socketConnection *websocket.Conn) {
 
 func initializeDependencies() {
 	dependency.RegisterGenerator(dependency.ServiceMapGeneratorDependency, func() interface{} { return servicemap.GetDefaultServiceMapInstance() })
-	dependency.RegisterGenerator(dependency.OasGeneratorDependency, func() interface{} { return oas.GetDefaultOasGeneratorInstance(nil) })
+	dependency.RegisterGenerator(dependency.OasGeneratorDependency, func() interface{} { return oas.GetDefaultOasGeneratorInstance() })
 	dependency.RegisterGenerator(dependency.EntriesProvider, func() interface{} { return &entries.BasenineEntriesProvider{} })
 	dependency.RegisterGenerator(dependency.EntriesSocketStreamer, func() interface{} { return &api.BasenineEntryStreamer{} })
 	dependency.RegisterGenerator(dependency.EntryStreamerSocketConnector, func() interface{} { return &api.DefaultEntryStreamerSocketConnector{} })

agent/pkg/controllers/oas_controller_test.go

@@ -58,12 +58,12 @@ func getRecorderAndContext() (*httptest.ResponseRecorder, *gin.Context) {
 		receiveBuffer: bytes.NewBufferString("\n"),
 	}
 	dependency.RegisterGenerator(dependency.OasGeneratorDependency, func() interface{} {
-		return oas.GetDefaultOasGeneratorInstance(dummyConn)
+		return oas.GetDefaultOasGeneratorInstance()
 	})
 
 	recorder := httptest.NewRecorder()
 	c, _ := gin.CreateTestContext(recorder)
-	oas.GetDefaultOasGeneratorInstance(dummyConn).Start()
-	oas.GetDefaultOasGeneratorInstance(dummyConn).GetServiceSpecs().Store("some", oas.NewGen("some"))
+	oas.GetDefaultOasGeneratorInstance().Start(dummyConn)
+	oas.GetDefaultOasGeneratorInstance().GetServiceSpecs().Store("some", oas.NewGen("some"))
 	return recorder, c
 }

agent/pkg/oas/oas_generator.go

@@ -19,12 +19,11 @@ var (
 )
 
 type OasGenerator interface {
-	Start()
+	Start(conn *basenine.Connection)
 	Stop()
 	IsStarted() bool
-	Reset()
 	GetServiceSpecs() *sync.Map
-	SetEntriesQuery(query string)
+	SetEntriesQuery(query string) bool
 }
 
 type defaultOasGenerator struct {
@@ -36,23 +35,41 @@ type defaultOasGenerator struct {
 	entriesQuery string
 }
 
-func GetDefaultOasGeneratorInstance(conn *basenine.Connection) *defaultOasGenerator {
+func GetDefaultOasGeneratorInstance() *defaultOasGenerator {
 	syncOnce.Do(func() {
-		instance = NewDefaultOasGenerator(conn)
+		instance = NewDefaultOasGenerator()
 		logger.Log.Debug("OAS Generator Initialized")
 	})
 	return instance
 }
 
-func (g *defaultOasGenerator) Start() {
+func (g *defaultOasGenerator) Start(conn *basenine.Connection) {
 	if g.started {
 		return
 	}
+
+	if g.dbConn == nil {
+		if conn == nil {
+			logger.Log.Infof("Creating new DB connection for OAS generator to address %s:%s", shared.BasenineHost, shared.BaseninePort)
+			newConn, err := basenine.NewConnection(shared.BasenineHost, shared.BaseninePort)
+			if err != nil {
+				logger.Log.Error("Error connecting to DB for OAS generator, err: %v", err)
+				return
+			}
+
+			conn = newConn
+		}
+
+		g.dbConn = conn
+	}
+
 	ctx, cancel := context.WithCancel(context.Background())
 	g.cancel = cancel
 	g.ctx = ctx
 	g.serviceSpecs = &sync.Map{}
+
 	g.started = true
+
 	go g.runGenerator()
 }
 
@@ -60,8 +77,15 @@ func (g *defaultOasGenerator) Stop() {
 	if !g.started {
 		return
 	}
+
+	if g.dbConn != nil {
+		g.dbConn.Close()
+		g.dbConn = nil
+	}
+
 	g.cancel()
-	g.Reset()
+	g.reset()
+
 	g.started = false
 }
 
@@ -70,7 +94,7 @@ func (g *defaultOasGenerator) IsStarted() bool {
 }
 
 func (g *defaultOasGenerator) runGenerator() {
-	// Make []byte channels to recieve the data and the meta
+	// Make []byte channels to receive the data and the meta
 	dataChan := make(chan []byte)
 	metaChan := make(chan []byte)
 
@@ -81,6 +105,8 @@ func (g *defaultOasGenerator) runGenerator() {
 		select {
 		case <-g.ctx.Done():
 			logger.Log.Infof("OAS Generator was canceled")
+			close(dataChan)
+			close(metaChan)
 			return
 
 		case metaBytes, ok := <-metaChan:
@@ -168,7 +194,7 @@ func (g *defaultOasGenerator) getGen(dest string, urlStr string) *SpecGen {
 	return gen
 }
 
-func (g *defaultOasGenerator) Reset() {
+func (g *defaultOasGenerator) reset() {
 	g.serviceSpecs = &sync.Map{}
 }
 
@@ -176,25 +202,18 @@ func (g *defaultOasGenerator) GetServiceSpecs() *sync.Map {
 	return g.serviceSpecs
 }
 
-func (g *defaultOasGenerator) SetEntriesQuery(query string) {
+func (g *defaultOasGenerator) SetEntriesQuery(query string) bool {
+	changed := g.entriesQuery != query
 	g.entriesQuery = query
+	return changed
 }
 
-func NewDefaultOasGenerator(conn *basenine.Connection) *defaultOasGenerator {
-	if conn == nil {
-		logger.Log.Infof("Creating new DB connection for OAS generator to address %s:%s", shared.BasenineHost, shared.BaseninePort)
-		newConn, err := basenine.NewConnection(shared.BasenineHost, shared.BaseninePort)
-		if err != nil {
-			panic(err)
-		}
-		conn = newConn
-	}
-
+func NewDefaultOasGenerator() *defaultOasGenerator {
 	return &defaultOasGenerator{
 		started:      false,
 		ctx:          nil,
 		cancel:       nil,
 		serviceSpecs: nil,
-		dbConn:       conn,
+		dbConn:       nil,
 	}
 }

agent/pkg/oas/oas_generator_test.go

@@ -3,14 +3,11 @@ package oas
 import (
 	"encoding/json"
 	"github.com/up9inc/mizu/agent/pkg/har"
-	"sync"
 	"testing"
 )
 
 func TestOASGen(t *testing.T) {
 	gen := new(defaultOasGenerator)
-	gen.dbConn = GetFakeDBConn(`{"startedDateTime": "20000101","request": {"url": "https://host/path", "method": "GET"}, "response": {"status": 200}}`)
-	gen.serviceSpecs = &sync.Map{}
 
 	e := new(har.Entry)
 	err := json.Unmarshal([]byte(`{"startedDateTime": "20000101","request": {"url": "https://host/path", "method": "GET"}, "response": {"status": 200}}`), e)
@@ -22,7 +19,9 @@ func TestOASGen(t *testing.T) {
 		Destination: "some",
 		Entry:       *e,
 	}
-	gen.Start()
+
+	dummyConn := GetFakeDBConn(`{"startedDateTime": "20000101","request": {"url": "https://host/path", "method": "GET"}, "response": {"status": 200}}`)
+	gen.Start(dummyConn)
 	gen.handleHARWithSource(ews)
 	g, ok := gen.serviceSpecs.Load("some")
 	if !ok {

agent/pkg/oas/specgen_test.go

@@ -61,8 +61,7 @@ func TestEntries(t *testing.T) {
 		t.FailNow()
 	}
 
-	dummyConn := GetFakeDBConn("\n")
-	gen := NewDefaultOasGenerator(dummyConn)
+	gen := NewDefaultOasGenerator()
 	gen.serviceSpecs = new(sync.Map)
 	loadStartingOAS("test_artifacts/catalogue.json", "catalogue", gen.serviceSpecs)
 	loadStartingOAS("test_artifacts/trcc.json", "trcc-api-service", gen.serviceSpecs)
@@ -136,8 +135,7 @@ func TestEntries(t *testing.T) {
 }
 
 func TestFileSingle(t *testing.T) {
-	dummyConn := GetFakeDBConn("\n")
-	gen := NewDefaultOasGenerator(dummyConn)
+	gen := NewDefaultOasGenerator()
 	gen.serviceSpecs = new(sync.Map)
 	// loadStartingOAS()
 	file := "test_artifacts/params.har"
@@ -227,8 +225,7 @@ func loadStartingOAS(file string, label string, specs *sync.Map) {
 }
 
 func TestEntriesNegative(t *testing.T) {
-	dummyConn := GetFakeDBConn("\n")
-	gen := NewDefaultOasGenerator(dummyConn)
+	gen := NewDefaultOasGenerator()
 	gen.serviceSpecs = new(sync.Map)
 	files := []string{"invalid"}
 	_, err := feedEntries(files, false, gen)
@@ -239,8 +236,7 @@ func TestEntriesNegative(t *testing.T) {
 }
 
 func TestEntriesPositive(t *testing.T) {
-	dummyConn := GetFakeDBConn("\n")
-	gen := NewDefaultOasGenerator(dummyConn)
+	gen := NewDefaultOasGenerator()
 	gen.serviceSpecs = new(sync.Map)
 	files := []string{"test_artifacts/params.har"}
 	_, err := feedEntries(files, false, gen)

tap/api/api.go

@@ -161,7 +161,7 @@ type Entry struct {
 	Capture                Capture                `json:"capture"`
 	Source                 *TCP                   `json:"src"`
 	Destination            *TCP                   `json:"dst"`
-	Namespace              string                 `json:"namespace,omitempty"`
+	Namespace              string                 `json:"namespace"`
 	Outgoing               bool                   `json:"outgoing"`
 	Timestamp              int64                  `json:"timestamp"`
 	StartTime              time.Time              `json:"startTime"`

tap/extensions/http/http2_assembler.go

@@ -28,26 +28,6 @@ const protoMinorHTTP2 = 0
 
 var maxHTTP2DataLen = 1 * 1024 * 1024 // 1MB
 
-var grpcStatusCodes = []string{
-	"OK",
-	"CANCELLED",
-	"UNKNOWN",
-	"INVALID_ARGUMENT",
-	"DEADLINE_EXCEEDED",
-	"NOT_FOUND",
-	"ALREADY_EXISTS",
-	"PERMISSION_DENIED",
-	"RESOURCE_EXHAUSTED",
-	"FAILED_PRECONDITION",
-	"ABORTED",
-	"OUT_OF_RANGE",
-	"UNIMPLEMENTED",
-	"INTERNAL",
-	"UNAVAILABLE",
-	"DATA_LOSS",
-	"UNAUTHENTICATED",
-}
-
 type messageFragment struct {
 	headers []hpack.HeaderField
 	data    []byte
@@ -142,18 +122,8 @@ func (ga *Http2Assembler) readMessage() (streamID uint32, messageHTTP1 interface
 
 	// gRPC detection
 	grpcStatus := headersHTTP1.Get("Grpc-Status")
-	if grpcStatus != "" {
+	if grpcStatus != "" || strings.Contains(headersHTTP1.Get("Content-Type"), "application/grpc") {
 		isGrpc = true
-		status = grpcStatus
-	}
-
-	if strings.Contains(headersHTTP1.Get("Content-Type"), "application/grpc") {
-		isGrpc = true
-		grpcPath := headersHTTP1.Get(":path")
-		pathSegments := strings.Split(grpcPath, "/")
-		if len(pathSegments) > 0 {
-			method = pathSegments[len(pathSegments)-1]
-		}
 	}
 
 	if method != "" {

tap/extensions/http/main.go

@@ -248,11 +248,6 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, resolvedSource string,
 	reqDetails["_queryStringMerged"] = mapSliceMergeRepeatedKeys(reqDetails["_queryString"].([]interface{}))
 	reqDetails["queryString"] = mapSliceRebuildAsMap(reqDetails["_queryStringMerged"].([]interface{}))
 
-	statusCode := int(resDetails["status"].(float64))
-	if item.Protocol.Abbreviation == "gRPC" {
-		resDetails["statusText"] = grpcStatusCodes[statusCode]
-	}
-
 	elapsedTime := item.Pair.Response.CaptureTime.Sub(item.Pair.Request.CaptureTime).Round(time.Millisecond).Milliseconds()
 	if elapsedTime < 0 {
 		elapsedTime = 0

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

@@ -89,12 +89,13 @@ const EntryTitle: React.FC<any> = ({ protocol, data, elapsedTime }) => {
     </div>;
 };
 
-const EntrySummary: React.FC<any> = ({ entry }) => {
+const EntrySummary: React.FC<any> = ({ entry, namespace }) => {
     return <EntryItem
         key={`entry-${entry.id}`}
         entry={entry}
         style={{}}
         headingMode={true}
+        namespace={namespace}
     />;
 };
 
@@ -140,7 +141,7 @@ export const EntryDetailed = () => {
             data={entryData.data}
             elapsedTime={entryData.data.elapsedTime}
         />}
-        {!isLoading && entryData && <EntrySummary entry={entryData.base} />}
+        {!isLoading && entryData && <EntrySummary entry={entryData.base} namespace={entryData.data.namespace} />}
         <React.Fragment>
             {!isLoading && entryData && <EntryViewer
                 representation={entryData.representation}

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

@@ -52,6 +52,7 @@ interface EntryProps {
     entry: Entry;
     style: object;
     headingMode: boolean;
+    namespace?: string;
 }
 
 enum CaptureTypes {
@@ -62,7 +63,7 @@ enum CaptureTypes {
     Ebpf = "ebpf",
 }
 
-export const EntryItem: React.FC<EntryProps> = ({entry, style, headingMode}) => {
+export const EntryItem: React.FC<EntryProps> = ({entry, style, headingMode, namespace}) => {
 
     const [focusedEntryId, setFocusedEntryId] = useRecoilState(focusedEntryIdAtom);
     const [queryState, setQuery] = useRecoilState(queryAtom);
@@ -224,6 +225,19 @@ export const EntryItem: React.FC<EntryProps> = ({entry, style, headingMode}) =>
                 : ""
             }
             <div className={styles.separatorRight}>
+                {headingMode ? <Queryable
+                        query={`namespace == "${namespace}"`}
+                        displayIconOnMouseOver={true}
+                        flipped={true}
+                        iconStyle={{marginRight: "16px"}}
+                >
+                    <span
+                        className={`${styles.tcpInfo} ${styles.ip}`}
+                        title="Namespace"
+                    >
+                        {namespace}
+                    </span>
+                </Queryable> : null}
                 <Queryable
                         query={`src.ip == "${entry.src.ip}"`}
                         displayIconOnMouseOver={true}

ui-common/src/components/UI/StatusCode.tsx

@@ -37,9 +37,9 @@ export function getClassification(statusCode: number): string {
 
     // 1 - 16 HTTP/2 (gRPC) status codes
     // 2xx - 5xx HTTP/1.x status codes
-    if ((statusCode >= 200 && statusCode <= 399) || statusCode === 0) {
+    if (statusCode >= 200 && statusCode <= 399) {
         classification = StatusCodeClassification.SUCCESS;
-    } else if (statusCode >= 400 || (statusCode >= 1 && statusCode <= 16)) {
+    } else if (statusCode >= 400) {
         classification = StatusCodeClassification.FAILURE;
     }