util/iptables: check for and use new iptables-restore 'wait' argument

iptables-restore did not previously perform any locking, meaning that
when callers (like kube-proxy) asked iptables-restore to write large
numbers of rules, the iptables-restore process might run in parallel
with other 'iptables' invocations in kubelet (hostports), docker,
and other software.  This causes errors like:

"CNI request failed with status 400: 'Failed to ensure that nat chain
POSTROUTING jumps to MASQUERADE: error checking rule: exit status 4:
iptables: Resource temporarily  unavailable."

or from Docker

"Failed to allocate and map port 1095-1095: iptables failed:
iptables --wait -t nat -A DOCKER -p tcp -d 0/0 --dport 1095
-j DNAT --to-destination 10.1.0.2:1095 ! -i lbr0: iptables:
Resource temporarily unavailable.\n (exit status 4)"

iptables-restore "wait" functionality was added in iptables git
commit 999eaa241212d3952ddff39a99d0d55a74e3639e but is NOT YET
in a released version of iptables.

See also https://bugzilla.redhat.com/show_bug.cgi?id=1417234

pkg/util/iptables/iptables.go

@@ -124,12 +124,13 @@ const MinWait2Version = "1.4.22"
 
 // runner implements Interface in terms of exec("iptables").
 type runner struct {
-	mu       sync.Mutex
-	exec     utilexec.Interface
-	dbus     utildbus.Interface
-	protocol Protocol
-	hasCheck bool
-	waitFlag []string
+	mu              sync.Mutex
+	exec            utilexec.Interface
+	dbus            utildbus.Interface
+	protocol        Protocol
+	hasCheck        bool
+	waitFlag        []string
+	restoreWaitFlag []string
 
 	reloadFuncs []func()
 	signal      chan *godbus.Signal
@@ -142,12 +143,14 @@ func New(exec utilexec.Interface, dbus utildbus.Interface, protocol Protocol) In
 		glog.Warningf("Error checking iptables version, assuming version at least %s: %v", MinCheckVersion, err)
 		vstring = MinCheckVersion
 	}
+
 	runner := &runner{
-		exec:     exec,
-		dbus:     dbus,
-		protocol: protocol,
-		hasCheck: getIPTablesHasCheckCommand(vstring),
-		waitFlag: getIPTablesWaitFlag(vstring),
+		exec:            exec,
+		dbus:            dbus,
+		protocol:        protocol,
+		hasCheck:        getIPTablesHasCheckCommand(vstring),
+		waitFlag:        getIPTablesWaitFlag(vstring),
+		restoreWaitFlag: getIPTablesRestoreWaitFlag(exec),
 	}
 	runner.connectToFirewallD()
 	return runner
@@ -335,8 +338,9 @@ func (runner *runner) restoreInternal(args []string, data []byte, flush FlushFla
 	}
 
 	// run the command and return the output or an error including the output and error
-	glog.V(4).Infof("running iptables-restore %v", args)
-	cmd := runner.exec.Command(cmdIPTablesRestore, args...)
+	fullArgs := append(runner.restoreWaitFlag, args...)
+	glog.V(4).Infof("running iptables-restore %v", fullArgs)
+	cmd := runner.exec.Command(cmdIPTablesRestore, fullArgs...)
 	cmd.SetStdin(bytes.NewBuffer(data))
 	b, err := cmd.CombinedOutput()
 	if err != nil {
@@ -519,6 +523,46 @@ func getIPTablesVersionString(exec utilexec.Interface) (string, error) {
 	return match[1], nil
 }
 
+// Checks if iptables-restore has a "wait" flag
+// --wait support landed in v1.6.1+ right before --version support, so
+// any version of iptables-restore that supports --version will also
+// support --wait
+func getIPTablesRestoreWaitFlag(exec utilexec.Interface) []string {
+	vstring, err := getIPTablesRestoreVersionString(exec)
+	if err != nil || vstring == "" {
+		glog.V(3).Infof("couldn't get iptables-restore version; assuming it doesn't support --wait")
+		return nil
+	}
+	if _, err := utilversion.ParseGeneric(vstring); err != nil {
+		glog.V(3).Infof("couldn't parse iptables-restore version; assuming it doesn't support --wait")
+		return nil
+	}
+
+	return []string{"--wait=2"}
+}
+
+// getIPTablesRestoreVersionString runs "iptables-restore --version" to get the version string
+// in the form "X.X.X"
+func getIPTablesRestoreVersionString(exec utilexec.Interface) (string, error) {
+	// this doesn't access mutable state so we don't need to use the interface / runner
+
+	// iptables-restore hasn't always had --version, and worse complains
+	// about unrecognized commands but doesn't exit when it gets them.
+	// Work around that by setting stdin to nothing so it exits immediately.
+	cmd := exec.Command(cmdIPTablesRestore, "--version")
+	cmd.SetStdin(bytes.NewReader([]byte{}))
+	bytes, err := cmd.CombinedOutput()
+	if err != nil {
+		return "", err
+	}
+	versionMatcher := regexp.MustCompile("v([0-9]+(\\.[0-9]+)+)")
+	match := versionMatcher.FindStringSubmatch(string(bytes))
+	if match == nil {
+		return "", fmt.Errorf("no iptables version found in string: %s", bytes)
+	}
+	return match[1], nil
+}
+
 // goroutine to listen for D-Bus signals
 func (runner *runner) dbusSignalHandler(bus utildbus.Connection) {
 	firewalld := bus.Object(firewalldName, firewalldPath)