Guarantee the order of rules.

pkg/agent/config.go

@@ -21,7 +21,6 @@ import (
 	"time"
 
 	debpkg "github.com/bep/debounce"
-	"github.com/coreos/go-iptables/iptables"
 	"github.com/spf13/pflag"
 	"k8s.io/klog/v2/klogr"
 	"k8s.io/utils/exec"
@@ -174,21 +173,9 @@ func (cfg Config) Manager() (*Manager, error) {
 		return nil, err
 	}
 
-	ipt, err := iptables.New()
-	if err != nil {
-		return nil, err
-	}
-
-	ipt6, err := iptables.NewWithProtocol(iptables.ProtocolIPv6)
-	if err != nil {
-		return nil, err
-	}
-
 	m := &Manager{
 		Config: cfg,
 		tm:     tm,
-		ipt4:   ipt,
-		ipt6:   ipt6,
 		log:    klogr.New().WithName("manager"),
 
 		events:        make(chan struct{}),

pkg/agent/iptables.go

@@ -15,34 +15,85 @@
 package agent
 
 import (
-	"fmt"
 	"strings"
 
-	"github.com/coreos/go-iptables/iptables"
 	"k8s.io/apimachinery/pkg/util/sets"
 
 	"github.com/fabedge/fabedge/pkg/util/ipset"
+	"github.com/fabedge/fabedge/pkg/util/iptables"
 )
 
 type IPSet struct {
 	IPSet    *ipset.IPSet
 	EntrySet sets.String
 }
 
+var jumpChains = []iptables.JumpChain{
+	{Table: iptables.TableFilter, SrcChain: iptables.ChainForward, DstChain: iptables.ChainFabEdgeForward, Position: iptables.Append},
+	{Table: iptables.TableNat, SrcChain: iptables.ChainPostRouting, DstChain: iptables.ChainFabEdgeNatOutgoing, Position: iptables.Prepend},
+}
+
+func buildRuleData(ipsetName string, subnets []string) []byte {
+	var builder strings.Builder
+	builder.WriteString(`
+*filter
+:FABEDGE-FORWARD - [0:0]
+
+`)
+
+	for _, subnet := range subnets {
+		builder.WriteString("-A FABEDGE-FORWARD -s ")
+		builder.WriteString(subnet)
+		builder.WriteString(" -j ACCEPT\n")
+
+		builder.WriteString("-A FABEDGE-FORWARD -d ")
+		builder.WriteString(subnet)
+		builder.WriteString(" -j ACCEPT\n")
+	}
+
+	builder.WriteString(`COMMIT
+
+*nat
+:FABEDGE-NAT-OUTGOING - [0:0]
+
+`)
+
+	for _, subnet := range subnets {
+		builder.WriteString("-A FABEDGE-NAT-OUTGOING -s ")
+		builder.WriteString(subnet)
+		builder.WriteString(" -m set --match-set ")
+		builder.WriteString(ipsetName)
+		builder.WriteString(" dst -j RETURN\n")
+
+		builder.WriteString("-A FABEDGE-NAT-OUTGOING -s ")
+		builder.WriteString(subnet)
+		builder.WriteString(" -d ")
+		builder.WriteString(subnet)
+		builder.WriteString(" -j RETURN\n")
+
+		builder.WriteString("-A FABEDGE-NAT-OUTGOING -s ")
+		builder.WriteString(subnet)
+		builder.WriteString(" -j MASQUERADE\n")
+	}
+
+	builder.WriteString("COMMIT\n")
+
+	return []byte(builder.String())
+}
+
 func (m *Manager) ensureIPTablesRules() error {
 	current := m.getCurrentEndpoint()
 
 	peerIPSet4, peerIPSet6 := m.getAllPeerCIDRs()
 	subnetsIP4, subnetsIP6 := classifySubnets(current.Subnets)
 
 	configs := []struct {
-		ipt           *iptables.IPTables
 		peerIPSet     IPSet
 		loopbackIPSet IPSet
 		subnets       []string
+		helper        iptables.ApplierCleaner
 	}{
 		{
-			ipt: m.ipt4,
 			peerIPSet: IPSet{
 				IPSet: &ipset.IPSet{
 					Name:       IPSetFabEdgePeerCIDR,
@@ -52,9 +103,9 @@ func (m *Manager) ensureIPTablesRules() error {
 				EntrySet: peerIPSet4,
 			},
 			subnets: subnetsIP4,
+			helper:  iptables.NewApplierCleaner(iptables.ProtocolIPv4, jumpChains, buildRuleData(IPSetFabEdgePeerCIDR, subnetsIP4)),
 		},
 		{
-			ipt: m.ipt6,
 			peerIPSet: IPSet{
 				IPSet: &ipset.IPSet{
 					Name:       IPSetFabEdgePeerCIDR6,
@@ -64,98 +115,20 @@ func (m *Manager) ensureIPTablesRules() error {
 				EntrySet: peerIPSet6,
 			},
 			subnets: subnetsIP6,
+			helper:  iptables.NewApplierCleaner(iptables.ProtocolIPv6, jumpChains, buildRuleData(IPSetFabEdgePeerCIDR6, subnetsIP6)),
 		},
 	}
 
-	clearOutgoingChain := !m.areSubnetsEqual(current.Subnets, m.lastSubnets)
 	for _, c := range configs {
-		if err := m.ensureIPForwardRules(c.ipt, c.subnets); err != nil {
-			return err
-		}
-
-		if m.MASQOutgoing {
-			if err := m.configureOutboundRules(c.ipt, c.peerIPSet, c.subnets, clearOutgoingChain); err != nil {
-				return err
-			}
-		}
-	}
-	// must be done after configureOutboundRules are executed
-	m.lastSubnets = current.Subnets
-
-	return nil
-}
-
-func (m *Manager) ensureIPForwardRules(ipt *iptables.IPTables, subnets []string) error {
-	if err := ensureChain(ipt, TableFilter, ChainFabEdgeForward); err != nil {
-		m.log.Error(err, "failed to check or create iptables chain", "table", TableFilter, "chain", ChainFabEdgeForward)
-		return err
-	}
-
-	ensureRule := ipt.AppendUnique
-	if err := ensureRule(TableFilter, ChainForward, "-j", ChainFabEdgeForward); err != nil {
-		m.log.Error(err, "failed to check or add rule", "table", TableFilter, "chain", ChainForward, "rule", "-j FABEDGE")
-		return err
-	}
-
-	// subnets won't change most of the time, and is append-only, so for now we don't need
-	// to handle removing old subnet
-	for _, subnet := range subnets {
-		if err := ensureRule(TableFilter, ChainFabEdgeForward, "-s", subnet, "-j", "ACCEPT"); err != nil {
-			m.log.Error(err, "failed to check or add rule", "table", TableFilter, "chain", ChainFabEdgeForward, "rule", fmt.Sprintf("-s %s -j ACCEPT", subnet))
-			return err
-		}
-
-		if err := ensureRule(TableFilter, ChainFabEdgeForward, "-d", subnet, "-j", "ACCEPT"); err != nil {
-			m.log.Error(err, "failed to check or add rule", "table", TableFilter, "chain", ChainFabEdgeForward, "rule", fmt.Sprintf("-d %s -j ACCEPT", subnet))
-			return err
-		}
-	}
-
-	return nil
-}
-
-// outbound NAT from pods to outside the cluster
-func (m *Manager) configureOutboundRules(ipt *iptables.IPTables, peerIPSet IPSet, subnets []string, clearFabEdgeNatOutgoingChain bool) error {
-	if clearFabEdgeNatOutgoingChain {
-		m.log.V(3).Info("Subnets are changed, clear iptables chain FABEDGE-NAT-OUTGOING")
-		if err := ipt.ClearChain(TableNat, ChainFabEdgeNatOutgoing); err != nil {
-			m.log.Error(err, "failed to check or add rule", "table", TableNat, "chain", ChainFabEdgeNatOutgoing)
-			return err
-		}
-	} else {
-		if err := ensureChain(ipt, TableNat, ChainFabEdgeNatOutgoing); err != nil {
-			m.log.Error(err, "failed to check or add rule", "table", TableNat, "chain", ChainFabEdgeNatOutgoing)
+		if err := m.ipset.EnsureIPSet(c.peerIPSet.IPSet, c.peerIPSet.EntrySet); err != nil {
+			m.log.Error(err, "failed to sync ipset", "ipsetName", c.peerIPSet.IPSet.Name)
 			return err
 		}
-	}
-
-	if err := m.ipset.EnsureIPSet(peerIPSet.IPSet, peerIPSet.EntrySet); err != nil {
-		m.log.Error(err, "failed to sync ipset", "ipsetName", peerIPSet.IPSet.Name)
-		return err
-	}
 
-	for _, subnet := range subnets {
-		m.log.V(3).Info("configure outgoing NAT iptables rules")
-
-		ensureRule := ipt.AppendUnique
-		if err := ensureRule(TableNat, ChainFabEdgeNatOutgoing, "-s", subnet, "-m", "set", "--match-set", peerIPSet.IPSet.Name, "dst", "-j", "RETURN"); err != nil {
-			m.log.Error(err, "failed to append rule", "table", TableNat, "chain", ChainFabEdgeNatOutgoing, "rule", fmt.Sprintf("-s %s -m set --match-set %s dst -j RETURN", subnet, peerIPSet.IPSet.Name))
-			continue
-		}
-
-		if err := ensureRule(TableNat, ChainFabEdgeNatOutgoing, "-s", subnet, "-d", subnet, "-j", "RETURN"); err != nil {
-			m.log.Error(err, "failed to append rule", "table", TableNat, "chain", ChainFabEdgeNatOutgoing, "rule", fmt.Sprintf("-s %s -d %s -j RETURN", subnet, subnet))
-			continue
-		}
-
-		if err := ensureRule(TableNat, ChainFabEdgeNatOutgoing, "-s", subnet, "-j", ChainMasquerade); err != nil {
-			m.log.Error(err, "failed to append rule", "table", TableNat, "chain", ChainFabEdgeNatOutgoing, "rule", fmt.Sprintf("-s %s -j %s", subnet, ChainMasquerade))
-			continue
-		}
-
-		if err := ensureRule(TableNat, ChainPostRouting, "-j", ChainFabEdgeNatOutgoing); err != nil {
-			m.log.Error(err, "failed to append rule", "table", TableNat, "chain", ChainPostRouting, "rule", fmt.Sprintf("-j %s", ChainFabEdgeNatOutgoing))
-			continue
+		if err := c.helper.Apply(); err != nil {
+			m.log.Error(err, "failed to sync iptables rules")
+		} else {
+			m.log.V(5).Info("iptables rules is synced")
 		}
 	}
 
@@ -176,19 +149,6 @@ func (m *Manager) areSubnetsEqual(sa1, sa2 []string) bool {
 	return true
 }
 
-func ensureChain(ipt *iptables.IPTables, table, chain string) error {
-	exists, err := ipt.ChainExists(table, chain)
-	if err != nil {
-		return err
-	}
-
-	if exists {
-		return nil
-	}
-
-	return ipt.NewChain(table, chain)
-}
-
 func (m *Manager) getAllPeerCIDRs() (cidrSet4, cidrSet6 sets.String) {
 	cidrSet4, cidrSet6 = sets.NewString(), sets.NewString()
 

pkg/agent/manager.go

@@ -24,7 +24,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/coreos/go-iptables/iptables"
 	"github.com/go-logr/logr"
 	"k8s.io/apimachinery/pkg/util/sets"
 
@@ -37,15 +36,8 @@ import (
 )
 
 const (
-	TableFilter             = "filter"
-	TableNat                = "nat"
-	ChainForward            = "FORWARD"
-	ChainPostRouting        = "POSTROUTING"
-	ChainMasquerade         = "MASQUERADE"
-	ChainFabEdgeForward     = "FABEDGE-FORWARD"
-	ChainFabEdgeNatOutgoing = "FABEDGE-NAT-OUTGOING"
-	IPSetFabEdgePeerCIDR    = "FABEDGE-PEER-CIDR"
-	IPSetFabEdgePeerCIDR6   = "FABEDGE-PEER-CIDR6"
+	IPSetFabEdgePeerCIDR  = "FABEDGE-PEER-CIDR"
+	IPSetFabEdgePeerCIDR6 = "FABEDGE-PEER-CIDR6"
 )
 
 type Manager struct {
@@ -54,10 +46,8 @@ type Manager struct {
 	ipvs    ipvs.Interface
 	ipset   ipset.Interface
 
-	tm   tunnel.Manager
-	ipt4 *iptables.IPTables
-	ipt6 *iptables.IPTables
-	log  logr.Logger
+	tm  tunnel.Manager
+	log logr.Logger
 
 	currentEndpoint  Endpoint
 	mediatorEndpoint *Endpoint

pkg/cloud-agent/cloud_agent.go

@@ -23,7 +23,6 @@ import (
 	"time"
 
 	"github.com/bep/debounce"
-	"github.com/coreos/go-iptables/iptables"
 	flag "github.com/spf13/pflag"
 	"github.com/vishvananda/netlink"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
@@ -108,15 +107,8 @@ func Execute() {
 }
 
 func NewCloudAgent() (*CloudAgent, error) {
-	iph, err := newIptableHandler(iptables.ProtocolIPv4)
-	if err != nil {
-		return nil, err
-	}
-
-	iph6, err := newIptableHandler(iptables.ProtocolIPv6)
-	if err != nil {
-		return nil, err
-	}
+	iph, _ := newIptableHandler()
+	iph6, _ := newIp6tableHandler()
 
 	if iph == nil && iph6 == nil {
 		return nil, fmt.Errorf("at lease one iptablesHandler is required")
@@ -147,10 +139,14 @@ func (a *CloudAgent) addAndSaveRoutes(cp routing.ConnectorPrefixes) {
 	}
 
 	routes := a.syncRoutes(cp.LocalPrefixes, cp.RemotePrefixes)
+
 	routes = append(routes, a.syncRoutes(cp.LocalPrefixes6, cp.RemotePrefixes6)...)
 
-	whitelist := sets.NewString(cp.RemotePrefixes...)
-	whitelist.Insert(cp.RemotePrefixes6...)
+	whitelist := sets.NewString()
+	for _, route := range routes {
+		whitelist.Insert(route.Dst.String())
+	}
+
 	if err := routeutil.PurgeStrongSwanRoutes(routeutil.NewDstWhitelist(whitelist)); err != nil {
 		logger.Error(err, "failed to purge stale routes in strongswan table")
 	}