Do our best not to invoke iptables concurrently if --wait is unsupported

We encountered a situation where concurrent invocations of the docker daemon on a machine with an older version of iptables led to nondeterministic errors related to simultaenous invocations of iptables.

While this is best resolved by upgrading iptables itself, the particular situation would have been avoided if the docker daemon simply took care not to concurrently invoke iptables. Of course, external processes could also cause iptables to fail in this way, but invoking docker in parallel seems like a pretty common case.

Signed-off-by: Aaron Davidson <aaron@databricks.com>

iptables/iptables.go

@@ -8,6 +8,7 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
+	"sync"
 
 	"github.com/Sirupsen/logrus"
 )
@@ -25,8 +26,10 @@ const (
 )
 
 var (
-	iptablesPath        string
-	supportsXlock       = false
+	iptablesPath  string
+	supportsXlock = false
+	// used to lock iptables commands if xtables lock is not supported
+	bestEffortLock      sync.Mutex
 	ErrIptablesNotFound = errors.New("Iptables not found")
 )
 
@@ -288,6 +291,9 @@ func Raw(args ...string) ([]byte, error) {
 	}
 	if supportsXlock {
 		args = append([]string{"--wait"}, args...)
+	} else {
+		bestEffortLock.Lock()
+		defer bestEffortLock.Unlock()
 	}
 
 	logrus.Debugf("%s, %v", iptablesPath, args)

iptables/iptables_test.go

@@ -5,6 +5,7 @@ import (
 	"os/exec"
 	"strconv"
 	"strings"
+	"sync"
 	"testing"
 )
 
@@ -169,6 +170,45 @@ func TestOutput(t *testing.T) {
 	}
 }
 
+func TestConcurrencyWithWait(t *testing.T) {
+	RunConcurrencyTest(t, true)
+}
+
+func TestConcurrencyNoWait(t *testing.T) {
+	RunConcurrencyTest(t, false)
+}
+
+// Runs 10 concurrent rule additions. This will fail if iptables
+// is actually invoked simultaneously without --wait.
+// Note that if iptables does not support the xtable lock on this
+// system, then allowXlock has no effect -- it will always be off.
+func RunConcurrencyTest(t *testing.T, allowXlock bool) {
+	var wg sync.WaitGroup
+
+	if !allowXlock && supportsXlock {
+		supportsXlock = false
+		defer func() { supportsXlock = true }()
+	}
+
+	ip := net.ParseIP("192.168.1.1")
+	port := 1234
+	dstAddr := "172.17.0.1"
+	dstPort := 4321
+	proto := "tcp"
+
+	for i := 0; i < 10; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			err := natChain.Forward(Append, ip, port, proto, dstAddr, dstPort)
+			if err != nil {
+				t.Fatal(err)
+			}
+		}()
+	}
+	wg.Wait()
+}
+
 func TestCleanup(t *testing.T) {
 	var err error
 	var rules []byte