Skip to content

Commit 188a291

Browse files
jotakjotak
committed
NETOBSERV-2126: reintroduce merge option, and cleanup old duplicate code
1 parent ec2f365 commit 188a291

File tree

23 files changed

+210
-292
lines changed

23 files changed

+210
-292
lines changed

‎config/sample-config.yaml‎

Lines changed: 0 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,6 @@ loki:
1313
- DstK8S_Type
1414
- K8S_FlowLayer
1515
- FlowDirection
16-
# - Duplicate
1716
# - _RecordType
1817
# - K8S_ClusterName
1918
# - SrcK8S_Zone
@@ -79,9 +78,6 @@ frontend:
7978
alertNamespaces:
8079
- netobserv
8180
sampling: 50
82-
deduper:
83-
mark: true
84-
merge: false
8581
# The following configuration is taken from Network Observability Operator
8682
# see https://github.com/netobserv/network-observability-operator/blob/main/controllers/consoleplugin/config/static-frontend-config.yaml
8783
panels:
@@ -1474,9 +1470,6 @@ frontend:
14741470
- name: IcmpCode
14751471
type: number
14761472
description: ICMP code
1477-
- name: Duplicate
1478-
type: boolean
1479-
description: Indicates if this flow was also captured from another interface on the same host
14801473
- name: FlowDirection
14811474
type: number
14821475
description: |

‎mocks/loki/flow_records.json‎

Lines changed: 50 additions & 50 deletions
Large diffs are not rendered by default.

‎mocks/loki/flow_records_has_dropped.json‎

Lines changed: 50 additions & 50 deletions
Large diffs are not rendered by default.

‎mocks/loki/flow_records_sent.json‎

Lines changed: 50 additions & 50 deletions
Large diffs are not rendered by default.

‎mocks/updateMocks.sh‎

Lines changed: 14 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -15,35 +15,35 @@ curl 'http://localhost:3100/loki/api/v1/query_range?query=\{app=%22netobserv-flo
1515

