NETOBSERV-2126: reintroduce merge option, and cleanup old dedup code

config/sample-config.yaml

@@ -13,7 +13,6 @@ loki:
     - DstK8S_Type
     - K8S_FlowLayer
     - FlowDirection
-    #    - Duplicate
     #    - _RecordType
     #    - K8S_ClusterName
     #    - SrcK8S_Zone
@@ -79,9 +78,6 @@ frontend:
   alertNamespaces:
     - netobserv
   sampling: 50
-  deduper:
-    mark: true
-    merge: false
   # The following configuration is taken from Network Observability Operator
   # see https://github.com/netobserv/network-observability-operator/blob/main/controllers/consoleplugin/config/static-frontend-config.yaml
   panels:
@@ -1474,9 +1470,6 @@ frontend:
     - name: IcmpCode
       type: number
       description: ICMP code
-    - name: Duplicate
-      type: boolean
-      description: Indicates if this flow was also captured from another interface on the same host
     - name: FlowDirection
       type: number
       description: |

mocks/updateMocks.sh

@@ -15,35 +15,35 @@ curl 'http://localhost:3100/loki/api/v1/query_range?query=\{app=%22netobserv-flo
 
 # flow_metrics_*.json contains queries result for overview / topology display
 echo 'Getting metrics'
-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'\
+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'\
  | jq > ./loki/flow_metrics_app.json
-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'\
+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'\
  | jq > ./loki/flow_metrics_cluster.json
- 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'\
+ 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'\
  | jq > ./loki/flow_metrics_zone.json
-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'\
+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'\
  | jq > ./loki/flow_metrics_host.json
-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'\
+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'\
  | jq > ./loki/flow_metrics_namespace.json
-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'\
+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'\
  | jq > ./loki/flow_metrics_owner.json
-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'\
+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'\
  | jq > ./loki/flow_metrics_resource.json
 
 echo 'Getting dropped metrics'
-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'\
+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'\
  | jq > ./loki/flow_metrics_dropped_app.json
-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'\
+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'\
  | jq > ./loki/flow_metrics_dropped_state.json
-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'\
+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'\
  | jq > ./loki/flow_metrics_dropped_cause.json
-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'\
+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'\
  | jq > ./loki/flow_metrics_dropped_host.json
-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'\
+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'\
  | jq > ./loki/flow_metrics_dropped_namespace.json
-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'\
+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'\
  | jq > ./loki/flow_metrics_dropped_owner.json
-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'\
+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'\
  | jq > ./loki/flow_metrics_dropped_resource.json
 
 # namespaces.json contains label values for autocomplete

pkg/config/config.go

@@ -120,11 +120,6 @@ type FieldConfig struct {
 	Filter string `yaml:"filter,omitempty" json:"filter,omitempty"`
 }
 
