From 0f8afa6b5f9a9eb1d4b4d5c0833f4470e87c00e1 Mon Sep 17 00:00:00 2001
From: Julien Pinsonneau <jpinsonn@redhat.com>
Date: Mon, 19 Aug 2024 17:30:09 +0200
Subject: [PATCH 1/2] manage services case

---
 pkg/handler/topology.go      |  2 +-
 pkg/handler/topology_test.go | 10 +++++-----
 pkg/server/server_test.go    |  6 +++---
 3 files changed, 9 insertions(+), 9 deletions(-)

diff --git a/pkg/handler/topology.go b/pkg/handler/topology.go
index 4a53616c4..00e1d88f4 100644
--- a/pkg/handler/topology.go
+++ b/pkg/handler/topology.go
@@ -234,7 +234,7 @@ func expandQueries(queries filters.MultiQueries, namespace string, isForProm fun
 	}
 	q2 := filters.SingleQuery{
 		filters.NewMatch(fields.FlowDirection, `"`+string(constants.Egress)+`"`),
-		filters.NewMatch(fields.DstOwnerName, `""`),
+		filters.NewMatch(fields.DstOwnerType, `"","Service"`),
 	}
 
 	shouldSkip := func(q filters.SingleQuery) bool {
diff --git a/pkg/handler/topology_test.go b/pkg/handler/topology_test.go
index ba450e95e..9cb1554a6 100644
--- a/pkg/handler/topology_test.go
+++ b/pkg/handler/topology_test.go
@@ -30,7 +30,7 @@ func TestSplitForReportersMerge(t *testing.T) {
 	}, res[0])
 	assert.Equal(t, filters.SingleQuery{
 		filters.NewMatch("FlowDirection", `"`+string(constants.Egress)+`"`),
-		filters.NewMatch("DstK8S_OwnerName", `""`),
+		filters.NewMatch("DstK8S_OwnerType", `"","Service"`),
 		filters.NewMatch("srcns", "a"),
 		filters.NewMatch("dstns", "b"),
 	}, res[1])
