Introduce (global) default config file locking

Adds LockDefault() and UnlockDefault() which lock an associate file
mapping to the same default containers file that Write() selects.
This functionality allows for multiple read-and-write operations
to be atomically executed by serializing access from multiple
updaters.

As an example, if two parallel updaters are inserting a different
entry into a map, one will likely be omitted from being
overwritten with data from a stale read. Instead, using these
functions, callers can coordinate, ensuring writes are never
interposed between another caller's read-update pattern.

Signed-off-by: Jason T. Greene <[email protected]>

Makefile

@@ -55,9 +55,16 @@ endif
 bin/netavark-testplugin:
 	$(GO_BUILD) -o $@ ./libnetwork/netavark/testplugin/
 
+.PHONY: bin/flocksim
+bin/flocksim:
+	$(GO_BUILD) -o $@ ./cmd/flocksim/
+
 .PHONY: netavark-testplugin
 netavark-testplugin: bin/netavark-testplugin
 
+.PHONY: flocksim
+flocksim: bin/flocksim
+
 .PHONY: docs
 docs:
 	$(MAKE) -C docs
@@ -97,7 +104,7 @@ install:
 test: test-unit
 
 .PHONY: test-unit
-test-unit: netavark-testplugin
+test-unit: netavark-testplugin flocksim
 	go test --tags $(BUILDTAGS) -v ./libimage
 	go test --tags $(BUILDTAGS) -v ./libnetwork/...
 	go test --tags $(BUILDTAGS) -v ./pkg/...
@@ -111,6 +118,7 @@ clean: ## Clean artifacts
 	$(MAKE) -C docs clean
 	find . -name \*~ -delete
 	find . -name \#\* -delete
+	rm -rf bin
 
 .PHONY: seccomp.json
 seccomp.json: $(sources)

cmd/flocksim/main.go

@@ -0,0 +1,30 @@
+package main
+
+import (
+	"bufio"
+	"fmt"
+	"os"
+
+	"github.com/containers/storage/pkg/lockfile"
+)
+
+// flocksim is a testing tool used by the config lock tests
+func main() {
+	if len(os.Args) < 2 {
+		fmt.Printf("Usage: %s <path to lock>\n", os.Args[0])
+		os.Exit(1)
+	}
+
+	lock, err := lockfile.GetLockFile(os.Args[1])
+	if err != nil {
+		fmt.Fprintln(os.Stderr, err.Error())
+		os.Exit(1)
+	}
+
+	lock.Lock()
+	fmt.Println("acquired lock, hit enter to release")
+
+	reader := bufio.NewReader(os.Stdin)
+	_, _ = reader.ReadString('\n')
+	lock.Unlock()
+}

pkg/config/config.go

