Merge pull request #429 from yanjianbo1983/improve-connector-iptables

Improve connector iptables

go.mod

@@ -7,7 +7,6 @@ require (
 	github.com/bep/debounce v1.2.0
 	github.com/coredns/caddy v1.1.1
 	github.com/coredns/coredns v1.8.0
-	github.com/coreos/go-iptables v0.6.0
 	github.com/davecgh/go-spew v1.1.1
 	github.com/fsnotify/fsnotify v1.4.9
 	github.com/go-chi/chi/v5 v5.0.0
@@ -138,7 +137,7 @@ require (
 	go.opentelemetry.io/otel/sdk/metric v0.20.0 // indirect
 	go.opentelemetry.io/otel/trace v0.20.0 // indirect
 	go.opentelemetry.io/proto/otlp v0.7.0 // indirect
-	go.uber.org/atomic v1.7.0 // indirect
+	go.uber.org/atomic v1.7.0
 	go.uber.org/multierr v1.6.0 // indirect
 	go.uber.org/zap v1.17.0 // indirect
 	golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d // indirect

go.sum

@@ -177,8 +177,6 @@ github.com/coredns/corefile-migration v1.0.12/go.mod h1:NJOI8ceUF/NTgEwtjD+TUq3/
 github.com/coreos/bbolt v1.3.2/go.mod h1:iRUV2dpdMOn7Bo10OQBFzIJO9kkE559Wcmn+qkEiiKk=
 github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
 github.com/coreos/etcd v3.3.13+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
-github.com/coreos/go-iptables v0.6.0 h1:is9qnZMPYjLd8LYqmm/qlE+wwEgJIkTYdhV3rfZo4jk=
-github.com/coreos/go-iptables v0.6.0/go.mod h1:Qe8Bv2Xik5FyTXwgIbLAnv2sWSBmvWdFETJConOQ//Q=
 github.com/coreos/go-oidc v2.1.0+incompatible/go.mod h1:CgnwVTmzoESiwO9qyAFEMiHoZ1nMCKZlZ9V6mm3/LKc=
 github.com/coreos/go-semver v0.2.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
 github.com/coreos/go-semver v0.3.0 h1:wkHLiw0WNATZnSG7epLsujiMCgPAc9xhjJ4tgnAxmfM=

pkg/connector/manager.go

@@ -103,6 +103,7 @@ func (c Config) Manager() (*Manager, error) {
 	tm, err := strongswan.New(
 		strongswan.SocketFile(c.ViciSocket),
 		strongswan.StartAction("none"),
+		strongswan.InitTimeout(10),
 	)
 	if err != nil {
 		return nil, err
@@ -180,8 +181,6 @@ func (m *Manager) notify() {
 func (m *Manager) Start() {
 	about.DisplayVersion()
 
-	m.removeAllChains()
-
 	go m.runLeaderElection()
 	go m.runHTTPServer()
 	go m.workLoop()
@@ -327,12 +326,15 @@ func (m *Manager) workLoop() {
 			continue
 		}
 
-		m.maintainTunnels()
 		m.maintainRoutes()
 
 		m.iptHandler.maintainIPTables()
 		m.ipt6Handler.maintainIPTables()
 		m.broadcastConnectorPrefixes()
+
+		// maintainTunnels may last for minutes, so put it at the end, otherwise it may cause error, such as wrong iptables
+		// rules and wrong routes are generated after isLeader is set to false
+		m.maintainTunnels()
 	}
 }
 

pkg/util/iptables/iptables.go

@@ -84,12 +84,15 @@ type Interface interface {
 	utiliptables.Interface
 	// CreateChains create custom chains and insert them in specified positions
 	CreateChains(chains []JumpChain) error
-	// FlushAllChains flush rules of all custom chains
-	FlushAllChains(chains []JumpChain) error
-	// RemoveAllChains flush rules of all custom chains and remove them all
-	RemoveAllChains(chains []JumpChain) error
-	// RemoveChain flush rules of specified chain from specified table and remove the chain
-	RemoveChain(table Table, chain Chain) error
+	// SafeFlushChain flush rules of all custom chains, it won't return error if chain doesn't exist
+	SafeFlushChain(table Table, chain Chain) error
+	// FlushChains flush rules of all custom chains
+	FlushChains(chains []JumpChain) error
+	// DeleteChains flush rules of all custom chains and remove them all
+	DeleteChains(chains []JumpChain) error
+	// SafeDeleteChain flush rules of specified chain from specified table and delete the chain,
+	// it won't return error if chain doesn't exist
+	SafeDeleteChain(chain JumpChain) error
 	// NewApplierCleaner create a ApplierCleaner with specified custom chains and rules
 	NewApplierCleaner(chains []JumpChain, rulesData []byte) ApplierCleaner
 }
@@ -150,36 +153,58 @@ func (h *iptablesHelper) CreateChains(chains []JumpChain) error {
 	return utilerrors.NewAggregate(errors)
 }
 
-func (h *iptablesHelper) RemoveAllChains(chains []JumpChain) error {
+func (h *iptablesHelper) DeleteChains(chains []JumpChain) error {
 	var errors []error
 
 	for _, chain := range chains {
-		if err := h.RemoveChain(chain.Table, chain.DstChain); err != nil {
+		if err := h.SafeDeleteChain(chain); err != nil {
 			errors = append(errors, err)
 		}
 	}
 
 	return utilerrors.NewAggregate(errors)
 }
 
-func (h *iptablesHelper) FlushAllChains(chains []JumpChain) error {
+func (h *iptablesHelper) FlushChains(chains []JumpChain) error {
 	var errors []error
 
 	for _, chain := range chains {
-		if err := h.ipt.FlushChain(chain.Table, chain.DstChain); err != nil {
+		if err := h.SafeFlushChain(chain.Table, chain.DstChain); err != nil {
 			errors = append(errors, err)
 		}
 	}
 
 	return utilerrors.NewAggregate(errors)
 }
 
-func (h *iptablesHelper) RemoveChain(table Table, chain Chain) error {
-	if err := h.FlushChain(table, chain); err != nil {
+func (h *iptablesHelper) SafeFlushChain(table Table, chain Chain) error {
+	exists, err := h.ChainExists(table, chain)
+	if exists {
+		return h.ipt.FlushChain(table, chain)
+	} else if err != nil && isNotFoundError(err) {
+		return nil
+	} else {
 		return err
 	}
+}
+
+func (h *iptablesHelper) SafeDeleteChain(chain JumpChain) error {
+	exists, err := h.ChainExists(chain.Table, chain.DstChain)
+	if exists {
+		if err = h.DeleteRule(chain.Table, chain.SrcChain, "-j", string(chain.DstChain)); err != nil {
+			return err
+		}
 
-	return h.ipt.DeleteChain(table, chain)
+		if err = h.FlushChain(chain.Table, chain.DstChain); err != nil {
+			return err
+		}
+
+		return h.ipt.DeleteChain(chain.Table, chain.DstChain)
+	} else if err != nil && isNotFoundError(err) {
+		return nil
+	} else {
+		return err
+	}
 }
 
 func (h *iptablesHelper) NewApplierCleaner(chains []JumpChain, rulesData []byte) ApplierCleaner {
@@ -199,9 +224,20 @@ func (ac *applierCleaner) Apply() error {
 }
 
 func (ac *applierCleaner) Flush() error {
-	return ac.helper.FlushAllChains(ac.chains)
+	return ac.helper.FlushChains(ac.chains)
 }
 
 func (ac *applierCleaner) Remove() error {
-	return ac.helper.RemoveAllChains(ac.chains)
+	return ac.helper.DeleteChains(ac.chains)
+}
+
+func isNotFoundError(err error) bool {
+	if utiliptables.IsNotFoundError(err) {
+		return true
+	}
+
+	if ee, isExitError := err.(utilexec.ExitError); isExitError {
+		return ee.ExitStatus() == 1
+	}
+	return false
 }