1616
# flow_metrics_*.json contains queries result for overview / topology display
1717
echo 'Getting metrics'
18-
curl 'http://localhost:3100/loki/api/v1/query_range?query=topk(5,sum%20by(app)%20(rate(\{app=%22netobserv-flowcollector%22,FlowDirection=%221%22\}!~`Duplicate%22:true`|json|unwrap%20Packets|__error__=%22%22\[720s\])))&limit=5&step=360s'\
18+
curl 'http://localhost:3100/loki/api/v1/query_range?query=topk(5,sum%20by(app)%20(rate(\{app=%22netobserv-flowcollector%22,FlowDirection=%221%22\}|json|unwrap%20Packets|__error__=%22%22\[720s\])))&limit=5&step=360s'\
1919
| jq > ./loki/flow_metrics_app.json
20-
curl 'http://localhost:3100/loki/api/v1/query_range?query=topk(50,sum%20by(K8S_ClusterName)%20(rate(\{app=%22netobserv-flowcollector%22,FlowDirection=%221%22\}!~`Duplicate%22:true`|json|unwrap%20Packets|__error__=%22%22\[720s\])))&limit=50&step=360s'\
20+
curl 'http://localhost:3100/loki/api/v1/query_range?query=topk(50,sum%20by(K8S_ClusterName)%20(rate(\{app=%22netobserv-flowcollector%22,FlowDirection=%221%22\}|json|unwrap%20Packets|__error__=%22%22\[720s\])))&limit=50&step=360s'\
2121
| jq > ./loki/flow_metrics_cluster.json
22-
curl 'http://localhost:3100/loki/api/v1/query_range?query=topk(50,sum%20by(SrcK8S_Zone,DstK8S_Zone)%20(rate(\{app=%22netobserv-flowcollector%22,FlowDirection=%221%22\}!~`Duplicate%22:true`|json|unwrap%20Packets|__error__=%22%22\[720s\])))&limit=50&step=360s'\
22+
curl 'http://localhost:3100/loki/api/v1/query_range?query=topk(50,sum%20by(SrcK8S_Zone,DstK8S_Zone)%20(rate(\{app=%22netobserv-flowcollector%22,FlowDirection=%221%22\}|json|unwrap%20Packets|__error__=%22%22\[720s\])))&limit=50&step=360s'\
2323
| jq > ./loki/flow_metrics_zone.json
24-
curl 'http://localhost:3100/loki/api/v1/query_range?query=topk(50,sum%20by(SrcK8S_HostName,DstK8S_HostName)%20(rate(\{app=%22netobserv-flowcollector%22,FlowDirection=%221%22\}!~`Duplicate%22:true`|json|unwrap%20Packets|__error__=%22%22\[720s\])))&limit=50&step=360s'\
24+
curl 'http://localhost:3100/loki/api/v1/query_range?query=topk(50,sum%20by(SrcK8S_HostName,DstK8S_HostName)%20(rate(\{app=%22netobserv-flowcollector%22,FlowDirection=%221%22\}|json|unwrap%20Packets|__error__=%22%22\[720s\])))&limit=50&step=360s'\
2525
| jq > ./loki/flow_metrics_host.json
26-
curl 'http://localhost:3100/loki/api/v1/query_range?query=topk(50,sum%20by(SrcK8S_Namespace,DstK8S_Namespace)%20(rate(\{app=%22netobserv-flowcollector%22,FlowDirection=%221%22\}!~`Duplicate%22:true`|json|unwrap%20Packets|__error__=%22%22\[720s\])))&limit=50&step=360s'\
26+
curl 'http://localhost:3100/loki/api/v1/query_range?query=topk(50,sum%20by(SrcK8S_Namespace,DstK8S_Namespace)%20(rate(\{app=%22netobserv-flowcollector%22,FlowDirection=%221%22\}|json|unwrap%20Packets|__error__=%22%22\[720s\])))&limit=50&step=360s'\
2727
| jq > ./loki/flow_metrics_namespace.json
28-
curl 'http://localhost:3100/loki/api/v1/query_range?query=topk(50,sum%20by(SrcK8S_OwnerName,SrcK8S_OwnerType,DstK8S_OwnerName,DstK8S_OwnerType,SrcK8S_Namespace,DstK8S_Namespace)%20(rate(\{app=%22netobserv-flowcollector%22,FlowDirection=%221%22\}!~`Duplicate%22:true`|json|unwrap%20Packets|__error__=%22%22\[720s\])))&limit=50&step=360s'\
28+
curl 'http://localhost:3100/loki/api/v1/query_range?query=topk(50,sum%20by(SrcK8S_OwnerName,SrcK8S_OwnerType,DstK8S_OwnerName,DstK8S_OwnerType,SrcK8S_Namespace,DstK8S_Namespace)%20(rate(\{app=%22netobserv-flowcollector%22,FlowDirection=%221%22\}|json|unwrap%20Packets|__error__=%22%22\[720s\])))&limit=50&step=360s'\
2929
| jq > ./loki/flow_metrics_owner.json
30-
curl 'http://localhost:3100/loki/api/v1/query_range?query=topk(50,sum%20by(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)%20(rate(\{app=%22netobserv-flowcollector%22,FlowDirection=%221%22\}!~`Duplicate%22:true`|json|unwrap%20Packets|__error__=%22%22\[720s\])))&limit=50&step=360s'\
30+
curl 'http://localhost:3100/loki/api/v1/query_range?query=topk(50,sum%20by(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)%20(rate(\{app=%22netobserv-flowcollector%22,FlowDirection=%221%22\}|json|unwrap%20Packets|__error__=%22%22\[720s\])))&limit=50&step=360s'\
3131
| jq > ./loki/flow_metrics_resource.json
3232

