gitea-admin/proxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: serialize Docker client recovery mutations

Browse Source
This commit is contained in:
proxyctl coordinator
2026-07-17 03:41:03 +08:00
parent 692a715248
commit 4c7a9c36e8
3 changed files with 209 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/dockerclient/dockerclient.go
+27 -10
View File
@@ -104,6 +104,7 @@ type FileSystem interface {
ReadFile(string) ([]byte, error)
AtomicWrite(string, []byte, os.FileMode, *Ownership) error
Remove(string) error
WithLock(string, *Ownership, func() error) error
}
type OSFileSystem struct{ TrustedRoots []string }
@@ -132,6 +133,9 @@ func (f OSFileSystem) Remove(path string) error { return f.store(pat
func (f OSFileSystem) AtomicWrite(path string, data []byte, mode os.FileMode, owner *Ownership) error {
return f.store(path, owner).AtomicWrite(path, data, mode)
}
func (f OSFileSystem) WithLock(path string, owner *Ownership, action func() error) error {
return f.store(path, owner).WithLock(path, action)
}
type priorValue struct {
Present bool `json:"present"`
@@ -191,6 +195,7 @@ func (a *Adapter) usablePath(path string) bool {
}
func (a *Adapter) ConfigPath() string { return a.configuration.ConfigFile }
func (a *Adapter) RecoveryPath() string { return a.configuration.RecoveryFile }
func (a *Adapter) LockPath() string { return a.configuration.RecoveryFile + ".lock" }
func (a *Adapter) Inspect(_ context.Context, active *profile.Profile) target.Inspection {
metadata, managed, metadataErr := a.readRecovery()
@@ -221,17 +226,29 @@ func (a *Adapter) SetDesired(ctx context.Context, enabled bool, active *profile.
if enabled && active == nil {
return target.OperationResult{Inspection: a.Inspect(ctx, active), Err: errors.New("no Active Profile; open Profile management to create and activate one")}
}
if enabled && a.Inspect(ctx, active).State == target.Enabled {
return target.OperationResult{Inspection: a.Inspect(ctx, active)}
}
var err error
if enabled {
err = a.enable(active)
} else {
err = a.disable()
}
inspection := a.Inspect(ctx, active)
var inspection target.Inspection
err := a.files.WithLock(a.LockPath(), a.ownership(), func() error {
if enabled {
inspection = a.Inspect(ctx, active)
if inspection.State == target.Enabled {
return nil
}
if err := a.enable(active); err != nil {
inspection = a.Inspect(ctx, active)
return err
}
} else if err := a.disable(); err != nil {
inspection = a.Inspect(ctx, active)
return err
}
inspection = a.Inspect(ctx, active)
return nil
})
if err != nil {
if inspection.Evidence == "" {
inspection = a.Inspect(ctx, active)
}
err = redactError(err, active)
}
return target.OperationResult{Inspection: inspection, Err: err}
internal/dockerclient/dockerclient_test.go
+151 -2
View File
@@ -5,9 +5,13 @@ import (
"encoding/json"
"errors"
"os"
"os/exec"
"path/filepath"
"strings"
"sync"
"syscall"
"testing"
"time"
"proxyctl/internal/dockerclient"
"proxyctl/internal/profile"
@@ -239,6 +243,92 @@ func TestUnmergeableJSONPreventsMutation(t *testing.T) {
}
}
func TestConcurrentMutationsSerializeRecoveryAndConfiguration(t *testing.T) {
files := &interleavingFS{FileSystem: dockerclient.OSFileSystem{}, firstRecovery: make(chan struct{}), secondRecovery: make(chan struct{}), secondConfig: make(chan struct{})}
a := newAdapterWithFS(t, files)
first := *activeProfile()
first.Name = "first"
first.HTTPURL = "http://first.example:8080"
second := *activeProfile()
second.Name = "second"
second.HTTPURL = "http://second.example:8080"
firstDone := make(chan target.OperationResult, 1)
go func() { firstDone <- a.SetDesired(context.Background(), true, &first) }()
select {
case <-files.firstRecovery:
case <-time.After(5 * time.Second):
t.Fatal("first mutation did not reach recovery write")
}
secondDone := make(chan target.OperationResult, 1)
go func() { secondDone <- a.SetDesired(context.Background(), true, &second) }()
for name, done := range map[string]<-chan target.OperationResult{"first": firstDone, "second": secondDone} {
select {
case result := <-done:
if result.Err != nil {
t.Fatalf("%s mutation: %v", name, result.Err)
}
case <-time.After(5 * time.Second):
t.Fatalf("%s mutation did not complete", name)
}
}
if got := a.Inspect(context.Background(), &second); got.State != target.Enabled {
t.Fatalf("final state = %s (%s), want second mutation consistently enabled", got.State, got.Evidence)
}
}
func TestDockerClientLockSerializesAcrossProcesses(t *testing.T) {
if root := os.Getenv("PROXYCTL_DOCKER_LOCK_HELPER_ROOT"); root != "" {
a := dockerclient.New(dockerclient.Configuration{
ConfigFile: filepath.Join(root, ".docker", "config.json"), RecoveryFile: filepath.Join(root, ".state", "proxyctl", "docker-containers.json"),
}, dockerclient.OSFileSystem{})
if result := a.SetDesired(context.Background(), true, activeProfile()); result.Err != nil {
t.Fatal(result.Err)
}
return
}
a := newAdapter(t, nil)
if err := os.MkdirAll(filepath.Dir(a.LockPath()), 0o700); err != nil {
t.Fatal(err)
}
lock, err := os.OpenFile(a.LockPath(), os.O_CREATE|os.O_RDWR, 0o600)
if err != nil {
t.Fatal(err)
}
defer lock.Close()
if err := syscall.Flock(int(lock.Fd()), syscall.LOCK_EX); err != nil {
t.Fatal(err)
}
root := filepath.Dir(filepath.Dir(filepath.Dir(a.RecoveryPath())))
command := exec.Command(os.Args[0], "-test.run=^TestDockerClientLockSerializesAcrossProcesses$")
command.Env = append(os.Environ(), "PROXYCTL_DOCKER_LOCK_HELPER_ROOT="+root)
done := make(chan error, 1)
go func() { done <- command.Run() }()
select {
case err := <-done:
t.Fatalf("child mutation did not wait for Docker client lock: %v", err)
case <-time.After(100 * time.Millisecond):
}
if err := syscall.Flock(int(lock.Fd()), syscall.LOCK_UN); err != nil {
t.Fatal(err)
}
select {
case err := <-done:
if err != nil {
t.Fatalf("child mutation after unlock: %v", err)
}
case <-time.After(5 * time.Second):
_ = command.Process.Kill()
t.Fatal("child mutation remained blocked after unlock")
}
if got := a.Inspect(context.Background(), activeProfile()); got.State != target.Enabled {
t.Fatalf("state after serialized child mutation = %s (%s)", got.State, got.Evidence)
}
}
func TestWritesAreOwnerOnlyAtomicAndIdempotent(t *testing.T) {
files := &countingFS{FileSystem: dockerclient.OSFileSystem{}}
a := newAdapterWithFS(t, files)
@@ -252,7 +342,7 @@ func TestWritesAreOwnerOnlyAtomicAndIdempotent(t *testing.T) {
if files.writes != writes {
t.Fatalf("idempotent enable performed %d additional writes", files.writes-writes)
}
for _, path := range []string{a.ConfigPath(), a.RecoveryPath()} {
for _, path := range []string{a.ConfigPath(), a.RecoveryPath(), a.LockPath()} {
info, err := os.Stat(path)
if err != nil {
t.Fatal(err)
@@ -274,7 +364,7 @@ func TestOwnedWritesUseConfigurationOwner(t *testing.T) {
if got := a.SetDesired(context.Background(), true, activeProfile()); got.Err != nil {
t.Fatal(got.Err)
}
if len(files.owners) != 2 {
if len(files.owners) != 3 {
t.Fatalf("owners = %#v", files.owners)
}
for _, owner := range files.owners {
@@ -448,6 +538,60 @@ func (f *countingFS) AtomicWrite(path string, data []byte, mode os.FileMode, own
return f.FileSystem.AtomicWrite(path, data, mode, owner)
}
type interleavingFS struct {
dockerclient.FileSystem
mu sync.Mutex
recoveryWrites, configWrites int
firstRecovery, secondRecovery, secondConfig chan struct{}
}
func (f *interleavingFS) AtomicWrite(path string, data []byte, mode os.FileMode, owner *dockerclient.Ownership) error {
f.mu.Lock()
isRecovery := strings.HasSuffix(path, "docker-containers.json")
if isRecovery {
f.recoveryWrites++
call := f.recoveryWrites
if call == 1 {
close(f.firstRecovery)
}
} else {
f.configWrites++
}
recoveryCall, configCall := f.recoveryWrites, f.configWrites
f.mu.Unlock()
if isRecovery && recoveryCall == 1 {
select {
case <-f.secondRecovery:
case <-time.After(100 * time.Millisecond):
}
err := f.FileSystem.AtomicWrite(path, data, mode, owner)
return err
}
if isRecovery && recoveryCall == 2 {
err := f.FileSystem.AtomicWrite(path, data, mode, owner)
close(f.secondRecovery)
return err
}
if !isRecovery && configCall == 1 {
select {
case <-f.secondConfig:
case <-time.After(100 * time.Millisecond):
}
return f.FileSystem.AtomicWrite(path, data, mode, owner)
}
if !isRecovery && configCall == 2 {
err := f.FileSystem.AtomicWrite(path, data, mode, owner)
close(f.secondConfig)
return err
}
return f.FileSystem.AtomicWrite(path, data, mode, owner)
}
func (f *interleavingFS) WithLock(path string, owner *dockerclient.Ownership, action func() error) error {
return f.FileSystem.WithLock(path, owner, action)
}
type ownershipFS struct {
dockerclient.FileSystem
owners []*dockerclient.Ownership
@@ -457,3 +601,8 @@ func (f *ownershipFS) AtomicWrite(path string, data []byte, mode os.FileMode, ow
f.owners = append(f.owners, owner)
return f.FileSystem.AtomicWrite(path, data, mode, nil)
}
func (f *ownershipFS) WithLock(path string, owner *dockerclient.Ownership, action func() error) error {
f.owners = append(f.owners, owner)
return f.FileSystem.WithLock(path, nil, action)
}
internal/securefs/securefs.go
+31
View File
@@ -146,6 +146,37 @@ func (s Store) AtomicWrite(path string, data []byte, mode os.FileMode) error {
return directory.Sync()
}
// WithLock serializes an action across processes using a descriptor-anchored,
// owner-restricted advisory lock file.
func (s Store) WithLock(path string, action func() error) error {
directory, base, err := s.openParent(path, true)
if err != nil {
return err
}
defer directory.Close()
lock, err := OpenAt(directory, base, unix.O_CREAT|unix.O_RDWR, 0o600)
if err != nil {
return err
}
defer lock.Close()
if err := regular(lock); err != nil {
return err
}
if s.Owner != nil && (s.Owner.UID != os.Geteuid() || s.Owner.GID != os.Getegid()) {
if err := lock.Chown(s.Owner.UID, s.Owner.GID); err != nil {
return err
}
}
if err := lock.Chmod(0o600); err != nil {
return err
}
if err := unix.Flock(int(lock.Fd()), unix.LOCK_EX); err != nil {
return err
}
defer func() { _ = unix.Flock(int(lock.Fd()), unix.LOCK_UN) }()
return action()
}
func (s Store) Remove(path string) error {
directory, base, err := s.openParent(path, false)
if err != nil {