Adds a CANARY chain and Monitor loop instead of reapplying iptables

net/expose.go

@@ -103,13 +103,8 @@ func addNatRule(ipt *iptables.IPTables, rulespec ...string) error {
 	// Loop until we get an exit code other than "temporarily unavailable"
 	for {
 		if err := ipt.AppendUnique("nat", "WEAVE", rulespec...); err != nil {
-			if ierr, ok := err.(*iptables.Error); ok {
-				if status, ok := ierr.ExitError.Sys().(syscall.WaitStatus); ok {
-					// (magic exit code 4 found in iptables source code; undocumented)
-					if status.ExitStatus() == 4 {
-						continue
-					}
-				}
+			if isResourceError(err) {
+				continue
 			}
 			return err
 		}

net/iptables.go

@@ -2,9 +2,13 @@ package net
 
 import (
 	"strings"
+	"syscall"
+	"time"
 
 	"github.com/coreos/go-iptables/iptables"
 	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	utilwait "k8s.io/apimachinery/pkg/util/wait"
 )
 
 // AddChainWithRules creates a chain and appends given rules to it.
@@ -104,3 +108,117 @@ func ensureRulesAtTop(table, chain string, rulespecs [][]string, ipt *iptables.I
 
 	return nil
 }
+
+func chainExists(ipt *iptables.IPTables, table string, chain string) (bool, error) {
+	existingChains, err := ipt.ListChains(table)
+	if err != nil {
+		return false, errors.Wrapf(err, "ipt.ListChains(%s)", table)
+	}
+	chainMap := make(map[string]struct{})
+	for _, c := range existingChains {
+		chainMap[c] = struct{}{}
+	}
+
+	_, found := chainMap[chain]
+	return found, nil
+}
+
+const (
+	// Max time we wait for an iptables flush to complete after we notice it has started
+	iptablesFlushTimeout = 5 * time.Second
+	// How often we poll while waiting for an iptables flush to complete
+	iptablesFlushPollTime = 100 * time.Millisecond
+)
+
+// Monitor is part of Interface
+func Monitor(log *logrus.Logger, canary string, tables []string, reloadFunc func(), interval time.Duration, stopCh <-chan struct{}) {
+	ipt, err := iptables.New()
+	if err != nil {
+		log.Errorf("creating iptables object while initializing iptable Monitoring %s", err)
+		return
+	}
+
+	for {
+		_ = PollImmediateUntil(interval, func() (bool, error) {
+			for _, table := range tables {
+				if err := ensureChains(ipt, table, canary); err != nil {
+					log.Warningf("Could not set up iptables canary %s/%s: %v", table, canary, err)
+					return false, nil
+				}
+			}
+			return true, nil
+		}, stopCh)
+
+		// Poll until stopCh is closed or iptables is flushed
+		err = utilwait.PollUntil(interval, func() (bool, error) {
+			if exists, err := chainExists(ipt, tables[0], canary); exists {
+				return false, nil
+			} else if isResourceError(err) {
+				log.Warningf("Could not check for iptables canary %s/%s: %v", tables[0], canary, err)
+				return false, nil
+			}
+			log.Infof("iptables canary %s/%s deleted", tables[0], canary)
+
+			// Wait for the other canaries to be deleted too before returning
+			// so we don't start reloading too soon.
+			err := utilwait.PollImmediate(iptablesFlushPollTime, iptablesFlushTimeout, func() (bool, error) {
+				for i := 1; i < len(tables); i++ {
+					if exists, err := chainExists(ipt, tables[i], canary); exists || isResourceError(err) {
+						return false, nil
+					}
+				}
+				return true, nil
+			})
+			if err != nil {
+				log.Warning("Inconsistent iptables state detected.")
+			}
+			return true, nil
+		}, stopCh)
+
+		if err != nil {
+			// stopCh was closed
+			for _, table := range tables {
+				_ = ipt.DeleteChain(table, canary)
+			}
+			return
+		}
+
+		log.Infof("Reloading after iptables flush")
+		reloadFunc()
+	}
+}
+
+const iptablesStatusResourceProblem = 4
+
+// isResourceError returns true if the error indicates that iptables ran into a "resource
+// problem" and was unable to attempt the request. In particular, this will be true if it
+// times out trying to get the iptables lock.
+func isResourceError(err error) bool {
+	if ierr, ok := err.(*iptables.Error); ok {
+		if status, ok := ierr.ExitError.Sys().(syscall.WaitStatus); ok {
+			return status.ExitStatus() == iptablesStatusResourceProblem
+		}
+	}
+
+	return false
+}
+
+// PollImmediateUntil tries a condition func until it returns true, an error or stopCh is closed.
+//
+// PollImmediateUntil runs the 'condition' before waiting for the interval.
+// 'condition' will always be invoked at least once.
+func PollImmediateUntil(interval time.Duration, condition utilwait.ConditionFunc, stopCh <-chan struct{}) error {
+	done, err := condition()
+	if err != nil {
+		return err
+	}
+	if done {
+		return nil
+	}
+	select {
+	case <-stopCh:
+		return utilwait.ErrWaitTimeout
+	default:
+		return utilwait.PollUntil(interval, condition, stopCh)
+	}
+}

prog/weaver/main.go

@@ -17,6 +17,7 @@ import (
 	"github.com/weaveworks/common/mflagext"
 	"github.com/weaveworks/common/signals"
 	"github.com/weaveworks/mesh"
+	"k8s.io/apimachinery/pkg/util/wait"
 
 	"github.com/weaveworks/weave/common"
 	"github.com/weaveworks/weave/common/docker"
@@ -519,23 +520,16 @@ func main() {
 
 	if iptablesRefresh > 0 {
 		Log.Printf("Entering iptable refresh loop (interval %v)", iptablesRefresh)
-		ticker := time.NewTicker(iptablesRefresh).C
-		go func() {
-			for {
-				select {
-				case <-ticker:
-					Log.Debug("Re-configuring iptables")
-					err := weavenet.ConfigureIPTables(&bridgeConfig, ips)
-					if err != nil {
-						Log.Errorf("Error configuring iptables: %s", err)
-					}
-
-					Log.Debug("Beginning re-exposure...")
-					weavenet.Reexpose(&bridgeConfig, Log)
-					Log.Debug("... re-exposure complete.")
-				}
+		applyIPTables := func() {
+			Log.Info("Re-configuring iptables")
+			err := weavenet.ConfigureIPTables(&bridgeConfig, ips)
+			if err != nil {
+				Log.Errorf("Error configuring iptables: %s", err)
 			}
-		}()
+
+			weavenet.Reexpose(&bridgeConfig, Log)
+		}
+		go weavenet.Monitor(Log, "WEAVE-CANARY", []string{"mangle", "nat", "filter"}, applyIPTables, iptablesRefresh, wait.NeverStop)
 	} else {
 		Log.Printf("Not refreshing iptables (interval: %v)", iptablesRefresh)
 	}

test/130_expose_2_test.sh

@@ -27,10 +27,6 @@ exec_on1() {
     assert_raises "exec_on  $HOST1 $@"
 }
 
-wait_for_iptable_refresh() {
-    sleep 2
-}
-
 # Containers in the same subnet should be able to talk; different subnet not.
 check_container_connectivity() {
     exec_on1 "c1 $PING $C2"
@@ -43,7 +39,7 @@ check_container_connectivity() {
 
 start_suite "exposing weave network to host"
 
-weave_on $HOST1 launch --ipalloc-range $UNIVERSE --iptables-refresh-interval=1s
+weave_on $HOST1 launch --ipalloc-range $UNIVERSE
 
 start_container $HOST1 $C1/24 --name=c1
 start_container $HOST1 $C2/24 --name=c2
@@ -65,20 +61,11 @@ weave_on1 "expose $EXP/24"
 run_on1   "! $PING $C1"
 run_on1   "  $PING $C3"
 run_on1   "! $PING $C5"
-wait_for_iptable_refresh
-run_on1   "! $PING $C1"
-run_on1   "  $PING $C3"
-run_on1   "! $PING $C5"
-
 weave_on1 "expose"
 weave_on1 "expose"
 run_on1   "! $PING $C1"
 run_on1   "  $PING $C3"
 run_on1   "  $PING $C5"
-wait_for_iptable_refresh
-run_on1   "! $PING $C1"
-run_on1   "  $PING $C3"
-run_on1   "  $PING $C5"
 check_container_connectivity
 
 # no host connectivity after 'hide', and idempotence of 'hide'
@@ -86,15 +73,9 @@ weave_on1 "hide $EXP/24"
 weave_on1 "hide $EXP/24"
 run_on1   "! $PING $C3"
 run_on1   "  $PING $C5"
-wait_for_iptable_refresh
-run_on1   "! $PING $C3"
-run_on1   "  $PING $C5"
-
 weave_on1 "hide"
 weave_on1 "hide"
 run_on1   "! $PING $C5"
-wait_for_iptable_refresh
-run_on1   "! $PING $C5"
 
 # remote host connectivity after 'expose'
 docker_on $HOST1 exec -d c3 nc -l -p $PORT

test/135_expose_iptables_refresh.sh

@@ -29,7 +29,7 @@ start_suite "exposing weave network to host"
 ## Check no refreshing
 weave_on1 "launch --iptables-refresh-interval=0s"
 IPT_BEFORE=$(get_weave_iptable_rules)
-run_on1 "sudo iptables -t nat -D POSTROUTING -j WEAVE"
+run_on1 "sudo iptables -t nat -X WEAVE-CANARY"
 wait_for_iptable_refresh
 IPT_AFTER=$(get_weave_iptable_rules)
 assert_raises "diff <(echo $IPT_BEFORE) <(echo $IPT_AFTER)" 1
@@ -38,7 +38,7 @@ stop_weave_on1
 ## Check refreshing
 weave_on1 "launch --iptables-refresh-interval=1s"
 IPT_BEFORE=$(get_weave_iptable_rules)
-run_on1 "sudo iptables -t nat -D POSTROUTING -j WEAVE"
+run_on1 "sudo iptables -t nat -X WEAVE-CANARY"
 wait_for_iptable_refresh
 IPT_AFTER=$(get_weave_iptable_rules)
 assert_raises "diff <(echo $IPT_BEFORE) <(echo $IPT_AFTER)" 0
@@ -50,7 +50,7 @@ stop_weave_on1
 weave_on1 "launch --iptables-refresh-interval=0s"
 weave_on1 "expose"
 IPT_BEFORE=$(get_weave_iptable_rules)
-run_on1 "sudo iptables -D FORWARD -o weave -j WEAVE-EXPOSE"
+run_on1 "sudo iptables -t nat -X WEAVE-CANARY"
 wait_for_iptable_refresh
 IPT_AFTER=$(get_weave_iptable_rules)
 assert_raises "diff <(echo $IPT_BEFORE) <(echo $IPT_AFTER)" 1
@@ -60,7 +60,7 @@ stop_weave_on1
 weave_on1 "launch --iptables-refresh-interval=1s"
 weave_on1 "expose"
 IPT_BEFORE=$(get_weave_iptable_rules)
-run_on1 "sudo iptables -D FORWARD -o weave -j WEAVE-EXPOSE"
+run_on1 "sudo iptables -t nat -X WEAVE-CANARY"
 wait_for_iptable_refresh
 IPT_AFTER=$(get_weave_iptable_rules)
 assert_raises "diff <(echo $IPT_BEFORE) <(echo $IPT_AFTER)" 0