Split felix server into watcher/handler

This patch splits the felix server in two pieces:
- a felix watcher placed under `agent/watchers/felix`
- a felix server placed under `agent/felix`

The former will have only the responsibility of watching
and submitting events into a single event queue.
The latter will receive the event in a single goroutine
and proceed to program VPP as a single thred.

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 <[email protected]>

Author: sknat

calico-vpp-agent/cmd/calico_vpp_dataplane.go

@@ -147,11 +147,9 @@ func main() {
 	serviceServer := services.NewServiceServer(vpp, k8sclient, log.WithFields(logrus.Fields{"component": "services"}))
 	prometheusServer := prometheus.NewPrometheusServer(vpp, log.WithFields(logrus.Fields{"component": "prometheus"}))
 	localSIDWatcher := watchers.NewLocalSIDWatcher(vpp, clientv3, log.WithFields(logrus.Fields{"subcomponent": "localsid-watcher"}))
-	felixServer, err := felix.NewFelixServer(vpp, log.WithFields(logrus.Fields{"component": "policy"}))
-	if err != nil {
-		log.Fatalf("Failed to create policy server %s", err)
-	}
-	err = felix.InstallFelixPlugin()
+	felixServer := felix.NewFelixServer(vpp, clientv3, log.WithFields(logrus.Fields{"component": "policy"}))
+	felixWatcher := watchers.NewFelixWatcher(felixServer.GetFelixServerEventChan(), log.WithFields(logrus.Fields{"component": "felix watcher"}))
+	err = watchers.InstallFelixPlugin()
 	if err != nil {
 		log.Fatalf("could not install felix plugin: %s", err)
 	}
@@ -168,10 +166,12 @@ func main() {
 	peerWatcher.SetBGPConf(bgpConf)
 	routingServer.SetBGPConf(bgpConf)
 	serviceServer.SetBGPConf(bgpConf)
+	felixServer.SetBGPConf(bgpConf)
 
 	watchDog := watchdog.NewWatchDog(log.WithFields(logrus.Fields{"component": "watchDog"}), &t)
 	Go(felixServer.ServeFelix)
 	felixConfig := watchDog.Wait(felixServer.FelixConfigChan, "Waiting for FelixConfig to be provided by the calico pod")
+	Go(felixWatcher.WatchFelix)
 	ourBGPSpec := watchDog.Wait(felixServer.GotOurNodeBGPchan, "Waiting for bgp spec to be provided on node add")
 	// check if the watchDog timer has issued the t.Kill() which would mean we are dead
 	if !t.Alive() {

calico-vpp-agent/cni/cni_pod_test.go

@@ -35,7 +35,6 @@ import (
 	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/common"
 	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/tests/mocks"
 	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/testutils"
-	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/watchers"
 	"github.com/projectcalico/vpp-dataplane/v3/config"
 	"github.com/projectcalico/vpp-dataplane/v3/vpplink"
 	"github.com/projectcalico/vpp-dataplane/v3/vpplink/types"
@@ -323,7 +322,7 @@ var _ = Describe("Pod-related functionality of CNI", func() {
 
 			Context("With MultiNet configuration (and multinet VRF and loopback already configured)", func() {
 				var (
-					networkDefinition *watchers.NetworkDefinition
+					networkDefinition *common.NetworkDefinition
 					pubSubHandlerMock *mocks.PubSubHandlerMock
 				)
 
@@ -355,9 +354,9 @@ var _ = Describe("Pod-related functionality of CNI", func() {
 					}
 					// NetworkDefinition CRD information caught by NetWatcher and send with additional information
 					// (VRF and loopback created by watcher) to the cni server as common.NetAdded CalicoVPPEvent
-					networkDefinition = &watchers.NetworkDefinition{
-						VRF:    watchers.VRF{Tables: tables},
-						PodVRF: watchers.VRF{Tables: podTables},
+					networkDefinition = &common.NetworkDefinition{
+						VRF:    common.VRF{Tables: tables},
+						PodVRF: common.VRF{Tables: podTables},
 						Vni:    uint32(0), // important only for VXLAN tunnel going out of node
 						Name:   networkName,
 						Range:  "10.1.1.0/24", // IP range for secondary network defined by multinet

calico-vpp-agent/cni/cni_server.go

@@ -36,7 +36,6 @@ import (
 	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/cni/model"
 	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/cni/podinterface"
 	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/common"
-	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/watchers"
 	"github.com/projectcalico/vpp-dataplane/v3/config"
 	"github.com/projectcalico/vpp-dataplane/v3/vpplink"
 	"github.com/projectcalico/vpp-dataplane/v3/vpplink/types"
@@ -53,7 +52,7 @@ type Server struct {
 
 	podInterfaceMap map[string]model.LocalPodSpec
 	lock            sync.Mutex /* protects Add/DelVppInterace/RescanState */
-	cniEventChan    chan common.CalicoVppEvent
+	cniEventChan    chan any
 
 	memifDriver    *podinterface.MemifPodInterfaceDriver
 	tuntapDriver   *podinterface.TunTapPodInterfaceDriver
@@ -65,7 +64,7 @@ type Server struct {
 	RedirectToHostClassifyTableIndex uint32
 
 	networkDefinitions   sync.Map
-	cniMultinetEventChan chan common.CalicoVppEvent
+	cniMultinetEventChan chan any
 	nodeBGPSpec          *common.LocalNodeSpec
 }
 
@@ -96,9 +95,9 @@ func (s *Server) Add(ctx context.Context, request *cniproto.AddRequest) (*cnipro
 		if !ok {
 			return nil, fmt.Errorf("trying to create a pod in an unexisting network %s", podSpec.NetworkName)
 		} else {
-			networkDefinition, ok := value.(*watchers.NetworkDefinition)
+			networkDefinition, ok := value.(*common.NetworkDefinition)
 			if !ok || networkDefinition == nil {
-				panic("Value is not of type *watchers.NetworkDefinition")
+				panic("Value is not of type *common.NetworkDefinition")
 			}
 			_, route, err := net.ParseCIDR(networkDefinition.Range)
 			if err == nil {
@@ -283,7 +282,7 @@ func NewCNIServer(vpp *vpplink.VppLink, felixServerIpam common.FelixServerIpam,
 		log: log,
 
 		felixServerIpam: felixServerIpam,
-		cniEventChan:    make(chan common.CalicoVppEvent, common.ChanSize),
+		cniEventChan:    make(chan any, common.ChanSize),
 
 		grpcServer:      grpc.NewServer(),
 		podInterfaceMap: make(map[string]model.LocalPodSpec),
@@ -292,7 +291,7 @@ func NewCNIServer(vpp *vpplink.VppLink, felixServerIpam common.FelixServerIpam,
 		vclDriver:       podinterface.NewVclPodInterfaceDriver(vpp, log),
 		loopbackDriver:  podinterface.NewLoopbackPodInterfaceDriver(vpp, log),
 
-		cniMultinetEventChan: make(chan common.CalicoVppEvent, common.ChanSize),
+		cniMultinetEventChan: make(chan any, common.ChanSize),
 	}
 	reg := common.RegisterHandler(server.cniEventChan, "CNI server events")
 	reg.ExpectEvents(
@@ -313,7 +312,11 @@ forloop:
 		select {
 		case <-t.Dying():
 			break forloop
-		case evt := <-s.cniEventChan:
+		case msg := <-s.cniEventChan:
+			evt, ok := msg.(common.CalicoVppEvent)
+			if !ok {
+				continue
+			}
 			switch evt.Type {
 			case common.FelixConfChanged:
 				if new, _ := evt.New.(*felixConfig.Config); new != nil {
@@ -434,21 +437,25 @@ func (s *Server) ServeCNI(t *tomb.Tomb) error {
 				case <-t.Dying():
 					s.log.Warn("Cni server asked to exit")
 					return
-				case event := <-s.cniMultinetEventChan:
+				case msg := <-s.cniMultinetEventChan:
+					event, ok := msg.(common.CalicoVppEvent)
+					if !ok {
+						continue
+					}
 					switch event.Type {
 					case common.NetsSynced:
 						netsSynced <- true
 					case common.NetAddedOrUpdated:
-						netDef, ok := event.New.(*watchers.NetworkDefinition)
+						netDef, ok := event.New.(*common.NetworkDefinition)
 						if !ok {
-							s.log.Errorf("event.New is not a *watchers.NetworkDefinition %v", event.New)
+							s.log.Errorf("event.New is not a *common.NetworkDefinition %v", event.New)
 							continue
 						}
 						s.networkDefinitions.Store(netDef.Name, netDef)
 					case common.NetDeleted:
-						netDef, ok := event.Old.(*watchers.NetworkDefinition)
+						netDef, ok := event.Old.(*common.NetworkDefinition)
 						if !ok {
-							s.log.Errorf("event.Old is not a *watchers.NetworkDefinition %v", event.Old)
+							s.log.Errorf("event.Old is not a *common.NetworkDefinition %v", event.Old)
 							continue
 						}
 						s.networkDefinitions.Delete(netDef.Name)
@@ -488,6 +495,6 @@ func (s *Server) ServeCNI(t *tomb.Tomb) error {
 
 // ForceAddingNetworkDefinition will add another NetworkDefinition to this CNI server.
 // The usage is mainly for testing purposes.
-func (s *Server) ForceAddingNetworkDefinition(networkDefinition *watchers.NetworkDefinition) {
+func (s *Server) ForceAddingNetworkDefinition(networkDefinition *common.NetworkDefinition) {
 	s.networkDefinitions.Store(networkDefinition.Name, networkDefinition)
 }

calico-vpp-agent/cni/network_vpp.go

@@ -24,7 +24,6 @@ import (
 
 	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/cni/model"
 	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/common"
-	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/watchers"
 	"github.com/projectcalico/vpp-dataplane/v3/config"
 	"github.com/projectcalico/vpp-dataplane/v3/vpplink"
 	"github.com/projectcalico/vpp-dataplane/v3/vpplink/types"
@@ -252,9 +251,9 @@ func (s *Server) AddVppInterface(podSpec *model.LocalPodSpec, doHostSideConf boo
 		if !ok {
 			s.log.Errorf("network not found %s", podSpec.NetworkName)
 		} else {
-			networkDefinition, ok := value.(*watchers.NetworkDefinition)
+			networkDefinition, ok := value.(*common.NetworkDefinition)
 			if !ok || networkDefinition == nil {
-				panic("networkDefinition not of type *watchers.NetworkDefinition")
+				panic("networkDefinition not of type *common.NetworkDefinition")
 			}
 			vni = networkDefinition.Vni
 		}
@@ -324,9 +323,9 @@ func (s *Server) DelVppInterface(podSpec *model.LocalPodSpec) {
 		if !ok {
 			deleteLocalPodAddress = false
 		} else {
-			networkDefinition, ok := value.(*watchers.NetworkDefinition)
+			networkDefinition, ok := value.(*common.NetworkDefinition)
 			if !ok || networkDefinition == nil {
-				panic("networkDefinition not of type *watchers.NetworkDefinition")
+				panic("networkDefinition not of type *common.NetworkDefinition")
 			}
 			vni = networkDefinition.Vni
 		}

calico-vpp-agent/cni/network_vpp_routes.go

@@ -20,7 +20,6 @@ import (
 
 	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/cni/model"
 	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/common"
-	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/watchers"
 	"github.com/projectcalico/vpp-dataplane/v3/vpplink"
 	"github.com/projectcalico/vpp-dataplane/v3/vpplink/types"
 )
@@ -37,9 +36,9 @@ func (s *Server) RoutePodInterface(podSpec *model.LocalPodSpec, stack *vpplink.C
 			if !ok {
 				s.log.Errorf("network not found %s", podSpec.NetworkName)
 			} else {
-				networkDefinition, ok := value.(*watchers.NetworkDefinition)
+				networkDefinition, ok := value.(*common.NetworkDefinition)
 				if !ok || networkDefinition == nil {
-					panic("networkDefinition not of type *watchers.NetworkDefinition")
+					panic("networkDefinition not of type *common.NetworkDefinition")
 				}
 				table = networkDefinition.VRF.Tables[idx]
 			}
@@ -88,9 +87,9 @@ func (s *Server) UnroutePodInterface(podSpec *model.LocalPodSpec, swIfIndex uint
 			if !ok {
 				s.log.Errorf("network not found %s", podSpec.NetworkName)
 			} else {
-				networkDefinition, ok := value.(*watchers.NetworkDefinition)
+				networkDefinition, ok := value.(*common.NetworkDefinition)
 				if !ok || networkDefinition == nil {
-					panic("networkDefinition not of type *watchers.NetworkDefinition")
+					panic("networkDefinition not of type *common.NetworkDefinition")
 				}
 				table = networkDefinition.VRF.Tables[idx]
 			}
@@ -215,9 +214,9 @@ func (s *Server) CreatePodVRF(podSpec *model.LocalPodSpec, stack *vpplink.Cleanu
 			if !ok {
 				return errors.Errorf("network not found %s", podSpec.NetworkName)
 			}
-			networkDefinition, ok := value.(*watchers.NetworkDefinition)
+			networkDefinition, ok := value.(*common.NetworkDefinition)
 			if !ok || networkDefinition == nil {
-				panic("networkDefinition not of type *watchers.NetworkDefinition")
+				panic("networkDefinition not of type *common.NetworkDefinition")
 			}
 			vrfIndex = networkDefinition.PodVRF.Tables[idx]
 		}
@@ -375,9 +374,9 @@ func (s *Server) DeletePodVRF(podSpec *model.LocalPodSpec) {
 			if !ok {
 				s.log.Errorf("network not found %s", podSpec.NetworkName)
 			} else {
-				networkDefinition, ok := value.(*watchers.NetworkDefinition)
+				networkDefinition, ok := value.(*common.NetworkDefinition)
 				if !ok || networkDefinition == nil {
-					panic("networkDefinition not of type *watchers.NetworkDefinition")
+					panic("networkDefinition not of type *common.NetworkDefinition")
 				}
 				vrfIndex = networkDefinition.PodVRF.Tables[idx]
 			}

calico-vpp-agent/common/pubsub.go

@@ -85,31 +85,27 @@ type PubSubHandlerRegistration struct {
 	/* Name for the registration, for logging & debugging */
 	name string
 	/* Channel where to send events */
-	channel chan CalicoVppEvent
+	channel chan any
 	/* Receive only these events. If empty we'll receive all */
 	expectedEvents map[CalicoVppEventType]bool
-	/* Receive all events */
-	expectAllEvents bool
 }
 
 func (reg *PubSubHandlerRegistration) ExpectEvents(eventTypes ...CalicoVppEventType) {
 	for _, eventType := range eventTypes {
 		reg.expectedEvents[eventType] = true
 	}
-	reg.expectAllEvents = false
 }
 
 type PubSub struct {
 	log                        *log.Entry
 	pubSubHandlerRegistrations []*PubSubHandlerRegistration
 }
 
-func RegisterHandler(channel chan CalicoVppEvent, name string) *PubSubHandlerRegistration {
+func RegisterHandler(channel chan any, name string) *PubSubHandlerRegistration {
 	reg := &PubSubHandlerRegistration{
-		channel:         channel,
-		name:            name,
-		expectedEvents:  make(map[CalicoVppEventType]bool),
-		expectAllEvents: true, /* By default receive everything, unless we ask for a filter */
+		channel:        channel,
+		name:           name,
+		expectedEvents: make(map[CalicoVppEventType]bool),
 	}
 	ThePubSub.pubSubHandlerRegistrations = append(ThePubSub.pubSubHandlerRegistrations, reg)
 	return reg
@@ -128,7 +124,7 @@ func redactPassword(event CalicoVppEvent) string {
 func SendEvent(event CalicoVppEvent) {
 	ThePubSub.log.Debugf("Broadcasting event %s", redactPassword(event))
 	for _, reg := range ThePubSub.pubSubHandlerRegistrations {
-		if reg.expectAllEvents || reg.expectedEvents[event.Type] {
+		if reg.expectedEvents[event.Type] {
 			reg.channel <- event
 		}
 	}

calico-vpp-agent/common/types.go

@@ -0,0 +1,72 @@
+// Copyright (C) 2025 Cisco Systems Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package common
+
+import (
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+type VRF struct {
+	Tables [2]uint32 // one for ipv4, one for ipv6
+}
+
+type NetworkDefinition struct {
+	// VRF is the main table used for the corresponding physical network
+	VRF VRF
+	// PodVRF is the table used for the pods in the corresponding physical network
+	PodVRF              VRF
+	Vni                 uint32
+	PhysicalNetworkName string
+	Name                string
+	Range               string
+	NetAttachDefs       string
+}
+
+// FelixSocketSyncState describes the status of the
+// felix socket connection. It applies mostly to policies
+type FelixSocketSyncState int
+
+const (
+	StateDisconnected FelixSocketSyncState = iota
+	StateConnected
+	StateSyncing
+	StateInSync
+)
+
+func (state FelixSocketSyncState) IsPending() bool {
+	return state != StateInSync
+}
+
+// FelixSocketStateChanged is emitted when the state
+// of the socket changed. Typically connection and disconnection.
+type FelixSocketStateChanged struct {
+	NewState FelixSocketSyncState
+}
+
+type ServiceAndEndpoints struct {
+	Service   *v1.Service
+	Endpoints *v1.Endpoints
+}
+
+type ServiceEndpointsUpdate struct {
+	New *ServiceAndEndpoints
+	Old *ServiceAndEndpoints
+}
+
+type ServiceEndpointsDelete struct {
+	Meta *metav1.ObjectMeta
+}