[WIP] converge cnat refactoring

calico-vpp-agent/cni/cni_pod_test.go

@@ -60,6 +60,8 @@ var _ = Describe("Pod-related functionality of CNI", func() {
 
 	BeforeEach(func() {
 		log = logrus.New()
+		nodeIP4String := net.ParseIP("6.6.6.6")
+		nodeIP6String := net.ParseIP("2001:db8::68")
 		testutils.StartVPP()
 		vpp, _ = testutils.ConfigureVPP(log)
 		// setup connectivity server (functionality target of tests)
@@ -71,6 +73,7 @@ var _ = Describe("Pod-related functionality of CNI", func() {
 		cniServer = cni.NewCNIServer(vpp, ipamStub, log.WithFields(logrus.Fields{"component": "cni"}))
 		cniServer.SetFelixConfig(&felixconfig.Config{})
 		cniServer.FetchBufferConfig()
+		vpp.CnatSetSnatAddresses(nodeIP4String, nodeIP6String)
 	})
 
 	Describe("Addition of the pod", func() {

calico-vpp-agent/cni/podinterface/common.go

@@ -105,7 +105,7 @@ func (i *PodInterfaceDriverData) DoPodIfNatConfiguration(podSpec *model.LocalPod
 		stack.Push(i.vpp.RemovePodInterface, swIfIndex)
 	}
 
-	err = i.vpp.CnatEnableFeatures(swIfIndex)
+	err = i.vpp.CnatEnableFeatures(swIfIndex, true)
 	if err != nil {
 		return errors.Wrapf(err, "error configuring nat on pod interface")
 	}
@@ -129,6 +129,10 @@ func (i *PodInterfaceDriverData) DoPodInterfaceConfiguration(podSpec *model.Loca
 		}
 	}
 
+	err = i.vpp.EnableCnatSNATOnInterfaceVRF(swIfIndex)
+	if err != nil {
+		return errors.Wrapf(err, "error configuring cnat snat on pod VRF")
+	}
 	if !*ifSpec.IsL3 {
 		/* L2 */
 		err = i.vpp.SetPromiscOn(swIfIndex)

calico-vpp-agent/connectivity/connectivity_server.go

@@ -137,6 +137,21 @@ func (s *ConnectivityServer) updateAllIPConnectivity() {
 	}
 }
 
+func (s *ConnectivityServer) configureRemoteNodeSnat(node *common.LocalNodeSpec, isAdd bool) {
+	if node.IPv4Address != nil {
+		err := s.vpp.CnatAddDelSnatPrefix(common.ToMaxLenCIDR(node.IPv4Address.IP), isAdd)
+		if err != nil {
+			s.log.Errorf("error configuring snat prefix for current node (%v): %v", node.IPv4Address.IP, err)
+		}
+	}
+	if node.IPv6Address != nil {
+		err := s.vpp.CnatAddDelSnatPrefix(common.ToMaxLenCIDR(node.IPv6Address.IP), isAdd)
+		if err != nil {
+			s.log.Errorf("error configuring snat prefix for current node (%v): %v", node.IPv6Address.IP, err)
+		}
+	}
+}
+
 func (s *ConnectivityServer) ServeConnectivity(t *tomb.Tomb) error {
 	/**
 	 * There might be leftover state in VPP in case we restarted
@@ -214,6 +229,7 @@ func (s *ConnectivityServer) ServeConnectivity(t *tomb.Tomb) error {
 							delete(s.nodeByAddr, old.IPv6Address.IP.String())
 						}
 					}
+					s.configureRemoteNodeSnat(old, false /* isAdd */)
 				}
 				if evt.New != nil {
 					new, ok := evt.New.(*common.LocalNodeSpec)
@@ -227,6 +243,7 @@ func (s *ConnectivityServer) ServeConnectivity(t *tomb.Tomb) error {
 							s.nodeByAddr[new.IPv6Address.IP.String()] = *new
 						}
 					}
+					s.configureRemoteNodeSnat(new, true /* isAdd */)
 				}
 			case common.FelixConfChanged:
 				old, ok := evt.Old.(*felixConfig.Config)

calico-vpp-agent/connectivity/ipip.go

@@ -127,7 +127,7 @@ func (p *IpipProvider) AddConnectivity(cn *common.NodeConnectivity) error {
 			return errors.Wrapf(err, "Error enabling gso for ipip interface")
 		}
 
-		err = p.vpp.CnatEnableFeatures(swIfIndex)
+		err = p.vpp.CnatEnableFeatures(swIfIndex, true)
 		if err != nil {
 			p.errorCleanup(tunnel)
 			return errors.Wrapf(err, "Error enabling nat for ipip interface")

calico-vpp-agent/connectivity/ipsec.go

@@ -197,7 +197,7 @@ func (p *IpsecProvider) createIPSECTunnel(tunnel *IpsecTunnel, psk string, stack
 		return errors.Wrapf(err, "Error enabling gso for ipip interface")
 	}
 
-	err = p.vpp.CnatEnableFeatures(swIfIndex)
+	err = p.vpp.CnatEnableFeatures(swIfIndex, true)
 	if err != nil {
 		return errors.Wrapf(err, "Error enabling nat for ipip interface")
 	}

calico-vpp-agent/connectivity/vxlan.go

@@ -177,7 +177,7 @@ func (p *VXLanProvider) AddConnectivity(cn *common.NodeConnectivity) error {
 			return errors.Wrapf(err, "Error enabling gso for vxlan interface")
 		}
 
-		err = p.vpp.CnatEnableFeatures(swIfIndex)
+		err = p.vpp.CnatEnableFeatures(swIfIndex, true)
 		if err != nil {
 			// TODO : delete tunnel
 			return errors.Wrapf(err, "Error enabling nat for vxlan interface")

calico-vpp-agent/connectivity/wireguard.go

@@ -217,7 +217,7 @@ func (p *WireguardProvider) createWireguardTunnels() error {
 				return errors.Wrapf(err, "Error enabling gso for wireguard interface")
 			}
 
-			err = p.vpp.CnatEnableFeatures(swIfIndex)
+			err = p.vpp.CnatEnableFeatures(swIfIndex, true)
 			if err != nil {
 				p.errorCleanup(tunnel)
 				return errors.Wrapf(err, "Error enabling nat for wireguard interface")

calico-vpp-agent/felix/felix_server.go

@@ -1215,21 +1215,17 @@ func (s *Server) handleWireguardEndpointRemove(msg *proto.WireguardEndpointRemov
 }
 
 func (s *Server) onNodeUpdated(old *common.LocalNodeSpec, node *common.LocalNodeSpec) (err error) {
-	// This is used by the routing server to process Wireguard key updates
-	// As a result we only send an event when a node is updated, not when it is added or deleted
-	common.SendEvent(common.CalicoVppEvent{
-		Type: common.PeerNodeStateChanged,
-		Old:  old,
-		New:  node,
-	})
 	change := common.GetIPNetChangeType(old.IPv4Address, node.IPv4Address) | common.GetIPNetChangeType(old.IPv6Address, node.IPv6Address)
 	if change&(common.ChangeDeleted|common.ChangeUpdated) != 0 && node.Name == *config.NodeName {
 		// restart if our BGP config changed
 		return NodeWatcherRestartError{}
 	}
 	if change != common.ChangeSame {
-		s.configureRemoteNodeSnat(old, false /* isAdd */)
-		s.configureRemoteNodeSnat(node, true /* isAdd */)
+		common.SendEvent(common.CalicoVppEvent{
+			Type: common.PeerNodeStateChanged,
+			Old:  old,
+			New:  node,
+		})
 	}
 
 	return nil
@@ -1242,12 +1238,21 @@ func (s *Server) onNodeAdded(node *common.LocalNodeSpec) (err error) {
 			/* We found a BGP Spec that seems valid enough */
 			s.GotOurNodeBGPchan <- node
 		}
+		ip4 := net.IP{}
+		ip6 := net.IP{}
 		if node.IPv4Address != nil {
 			s.ip4 = &node.IPv4Address.IP
+			ip4 = node.IPv4Address.IP
 		}
 		if node.IPv6Address != nil {
 			s.ip6 = &node.IPv6Address.IP
+			ip6 = node.IPv6Address.IP
+		}
+		err = s.vpp.CnatSetSnatAddresses(ip4, ip6)
+		if err != nil {
+			s.log.Errorf("Failed to configure SNAT addresses %v", err)
 		}
+
 		err = s.createAllowFromHostPolicy()
 		if err != nil {
 			return errors.Wrap(err, "Error in creating AllowFromHostPolicy")
@@ -1262,26 +1267,10 @@ func (s *Server) onNodeAdded(node *common.LocalNodeSpec) (err error) {
 		Type: common.PeerNodeStateChanged,
 		New:  node,
 	})
-	s.configureRemoteNodeSnat(node, true /* isAdd */)
 
 	return nil
 }
 
-func (s *Server) configureRemoteNodeSnat(node *common.LocalNodeSpec, isAdd bool) {
-	if node.IPv4Address != nil {
-		err := s.vpp.CnatAddDelSnatPrefix(common.ToMaxLenCIDR(node.IPv4Address.IP), isAdd)
-		if err != nil {
-			s.log.Errorf("error configuring snat prefix for current node (%v): %v", node.IPv4Address.IP, err)
-		}
-	}
-	if node.IPv6Address != nil {
-		err := s.vpp.CnatAddDelSnatPrefix(common.ToMaxLenCIDR(node.IPv6Address.IP), isAdd)
-		if err != nil {
-			s.log.Errorf("error configuring snat prefix for current node (%v): %v", node.IPv6Address.IP, err)
-		}
-	}
-}
-
 func (s *Server) onNodeDeleted(old *common.LocalNodeSpec, node *common.LocalNodeSpec) error {
 	common.SendEvent(common.CalicoVppEvent{
 		Type: common.PeerNodeStateChanged,
@@ -1292,7 +1281,6 @@ func (s *Server) onNodeDeleted(old *common.LocalNodeSpec, node *common.LocalNode
 		return NodeWatcherRestartError{}
 	}
 
-	s.configureRemoteNodeSnat(old, false /* isAdd */)
 	return nil
 }
 
@@ -1315,8 +1303,8 @@ func (s *Server) handleIpamPoolUpdate(msg *proto.IPAMPoolUpdate, pending bool) (
 		if newIpamPool.GetCidr() != oldIpamPool.GetCidr() ||
 			newIpamPool.GetMasquerade() != oldIpamPool.GetMasquerade() {
 			var err, err2 error
-			err = s.addDelSnatPrefix(oldIpamPool, false /* isAdd */)
-			err2 = s.addDelSnatPrefix(newIpamPool, true /* isAdd */)
+			err = s.addDelSnatPrefixForIPPool(oldIpamPool, false /* isAdd */)
+			err2 = s.addDelSnatPrefixForIPPool(newIpamPool, true /* isAdd */)
 			if err != nil || err2 != nil {
 				return errors.Errorf("error updating snat prefix del:%s, add:%s", err, err2)
 			}
@@ -1330,7 +1318,7 @@ func (s *Server) handleIpamPoolUpdate(msg *proto.IPAMPoolUpdate, pending bool) (
 		s.log.Infof("Adding pool: %s, nat:%t", msg.GetId(), newIpamPool.GetMasquerade())
 		s.ippoolmap[msg.GetId()] = newIpamPool
 		s.log.Debugf("Pool %v Added, handler called", msg)
-		err = s.addDelSnatPrefix(newIpamPool, true /* isAdd */)
+		err = s.addDelSnatPrefixForIPPool(newIpamPool, true /* isAdd */)
 		if err != nil {
 			return errors.Wrap(err, "error handling ipam add")
 		}
@@ -1356,7 +1344,7 @@ func (s *Server) handleIpamPoolRemove(msg *proto.IPAMPoolRemove, pending bool) (
 		delete(s.ippoolmap, msg.GetId())
 		s.log.Infof("Deleting pool: %s", msg.GetId())
 		s.log.Debugf("Pool %s deleted, handler called", oldIpamPool.Cidr)
-		err = s.addDelSnatPrefix(oldIpamPool, false /* isAdd */)
+		err = s.addDelSnatPrefixForIPPool(oldIpamPool, false /* isAdd */)
 		if err != nil {
 			return errors.Wrap(err, "error handling ipam deletion")
 		}
@@ -1401,12 +1389,12 @@ func ipamPoolEquals(a *proto.IPAMPool, b *proto.IPAMPool) bool {
 	return true
 }
 
-// addDelSnatPrefix configures IP Pool prefixes so that we don't source-NAT the packets going
+// addDelSnatPrefixForIPPool configures IP Pool prefixes so that we don't source-NAT the packets going
 // to these addresses. All the IP Pools prefixes are configured that way so that pod <-> pod
 // communications are never source-nated in the cluster
 // Note(aloaugus) - I think the iptables dataplane behaves differently and uses the k8s level
 // pod CIDR for this rather than the individual pool prefixes
-func (s *Server) addDelSnatPrefix(pool *proto.IPAMPool, isAdd bool) (err error) {
+func (s *Server) addDelSnatPrefixForIPPool(pool *proto.IPAMPool, isAdd bool) (err error) {
 	_, ipNet, err := net.ParseCIDR(pool.GetCidr())
 	if err != nil {
 		return errors.Wrapf(err, "Couldn't parse pool CIDR %s", pool.Cidr)

calico-vpp-agent/services/service_server.go

@@ -305,10 +305,6 @@ func serviceID(meta *metav1.ObjectMeta) string {
 }
 
 func (s *Server) configureSnat() (err error) {
-	err = s.vpp.CnatSetSnatAddresses(s.getNodeIP(false /* isv6 */), s.getNodeIP(true /* isv6 */))
-	if err != nil {
-		s.log.Errorf("Failed to configure SNAT addresses %v", err)
-	}
 	nodeIP4, nodeIP6 := common.GetBGPSpecAddresses(s.nodeBGPSpec)
 	if nodeIP6 != nil {
 		err = s.vpp.CnatAddSnatPrefix(common.FullyQualified(*nodeIP6))
@@ -328,6 +324,23 @@ func (s *Server) configureSnat() (err error) {
 			s.log.Errorf("Failed to Add Service CIDR %s %v", serviceCIDR, err)
 		}
 	}
+	err = s.vpp.SetK8sSnatPolicy()
+	if err != nil {
+		return errors.Wrap(err, "Error configuring cnat source policy")
+	}
+	for _, uplink := range common.VppManagerInfo.UplinkStatuses {
+		// register vpptap0
+		err = s.vpp.RegisterPodInterface(uplink.TapSwIfIndex)
+		if err != nil {
+			return errors.Wrap(err, "error configuring vpptap0 as pod intf")
+		}
+
+		err = s.vpp.RegisterHostInterface(uplink.TapSwIfIndex)
+		if err != nil {
+			return errors.Wrap(err, "error configuring vpptap0 as host intf")
+		}
+	}
+
 	return nil
 }
 
@@ -498,11 +511,6 @@ func (s *Server) ServeService(t *tomb.Tomb) error {
 		})
 	}
 
-	err = s.vpp.CnatPurge()
-	if err != nil {
-		return err
-	}
-
 	if *config.GetCalicoVppDebug().ServicesEnabled {
 		s.t.Go(func() error { s.serviceInformer.Run(t.Dying()); return nil })
 		s.t.Go(func() error { s.endpointInformer.Run(t.Dying()); return nil })

vpp-manager/vpp_runner.go

@@ -460,7 +460,7 @@ func (v *VppRunner) configureVppUplinkInterface(
 		return errors.Wrap(err, "Error disabling ipv6 RA on uplink interface")
 	}
 
-	err = v.vpp.CnatEnableFeatures(ifSpec.SwIfIndex)
+	err = v.vpp.CnatEnableFeatures(ifSpec.SwIfIndex, true)
 	if err != nil {
 		return errors.Wrap(err, "Error configuring NAT on uplink interface")
 	}
@@ -606,21 +606,11 @@ func (v *VppRunner) configureVppUplinkInterface(
 		log.Errorf("Error SetInterfaceRxMode on vpptap0 %v", err)
 	}
 
-	err = v.vpp.CnatEnableFeatures(tapSwIfIndex)
+	err = v.vpp.CnatEnableFeatures(tapSwIfIndex, true)
 	if err != nil {
 		return errors.Wrap(err, "Error configuring NAT on vpptap0")
 	}
 
-	err = v.vpp.RegisterPodInterface(tapSwIfIndex)
-	if err != nil {
-		return errors.Wrap(err, "error configuring vpptap0 as pod intf")
-	}
-
-	err = v.vpp.RegisterHostInterface(tapSwIfIndex)
-	if err != nil {
-		return errors.Wrap(err, "error configuring vpptap0 as host intf")
-	}
-
 	// Linux side tap setup
 	link, err := netlink.LinkByName(ifSpec.InterfaceName)
 	if err != nil {
@@ -659,11 +649,6 @@ func (v *VppRunner) doVppGlobalConfiguration() (err error) {
 		return errors.Wrap(err, "Error creating static VRFs in VPP")
 	}
 
-	err = v.vpp.SetK8sSnatPolicy()
-	if err != nil {
-		return errors.Wrap(err, "Error configuring cnat source policy")
-	}
-
 	err = v.vpp.ConfigureNeighborsV4(&types.NeighborConfig{
 		MaxNumber: *config.GetCalicoVppInitialConfig().IP4NeighborsMaxNumber,
 		MaxAge:    *config.GetCalicoVppInitialConfig().IP4NeighborsMaxAge,