ICMP Rule with NodeLatencyMonitor

build/yamls/antrea.yml

@@ -4078,7 +4078,7 @@ data:
     #  L7FlowExporter: false
 
     # Enable NodeLatencyMonitor to monitor the latency between Nodes.
-    #  NodeLatencyMonitor: false
+      NodeLatencyMonitor: true
 
     # Allow users to initiate BGP process on selected Kubernetes Nodes and advertise Service IPs, Pod IPs and Egress IPs to
     # remote BGP peers.

cmd/antrea-agent/agent.go

@@ -154,6 +154,8 @@
 	enableBridgingMode := enableAntreaIPAM && o.config.EnableBridgingMode
 	l7NetworkPolicyEnabled := features.DefaultFeatureGate.Enabled(features.L7NetworkPolicy)
 	nodeNetworkPolicyEnabled := features.DefaultFeatureGate.Enabled(features.NodeNetworkPolicy)
+	klog.Infof("DBUG: featuregates: %v", o.config.FeatureGates[string(features.NodeLatencyMonitor)])
+	nodeLatencyMonitorEnabled := o.config.FeatureGates[string(features.NodeLatencyMonitor)]
 	l7FlowExporterEnabled := features.DefaultFeatureGate.Enabled(features.L7FlowExporter)
 	enableMulticlusterGW := features.DefaultFeatureGate.Enabled(features.Multicluster) && o.config.Multicluster.EnableGateway
 	_, multiclusterEncryptionMode := config.GetTrafficEncryptionModeFromStr(o.config.Multicluster.TrafficEncryptionMode)
@@ -241,10 +243,11 @@
 		o.config.AntreaProxy.ProxyAll,
 		connectUplinkToBridge,
 		nodeNetworkPolicyEnabled,
+		nodeLatencyMonitorEnabled,
 		multicastEnabled,
 		o.config.SNATFullyRandomPorts,
 		*o.config.Egress.SNATFullyRandomPorts,
 		serviceCIDRProvider)
 	if err != nil {
 		return fmt.Errorf("error creating route client: %v", err)
 	}

pkg/agent/monitortool/monitor.go

@@ -183,6 +183,7 @@ func (m *NodeLatencyMonitor) onNodeDelete(obj interface{}) {
 
 // onNodeLatencyMonitorAdd is the event handler for adding NodeLatencyMonitor.
 func (m *NodeLatencyMonitor) onNodeLatencyMonitorAdd(obj interface{}) {
+	// TODO maybe add the change here
 	nlm := obj.(*v1alpha1.NodeLatencyMonitor)
 	klog.V(4).InfoS("NodeLatencyMonitor added", "NodeLatencyMonitor", klog.KObj(nlm))
 

pkg/agent/route/route_linux.go

@@ -122,12 +122,14 @@ type Client struct {
 	// markToSNATIP caches marks to SNAT IPs. It's used in Egress feature.
 	markToSNATIP sync.Map
 	// iptablesInitialized is used to notify when iptables initialization is done.
-	iptablesInitialized      chan struct{}
-	proxyAll                 bool
-	connectUplinkToBridge    bool
-	multicastEnabled         bool
-	isCloudEKS               bool
-	nodeNetworkPolicyEnabled bool
+	iptablesInitialized       chan struct{}
+	proxyAll                  bool
+	connectUplinkToBridge     bool
+	multicastEnabled          bool
+	isCloudEKS                bool
+	nodeNetworkPolicyEnabled  bool
+	nodeLatencyMonitorEnabled bool
+	networkPolicyOnlyMode     bool
 	// serviceRoutes caches ip routes about Services.
 	serviceRoutes sync.Map
 	// serviceExternalIPReferences tracks the references of Service IP. The key is the Service IP and the value is
@@ -170,6 +172,7 @@ func NewClient(networkConfig *config.NetworkConfig,
 	proxyAll bool,
 	connectUplinkToBridge bool,
 	nodeNetworkPolicyEnabled bool,
+	nodeLatencyMonitorEnabled bool,
 	multicastEnabled bool,
 	nodeSNATRandomFully bool,
 	egressSNATRandomFully bool,
@@ -183,6 +186,7 @@ func NewClient(networkConfig *config.NetworkConfig,
 		multicastEnabled:            multicastEnabled,
 		connectUplinkToBridge:       connectUplinkToBridge,
 		nodeNetworkPolicyEnabled:    nodeNetworkPolicyEnabled,
+		nodeLatencyMonitorEnabled:   nodeLatencyMonitorEnabled,
 		ipset:                       ipset.NewClient(),
 		netlink:                     &netlink.Handle{},
 		isCloudEKS:                  env.IsCloudEKS(),
@@ -265,6 +269,9 @@ func (c *Client) Initialize(nodeConfig *config.NodeConfig, done func()) error {
 	if c.nodeNetworkPolicyEnabled {
 		c.initNodeNetworkPolicy()
 	}
+	if c.nodeLatencyMonitorEnabled {
+		c.initNodeLatency()
+	}
 
 	return nil
 }
@@ -675,7 +682,7 @@ func (c *Client) syncIPTables() error {
 	if c.proxyAll {
 		jumpRules = append(jumpRules, jumpRule{iptables.NATTable, iptables.OutputChain, antreaOutputChain, "Antrea: jump to Antrea output rules", true})
 	}
-	if c.nodeNetworkPolicyEnabled {
+	if c.nodeNetworkPolicyEnabled || c.nodeLatencyMonitorEnabled {
 		jumpRules = append(jumpRules, jumpRule{iptables.FilterTable, iptables.InputChain, antreaInputChain, "Antrea: jump to Antrea input rules", false})
 		jumpRules = append(jumpRules, jumpRule{iptables.FilterTable, iptables.OutputChain, antreaOutputChain, "Antrea: jump to Antrea output rules", false})
 	}
@@ -1198,6 +1205,38 @@ func (c *Client) initNodeNetworkPolicy() {
 	}
 }
 
+func (c *Client) initNodeLatency() {
+	gateway := "antrea-gw0"
+	if c.networkConfig.TrafficEncapMode.String() == "networkPolicyOnly" {
+		gateway = "transport"
+	}
+	antreaInputChainRules := []string{
+		iptables.NewRuleBuilder(antreaInputChain).
+			MatchInputInterface(gateway).
+			SetComment("Antrea: allow ICMP packets from NodeLatencyMonitor").
+			SetTarget(iptables.AcceptTarget).
+			Done().
+			GetRule(),
+	}
+	antreaOutputChainRules := []string{
+		iptables.NewRuleBuilder(antreaOutputChain).
+			MatchOutputInterface(gateway).
+			SetComment("Antrea: allow egress packets from NodeLatencyMonitor").
+			SetTarget(iptables.AcceptTarget).
+			Done().
+			GetRule(),
+	}
+
+	if c.networkConfig.IPv6Enabled {
+		c.nodeNetworkPolicyIPTablesIPv6.Store(antreaInputChain, antreaInputChainRules)
+		c.nodeNetworkPolicyIPTablesIPv6.Store(antreaOutputChain, antreaOutputChainRules)
+	}
+	if c.networkConfig.IPv4Enabled {
+		c.nodeNetworkPolicyIPTablesIPv4.Store(antreaInputChain, antreaInputChainRules)
+		c.nodeNetworkPolicyIPTablesIPv4.Store(antreaOutputChain, antreaOutputChainRules)
+	}
+}
+
 // Reconcile removes orphaned podCIDRs from ipset and removes routes to orphaned podCIDRs
 // based on the desired podCIDRs.
 func (c *Client) Reconcile(podCIDRs []string) error {