diff --git a/pkg/agent/agent.go b/pkg/agent/agent.go
index cfab9ffce7ae8b157b271d7e381b5395fc88181f..ceada0ae20bb0560661151c5a98a0c62aaf7c9f5 100644
--- a/pkg/agent/agent.go
+++ b/pkg/agent/agent.go
@@ -422,7 +422,7 @@ func (f *Flows) buildAndStartPipeline(ctx context.Context) (*node.Terminal[[]*fl
rbTracer.SendsTo(accounter)
if f.cfg.Deduper == DeduperFirstCome {
- deduper := node.AsMiddle(flow.Dedupe(f.cfg.DeduperFCExpiry, f.cfg.DeduperJustMark, f.cfg.DeduperMerge),
+ deduper := node.AsMiddle(flow.Dedupe(f.cfg.DeduperFCExpiry, f.cfg.DeduperJustMark, f.cfg.DeduperMerge, f.interfaceNamer),
node.ChannelBufferLen(f.cfg.BuffersLength))
mapTracer.SendsTo(deduper)
accounter.SendsTo(deduper)
diff --git a/pkg/flow/deduper.go b/pkg/flow/deduper.go
index 85d6d91fd417fc79c5343d659b2adcbf67533138..d8cb8ea4bb67a07fe48d61628838255035705e54 100644
--- a/pkg/flow/deduper.go
+++ b/pkg/flow/deduper.go
@@ -8,7 +8,6 @@ import (
"github.com/sirupsen/logrus"
"github.com/netobserv/netobserv-ebpf-agent/pkg/ebpf"
- "github.com/netobserv/netobserv-ebpf-agent/pkg/utils"
)
var dlog = logrus.WithField("component", "flow/Deduper")
@@ -40,7 +39,7 @@ type entry struct {
// (no activity for it during the expiration time)
// The justMark argument tells that the deduper should not drop the duplicate flows but
// set their Duplicate field.
-func Dedupe(expireTime time.Duration, justMark, mergeDup bool) func(in <-chan []*Record, out chan<- []*Record) {
+func Dedupe(expireTime time.Duration, justMark, mergeDup bool, ifaceNamer InterfaceNamer) func(in <-chan []*Record, out chan<- []*Record) {
cache := &deduperCache{
expire: expireTime,
entries: list.New(),
@@ -51,7 +50,7 @@ func Dedupe(expireTime time.Duration, justMark, mergeDup bool) func(in <-chan []
cache.removeExpired()
fwd := make([]*Record, 0, len(records))
for _, record := range records {
- cache.checkDupe(record, justMark, mergeDup, &fwd)
+ cache.checkDupe(record, justMark, mergeDup, &fwd, ifaceNamer)
}
if len(fwd) > 0 {
out <- fwd
@@ -61,7 +60,7 @@ func Dedupe(expireTime time.Duration, justMark, mergeDup bool) func(in <-chan []
}
// checkDupe check current record if its already available nad if not added to fwd records list
-func (c *deduperCache) checkDupe(r *Record, justMark, mergeDup bool, fwd *[]*Record) {
+func (c *deduperCache) checkDupe(r *Record, justMark, mergeDup bool, fwd *[]*Record, ifaceNamer InterfaceNamer) {
mergeEntry := make(map[string]uint8)
rk := r.Id
// zeroes fields from key that should be ignored from the flow comparison
@@ -95,7 +94,7 @@ func (c *deduperCache) checkDupe(r *Record, justMark, mergeDup bool, fwd *[]*Rec
*fwd = append(*fwd, r)
}
if mergeDup {
- ifName := utils.GetInterfaceName(r.Id.IfIndex)
+ ifName := ifaceNamer(int(r.Id.IfIndex))
mergeEntry[ifName] = r.Id.Direction
if dupEntryNew(*fEntry.dupList, mergeEntry) {
*fEntry.dupList = append(*fEntry.dupList, mergeEntry)
@@ -122,7 +121,7 @@ func (c *deduperCache) checkDupe(r *Record, justMark, mergeDup bool, fwd *[]*Rec
expiryTime: timeNow().Add(c.expire),
}
if mergeDup {
- ifName := utils.GetInterfaceName(r.Id.IfIndex)
+ ifName := ifaceNamer(int(r.Id.IfIndex))
mergeEntry[ifName] = r.Id.Direction
r.DupList = append(r.DupList, mergeEntry)
e.dupList = &r.DupList
diff --git a/pkg/flow/deduper_test.go b/pkg/flow/deduper_test.go
index 1bd9cb907d7a945f7cf9262cae8f273248bba3c7..e054191b083149390e49fc946dddac68ac9235b7 100644
--- a/pkg/flow/deduper_test.go
+++ b/pkg/flow/deduper_test.go
@@ -1,13 +1,13 @@
package flow
import (
+ "net"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/netobserv/netobserv-ebpf-agent/pkg/ebpf"
- "github.com/netobserv/netobserv-ebpf-agent/pkg/utils"
)
var (
@@ -70,7 +70,7 @@ func TestDedupe(t *testing.T) {
input := make(chan []*Record, 100)
output := make(chan []*Record, 100)
- go Dedupe(time.Minute, false, false)(input, output)
+ go Dedupe(time.Minute, false, false, interfaceNamer)(input, output)
input <- []*Record{
oneIf2, // record 1 at interface 2: should be accepted
@@ -108,7 +108,7 @@ func TestDedupe_EvictFlows(t *testing.T) {
input := make(chan []*Record, 100)
output := make(chan []*Record, 100)
- go Dedupe(15*time.Second, false, false)(input, output)
+ go Dedupe(15*time.Second, false, false, interfaceNamer)(input, output)
// Should only accept records 1 and 2, at interface 1
input <- []*Record{oneIf1, twoIf1, oneIf2}
@@ -143,7 +143,7 @@ func TestDedupeMerge(t *testing.T) {
input := make(chan []*Record, 100)
output := make(chan []*Record, 100)
- go Dedupe(time.Minute, false, true)(input, output)
+ go Dedupe(time.Minute, false, true, interfaceNamer)(input, output)
input <- []*Record{
oneIf2, // record 1 at interface 2: should be accepted
@@ -155,10 +155,10 @@ func TestDedupeMerge(t *testing.T) {
expectedMap := []map[string]uint8{
{
- utils.GetInterfaceName(oneIf2.Id.IfIndex): oneIf2.Id.Direction,
+ interfaceNamer(int(oneIf2.Id.IfIndex)): oneIf2.Id.Direction,
},
{
- utils.GetInterfaceName(oneIf1.Id.IfIndex): oneIf1.Id.Direction,
+ interfaceNamer(int(oneIf1.Id.IfIndex)): oneIf1.Id.Direction,
},
}
@@ -174,3 +174,11 @@ type timerMock struct {
func (tm *timerMock) Now() time.Time {
return tm.now
}
+
+func interfaceNamer(ifIndex int) string {
+ iface, err := net.InterfaceByIndex(ifIndex)
+ if err != nil {
+ return "unknown"
+ }
+ return iface.Name
+}
diff --git a/pkg/utils/utils.go b/pkg/utils/utils.go
index a5eedfba6f03a992aa0f1419a2004c0d17c71637..f813fd9ca0e0406c4122c3249a5a2ae5bad2541a 100644
--- a/pkg/utils/utils.go
+++ b/pkg/utils/utils.go
@@ -91,11 +91,3 @@ func utsnameStr[T int8 | uint8](in []T) string {
}
return string(out)
}
-
-func GetInterfaceName(ifIndex uint32) string {
- iface, err := net.InterfaceByIndex(int(ifIndex))
- if err != nil {
- return "unknown"
- }
- return iface.Name
-}