feat: route reconciliation phase 2

pkg/bgp/path.go

@@ -0,0 +1,45 @@
+package bgp
+
+import (
+	"github.com/cloudnativelabs/kube-router/v2/pkg"
+	gobgpapi "github.com/osrg/gobgp/v3/api"
+	"k8s.io/klog/v2"
+)
+
+func PathChanged(path *gobgpapi.Path, pl PeerLister, rs pkg.RouteSyncer, tc pkg.TunnelCleaner) error {
+	klog.V(2).Infof("Path Looks Like: %s", path.String())
+	dst, nextHop, err := ParsePath(path)
+	if err != nil {
+		return err
+	}
+	tunnelName := tc.GenerateTunnelName(nextHop.String())
+
+	// If we've made it this far, then it is likely that the node is holding a destination route for this path already.
+	// If the path we've received from GoBGP is a withdrawal, we should clean up any lingering routes that may exist
+	// on the host (rather than creating a new one or updating an existing one), and then return.
+	if path.IsWithdraw {
+		klog.V(2).Infof("Removing route: '%s via %s' from peer in the routing table", dst, nextHop)
+
+		// The path might be withdrawn because the peer became unestablished or it may be withdrawn because just the
+		// path was withdrawn. Check to see if the peer is still established before deciding whether to clean the
+		// tunnel and tunnel routes or whether to just delete the destination route.
+		peerEstablished, err := IsPeerEstablished(pl, nextHop.String())
+		if err != nil {
+			klog.Errorf("encountered error while checking peer status: %v", err)
+		}
+		if err == nil && !peerEstablished {
+			klog.V(1).Infof("Peer '%s' was not found any longer, removing tunnel and routes",
+				nextHop.String())
+			// Also delete route from state map so that it doesn't get re-synced after deletion
+			rs.DelInjectedRoute(dst)
+			tc.CleanupTunnel(dst, tunnelName)
+			return nil
+		}
+
+		// Also delete route from state map so that it doesn't get re-synced after deletion
+		rs.DelInjectedRoute(dst)
+		return nil
+	}
+
+	return nil
+}

pkg/bgp/peer.go

@@ -0,0 +1,27 @@
+package bgp
+
+import (
+	"context"
+	"fmt"
+
+	api "github.com/osrg/gobgp/v3/api"
+)
+
+type PeerLister interface {
+	ListPeer(ctx context.Context, r *api.ListPeerRequest, fn func(*api.Peer)) error
+}
+
+func IsPeerEstablished(pl PeerLister, peerIP string) (bool, error) {
+	var peerConnected bool
+	peerFunc := func(peer *api.Peer) {
+		if peer.Conf.NeighborAddress == peerIP && peer.State.SessionState == api.PeerState_ESTABLISHED {
+			peerConnected = true
+		}
+	}
+	err := pl.ListPeer(context.Background(), &api.ListPeerRequest{Address: peerIP}, peerFunc)
+	if err != nil {
+		return false, fmt.Errorf("unable to list peers to see if tunnel & routes need to be removed: %v", err)
+	}
+
+	return peerConnected, nil
+}

pkg/routes/route_sync.go → pkg/controllers/routing/host_route_sync.go

@@ -1,4 +1,4 @@
-package routes
+package routing
 
 import (
 	"context"
@@ -7,6 +7,9 @@ import (
 	"sync"
 	"time"
 
+	"github.com/cloudnativelabs/kube-router/v2/pkg/bgp"
+	"github.com/cloudnativelabs/kube-router/v2/pkg/routes"
+
 	gobgpapi "github.com/osrg/gobgp/v3/api"
 	gobgp "github.com/osrg/gobgp/v3/pkg/server"
 	"github.com/vishvananda/netlink"
@@ -41,14 +44,6 @@ func (b BGPListError) Unwrap() error {
 	return b.err
 }
 
-// RouteSyncer is an interface that defines the methods needed to sync routes to the kernel's routing table
-type RouteSyncer interface {
-	AddInjectedRoute(dst *net.IPNet, route *netlink.Route)
-	DelInjectedRoute(dst *net.IPNet)
-	Run(stopCh <-chan struct{}, wg *sync.WaitGroup)
-	SyncLocalRouteTable()
-}
-
 // RouteSync is a struct that holds all of the information needed for syncing routes to the kernel's routing table
 type RouteSync struct {
 	routeTableStateMap       map[string]*netlink.Route
@@ -75,7 +70,7 @@ func (rs *RouteSync) DelInjectedRoute(dst *net.IPNet) {
 	if _, ok := rs.routeTableStateMap[dst.String()]; ok {
 		klog.V(3).Infof("Removing route for destination: %s", dst)
 		delete(rs.routeTableStateMap, dst.String())
-		err := DeleteByDestination(dst)
+		err := routes.DeleteByDestination(dst)
 		if err != nil {
 			klog.Errorf("Failed to cleanup routes: %v", err)
 		}
@@ -87,17 +82,16 @@ func (rs *RouteSync) checkCacheAgainstBGP() error {
 		routeMap := make(map[string]*netlink.Route, 0)
 		for _, p := range path {
 			klog.V(3).Infof("Path: %v", p)
-			/*
-			dst, nh, err := routing.ParseBGPPath(p)
+			dst, nh, err := bgp.ParsePath(p)
 			if err != nil {
 				klog.Warningf("Failed to parse BGP path, not failing so as to not block updating paths that are "+
 					"valid: %v", err)
 			}
 			routeMap[dst.String()] = &netlink.Route{
-				Dst: dst,
-				Gw:  nh,
-				Protocol: ZebraOriginator,
-			}*/
+				Dst:      dst,
+				Gw:       nh,
+				Protocol: routes.ZebraOriginator,
+			}
 		}
 		return routeMap
 	}
@@ -131,7 +125,6 @@ func (rs *RouteSync) checkCacheAgainstBGP() error {
 		}
 	}
 
-
 	return nil
 }
 
@@ -186,7 +179,7 @@ func NewRouteSyncer(syncPeriod time.Duration) *RouteSync {
 
 	// We substitute the RouteR* functions here so that we can easily monkey patch it in our unit tests
 	rs.routeReplacer = netlink.RouteReplace
-	rs.routeDeleter = DeleteByDestination
+	rs.routeDeleter = routes.DeleteByDestination
 	rs.routeAdder = netlink.RouteAdd
 
 	return &rs

pkg/routes/route_sync_test.go → pkg/controllers/routing/host_route_sync_test.go

@@ -1,4 +1,4 @@
-package routes
+package routing
 
 import (
 	"context"
@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/cloudnativelabs/kube-router/v2/pkg"
 	gobgpapi "github.com/osrg/gobgp/v3/api"
 	gobgp "github.com/osrg/gobgp/v3/pkg/server"
 	"github.com/stretchr/testify/assert"
@@ -60,7 +61,7 @@ func (mnl *mockNetlink) mockDestinationDelete(destinationSubnet *net.IPNet) erro
 	return nil
 }
 
-func (mnl *mockNetlink) waitForSyncLocalRouteToAcquireLock(syncer RouteSyncer) {
+func (mnl *mockNetlink) waitForSyncLocalRouteToAcquireLock(syncer pkg.RouteSyncer) {
 	// Launch syncLocalRouteTable in a separate goroutine so that we can try to inject a route into the map while it
 	// is syncing. Then wait on the wait group so that we know that syncLocalRouteTable has a hold on the lock when
 	// we try to use it in addInjectedRoute() below