node-neigh: Avoid flooding the same next hop

pkg/datapath/linux/node.go

@@ -21,6 +21,7 @@ import (
 	"net"
 	"os"
 	"reflect"
+	"time"
 
 	"github.com/cilium/cilium/pkg/cidr"
 	"github.com/cilium/cilium/pkg/counter"
@@ -53,32 +54,34 @@ const (
 )
 
 type linuxNodeHandler struct {
-	mutex                lock.Mutex
-	isInitialized        bool
-	nodeConfig           datapath.LocalNodeConfiguration
-	nodeAddressing       datapath.NodeAddressing
-	datapathConfig       DatapathConfiguration
-	nodes                map[nodeTypes.Identity]*nodeTypes.Node
-	enableNeighDiscovery bool
-	neighLock            lock.Mutex // protects neigh* fields below
-	neighDiscoveryLink   netlink.Link
-	neighNextHopByNode   map[nodeTypes.Identity]string // val = string(net.IP)
-	neighNextHopRefCount counter.StringCounter
-	neighByNextHop       map[string]*netlink.Neigh // key = string(net.IP)
-	wgAgent              datapath.WireguardAgent
+	mutex                  lock.Mutex
+	isInitialized          bool
+	nodeConfig             datapath.LocalNodeConfiguration
+	nodeAddressing         datapath.NodeAddressing
+	datapathConfig         DatapathConfiguration
+	nodes                  map[nodeTypes.Identity]*nodeTypes.Node
+	enableNeighDiscovery   bool
+	neighLock              lock.Mutex // protects neigh* fields below
+	neighDiscoveryLink     netlink.Link
+	neighNextHopByNode     map[nodeTypes.Identity]string // val = string(net.IP)
+	neighNextHopRefCount   counter.StringCounter
+	neighByNextHop         map[string]*netlink.Neigh // key = string(net.IP)
+	neighLastPingByNextHop map[string]time.Time      // key = string(net.IP)
+	wgAgent                datapath.WireguardAgent
 }
 
 // NewNodeHandler returns a new node handler to handle node events and
 // implement the implications in the Linux datapath
 func NewNodeHandler(datapathConfig DatapathConfiguration, nodeAddressing datapath.NodeAddressing, wgAgent datapath.WireguardAgent) datapath.NodeHandler {
 	return &linuxNodeHandler{
-		nodeAddressing:       nodeAddressing,
-		datapathConfig:       datapathConfig,
-		nodes:                map[nodeTypes.Identity]*nodeTypes.Node{},
-		neighNextHopByNode:   map[nodeTypes.Identity]string{},
-		neighNextHopRefCount: counter.StringCounter{},
-		neighByNextHop:       map[string]*netlink.Neigh{},
-		wgAgent:              wgAgent,
+		nodeAddressing:         nodeAddressing,
+		datapathConfig:         datapathConfig,
+		nodes:                  map[nodeTypes.Identity]*nodeTypes.Node{},
+		neighNextHopByNode:     map[nodeTypes.Identity]string{},
+		neighNextHopRefCount:   counter.StringCounter{},
+		neighByNextHop:         map[string]*netlink.Neigh{},
+		neighLastPingByNextHop: map[string]time.Time{},
+		wgAgent:                wgAgent,
 	}
 }
 
@@ -696,7 +699,6 @@ func (n *linuxNodeHandler) insertNeighbor(ctx context.Context, newNode *nodeType
 		return
 	}
 	nextHopStr := nextHopIPv4.String()
-
 	scopedLog = scopedLog.WithField(logfields.IPAddr, nextHopIPv4)
 
 	n.neighLock.Lock()
@@ -717,7 +719,8 @@ func (n *linuxNodeHandler) insertNeighbor(ctx context.Context, newNode *nodeType
 						logfields.LinkIndex:    neigh.LinkIndex,
 					}).WithError(err).Info("Unable to remove neighbor entry")
 				}
-				delete(n.neighByNextHop, nextHopStr)
+				delete(n.neighByNextHop, existingNextHopStr)
+				delete(n.neighLastPingByNextHop, existingNextHopStr)
 				if option.Config.NodePortHairpin {
 					neighborsmap.NeighRetire(net.ParseIP(existingNextHopStr))
 				}
@@ -733,6 +736,18 @@ func (n *linuxNodeHandler) insertNeighbor(ctx context.Context, newNode *nodeType
 
 	n.neighNextHopByNode[newNode.Identity()] = nextHopStr
 
+	if refresh {
+		if lastPing, found := n.neighLastPingByNextHop[nextHopStr]; found &&
+			time.Now().Sub(lastPing) < option.Config.ARPPingRefreshPeriod {
+
+			n.neighLock.Unlock()
+			// Last ping was issued less than option.Config.ARPPingRefreshPeriod
+			// ago, so skip it (e.g. to avoid ddos'ing the same GW if nodes are
+			// L3 connected)
+			return
+		}
+	}
+
 	n.neighLock.Unlock() // to allow concurrent arpings below
 
 	// nextHop hasn't been arpinged before OR we are refreshing neigh entry
@@ -748,6 +763,7 @@ func (n *linuxNodeHandler) insertNeighbor(ctx context.Context, newNode *nodeType
 	}
 
 	n.neighLock.Lock()
+	n.neighLastPingByNextHop[nextHopStr] = time.Now()
 	defer n.neighLock.Unlock()
 
 	if hwAddr != nil {
@@ -807,6 +823,7 @@ func (n *linuxNodeHandler) deleteNeighbor(oldNode *nodeTypes.Node) {
 
 		neigh, found := n.neighByNextHop[nextHopStr]
 		delete(n.neighByNextHop, nextHopStr)
+		delete(n.neighLastPingByNextHop, nextHopStr)
 
 		if found {
 			if err := netlink.NeighDel(neigh); err != nil {

pkg/datapath/linux/node_linux_test.go

@@ -1012,6 +1012,9 @@ func (s *linuxPrivilegedIPv4OnlyTestSuite) TestArpPingHandling(c *check.C) {
 	defer func() { option.Config.EnableNodePort = prevNP }()
 	option.Config.EnableNodePort = true
 	dpConfig := DatapathConfiguration{HostDevice: "veth0"}
+	prevARPPeriod := option.Config.ARPPingRefreshPeriod
+	defer func() { option.Config.ARPPingRefreshPeriod = prevARPPeriod }()
+	option.Config.ARPPingRefreshPeriod = time.Duration(10 * time.Millisecond)
 
 	linuxNodeHandler := NewNodeHandler(dpConfig, s.nodeAddressing, nil).(*linuxNodeHandler)
 	c.Assert(linuxNodeHandler, check.Not(check.IsNil))