Merge pull request networkservicemesh#598 from Nordix/pbr-sync-table-key

Improved recovery for policy base routing

pkg/kernel/networkservice/connectioncontextkernel/ipcontext/iprule/common.go

@@ -41,6 +41,7 @@ import (
 	"github.com/networkservicemesh/sdk/pkg/tools/log"
 
 	link "github.com/networkservicemesh/sdk-kernel/pkg/kernel"
+	"github.com/networkservicemesh/sdk-kernel/pkg/kernel/tools/nshandle"
 )
 
 func create(ctx context.Context, conn *networkservice.Connection, tableIDs *genericsync.Map[string, policies], nsRTableNextIDToConnID *genericsync.Map[NetnsRTableNextID, string]) error {
@@ -71,14 +72,20 @@ func create(ctx context.Context, conn *networkservice.Connection, tableIDs *gene
 
 		// Remove no longer existing policies
 		for tableID, policy := range toRemove {
-			if errRule := delRule(ctx, netlinkHandle, policy, tableID); errRule != nil {
+			if errRule := delRule(ctx, netlinkHandle, policy, tableID, l.Attrs().Index); errRule != nil {
 				return errRule
 			}
 			delete(ps, tableID)
 			tableIDs.Store(connID, ps)
 		}
+
+		// Get netns for key to namespace to routing tableID map
+		netNS, err := nshandle.FromURL(mechanism.GetNetNSURL())
+		if err != nil {
+			return err
+		}
+
 		// Add new policies
-		netNSInode := mechanism.GetNetNSInode()
 		for _, policy := range toAdd {
 			var tableID int
 			var nsrtid *NetnsRTableNextID
@@ -88,8 +95,12 @@ func create(ctx context.Context, conn *networkservice.Connection, tableIDs *gene
 				if err != nil {
 					return err
 				}
-				nsrtid = NewNetnsRTableNextID(netNSInode, tableID)
+				nsrtid = NewNetnsRTableNextID(netNS.UniqueId(), tableID)
 				storedConnID, _ := nsRTableNextIDToConnID.LoadOrStore(*nsrtid, connID)
+				log.FromContext(ctx).
+					WithField("nsrtid", *nsrtid).
+					WithField("ConnID", storedConnID).
+					Debug("storedTableID")
 				if connID == storedConnID {
 					// No other connection adding policy using this free routing table ID
 					break
@@ -273,15 +284,23 @@ func del(ctx context.Context, conn *networkservice.Connection, tableIDs *generic
 			return err
 		}
 		defer netlinkHandle.Close()
+		ifName := mechanism.GetInterfaceName()
+		l, err := netlinkHandle.LinkByName(ifName)
+		if err != nil {
+			return errors.Wrapf(err, "failed to find link %s", ifName)
+		}
 		ps, ok := tableIDs.LoadAndDelete(conn.GetId())
 		if ok {
-			netNSInode := mechanism.GetNetNSInode()
+			netNS, err := nshandle.FromURL(mechanism.GetNetNSURL())
+			if err != nil {
+				return err
+			}
 			for tableID := range ps {
-				nsrtid := NewNetnsRTableNextID(netNSInode, tableID)
+				nsrtid := NewNetnsRTableNextID(netNS.UniqueId(), tableID)
 				nsRTableNextIDToConnID.Delete(*nsrtid)
 			}
 			for tableID, policy := range ps {
-				if err := delRule(ctx, netlinkHandle, policy, tableID); err != nil {
+				if err := delRule(ctx, netlinkHandle, policy, tableID, l.Attrs().Index); err != nil {
 					return err
 				}
 			}
@@ -290,16 +309,11 @@ func del(ctx context.Context, conn *networkservice.Connection, tableIDs *generic
 	return nil
 }
 
-func delRule(ctx context.Context, handle *netlink.Handle, policy *networkservice.PolicyRoute, tableID int) error {
+func delRuleOnly(ctx context.Context, handle *netlink.Handle, policy *networkservice.PolicyRoute) error {
 	rule, err := policyToRule(policy)
 	if err != nil {
 		return err
 	}
-
-	if err := flushTable(ctx, handle, tableID); err != nil {
-		return err
-	}
-
 	now := time.Now()
 	if err := handle.RuleDel(rule); err != nil {
 		log.FromContext(ctx).
@@ -321,10 +335,17 @@ func delRule(ctx context.Context, handle *netlink.Handle, policy *networkservice
 	return nil
 }
 
-func flushTable(ctx context.Context, handle *netlink.Handle, tableID int) error {
+func delRule(ctx context.Context, handle *netlink.Handle, policy *networkservice.PolicyRoute, tableID, linkIndex int) error {
+	if err := flushTable(ctx, handle, tableID, linkIndex); err != nil {
+		return err
+	}
+	return delRuleOnly(ctx, handle, policy)
+}
+func flushTable(ctx context.Context, handle *netlink.Handle, tableID, linkIndex int) error {
 	routes, err := handle.RouteListFiltered(netlink.FAMILY_ALL,
 		&netlink.Route{
-			Table: tableID,
+			Table:     tableID,
+			LinkIndex: linkIndex,
 		},
 		netlink.RT_FILTER_TABLE)
 	if err != nil {
@@ -335,6 +356,9 @@ func flushTable(ctx context.Context, handle *netlink.Handle, tableID int) error
 		if err != nil {
 			return errors.Wrapf(err, "failed to delete route")
 		}
+		log.FromContext(ctx).
+			WithField("Route", &routes[i]).
+			WithField("netlink", "RouteDel").Debug("completed")
 	}
 	log.FromContext(ctx).
 		WithField("tableID", tableID).

pkg/kernel/networkservice/connectioncontextkernel/ipcontext/iprule/heal.go

@@ -53,11 +53,18 @@ func recoverTableIDs(ctx context.Context, conn *networkservice.Connection, table
 		}
 		defer netlinkHandle.Close()
 
+		ifName := mechanism.GetInterfaceName()
+		l, err := netlinkHandle.LinkByName(ifName)
+		if err != nil {
+			return errors.Wrapf(err, "failed to find link %s", ifName)
+		}
+
 		podRules, err := netlinkHandle.RuleList(netlink.FAMILY_ALL)
 		if err != nil {
 			return errors.Wrap(err, "failed to get list of rules")
 		}
 
+		tableIDtoPolicyMap := make(map[int]*networkservice.PolicyRoute)
 		// try to find the corresponding missing policies in the network namespace of the pod
 		for _, policy := range conn.Context.IpContext.Policies {
 			policyRule, err := policyToRule(policy)
@@ -66,20 +73,42 @@ func recoverTableIDs(ctx context.Context, conn *networkservice.Connection, table
 			}
 			for i := range podRules {
 				if ruleEquals(&podRules[i], policyRule) {
+					tableIDtoPolicyMap[podRules[i].Table] = policy
 					log.FromContext(ctx).
 						WithField("From", policy.From).
 						WithField("IPProto", policy.Proto).
 						WithField("DstPort", policy.DstPort).
 						WithField("SrcPort", policy.SrcPort).
 						WithField("Table", podRules[i].Table).Debug("policy recovered")
-					err := delRule(ctx, netlinkHandle, policy, podRules[i].Table)
-					if err != nil {
-						return err
-					}
 					break
 				}
 			}
 		}
+
+		return deleteRemainders(ctx, netlinkHandle, tableIDtoPolicyMap, podRules, l)
+	}
+	return nil
+}
+
+func deleteRemainders(ctx context.Context, netlinkHandle *netlink.Handle, tableIDtoPolicyMap map[int]*networkservice.PolicyRoute, podRules []netlink.Rule, l netlink.Link) error {
+	for tableID, policy := range tableIDtoPolicyMap {
+		usage := 0
+		for i := range podRules {
+			if podRules[i].Table == tableID {
+				usage++
+			}
+		}
+		if usage == 1 {
+			err := delRule(ctx, netlinkHandle, policy, tableID, l.Attrs().Index)
+			if err != nil {
+				return err
+			}
+		} else {
+			err := delRuleOnly(ctx, netlinkHandle, policy)
+			if err != nil {
+				return err
+			}
+		}
 	}
 	return nil
 }