Policy route removal on connection update

* Based on memory, Policies are filtered: to add/to removed
* Doesn't support forwarder resiliency. If the forwarder restart, the
  existing policy routes are forgotten

Signed-off-by: Lionel Jouin <[email protected]>

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

@@ -21,6 +21,7 @@ package iprule
 
 import (
 	"context"
+	"fmt"
 	"strconv"
 	"time"
 
@@ -54,45 +55,72 @@ func create(ctx context.Context, conn *networkservice.Connection, tableIDs *Map,
 			return errors.WithStack(err)
 		}
 
-		for _, policy := range conn.Context.IpContext.Policies {
-			// Check if we already created required ip table
-			key := tableKey{
-				connId:   conn.GetId(),
-				from:     policy.From,
-				protocol: policy.Proto,
-				dstPort:  policy.DstPort,
-				srcPort:  policy.SrcPort,
-			}
-			tableID, ok := tableIDs.Load(key)
-			if !ok {
-				counter.Inc()
-				tableID = int(counter.Load())
+		ps, ok := tableIDs.Load(mechanism.GetNetNSURL())
+		if !ok {
+			ps = policies{
+				policies: make(map[int]*networkservice.PolicyRoute),
 			}
+			tableIDs.Store(mechanism.GetNetNSURL(), ps)
+		}
+		// Get policies to add and to remove
+		toAdd, toRemove := getPolicyDifferences(ps.policies, conn.Context.IpContext.Policies)
 
-			// If policy doesn't contain any route - add default
-			if len(policy.Routes) == 0 {
-				policy.Routes = append(policy.Routes, defaultRoute())
+		// Remove no longer existing policies
+		for tableID, policy := range toRemove {
+			if err := delRule(ctx, netlinkHandle, policy); err != nil {
+				return err
 			}
+			delete(ps.policies, tableID)
+			tableIDs.Store(mechanism.GetNetNSURL(), ps)
+		}
+		// Add new policies
+		for _, policy := range toAdd {
+			tableID := int(ps.counter.Inc())
 			for _, route := range policy.Routes {
 				if err := routeAdd(ctx, netlinkHandle, l, route, tableID); err != nil {
 					return err
 				}
 			}
-
-			if !ok {
-				// Check and delete old rules if they don't fit
-				_ = delOldRules(ctx, netlinkHandle, policy, tableID)
-				// Add new rule
-				if err := ruleAdd(ctx, netlinkHandle, policy, tableID); err != nil {
-					return err
-				}
-				tableIDs.Store(key, tableID)
+			if err := ruleAdd(ctx, netlinkHandle, policy, tableID); err != nil {
+				return err
 			}
+			ps.policies[tableID] = policy
+			tableIDs.Store(mechanism.GetNetNSURL(), ps)
 		}
 	}
 	return nil
 }
 
+func getPolicyDifferences(current map[int]*networkservice.PolicyRoute, new []*networkservice.PolicyRoute) ([]*networkservice.PolicyRoute, map[int]*networkservice.PolicyRoute) {
+	type table struct {
+		tableID     int
+		policyRoute *networkservice.PolicyRoute
+	}
+	toAdd := []*networkservice.PolicyRoute{}
+	toRemove := map[int]*networkservice.PolicyRoute{}
+	currentMap := make(map[string]*table)
+	for tableId, policy := range current {
+		currentMap[policyKey(policy)] = &table{
+			tableID:     tableId,
+			policyRoute: policy,
+		}
+	}
+	for _, policy := range new {
+		if _, ok := currentMap[policyKey(policy)]; !ok {
+			toAdd = append(toAdd, policy)
+		}
+		delete(currentMap, policyKey(policy))
+	}
+	for _, table := range currentMap {
+		toRemove[table.tableID] = table.policyRoute
+	}
+	return toAdd, toRemove
+}
+
+func policyKey(policy *networkservice.PolicyRoute) string {
+	return fmt.Sprintf("%s;%s;%s;%s", policy.DstPort, policy.SrcPort, policy.From, policy.Proto)
+}
+
 func policyToRule(policy *networkservice.PolicyRoute) (*netlink.Rule, error) {
 	rule := netlink.NewRule()
 	if policy.From != "" {
@@ -254,6 +282,8 @@ func delRule(ctx context.Context, handle *netlink.Handle, policy *networkservice
 		return errors.WithStack(err)
 	}
 
+	// TODO: Flush table
+
 	now := time.Now()
 	if err := handle.RuleDel(rule); err != nil {
 		log.FromContext(ctx).

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

@@ -16,17 +16,26 @@
 
 package iprule
 
-import "sync"
+import (
+	"sync"
 
-//go:generate go-syncmap -output table_map.gen.go -type Map<tableKey,int>
+	"github.com/networkservicemesh/api/pkg/api/networkservice"
+	"go.uber.org/atomic"
+)
 
-type tableKey struct {
-	connId   string
-	from     string
-	protocol string
-	dstPort  string
-	srcPort  string
+//go:generate go-syncmap -output table_map.gen.go -type Map<string,policies>
+type policies struct {
+	counter  atomic.Int32
+	policies map[int]*networkservice.PolicyRoute
 }
 
-// Map - sync.Map with key == tableKey and value == uint32
+func (p *policies) toSlice() []*networkservice.PolicyRoute {
+	policies := []*networkservice.PolicyRoute{}
+	for _, policy := range p.policies {
+		policies = append(policies, policy)
+	}
+	return policies
+}
+
+// Map - sync.Map with key == string (netNsURL) and value == policies
 type Map sync.Map