@@ -17,6 +17,7 @@ import (
 	"github.com/containers/common/pkg/capabilities"
 	"github.com/containers/common/pkg/util"
 	"github.com/containers/storage/pkg/ioutils"
+	"github.com/containers/storage/pkg/lockfile"
 	"github.com/containers/storage/pkg/unshare"
 	units "github.com/docker/go-units"
 	selinux "github.com/opencontainers/selinux/go-selinux"
@@ -1191,9 +1192,10 @@ func rootlessConfigPath() (string, error) {
 }
 
 var (
-	configErr   error
-	configMutex sync.Mutex
-	config      *Config
+	configErr      error
+	configMutex    sync.Mutex
+	config         *Config
+	configLockFile *lockfile.LockFile
 )
 
 // Default returns the default container config.
@@ -1221,6 +1223,72 @@ func defConfig() (*Config, error) {
 	return config, configErr
 }
 
+// Locks the config default file through an associated runtime file lock
+// This method enables callers to coordinate updates to the default
+// configuration file, and ensure atomicity across multiple operations.
+// Typically a caller would call this method before reading values,
+// perform permutations to the config, write the config, and finally
+// call UnlockDefault(). Since this lock is global, callers should
+// aim to minimize the length of time the lock is held.
+//
+// Multiple calls to this method will reference count the underlying
+// lock.
+func LockDefault() error {
+	lockFile, err := getLockFile()
+	if err != nil {
+		return err
+	}
+	lockFile.Lock()
+	return nil
+}
+
+// Unlocks the config default file through an associated runtime file lock.
+// The underlying lock is reference counted. The file will only become
+// unlocked when a corresponding number of calls to this method have been
+// made matching the number of preceding LockDefault() calls.
+func UnlockDefault() {
+	lockFile, err := getLockFile()
+	if err != nil {
+		return
+	}
+	lockFile.Unlock()
+}
+
+func getLockFilePath() (string, error) {
+	lockDir, err := util.GetRuntimeDir()
+	if err != nil {
+		return "", err
+	}
+
+	lockDir = filepath.Join(lockDir, "containers")
+	if err := os.MkdirAll(lockDir, 0o755); err != nil {
+		return "", err
+	}
+
+	return filepath.Join(lockDir, "config.lck"), nil
+}
+
+func getLockFile() (*lockfile.LockFile, error) {
+	configMutex.Lock()
+	defer configMutex.Unlock()
+
+	if configLockFile != nil {
+		return configLockFile, nil
+	}
+
+	path, err := getLockFilePath()
+	if err != nil {
+		return nil, err
+	}
+	lockFile, err := lockfile.GetLockFile(path)
+	if err != nil {
+		return nil, err
+	}
+
+	configLockFile = lockFile
+	return lockFile, nil
+}
+
 func Path() string {
 	if path := os.Getenv("CONTAINERS_CONF"); path != "" {
 		return path

pkg/config/config_test.go

@@ -1,13 +1,17 @@
 package config
 
 import (
+	"bufio"
 	"bytes"
+	"errors"
 	"fmt"
 	"io"
 	"os"
+	"os/exec"
 	"runtime"
 	"sort"
 	"strings"
+	"time"
 
 	"github.com/containers/common/pkg/apparmor"
 	"github.com/containers/common/pkg/capabilities"
@@ -1013,4 +1017,198 @@ env=["foo=bar"]`
 		gomega.Expect(config.Containers.BaseHostsFile).To(gomega.Equal("/etc/hosts2"))
 		gomega.Expect(config.Containers.EnableLabeledUsers).To(gomega.BeTrue())
 	})
+
+	It("container conf lock operations", func() {
+		// This test tests two competing goroutines which both add a connection
+		// definition at the same time. Through the use of LockDefault/UnlockDefault
+		// each goroutine should be serialized and correctly observe the other's
+		// state
+		const oneURI = "https://qa/run/one.sock"
+		const twoURI = "https://qa/run/two.sock"
+
+		if _, err := os.Stat("../../bin/flocksim"); err != nil {
+			Skip("flocksim not present")
+			return
+		}
+
+		defer os.Unsetenv("CONTAINERS_CONF")
+		f, err := os.CreateTemp("", "container-common-test")
+		gomega.Expect(err).ToNot(gomega.HaveOccurred())
+		defer f.Close()
+		defer os.Remove(f.Name())
+
+		os.Setenv("CONTAINERS_CONF", f.Name())
+		primaryConfig, err := Default()
+		gomega.Expect(err).ToNot(gomega.HaveOccurred())
+		err = primaryConfig.Write()
+		gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+		configPath := f.Name()
+		lockPath, err := getLockFilePath()
+		gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+		err = LockDefault()
+		gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+		result := make(chan error, 3)
+		go func() {
+			// Since the contract for lock is global, simulate another process
+			// through a manual direct lock.
+			cmd, inStream, outStream, errStream, err := startCommandWithStreams("../../bin/flocksim", lockPath)
+			if err != nil {
+				result <- err
+			}
+
+			output := singleLineSlurp(outStream)
+			errResult := allLineSlurp(errStream)
+
+			// Wait until we get a lock acquired message, if we don't an error was thrown
+			if len(<-output) == 0 {
+				_ = cmd.Wait()
+				result <- errors.New(<-errResult)
+				return
+			}
+
+			defer func() {
+				// Tell flocksim to shutdown
+				_, _ = inStream.Write([]byte("done\n"))
+				_ = inStream.Close()
+				result <- waitCommandWithTimeout(cmd, 10)
+			}()
+
+			// Indicate success, we have a lock
+			result <- nil
+
+			racingConfig := Config{}
+			err = readConfigFromFile(configPath, &racingConfig)
+			if err != nil {
+				result <- err
+				return
+			}
+
+			sd, ok := racingConfig.Engine.ServiceDestinations["one"]
+			if !ok {
+				result <- errors.New("expected service definition missing")
+				return
+			}
+
+			if sd.URI != oneURI {
+				result <- errors.New("service definition did not match expected URI")
+				return
+			}
+
+			racingConfig.Engine.ServiceDestinations = map[string]Destination{
+				"two": {
+					URI:      twoURI,
+					Identity: "/.ssh/id_two",
+				},
+			}
+
+			result <- racingConfig.Write()
+		}()
+
+		// Not mandatory, just yielding to increase chance of failure if problem
+		time.Sleep(1 * time.Second)
+
+		primaryConfig, err = ReadCustomConfig()
+		gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+		primaryConfig.Engine.ActiveService = "one"
+		primaryConfig.Engine.ServiceDestinations = map[string]Destination{
+			"one": {
+				URI:      oneURI,
+				Identity: "/.ssh/id_one",
+			},
+		}
+		err = primaryConfig.Write()
+		gomega.Expect(err).ToNot(gomega.HaveOccurred())
+		UnlockDefault()
+		// Verify goroutine acquired lock successfully
+		gomega.Expect(<-result).ToNot(gomega.HaveOccurred())
+
+		// Block until goroutine finishes
+		err = LockDefault()
+		gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+		// Verify goroutine's write was successful
+		gomega.Expect(<-result).ToNot(gomega.HaveOccurred())
+
+		// Verify goroutine exit was successful
+		gomega.Expect(<-result).ToNot(gomega.HaveOccurred())
+
+		primaryConfig, err = ReadCustomConfig()
+		gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+		sd, ok := primaryConfig.Engine.ServiceDestinations["two"]
+		gomega.Expect(ok).To(gomega.BeTrue())
+		gomega.Expect(sd.URI).To(gomega.Equal(twoURI))
+	})
 })
+
+func waitCommandWithTimeout(cmd *exec.Cmd, seconds int) error {
+	wChan := make(chan error, 1)
+	go func() {
+		wChan <- cmd.Wait()
+	}()
+	select {
+	case err := <-wChan:
+		return err
+	case <-time.After(time.Duration(seconds) * time.Second):
+		_ = cmd.Process.Kill()
+		return fmt.Errorf("Timeout running %s had to kill: %w", cmd.Args[0], <-wChan)
+	}
+}
+
+func startCommandWithStreams(executable string, args ...string) (*exec.Cmd, io.WriteCloser, io.ReadCloser, io.ReadCloser, error) {
+	cmd := exec.Command(executable, args...)
+	outStream, err := cmd.StdoutPipe()
+	if err != nil {
+		return nil, nil, nil, nil, err
+	}
+	errStream, err := cmd.StderrPipe()
+	if err != nil {
+		return nil, nil, nil, nil, err
+	}
+	inStream, err := cmd.StdinPipe()
+	if err != nil {
+		return nil, nil, nil, nil, err
+	}
+	err = cmd.Start()
+	if err != nil {
+		return nil, nil, nil, nil, err
+	}
+
+	return cmd, inStream, outStream, errStream, nil
+}
+
+func singleLineSlurp(stream io.ReadCloser) chan string {
+	outChan := make(chan string, 1)
+
+	go func() {
+		reader := bufio.NewReader(stream)
+		str, err := reader.ReadString('\n')
+		if err != nil {
+			str = ""
+			stream.Close()
+		}
+		outChan <- str
+	}()
+
+	return outChan
+}
+
+func allLineSlurp(stream io.ReadCloser) chan string {
+	outChan := make(chan string, 1)
+
+	go func() {
+		reader := bufio.NewReader(stream)
+		bytes, err := io.ReadAll(reader)
+		if err != nil {
+			bytes = []byte{}
+			stream.Close()
+		}
+		outChan <- string(bytes)
+	}()
+
+	return outChan
+}

pkg/util/util_darwin.go

@@ -0,0 +1,12 @@
+package util
+
+import "os"
+
+// getRuntimeDir returns the runtime directory
+func GetRuntimeDir() (string, error) {
+	tmpDir, ok := os.LookupEnv("TMPDIR")
+	if !ok {
+		tmpDir = "/tmp"
+	}
+	return tmpDir, nil
+}

pkg/util/util_supported.go

@@ -1,5 +1,5 @@
-//go:build linux || darwin || freebsd
-// +build linux darwin freebsd
+//go:build linux || freebsd
+// +build linux freebsd
 
 package util