-type Deduper struct {
-	Mark  bool `yaml:"mark" json:"mark"`
-	Merge bool `yaml:"merge" json:"merge"`
-}
-
 type Frontend struct {
 	BuildVersion    string        `yaml:"buildVersion" json:"buildVersion"`
 	BuildDate       string        `yaml:"buildDate" json:"buildDate"`
@@ -138,7 +133,6 @@ type Frontend struct {
 	AlertNamespaces []string      `yaml:"alertNamespaces" json:"alertNamespaces"`
 	Sampling        int           `yaml:"sampling" json:"sampling"`
 	Features        []string      `yaml:"features" json:"features"`
-	Deduper         Deduper       `yaml:"deduper" json:"deduper"`
 	Fields          []FieldConfig `yaml:"fields" json:"fields"`
 	DataSources     []string      `yaml:"dataSources" json:"dataSources"`
 	LokiMocks       bool          `yaml:"lokiMocks,omitempty" json:"lokiMocks,omitempty"`
@@ -190,10 +184,6 @@ func ReadFile(version, date, filename string) (*Config, error) {
 			},
 			QuickFilters: []QuickFilter{},
 			Features:     []string{},
-			Deduper: Deduper{
-				Mark:  false,
-				Merge: true,
-			},
 			Fields: []FieldConfig{
 				{Name: "TimeFlowEndMs", Type: "number"},
 				{Name: "SrcAddr", Type: "string"},

pkg/handler/flows.go

@@ -5,7 +5,6 @@ import (
 	"errors"
 	"net/http"
 	"net/url"
-	"slices"
 	"time"
 
 	"github.com/netobserv/network-observability-console-plugin/pkg/httpclient"
@@ -14,15 +13,13 @@ import (
 	"github.com/netobserv/network-observability-console-plugin/pkg/model"
 	"github.com/netobserv/network-observability-console-plugin/pkg/model/fields"
 	"github.com/netobserv/network-observability-console-plugin/pkg/model/filters"
-	"github.com/netobserv/network-observability-console-plugin/pkg/utils/constants"
 )
 
 const (
 	startTimeKey  = "startTime"
 	endTimeKey    = "endTime"
 	timeRangeKey  = "timeRange"
 	limitKey      = "limit"
-	dedupKey      = "dedup"
 	recordTypeKey = "recordType"
 	dataSourceKey = "dataSource"
 	filtersKey    = "filters"
@@ -75,10 +72,6 @@ func (h *Handlers) getFlows(ctx context.Context, lokiClient httpclient.Caller, p
 	if err != nil {
 		return nil, http.StatusBadRequest, err
 	}
-	dedup := params.Get(dedupKey) == "true"
-	if !h.Cfg.Frontend.Deduper.Mark || slices.Contains(constants.AnyConnectionType, string(recordType)) {
-		dedup = false
-	}
 	packetLoss, err := getPacketLoss(params)
 	if err != nil {
 		return nil, http.StatusBadRequest, err
@@ -107,7 +100,7 @@ func (h *Handlers) getFlows(ctx context.Context, lokiClient httpclient.Caller, p
 		// match any, and multiple filters => run in parallel then aggregate
 		var queries []string
 		for _, group := range filterGroups {
-			qb := loki.NewFlowQueryBuilder(&h.Cfg.Loki, start, end, limit, dedup, recordType, packetLoss)
+			qb := loki.NewFlowQueryBuilder(&h.Cfg.Loki, start, end, limit, recordType, packetLoss)
 			err := qb.Filters(group)
 			if err != nil {
 				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
 		}
 	} else {
 		// else, run all at once
-		qb := loki.NewFlowQueryBuilder(&h.Cfg.Loki, start, end, limit, dedup, recordType, packetLoss)
+		qb := loki.NewFlowQueryBuilder(&h.Cfg.Loki, start, end, limit, recordType, packetLoss)
 		if len(filterGroups) > 0 {
 			err := qb.Filters(filterGroups[0])
 			if err != nil {

pkg/handler/topology.go

@@ -83,7 +83,7 @@ func (h *Handlers) GetTopology(ctx context.Context) func(w http.ResponseWriter,
 }
 
 func (h *Handlers) extractTopologyQueryParams(params url.Values, ds constants.DataSource) (*loki.TopologyInput, filters.MultiQueries, v1.Range, int, error) {
-	in := loki.TopologyInput{DedupMark: h.Cfg.Frontend.Deduper.Mark, DataSource: ds}
+	in := loki.TopologyInput{DataSource: ds}
 	qr := v1.Range{}
 	var reqLimit int
 	var err error

pkg/loki/flow_query.go

@@ -36,7 +36,7 @@ type FlowQueryBuilder struct {
 	jsonFilters  [][]filters.LabelFilter
 }
 
-func NewFlowQueryBuilder(cfg *config.Loki, start, end, limit string, dedup bool,
+func NewFlowQueryBuilder(cfg *config.Loki, start, end, limit string,
 	recordType constants.RecordType, packetLoss constants.PacketLoss) *FlowQueryBuilder {
 	// Always use following stream selectors
 	labelFilters := []filters.LabelFilter{
@@ -56,14 +56,6 @@ func NewFlowQueryBuilder(cfg *config.Loki, start, end, limit string, dedup bool,
 	}
 
 	lineFilters := []filters.LineFilter{}
-	if dedup {
-		if cfg.IsLabel(fields.Duplicate) {
-			labelFilters = append(labelFilters, filters.NotStringLabelFilter(fields.Duplicate, "true"))
-		} else {
-			lineFilters = append(lineFilters, filters.NewEmptyLineFilter(fields.Duplicate, true, false, true).MatchTrue())
-		}
-	}
-
 	if packetLoss == constants.PacketLossDropped {
 		// match records that doesn't contains "Packets" field and 1+ packets dropped
 		// as FLP will ensure the filtering
@@ -95,7 +87,7 @@ func NewFlowQueryBuilder(cfg *config.Loki, start, end, limit string, dedup bool,
 }
 
 func NewFlowQueryBuilderWithDefaults(cfg *config.Loki) *FlowQueryBuilder {
-	return NewFlowQueryBuilder(cfg, "", "", "", false, constants.RecordTypeLog, constants.PacketLossAll)
+	return NewFlowQueryBuilder(cfg, "", "", "", constants.RecordTypeLog, constants.PacketLossAll)
 }
 
 func (q *FlowQueryBuilder) Filters(queryFilters filters.SingleQuery) error {

pkg/loki/topology_query.go

@@ -26,7 +26,6 @@ type TopologyInput struct {
 	PacketLoss     constants.PacketLoss
 	Aggregate      string
 	Groups         string
-	DedupMark      bool
 }
 
 type TopologyQueryBuilder struct {
@@ -36,17 +35,14 @@ type TopologyQueryBuilder struct {
 }
 
 func NewTopologyQuery(cfg *config.Loki, kl map[string][]string, in *TopologyInput) (*TopologyQueryBuilder, error) {
-	var dedup bool
 	var rt constants.RecordType
 	if slices.Contains(constants.AnyConnectionType, string(in.RecordType)) {
-		dedup = false
 		rt = "endConnection"
 	} else {
-		dedup = in.DedupMark
 		rt = "flowLog"
 	}
 
-	fqb := NewFlowQueryBuilder(cfg, in.Start, in.End, in.Top, dedup, rt, in.PacketLoss)
+	fqb := NewFlowQueryBuilder(cfg, in.Start, in.End, in.Top, rt, in.PacketLoss)
 	return &TopologyQueryBuilder{
 		FlowQueryBuilder:   fqb,
 		topology:           in,

pkg/loki/topology_query_test.go

@@ -39,15 +39,14 @@ func TestBuildTopologyQuery_SimpleAggregate(t *testing.T) {
 		RecordType:     constants.RecordTypeLog,
 		DataSource:     constants.DataSourceAuto,
 		Aggregate:      "namespace",
-		DedupMark:      true,
 	}
 	q, err := NewTopologyQuery(&lokiConfig, aggregateKeyLabels, &in)
 	require.NoError(t, err)
 	result := q.Build()
 	assert.Equal(
 		t,
 		"http://loki/loki/api/v1/query_range?query="+
-			"topk(50,sum by(SrcK8S_Namespace,DstK8S_Namespace)(rate({app=\"netobserv-flowcollector\"}!~`Duplicate\":true`|json|unwrap Bytes|__error__=\"\"[2m])))&start=(start)&limit=50&step=10s",
+			"topk(50,sum by(SrcK8S_Namespace,DstK8S_Namespace)(rate({app=\"netobserv-flowcollector\"}|json|unwrap Bytes|__error__=\"\"[2m])))&start=(start)&limit=50&step=10s",
 		result,
 	)
 }
@@ -65,15 +64,14 @@ func TestBuildTopologyQuery_GroupsAndAggregate(t *testing.T) {
 		DataSource:     constants.DataSourceAuto,
 		Aggregate:      "namespace",
 		Groups:         "hosts",
-		DedupMark:      true,
 	}
 	q, err := NewTopologyQuery(&lokiConfig, aggregateKeyLabels, &in)
 	require.NoError(t, err)
 	result := q.Build()
 	assert.Equal(
 		t,
 		"http://loki/loki/api/v1/query_range?query="+
-			"topk(50,sum by(SrcK8S_Namespace,DstK8S_Namespace,SrcK8S_HostName,DstK8S_HostName)(rate({app=\"netobserv-flowcollector\"}!~`Duplicate\":true`|json|unwrap Bytes|__error__=\"\"[2m])))&start=(start)&limit=50&step=10s",
+			"topk(50,sum by(SrcK8S_Namespace,DstK8S_Namespace,SrcK8S_HostName,DstK8S_HostName)(rate({app=\"netobserv-flowcollector\"}|json|unwrap Bytes|__error__=\"\"[2m])))&start=(start)&limit=50&step=10s",
 		result,
 	)
 }
@@ -90,15 +88,14 @@ func TestBuildTopologyQuery_CustomAggregate(t *testing.T) {
 		RecordType:     constants.RecordTypeLog,
 		DataSource:     constants.DataSourceAuto,
 		Aggregate:      "SomeField",
-		DedupMark:      true,
 	}
 	q, err := NewTopologyQuery(&lokiConfig, aggregateKeyLabels, &in)
 	require.NoError(t, err)
 	result := q.Build()
 	assert.Equal(
 		t,
 		"http://loki/loki/api/v1/query_range?query="+
-			"topk(50,sum by(SomeField)(rate({app=\"netobserv-flowcollector\"}!~`Duplicate\":true`|~`\"SomeField\"`|json|unwrap Bytes|__error__=\"\"[2m])))&start=(start)&limit=50&step=10s",
+			"topk(50,sum by(SomeField)(rate({app=\"netobserv-flowcollector\"}|~`\"SomeField\"`|json|unwrap Bytes|__error__=\"\"[2m])))&start=(start)&limit=50&step=10s",
 		result,
 	)
 }
@@ -115,15 +112,14 @@ func TestBuildTopologyQuery_CustomLabelAggregate(t *testing.T) {
 		RecordType:     constants.RecordTypeLog,
 		DataSource:     constants.DataSourceAuto,
 		Aggregate:      "FlowDirection",
-		DedupMark:      true,
 	}
 	q, err := NewTopologyQuery(&lokiConfig, aggregateKeyLabels, &in)
 	require.NoError(t, err)
 	result := q.Build()
 	assert.Equal(
 		t,
 		"http://loki/loki/api/v1/query_range?query="+
-			"topk(50,sum by(FlowDirection)(rate({app=\"netobserv-flowcollector\"}!~`Duplicate\":true`|json|unwrap Bytes|__error__=\"\"[2m])))&start=(start)&limit=50&step=10s",
+			"topk(50,sum by(FlowDirection)(rate({app=\"netobserv-flowcollector\"}|json|unwrap Bytes|__error__=\"\"[2m])))&start=(start)&limit=50&step=10s",
 		result,
 	)
 }

pkg/model/fields/fields.go

@@ -53,7 +53,6 @@ const (
 	DNSLatency             = "DnsLatencyMs"
 	DNSErrNo               = "DnsErrno"
 	DNSCode                = "DnsFlagsResponseCode"
-	Duplicate              = "Duplicate"
 	TimeFlowRTT            = "TimeFlowRttNs"
 	TCPFlags               = "Flags"
 	XlatSrcPort            = "XlatSrcPort"