3333
echo 'Getting dropped metrics'
34-
curl 'http://localhost:3100/loki/api/v1/query_range?query=topk(5,sum%20by(app)%20(rate(\{app=%22netobserv-flowcollector%22,FlowDirection=%221%22\}!~`Duplicate%22:true`|json|unwrap%20PktDropPackets|__error__=%22%22\[720s\])))&limit=5&step=360s'\
34+
curl 'http://localhost:3100/loki/api/v1/query_range?query=topk(5,sum%20by(app)%20(rate(\{app=%22netobserv-flowcollector%22,FlowDirection=%221%22\}|json|unwrap%20PktDropPackets|__error__=%22%22\[720s\])))&limit=5&step=360s'\
3535
| jq > ./loki/flow_metrics_dropped_app.json
36-
curl 'http://localhost:3100/loki/api/v1/query_range?query=topk(5,sum%20by(PktDropLatestState)%20(rate(\{app=%22netobserv-flowcollector%22,FlowDirection=%221%22\}!~`Duplicate%22:true`|json|unwrap%20PktDropPackets|__error__=%22%22\[720s\])))&limit=5&step=360s'\
36+
curl 'http://localhost:3100/loki/api/v1/query_range?query=topk(5,sum%20by(PktDropLatestState)%20(rate(\{app=%22netobserv-flowcollector%22,FlowDirection=%221%22\}|json|unwrap%20PktDropPackets|__error__=%22%22\[720s\])))&limit=5&step=360s'\
3737
| jq > ./loki/flow_metrics_dropped_state.json
38-
curl 'http://localhost:3100/loki/api/v1/query_range?query=topk(5,sum%20by(PktDropLatestDropCause)%20(rate(\{app=%22netobserv-flowcollector%22,FlowDirection=%221%22\}!~`Duplicate%22:true`|json|unwrap%20PktDropPackets|__error__=%22%22\[720s\])))&limit=5&step=360s'\
38+
curl 'http://localhost:3100/loki/api/v1/query_range?query=topk(5,sum%20by(PktDropLatestDropCause)%20(rate(\{app=%22netobserv-flowcollector%22,FlowDirection=%221%22\}|json|unwrap%20PktDropPackets|__error__=%22%22\[720s\])))&limit=5&step=360s'\
3939
| jq > ./loki/flow_metrics_dropped_cause.json
40-
curl 'http://localhost:3100/loki/api/v1/query_range?query=topk(50,sum%20by(SrcK8S_HostName,DstK8S_HostName)%20(rate(\{app=%22netobserv-flowcollector%22,FlowDirection=%221%22\}!~`Duplicate%22:true`|json|unwrap%20PktDropPackets|__error__=%22%22\[720s\])))&limit=50&step=360s'\
40+
curl 'http://localhost:3100/loki/api/v1/query_range?query=topk(50,sum%20by(SrcK8S_HostName,DstK8S_HostName)%20(rate(\{app=%22netobserv-flowcollector%22,FlowDirection=%221%22\}|json|unwrap%20PktDropPackets|__error__=%22%22\[720s\])))&limit=50&step=360s'\
4141
| jq > ./loki/flow_metrics_dropped_host.json
42-
curl 'http://localhost:3100/loki/api/v1/query_range?query=topk(50,sum%20by(SrcK8S_Namespace,DstK8S_Namespace)%20(rate(\{app=%22netobserv-flowcollector%22,FlowDirection=%221%22\}!~`Duplicate%22:true`|json|unwrap%20PktDropPackets|__error__=%22%22\[720s\])))&limit=50&step=360s'\
42+
curl 'http://localhost:3100/loki/api/v1/query_range?query=topk(50,sum%20by(SrcK8S_Namespace,DstK8S_Namespace)%20(rate(\{app=%22netobserv-flowcollector%22,FlowDirection=%221%22\}|json|unwrap%20PktDropPackets|__error__=%22%22\[720s\])))&limit=50&step=360s'\
4343
| jq > ./loki/flow_metrics_dropped_namespace.json
44-
curl 'http://localhost:3100/loki/api/v1/query_range?query=topk(50,sum%20by(SrcK8S_OwnerName,SrcK8S_OwnerType,DstK8S_OwnerName,DstK8S_OwnerType,SrcK8S_Namespace,DstK8S_Namespace)%20(rate(\{app=%22netobserv-flowcollector%22,FlowDirection=%221%22\}!~`Duplicate%22:true`|json|unwrap%20PktDropPackets|__error__=%22%22\[720s\])))&limit=50&step=360s'\
44+
curl 'http://localhost:3100/loki/api/v1/query_range?query=topk(50,sum%20by(SrcK8S_OwnerName,SrcK8S_OwnerType,DstK8S_OwnerName,DstK8S_OwnerType,SrcK8S_Namespace,DstK8S_Namespace)%20(rate(\{app=%22netobserv-flowcollector%22,FlowDirection=%221%22\}|json|unwrap%20PktDropPackets|__error__=%22%22\[720s\])))&limit=50&step=360s'\
4545
| jq > ./loki/flow_metrics_dropped_owner.json
46-
curl 'http://localhost:3100/loki/api/v1/query_range?query=topk(50,sum%20by(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)%20(rate(\{app=%22netobserv-flowcollector%22,FlowDirection=%221%22\}!~`Duplicate%22:true`|json|unwrap%20PktDropPackets|__error__=%22%22\[720s\])))&limit=50&step=360s'\
46+
curl 'http://localhost:3100/loki/api/v1/query_range?query=topk(50,sum%20by(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)%20(rate(\{app=%22netobserv-flowcollector%22,FlowDirection=%221%22\}|json|unwrap%20PktDropPackets|__error__=%22%22\[720s\])))&limit=50&step=360s'\
4747
| jq > ./loki/flow_metrics_dropped_resource.json
4848

