Refactor Windows SNAT flows for SNAT policy implementation

Separate common flows and Windows only flows for SNAT.
Add a new snatTable for looking up the SNAT IPs of external traffic.

pkg/agent/agent.go

@@ -294,9 +294,9 @@ func (i *Initializer) initOpenFlowPipeline() error {
 		return err
 	}
 
-	// On Windows platform, extra flows are needed to perform SNAT for the
-	// traffic to external network.
-	if err := i.initExternalConnectivityFlows(); err != nil {
+	// Install OpenFlow entries to enable Pod traffic to external IP
+	// addresses.
+	if err := i.ofClient.InstallExternalFlows(); err != nil {
 		klog.Errorf("Failed to install openflow entries for external connectivity: %v", err)
 		return err
 	}

pkg/agent/agent_linux.go

@@ -35,11 +35,6 @@ func (i *Initializer) initHostNetworkFlows() error {
 	return nil
 }
 
-// initExternalConnectivityFlows returns immediately on Linux. The corresponding functions are provided in routeClient.
-func (i *Initializer) initExternalConnectivityFlows() error {
-	return nil
-}
-
 // getTunnelLocalIP returns local_ip of tunnel port.
 // On linux platform, local_ip option is not needed.
 func (i *Initializer) getTunnelPortLocalIP() net.IP {

pkg/agent/agent_windows.go

@@ -188,19 +188,6 @@ func (i *Initializer) initHostNetworkFlows() error {
 	return nil
 }
 
-// initExternalConnectivityFlows installs OpenFlow entries to SNAT Pod traffic
-// using Node IP, and then Pod could communicate to the external IP addresses.
-func (i *Initializer) initExternalConnectivityFlows() error {
-	if i.nodeConfig.PodIPv4CIDR == nil {
-		return fmt.Errorf("Failed to find valid IPv4 PodCIDR")
-	}
-	// Install OpenFlow entries on the OVS to enable Pod traffic to communicate to external IP addresses.
-	if err := i.ofClient.InstallExternalFlows(); err != nil {
-		return err
-	}
-	return nil
-}
-
 // getTunnelLocalIP returns local_ip of tunnel port
 func (i *Initializer) getTunnelPortLocalIP() net.IP {
 	return i.nodeConfig.NodeIPAddr.IP

pkg/agent/openflow/client.go

@@ -150,10 +150,11 @@ type Client interface {
 	// This function is only used for Windows platform.
 	InstallBridgeUplinkFlows() error
 
-	// InstallExternalFlows sets up flows to enable Pods to communicate to the external IP addresses. The corresponding
-	// OpenFlow entries include: 1) identify the packets from local Pods to the external IP address, 2) mark the traffic
-	// in the connection tracking context, and 3) SNAT the packets with Node IP.
-	// This function is only used for Windows platform.
+	// InstallExternalFlows sets up flows to enable Pods to communicate to
+	// the external IP addresses. The flows identify the packets from local
+	// Pods to the external IP address, and mark the packets to be SNAT'd
+	// with the configured SNAT IPs. On Windows Node, the flows also perform
+	// SNAT with the Openflow NAT action.
 	InstallExternalFlows() error
 
 	// Disconnect disconnects the connection between client and OFSwitch.
@@ -465,19 +466,11 @@ func (c *client) UninstallServiceFlows(svcIP net.IP, svcPort uint16, protocol bi
 }
 
 func (c *client) InstallLoadBalancerServiceFromOutsideFlows(svcIP net.IP, svcPort uint16, protocol binding.Protocol) error {
-	c.replayMutex.RLock()
-	defer c.replayMutex.RUnlock()
-	var flows []binding.Flow
-	flows = append(flows, c.loadBalancerServiceFromOutsideFlow(svcIP, svcPort, protocol))
-	cacheKey := fmt.Sprintf("LoadBalancerService_%s_%d_%s", svcIP, svcPort, protocol)
-	return c.addFlows(c.serviceFlowCache, cacheKey, flows)
+	return c.installLoadBalancerServiceFromOutsideFlows(svcIP, svcPort, protocol)
 }
 
 func (c *client) UninstallLoadBalancerServiceFromOutsideFlows(svcIP net.IP, svcPort uint16, protocol binding.Protocol) error {
-	c.replayMutex.RLock()
-	defer c.replayMutex.RUnlock()
-	cacheKey := fmt.Sprintf("LoadBalancerService_%s_%d_%s", svcIP, svcPort, protocol)
-	return c.deleteFlows(c.serviceFlowCache, cacheKey)
+	return c.uninstallLoadBalancerServiceFromOutsideFlows(svcIP, svcPort, protocol)
 }
 
 func (c *client) InstallClusterServiceFlows() error {
@@ -557,13 +550,7 @@ func (c *client) InstallDefaultTunnelFlows() error {
 }
 
 func (c *client) InstallBridgeUplinkFlows() error {
-	flows := c.hostBridgeUplinkFlows(*c.nodeConfig.PodIPv4CIDR, cookie.Default)
-	c.hostNetworkingFlows = flows
-	if err := c.ofEntryOperations.AddAll(flows); err != nil {
-		return err
-	}
-	c.hostNetworkingFlows = flows
-	return nil
+	return c.installBridgeUplinkFlows()
 }
 
 func (c *client) initialize() error {
@@ -633,8 +620,8 @@ func (c *client) Initialize(roundInfo types.RoundInfo, nodeConfig *config.NodeCo
 func (c *client) InstallExternalFlows() error {
 	nodeIP := c.nodeConfig.NodeIPAddr.IP
 	podSubnet := c.nodeConfig.PodIPv4CIDR
-	flows := c.uplinkSNATFlows(cookie.SNAT)
-	flows = append(flows, c.snatFlows(nodeIP, *podSubnet, cookie.SNAT)...)
+	localGatewayMAC := c.nodeConfig.GatewayConfig.MAC
+	flows := c.externalFlows(nodeIP, *podSubnet, localGatewayMAC)
 	if err := c.ofEntryOperations.AddAll(flows); err != nil {
 		return fmt.Errorf("failed to install flows for external communication: %v", err)
 	}