portmap: add a mutex

Use a cross-proces mutex based in a lock file to make
the teardown portmap function idempotent.

The teardown portmap function is not idempotent if it
is executed in parallel because it has concurrency
issues modifying the iptables rules.

Signed-off-by: Antonio Ojea <[email protected]>

go.mod

@@ -18,6 +18,7 @@ require (
 	github.com/golang/protobuf v1.3.1 // indirect
 	github.com/j-keck/arping v0.0.0-20160618110441-2cf9dc699c56
 	github.com/juju/errors v0.0.0-20180806074554-22422dad46e1
+	github.com/juju/fslock v0.0.0-20160525022230-4d5c94c67b4b
 	github.com/juju/loggo v0.0.0-20190526231331-6e530bcce5d8 // indirect
 	github.com/juju/testing v0.0.0-20190613124551-e81189438503 // indirect
 	github.com/kr/pretty v0.1.0 // indirect

go.sum

@@ -30,6 +30,8 @@ github.com/j-keck/arping v0.0.0-20160618110441-2cf9dc699c56 h1:742eGXur0715JMq73
 github.com/j-keck/arping v0.0.0-20160618110441-2cf9dc699c56/go.mod h1:ymszkNOg6tORTn+6F6j+Jc8TOr5osrynvN6ivFWZ2GA=
 github.com/juju/errors v0.0.0-20180806074554-22422dad46e1 h1:wnhMXidtb70kDZCeLt/EfsVtkXS5c8zLnE9y/6DIRAU=
 github.com/juju/errors v0.0.0-20180806074554-22422dad46e1/go.mod h1:W54LbzXuIE0boCoNJfwqpmkKJ1O4TCTZMetAt6jGk7Q=
+github.com/juju/fslock v0.0.0-20160525022230-4d5c94c67b4b h1:FQ7+9fxhyp82ks9vAuyPzG0/vVbWwMwLJ+P6yJI5FN8=
+github.com/juju/fslock v0.0.0-20160525022230-4d5c94c67b4b/go.mod h1:HMcgvsgd0Fjj4XXDkbjdmlbI505rUPBs6WBMYg2pXks=
 github.com/juju/loggo v0.0.0-20190526231331-6e530bcce5d8 h1:UUHMLvzt/31azWTN/ifGWef4WUqvXk0iRqdhdy/2uzI=
 github.com/juju/loggo v0.0.0-20190526231331-6e530bcce5d8/go.mod h1:vgyd7OREkbtVEN/8IXZe5Ooef3LQePvuBm9UWj6ZL8U=
 github.com/juju/testing v0.0.0-20190613124551-e81189438503 h1:ZUgTbk8oHgP0jpMieifGC9Lv47mHn8Pb3mFX3/Ew4iY=

plugins/meta/portmap/chain.go

@@ -17,12 +17,16 @@ package main
 import (
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/containernetworking/plugins/pkg/utils"
 	"github.com/coreos/go-iptables/iptables"
+	"github.com/juju/fslock"
 	"github.com/mattn/go-shellwords"
 )
 
+const lockfile = "/tmp/portmap.lock"
+
 type chain struct {
 	table       string
 	name        string
@@ -67,6 +71,13 @@ func (c *chain) setup(ipt *iptables.IPTables) error {
 // teardown idempotently deletes a chain. It will not error if the chain doesn't exist.
 // It will first delete all references to this chain in the entryChains.
 func (c *chain) teardown(ipt *iptables.IPTables) error {
+	m := fslock.New(lockfile)
+
+	// Will block until lock can be acquired and timeout after 5 seconds
+	if err := m.LockWithTimeout(5 * time.Second); err != nil {
+		return fmt.Errorf("failed to acquire the lock file %s : %v", lockfile, err)
+	}
+	defer m.Unlock()
 	// flush the chain
 	// This will succeed *and create the chain* if it does not exist.
 	// If the chain doesn't exist, the next checks will fail.

plugins/meta/portmap/chain_test.go

@@ -18,6 +18,7 @@ import (
 	"fmt"
 	"math/rand"
 	"runtime"
+	"sync"
 
 	"github.com/containernetworking/plugins/pkg/ns"
 	"github.com/containernetworking/plugins/pkg/testutils"
@@ -32,6 +33,7 @@ const TABLE = "filter" // We'll monkey around here
 var _ = Describe("chain tests", func() {
 	var testChain chain
 	var ipt *iptables.IPTables
+	var testNs ns.NetNS
 	var cleanup func()
 
 	BeforeEach(func() {
@@ -41,7 +43,7 @@ var _ = Describe("chain tests", func() {
 		currNs, err := ns.GetCurrentNS()
 		Expect(err).NotTo(HaveOccurred())
 
-		testNs, err := testutils.NewNS()
+		testNs, err = testutils.NewNS()
 		Expect(err).NotTo(HaveOccurred())
 
 		tlChainName := fmt.Sprintf("cni-test-%d", rand.Intn(10000000))
@@ -195,4 +197,46 @@ var _ = Describe("chain tests", func() {
 			}
 		}
 	})
+
+	It("deletes chains idempotently in parallel", func() {
+		defer cleanup()
+		// number of parallel executions
+		N := 10
+		var wg sync.WaitGroup
+		err := testChain.setup(ipt)
+		Expect(err).NotTo(HaveOccurred())
+		errCh := make(chan error, N)
+		for i := 0; i < N; i++ {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				// Use the testing namespace
+				runtime.LockOSThread()
+				defer runtime.UnlockOSThread()
+				err = testNs.Set()
+				Expect(err).NotTo(HaveOccurred())
+				// teardown chain
+				errCh <- testChain.teardown(ipt)
+			}()
+		}
+		wg.Wait()
+		close(errCh)
+		errs := 0
+		for err := range errCh {
+			if err != nil {
+				errs++
+			}
+		}
+
+		Expect(errs).Should(BeNumerically("==", 0))
+
+		chains, err := ipt.ListChains(TABLE)
+		Expect(err).NotTo(HaveOccurred())
+		for _, chain := range chains {
+			if chain == testChain.name {
+				Fail("Chain was not deleted")
+			}
+		}
+
+	})
 })