4949
# namespaces.json contains label values for autocomplete

‎pkg/config/config.go‎

Lines changed: 0 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -120,11 +120,6 @@ type FieldConfig struct {
120120
Filter string `yaml:"filter,omitempty" json:"filter,omitempty"`
121121
}
122122

123-
type Deduper struct {
124-
Mark bool `yaml:"mark" json:"mark"`
125-
Merge bool `yaml:"merge" json:"merge"`
126-
}
127-
128123
type Frontend struct {
129124
BuildVersion string `yaml:"buildVersion" json:"buildVersion"`
130125
BuildDate string `yaml:"buildDate" json:"buildDate"`
@@ -138,7 +133,6 @@ type Frontend struct {
138133
AlertNamespaces []string `yaml:"alertNamespaces" json:"alertNamespaces"`
139134
Sampling int `yaml:"sampling" json:"sampling"`
140135
Features []string `yaml:"features" json:"features"`
141-
Deduper Deduper `yaml:"deduper" json:"deduper"`
142136
Fields []FieldConfig `yaml:"fields" json:"fields"`
143137
DataSources []string `yaml:"dataSources" json:"dataSources"`
144138
LokiMocks bool `yaml:"lokiMocks,omitempty" json:"lokiMocks,omitempty"`
@@ -190,10 +184,6 @@ func ReadFile(version, date, filename string) (*Config, error) {
190184
},
191185
QuickFilters: []QuickFilter{},
192186
Features: []string{},
193-
Deduper: Deduper{
194-
Mark: false,
195-
Merge: true,
196-
},
197187
Fields: []FieldConfig{
198188
{Name: "TimeFlowEndMs", Type: "number"},
199189
{Name: "SrcAddr", Type: "string"},

‎pkg/handler/flows.go‎

Lines changed: 2 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,6 @@ import (
55
"errors"
66
"net/http"
77
"net/url"
8-
"slices"
98
"time"
109

1110
"github.com/netobserv/network-observability-console-plugin/pkg/httpclient"
@@ -14,15 +13,13 @@ import (
1413
"github.com/netobserv/network-observability-console-plugin/pkg/model"
1514
"github.com/netobserv/network-observability-console-plugin/pkg/model/fields"
1615
"github.com/netobserv/network-observability-console-plugin/pkg/model/filters"
17-
"github.com/netobserv/network-observability-console-plugin/pkg/utils/constants"
1816
)
1917

2018
const (
2119
startTimeKey = "startTime"
2220
endTimeKey = "endTime"
2321
timeRangeKey = "timeRange"
2422
limitKey = "limit"
25-
dedupKey = "dedup"
2623
recordTypeKey = "recordType"
2724
dataSourceKey = "dataSource"
2825
filtersKey = "filters"
@@ -75,10 +72,6 @@ func (h *Handlers) getFlows(ctx context.Context, lokiClient httpclient.Caller, p
7572
if err != nil {
7673
return nil, http.StatusBadRequest, err
7774
}
78-
dedup := params.Get(dedupKey) == "true"
79-
if !h.Cfg.Frontend.Deduper.Mark || slices.Contains(constants.AnyConnectionType, string(recordType)) {
80-
dedup = false
81-
}
8275
packetLoss, err := getPacketLoss(params)
8376
if err != nil {
8477
return nil, http.StatusBadRequest, err
@@ -107,7 +100,7 @@ func (h *Handlers) getFlows(ctx context.Context, lokiClient httpclient.Caller, p
107100
// match any, and multiple filters => run in parallel then aggregate
108101
var queries []string
109102
for _, group := range filterGroups {
110-
qb := loki.NewFlowQueryBuilder(&h.Cfg.Loki, start, end, limit, dedup, recordType, packetLoss)
103+
qb := loki.NewFlowQueryBuilder(&h.Cfg.Loki, start, end, limit, recordType, packetLoss)
111104
err := qb.Filters(group)
112105
if err != nil {
113106
return nil, http.StatusBadRequest, errors.New("Can't build query: " + err.Error())
@@ -120,7 +113,7 @@ func (h *Handlers) getFlows(ctx context.Context, lokiClient httpclient.Caller, p
120113
}
121114
} else {
122115
// else, run all at once
123-
qb := loki.NewFlowQueryBuilder(&h.Cfg.Loki, start, end, limit, dedup, recordType, packetLoss)
116+
qb := loki.NewFlowQueryBuilder(&h.Cfg.Loki, start, end, limit, recordType, packetLoss)
124117
if len(filterGroups) > 0 {
125118
err := qb.Filters(filterGroups[0])
126119
if err != nil {

‎pkg/handler/topology.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -83,7 +83,7 @@ func (h *Handlers) GetTopology(ctx context.Context) func(w http.ResponseWriter,
8383
}
8484

8585
func (h *Handlers) extractTopologyQueryParams(params url.Values, ds constants.DataSource) (*loki.TopologyInput, filters.MultiQueries, v1.Range, int, error) {
86-
in := loki.TopologyInput{DedupMark: h.Cfg.Frontend.Deduper.Mark, DataSource: ds}
86+
in := loki.TopologyInput{DataSource: ds}
8787
qr := v1.Range{}
8888
var reqLimit int
8989
var err error

‎pkg/loki/flow_query.go‎

Lines changed: 2 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,7 @@ type FlowQueryBuilder struct {
3636
jsonFilters [][]filters.LabelFilter
3737
}
3838

39-
func NewFlowQueryBuilder(cfg *config.Loki, start, end, limit string, dedup bool,
39+
func NewFlowQueryBuilder(cfg *config.Loki, start, end, limit string,
4040
recordType constants.RecordType, packetLoss constants.PacketLoss) *FlowQueryBuilder {
4141
// Always use following stream selectors
4242
labelFilters := []filters.LabelFilter{
@@ -56,14 +56,6 @@ func NewFlowQueryBuilder(cfg *config.Loki, start, end, limit string, dedup bool,
5656
}
5757

5858
lineFilters := []filters.LineFilter{}
59-
if dedup {
60-
if cfg.IsLabel(fields.Duplicate) {
61-
labelFilters = append(labelFilters, filters.NotStringLabelFilter(fields.Duplicate, "true"))
62-
} else {
63-
lineFilters = append(lineFilters, filters.NewEmptyLineFilter(fields.Duplicate, true, false, true).MatchTrue())
64-
}
65-
}
66-
6759
if packetLoss == constants.PacketLossDropped {
6860
// match records that doesn't contains "Packets" field and 1+ packets dropped
6961
// as FLP will ensure the filtering
@@ -95,7 +87,7 @@ func NewFlowQueryBuilder(cfg *config.Loki, start, end, limit string, dedup bool,
9587
}
9688

9789
func NewFlowQueryBuilderWithDefaults(cfg *config.Loki) *FlowQueryBuilder {
98-
return NewFlowQueryBuilder(cfg, "", "", "", false, constants.RecordTypeLog, constants.PacketLossAll)
90+
return NewFlowQueryBuilder(cfg, "", "", "", constants.RecordTypeLog, constants.PacketLossAll)
9991
}
10092

10193
func (q *FlowQueryBuilder) Filters(queryFilters filters.SingleQuery) error {

‎pkg/loki/topology_query.go‎

Lines changed: 1 addition & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,6 @@ type TopologyInput struct {
2626
PacketLoss constants.PacketLoss
2727
Aggregate string
2828
Groups string
29-
DedupMark bool
3029
}
3130

3231
type TopologyQueryBuilder struct {
@@ -36,17 +35,14 @@ type TopologyQueryBuilder struct {
3635
}
3736

3837
func NewTopologyQuery(cfg *config.Loki, kl map[string][]string, in *TopologyInput) (*TopologyQueryBuilder, error) {
39-
var dedup bool
4038
var rt constants.RecordType
4139
if slices.Contains(constants.AnyConnectionType, string(in.RecordType)) {
42-
dedup = false
4340
rt = "endConnection"
4441
} else {
45-
dedup = in.DedupMark
4642
rt = "flowLog"
4743
}
4844

49-
fqb := NewFlowQueryBuilder(cfg, in.Start, in.End, in.Top, dedup, rt, in.PacketLoss)
45+
fqb := NewFlowQueryBuilder(cfg, in.Start, in.End, in.Top, rt, in.PacketLoss)
5046
return &TopologyQueryBuilder{
5147
FlowQueryBuilder: fqb,
5248
topology: in,

0 commit comments

Comments
 (0)