@@ -73,14 +73,14 @@ func TestExpand_ComplexQuery(t *testing.T) {
 	assert.Equal(t, filters.SingleQuery{
 		filters.NewMatch("SrcK8S_Namespace", `"my-namespace"`),
 		filters.NewMatch("FlowDirection", `"`+string(constants.Egress)+`"`),
-		filters.NewMatch("DstK8S_OwnerName", `""`),
+		filters.NewMatch("DstK8S_OwnerType", `"","Service"`),
 		filters.NewMatch("key1", "a"),
 		filters.NewMatch("key2", "b"),
 	}, res[4])
 	assert.Equal(t, filters.SingleQuery{
 		filters.NewMatch("DstK8S_Namespace", `"my-namespace"`),
 		filters.NewMatch("FlowDirection", `"`+string(constants.Egress)+`"`),
-		filters.NewMatch("DstK8S_OwnerName", `""`),
+		filters.NewMatch("DstK8S_OwnerType", `"","Service"`),
 		filters.NewMatch("key1", "a"),
 		filters.NewMatch("key2", "b"),
 	}, res[5])
@@ -104,14 +104,14 @@ func TestExpand_ComplexQuery(t *testing.T) {
 	assert.Equal(t, filters.SingleQuery{
 		filters.NewMatch("SrcK8S_Namespace", `"my-namespace"`),
 		filters.NewMatch("FlowDirection", `"`+string(constants.Egress)+`"`),
-		filters.NewMatch("DstK8S_OwnerName", `""`),
+		filters.NewMatch("DstK8S_OwnerType", `"","Service"`),
 		filters.NewMatch("key1", "c"),
 		filters.NewMatch("key2", "d"),
 	}, res[9])
 	assert.Equal(t, filters.SingleQuery{
 		filters.NewMatch("DstK8S_Namespace", `"my-namespace"`),
 		filters.NewMatch("FlowDirection", `"`+string(constants.Egress)+`"`),
-		filters.NewMatch("DstK8S_OwnerName", `""`),
+		filters.NewMatch("DstK8S_OwnerType", `"","Service"`),
 		filters.NewMatch("key1", "c"),
 		filters.NewMatch("key2", "d"),
 	}, res[10])
diff --git a/pkg/server/server_test.go b/pkg/server/server_test.go
index 18b2de749..a0669a4cf 100644
--- a/pkg/server/server_test.go
+++ b/pkg/server/server_test.go
@@ -339,7 +339,7 @@ func TestLokiConfigurationForTopology(t *testing.T) {
 		Loki: config.Loki{
 			URL:     lokiSvc.URL,
 			Timeout: config.Duration{Duration: time.Second},
-			Labels:  []string{fields.SrcNamespace, fields.DstNamespace, fields.SrcOwnerName, fields.DstOwnerName, fields.FlowDirection},
+			Labels:  []string{fields.SrcNamespace, fields.DstNamespace, fields.SrcOwnerName, fields.DstOwnerName, fields.SrcOwnerType, fields.DstOwnerType, fields.FlowDirection},
 		},
 		Frontend: config.Frontend{Deduper: config.Deduper{
 			Mark:  true,
@@ -361,7 +361,7 @@ func TestLokiConfigurationForTopology(t *testing.T) {
 	queries := []string{req1.URL.Query().Get("query"), req2.URL.Query().Get("query")}
 	expected := []string{
 		`topk(100,sum by(SrcK8S_Name,SrcK8S_Type,SrcK8S_OwnerName,SrcK8S_OwnerType,SrcK8S_Namespace,SrcAddr,SrcK8S_HostName,DstK8S_Name,DstK8S_Type,DstK8S_OwnerName,DstK8S_OwnerType,DstK8S_Namespace,DstAddr,DstK8S_HostName)(rate({app="netobserv-flowcollector",FlowDirection=~"^0$|^2$"}!~` + "`" + `Duplicate":true` + "`" + `|json|unwrap Bytes|__error__=""[1m])))`,
-		`topk(100,sum by(SrcK8S_Name,SrcK8S_Type,SrcK8S_OwnerName,SrcK8S_OwnerType,SrcK8S_Namespace,SrcAddr,SrcK8S_HostName,DstK8S_Name,DstK8S_Type,DstK8S_OwnerName,DstK8S_OwnerType,DstK8S_Namespace,DstAddr,DstK8S_HostName)(rate({app="netobserv-flowcollector",FlowDirection="1",DstK8S_OwnerName=""}!~` + "`" + `Duplicate":true` + "`" + `|json|unwrap Bytes|__error__=""[1m])))`,
+		`topk(100,sum by(SrcK8S_Name,SrcK8S_Type,SrcK8S_OwnerName,SrcK8S_OwnerType,SrcK8S_Namespace,SrcAddr,SrcK8S_HostName,DstK8S_Name,DstK8S_Type,DstK8S_OwnerName,DstK8S_OwnerType,DstK8S_Namespace,DstAddr,DstK8S_HostName)(rate({app="netobserv-flowcollector",FlowDirection="1",DstK8S_OwnerType=~"^$|^Service$"}!~` + "`" + `Duplicate":true` + "`" + `|json|unwrap Bytes|__error__=""[1m])))`,
 	}
 	// We don't predict the order so sort both actual and expected
 	sort.Strings(queries)
@@ -397,7 +397,7 @@ func TestLokiConfigurationForTableHistogram(t *testing.T) {
 		Loki: config.Loki{
 			URL:     lokiSvc.URL,
 			Timeout: config.Duration{Duration: time.Second},
-			Labels:  []string{fields.SrcNamespace, fields.DstNamespace, fields.SrcOwnerName, fields.DstOwnerName, fields.FlowDirection},
+			Labels:  []string{fields.SrcNamespace, fields.DstNamespace, fields.SrcOwnerName, fields.DstOwnerName, fields.SrcOwnerType, fields.DstOwnerType, fields.FlowDirection},
 		},
 		Frontend: config.Frontend{Deduper: config.Deduper{
 			Mark:  true,

From 609a6068e9e683e0d09bda295669976661034d4c Mon Sep 17 00:00:00 2001
From: Julien Pinsonneau <jpinsonn@redhat.com>
Date: Tue, 10 Sep 2024 09:37:29 +0200
Subject: [PATCH 2/2] use type instead of owner type

---
 pkg/handler/topology.go      |  2 +-
 pkg/handler/topology_test.go | 10 +++++-----
 pkg/server/server_test.go    |  6 +++---
 3 files changed, 9 insertions(+), 9 deletions(-)

diff --git a/pkg/handler/topology.go b/pkg/handler/topology.go
index 00e1d88f4..fc66bcea9 100644
--- a/pkg/handler/topology.go
+++ b/pkg/handler/topology.go
@@ -234,7 +234,7 @@ func expandQueries(queries filters.MultiQueries, namespace string, isForProm fun
 	}
 	q2 := filters.SingleQuery{
 		filters.NewMatch(fields.FlowDirection, `"`+string(constants.Egress)+`"`),
-		filters.NewMatch(fields.DstOwnerType, `"","Service"`),
+		filters.NewMatch(fields.DstType, `"","Service"`),
 	}
 
 	shouldSkip := func(q filters.SingleQuery) bool {
diff --git a/pkg/handler/topology_test.go b/pkg/handler/topology_test.go
index 9cb1554a6..a4ffc1131 100644
--- a/pkg/handler/topology_test.go
+++ b/pkg/handler/topology_test.go
@@ -30,7 +30,7 @@ func TestSplitForReportersMerge(t *testing.T) {
 	}, res[0])
 	assert.Equal(t, filters.SingleQuery{
 		filters.NewMatch("FlowDirection", `"`+string(constants.Egress)+`"`),
-		filters.NewMatch("DstK8S_OwnerType", `"","Service"`),
+		filters.NewMatch("DstK8S_Type", `"","Service"`),
 		filters.NewMatch("srcns", "a"),
 		filters.NewMatch("dstns", "b"),
 	}, res[1])
@@ -73,14 +73,14 @@ func TestExpand_ComplexQuery(t *testing.T) {
 	assert.Equal(t, filters.SingleQuery{
 		filters.NewMatch("SrcK8S_Namespace", `"my-namespace"`),
 		filters.NewMatch("FlowDirection", `"`+string(constants.Egress)+`"`),
-		filters.NewMatch("DstK8S_OwnerType", `"","Service"`),
+		filters.NewMatch("DstK8S_Type", `"","Service"`),
 		filters.NewMatch("key1", "a"),
 		filters.NewMatch("key2", "b"),
 	}, res[4])
 	assert.Equal(t, filters.SingleQuery{
 		filters.NewMatch("DstK8S_Namespace", `"my-namespace"`),
 		filters.NewMatch("FlowDirection", `"`+string(constants.Egress)+`"`),
-		filters.NewMatch("DstK8S_OwnerType", `"","Service"`),
+		filters.NewMatch("DstK8S_Type", `"","Service"`),
 		filters.NewMatch("key1", "a"),
 		filters.NewMatch("key2", "b"),
 	}, res[5])
@@ -104,14 +104,14 @@ func TestExpand_ComplexQuery(t *testing.T) {
 	assert.Equal(t, filters.SingleQuery{
 		filters.NewMatch("SrcK8S_Namespace", `"my-namespace"`),
 		filters.NewMatch("FlowDirection", `"`+string(constants.Egress)+`"`),
-		filters.NewMatch("DstK8S_OwnerType", `"","Service"`),
+		filters.NewMatch("DstK8S_Type", `"","Service"`),
 		filters.NewMatch("key1", "c"),
 		filters.NewMatch("key2", "d"),
 	}, res[9])
 	assert.Equal(t, filters.SingleQuery{
 		filters.NewMatch("DstK8S_Namespace", `"my-namespace"`),
 		filters.NewMatch("FlowDirection", `"`+string(constants.Egress)+`"`),
-		filters.NewMatch("DstK8S_OwnerType", `"","Service"`),
+		filters.NewMatch("DstK8S_Type", `"","Service"`),
 		filters.NewMatch("key1", "c"),
 		filters.NewMatch("key2", "d"),
 	}, res[10])
diff --git a/pkg/server/server_test.go b/pkg/server/server_test.go
index a0669a4cf..06d7ce6d3 100644
--- a/pkg/server/server_test.go
+++ b/pkg/server/server_test.go
@@ -339,7 +339,7 @@ func TestLokiConfigurationForTopology(t *testing.T) {
 		Loki: config.Loki{
 			URL:     lokiSvc.URL,
 			Timeout: config.Duration{Duration: time.Second},
-			Labels:  []string{fields.SrcNamespace, fields.DstNamespace, fields.SrcOwnerName, fields.DstOwnerName, fields.SrcOwnerType, fields.DstOwnerType, fields.FlowDirection},
+			Labels:  []string{fields.SrcNamespace, fields.DstNamespace, fields.SrcOwnerName, fields.DstOwnerName, fields.SrcType, fields.DstType, fields.FlowDirection},
 		},
 		Frontend: config.Frontend{Deduper: config.Deduper{
 			Mark:  true,
@@ -361,7 +361,7 @@ func TestLokiConfigurationForTopology(t *testing.T) {
 	queries := []string{req1.URL.Query().Get("query"), req2.URL.Query().Get("query")}
 	expected := []string{
 		`topk(100,sum by(SrcK8S_Name,SrcK8S_Type,SrcK8S_OwnerName,SrcK8S_OwnerType,SrcK8S_Namespace,SrcAddr,SrcK8S_HostName,DstK8S_Name,DstK8S_Type,DstK8S_OwnerName,DstK8S_OwnerType,DstK8S_Namespace,DstAddr,DstK8S_HostName)(rate({app="netobserv-flowcollector",FlowDirection=~"^0$|^2$"}!~` + "`" + `Duplicate":true` + "`" + `|json|unwrap Bytes|__error__=""[1m])))`,
-		`topk(100,sum by(SrcK8S_Name,SrcK8S_Type,SrcK8S_OwnerName,SrcK8S_OwnerType,SrcK8S_Namespace,SrcAddr,SrcK8S_HostName,DstK8S_Name,DstK8S_Type,DstK8S_OwnerName,DstK8S_OwnerType,DstK8S_Namespace,DstAddr,DstK8S_HostName)(rate({app="netobserv-flowcollector",FlowDirection="1",DstK8S_OwnerType=~"^$|^Service$"}!~` + "`" + `Duplicate":true` + "`" + `|json|unwrap Bytes|__error__=""[1m])))`,
+		`topk(100,sum by(SrcK8S_Name,SrcK8S_Type,SrcK8S_OwnerName,SrcK8S_OwnerType,SrcK8S_Namespace,SrcAddr,SrcK8S_HostName,DstK8S_Name,DstK8S_Type,DstK8S_OwnerName,DstK8S_OwnerType,DstK8S_Namespace,DstAddr,DstK8S_HostName)(rate({app="netobserv-flowcollector",FlowDirection="1",DstK8S_Type=~"^$|^Service$"}!~` + "`" + `Duplicate":true` + "`" + `|json|unwrap Bytes|__error__=""[1m])))`,
 	}
 	// We don't predict the order so sort both actual and expected
 	sort.Strings(queries)
@@ -397,7 +397,7 @@ func TestLokiConfigurationForTableHistogram(t *testing.T) {
 		Loki: config.Loki{
 			URL:     lokiSvc.URL,
 			Timeout: config.Duration{Duration: time.Second},
-			Labels:  []string{fields.SrcNamespace, fields.DstNamespace, fields.SrcOwnerName, fields.DstOwnerName, fields.SrcOwnerType, fields.DstOwnerType, fields.FlowDirection},
+			Labels:  []string{fields.SrcNamespace, fields.DstNamespace, fields.SrcOwnerName, fields.DstOwnerName, fields.SrcType, fields.DstType, fields.FlowDirection},
 		},
 		Frontend: config.Frontend{Deduper: config.Deduper{
 			Mark:  true,