When refreshing iptables, solve the interruption risk of communication services

agent/pkg/proxy/iptables.go

@@ -7,6 +7,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/thoas/go-funk"
 	"github.com/vishvananda/netlink"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -21,10 +22,10 @@ import (
 )
 
 const (
-	meshRootChain      utiliptables.Chain = "EDGE-MESH"
-	None                                  = "None"
-	labelCoreDNS                          = "k8s-app=kube-dns"
-	labelProxyEdgeMesh                    = "proxyby!=edgemesh"
+	meshRootChain        utiliptables.Chain = "EDGE-MESH"
+	None                                    = "None"
+	labelCoreDNS                            = "k8s-app=kube-dns"
+	labelNoProxyEdgeMesh                    = "noproxy=edgemesh"
 )
 
 type iptablesEnsureInfo struct {
@@ -56,20 +57,25 @@ type Proxier struct {
 	ignoreRules []iptablesEnsureInfo
 	proxyRules  []iptablesEnsureInfo
 	dnatRules   []iptablesEnsureInfo
+
+	invalidIgnoreRules []iptablesEnsureInfo
+	invalidProxyRules  []iptablesEnsureInfo
+	invalidDNATRules   []iptablesEnsureInfo
 }
 
 func NewProxier(subnet string, protoProxies []protocol.ProtoProxy, kubeClient kubernetes.Interface) (proxier *Proxier, err error) {
 	primaryProtocol := utiliptables.ProtocolIPv4
 	execer := utilexec.New()
 	iptInterface := utiliptables.New(execer, primaryProtocol)
 	proxier = &Proxier{
-		iptables:     iptInterface,
-		kubeClient:   kubeClient,
-		serviceCIDR:  subnet,
-		protoProxies: protoProxies,
-		ignoreRules:  make([]iptablesEnsureInfo, 2),
-		proxyRules:   make([]iptablesEnsureInfo, 2),
-		dnatRules:    make([]iptablesEnsureInfo, 2),
+		iptables:           iptInterface,
+		kubeClient:         kubeClient,
+		serviceCIDR:        subnet,
+		protoProxies:       protoProxies,
+		ignoreRules:        make([]iptablesEnsureInfo, 0),
+		proxyRules:         make([]iptablesEnsureInfo, 2),
+		dnatRules:          make([]iptablesEnsureInfo, 2),
+		invalidIgnoreRules: make([]iptablesEnsureInfo, 0),
 	}
 
 	// iptables rule cleaning and writing
@@ -87,7 +93,6 @@ func (proxier *Proxier) Start() {
 		for {
 			select {
 			case <-ticker.C:
-				proxier.FlushRules()
 				proxier.EnsureRules()
 			case <-beehiveContext.Done():
 				proxier.FlushRules()
@@ -139,7 +144,7 @@ func (proxier *Proxier) ignoreRuleByService(svc *corev1.Service) iptablesEnsureI
 }
 
 // createIgnoreRules exclude some services that must be ignored
-func (proxier *Proxier) createIgnoreRules() (ignoreRules []iptablesEnsureInfo, err error) {
+func (proxier *Proxier) createIgnoreRules() (ignoreRules, invalidIgnoreRules []iptablesEnsureInfo, err error) {
 	ignoreRulesIptablesEnsureMap := make(map[string]*corev1.Service)
 	// kube-apiserver service
 	kubeAPI, err := proxier.kubeClient.CoreV1().Services("default").Get(context.Background(), "kubernetes", metav1.GetOptions{})
@@ -172,23 +177,29 @@ func (proxier *Proxier) createIgnoreRules() (ignoreRules []iptablesEnsureInfo, e
 	}
 
 	// Other services we want to ignore...
-	otherIgnoreServiceList, err := proxier.kubeClient.CoreV1().Services("").List(context.Background(), metav1.ListOptions{LabelSelector: labelProxyEdgeMesh})
+	otherIgnoreServiceList, err := proxier.kubeClient.CoreV1().Services("").List(context.Background(), metav1.ListOptions{LabelSelector: labelNoProxyEdgeMesh})
 	if err != nil {
 		klog.Warningf("get Other ignore service failed: %s", err)
 	}
 	if err == nil && otherIgnoreServiceList != nil && len(otherIgnoreServiceList.Items) > 0 {
 		for _, item := range otherIgnoreServiceList.Items {
 			otherIgnoreService := item
-			klog.V(4).Infof("ignored not containing proxyby=edgemesh label service: %s", otherIgnoreService.Name)
+			klog.V(4).Infof("ignored containing noproxy=edgemesh label service: %s", otherIgnoreService.Name)
 			ignoreRulesIptablesEnsureMap[strings.Join([]string{item.Namespace, item.Name}, ".")] = &otherIgnoreService
 		}
 	}
 
 	for _, service := range ignoreRulesIptablesEnsureMap {
 		ignoreRules = append(ignoreRules, proxier.ignoreRuleByService(service))
 	}
+	// The go-funk library is used here for set operations and comparisons
+	for _, haveIgnoredRule := range proxier.ignoreRules {
+		if !funk.Contains(ignoreRules, haveIgnoredRule) {
+			invalidIgnoreRules = append(invalidIgnoreRules, haveIgnoredRule)
+		}
+	}
 
-	return ignoreRules, nil
+	return
 }
 
 // createProxyRules get proxy rules and DNAT rules
@@ -230,7 +241,7 @@ func (proxier *Proxier) createProxyRules() (proxyRules, dnatRules []iptablesEnsu
 		})
 	}
 
-	return proxyRules, dnatRules
+	return
 }
 
 // ensureRule ensures iptables rules exist
@@ -256,37 +267,29 @@ func (proxier *Proxier) EnsureRules() {
 	}
 
 	// recollect need to ignore rules
-	proxier.ignoreRules, err = proxier.createIgnoreRules()
+	proxier.ignoreRules, proxier.invalidIgnoreRules, err = proxier.createIgnoreRules()
 	if err != nil {
 		klog.Errorf("failed to create ignore rules: %v", err)
 	} else {
 		klog.V(5).Infof("ignore rules: %v", proxier.ignoreRules)
 	}
 
+	// clean the invalid ignore rules
+	err = proxier.setIgnoreRules("Delete", proxier.invalidIgnoreRules)
+	if err != nil {
+		klog.Errorf("clean the invalid ignore rules failed: %s", err)
+		return
+	} else {
+		klog.V(5).Infof("clean %d invalid ignore rules.", len(proxier.invalidIgnoreRules))
+	}
+
 	// ensure ignore rules
-	for _, rule := range proxier.ignoreRules {
-		if rule.extraArgs[0] == None {
-			headLessIps := rule.extraArgs[1:]
-			if len(headLessIps) > 0 {
-				for _, headLessIp := range headLessIps {
-					hlIp := headLessIp
-					args := append([]string{"-d"}, fmt.Sprintf("%s/32", hlIp), "-m", "comment", "--comment", rule.comment, "-j", string(rule.dstChain))
-					if _, err := proxier.iptables.EnsureRule(utiliptables.Append, rule.table, rule.srcChain, args...); err != nil {
-						klog.ErrorS(err, "Failed to ensure ignore rules", "table", rule.table, "srcChain", rule.srcChain, "dstChain", rule.dstChain)
-						return
-					}
-				}
-			}
-		} else {
-			args := append(rule.extraArgs,
-				"-m", "comment", "--comment", rule.comment,
-				"-j", string(rule.dstChain),
-			)
-			if _, err := proxier.iptables.EnsureRule(utiliptables.Append, rule.table, rule.srcChain, args...); err != nil {
-				klog.ErrorS(err, "Failed to ensure ignore rules", "table", rule.table, "srcChain", rule.srcChain, "dstChain", rule.dstChain)
-				return
-			}
-		}
+	err = proxier.setIgnoreRules("Ensure", proxier.ignoreRules)
+	if err != nil {
+		klog.Errorf("ensure ignore rules failed: %s", err)
+		return
+	} else {
+		klog.V(5).Infof("ensure %d ignore rules.", len(proxier.ignoreRules))
 	}
 
 	// recollect need to proxy rules
@@ -319,6 +322,51 @@ func (proxier *Proxier) EnsureRules() {
 	}
 }
 
+// setIgnoreRules Delete and Ensure ignore rule for EDGE-MESH chain
+func (proxier *Proxier) setIgnoreRules(ruleSetType string, ignoreRules []iptablesEnsureInfo) (err error) {
+	for _, ignoreRule := range ignoreRules {
+		if ignoreRule.extraArgs[0] == None {
+			headLessIps := ignoreRule.extraArgs[1:]
+			if len(headLessIps) > 0 {
+				for _, headLessIp := range headLessIps {
+					hlIp := headLessIp
+					args := append([]string{"-d"}, fmt.Sprintf("%s/32", hlIp), "-m", "comment", "--comment", ignoreRule.comment, "-j", string(ignoreRule.dstChain))
+					switch ruleSetType {
+					case "Ensure":
+						if _, err = proxier.iptables.EnsureRule(utiliptables.Prepend, ignoreRule.table, ignoreRule.srcChain, args...); err != nil {
+							klog.ErrorS(err, "failed to ensure ignore rules", "table", ignoreRule.table, "srcChain", ignoreRule.srcChain, "dstChain", ignoreRule.dstChain)
+						}
+					case "Delete":
+						if err = proxier.iptables.DeleteRule(ignoreRule.table, ignoreRule.srcChain, args...); err != nil {
+							klog.ErrorS(err, "failed to clean invalid ignore rules", "table", ignoreRule.table, "srcChain", ignoreRule.srcChain, "dstChain", ignoreRule.dstChain)
+						}
+					default:
+						return fmt.Errorf("incorrect parameter passing, ruleSetType must be Ensure or Delete")
+					}
+				}
+			}
+		} else {
+			args := append(ignoreRule.extraArgs,
+				"-m", "comment", "--comment", ignoreRule.comment,
+				"-j", string(ignoreRule.dstChain),
+			)
+			switch ruleSetType {
+			case "Ensure":
+				if _, err = proxier.iptables.EnsureRule(utiliptables.Prepend, ignoreRule.table, ignoreRule.srcChain, args...); err != nil {
+					klog.ErrorS(err, "failed to ensure ignore rules", "table", ignoreRule.table, "srcChain", ignoreRule.srcChain, "dstChain", ignoreRule.dstChain)
+				}
+			case "Delete":
+				if err = proxier.iptables.DeleteRule(ignoreRule.table, ignoreRule.srcChain, args...); err != nil {
+					klog.ErrorS(err, "failed to clean the invalid ignore rule", "table", ignoreRule.table, "srcChain", ignoreRule.srcChain, "dstChain", ignoreRule.dstChain)
+				}
+			default:
+				return fmt.Errorf("incorrect parameter passing, ruleSetType must be Ensure or Delete")
+			}
+		}
+	}
+	return nil
+}
+
 // FlushRules flush root chain and proxy chains
 func (proxier *Proxier) FlushRules() {
 	proxier.mu.Lock()

go.mod

@@ -19,6 +19,7 @@ require (
 	github.com/onsi/ginkgo v1.16.4
 	github.com/onsi/gomega v1.10.1
 	github.com/spf13/cobra v1.0.0
+	github.com/thoas/go-funk v0.9.1
 	github.com/vishvananda/netlink v1.1.0
 	istio.io/api v0.0.0-20210131044048-bfeb10697307
 	istio.io/client-go v0.0.0-20210218000043-b598dd019200

go.sum

@@ -1054,6 +1054,8 @@ github.com/syndtr/goleveldb v0.0.0-20181127023241-353a9fca669c/go.mod h1:Z4AUp2K
 github.com/syndtr/goleveldb v1.0.0/go.mod h1:ZVVdQEZoIme9iO1Ch2Jdy24qqXrMMOU6lpPAyBWyWuQ=
 github.com/tarm/serial v0.0.0-20180830185346-98f6abe2eb07/go.mod h1:kDXzergiv9cbyO7IOYJZWg1U88JhDg3PB6klq9Hg2pA=
 github.com/thecodeteam/goscaleio v0.1.0/go.mod h1:68sdkZAsK8bvEwBlbQnlLS+xU+hvLYM/iQ8KXej1AwM=
+github.com/thoas/go-funk v0.9.1 h1:O549iLZqPpTUQ10ykd26sZhzD+rmR5pWhuElrhbC20M=
+github.com/thoas/go-funk v0.9.1/go.mod h1:+IWnUfUmFO1+WVYQWQtIJHeRRdaIyyYglZN7xzUPe4Q=
 github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20170815181823-89b8d40f7ca8/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20190109142713-0ad062ec5ee5/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=

hack/verify-vendor.sh

@@ -36,6 +36,7 @@ ret=$(edgemesh::git::check_status)
 if [ ${ret} -eq 0 ]; then
 	echo "SUCCESS: Vendor Verified."
 else
+	git status
 	echo  "FAILED: Vendor Verify failed. Please run the command to check your directories: git status"
 	exit 1
 fi