Split Connectivity into watcher/handler under felix

This patch moves the Connectivity handlers in the main felix
loop to allow lockless access to the cache.

The intent is to move away from a model with multiple servers
replicating state and communicating over a pubsub. This being
prone to race conditions, deadlocks, and not providing many
benefits as scale & asynchronicity will not be a constraint
on nodes with relatively small number of pods (~100) as is k8s
default.

Signed-off-by: Nathan Skrzypczak <nathan.skrzypczak@gmail.com>

Author: sknat

calico-vpp-agent/cmd/calico_vpp_dataplane.go

@@ -26,7 +26,6 @@ import (
 	apipb "github.com/osrg/gobgp/v3/api"
 	bgpserver "github.com/osrg/gobgp/v3/pkg/server"
 	"github.com/pkg/errors"
-	felixconfig "github.com/projectcalico/calico/felix/config"
 	calicov3cli "github.com/projectcalico/calico/libcalico-go/lib/clientv3"
 	"github.com/sirupsen/logrus"
 	"google.golang.org/grpc"
@@ -35,7 +34,6 @@ import (
 	"k8s.io/client-go/rest"
 
 	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/common"
-	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/connectivity"
 	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/felix"
 	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/health"
 	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/routing"
@@ -158,7 +156,6 @@ func main() {
 	if err != nil {
 		log.Fatalf("could not install felix plugin: %s", err)
 	}
-	connectivityServer := connectivity.NewConnectivityServer(vpp, felixServer, clientv3, log.WithFields(logrus.Fields{"subcomponent": "connectivity"}))
 
 	/* Pubsub should now be registered */
 
@@ -185,14 +182,13 @@ func main() {
 	ticker := time.NewTicker(10 * time.Second)
 	defer ticker.Stop()
 
-	var felixConfig *felixconfig.Config
 	var ourBGPSpec *common.LocalNodeSpec
 	felixConfigReceived := false
 	bgpSpecReceived := false
 
 	for !felixConfigReceived || !bgpSpecReceived {
 		select {
-		case felixConfig = <-felixServer.FelixConfigChan:
+		case <-felixServer.GotFelixConfig():
 			felixConfigReceived = true
 			log.Info("FelixConfig received from calico pod")
 		case ourBGPSpec = <-felixServer.GotOurNodeBGPchan():
@@ -216,7 +212,6 @@ func main() {
 	log.Info("Felix configuration received")
 
 	prefixWatcher.SetOurBGPSpec(ourBGPSpec)
-	connectivityServer.SetOurBGPSpec(ourBGPSpec)
 	routingServer.SetOurBGPSpec(ourBGPSpec)
 	serviceServer.SetOurBGPSpec(ourBGPSpec)
 	localSIDWatcher.SetOurBGPSpec(ourBGPSpec)
@@ -234,15 +229,12 @@ func main() {
 		}
 	}
 
-	connectivityServer.SetFelixConfig(felixConfig)
-
 	Go(routeWatcher.WatchRoutes)
 	Go(linkWatcher.WatchLinks)
 	Go(bgpConfigurationWatcher.WatchBGPConfiguration)
 	Go(prefixWatcher.WatchPrefix)
 	Go(peerWatcher.WatchBGPPeers)
 	Go(bgpFilterWatcher.WatchBGPFilters)
-	Go(connectivityServer.ServeConnectivity)
 	Go(routingServer.ServeRouting)
 	Go(serviceServer.ServeService)
 	Go(cniServer.ServeCNI)

calico-vpp-agent/common/common.go

@@ -557,24 +557,6 @@ func FormatBGPConfiguration(conf *calicov3.BGPConfigurationSpec) string {
 	)
 }
 
-func FetchNDataThreads(vpp *vpplink.VppLink, log *logrus.Entry) int {
-	nVppWorkers, err := vpp.GetNumVPPWorkers()
-	if err != nil {
-		log.Panicf("Error getting number of VPP workers: %v", err)
-	}
-	nDataThreads := nVppWorkers
-	if config.GetCalicoVppIpsec().IpsecNbAsyncCryptoThread > 0 {
-		nDataThreads = nVppWorkers - config.GetCalicoVppIpsec().IpsecNbAsyncCryptoThread
-		if nDataThreads <= 0 {
-			log.Errorf("Couldn't fulfill request [crypto=%d total=%d]", config.GetCalicoVppIpsec().IpsecNbAsyncCryptoThread, nVppWorkers)
-			nDataThreads = nVppWorkers
-		}
-		log.Infof("Using ipsec workers [data=%d crypto=%d]", nDataThreads, nVppWorkers-nDataThreads)
-
-	}
-	return nDataThreads
-}
-
 func CompareIPList(newIPList, oldIPList []net.IP) (added []net.IP, deleted []net.IP, changed bool) {
 	oldIPListMap := make(map[string]bool)
 	newIPListMap := make(map[string]bool)

calico-vpp-agent/common/pubsub.go

@@ -27,7 +27,6 @@ const (
 	ChanSize = 500
 
 	PeerNodeStateChanged CalicoVppEventType = "PeerNodeStateChanged"
-	FelixConfChanged     CalicoVppEventType = "FelixConfChanged"
 	IpamConfChanged      CalicoVppEventType = "IpamConfChanged"
 	BGPConfChanged       CalicoVppEventType = "BGPConfChanged"
 
@@ -66,8 +65,6 @@ const (
 
 	IpamPoolUpdate CalicoVppEventType = "IpamPoolUpdate"
 	IpamPoolRemove CalicoVppEventType = "IpamPoolRemove"
-
-	WireguardPublicKeyChanged CalicoVppEventType = "WireguardPublicKeyChanged"
 )
 
 var (