f-denkena/rclone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docker serve: fix plugin timeout on restart when volumes have active mounts

Browse Source 
Previously, restoreState in NewDriver would restore volumes AND perform
FUSE mounts synchronously before the Docker plugin socket was created.
This caused two problems:

1. The monChan was created after restoreState, but mount() sends on
   monChan, causing a deadlock (send on nil channel blocks forever).

2. Even with the channel fix, slow or hanging mounts during state
   restore would block the server socket from being created, causing
   Docker to time out after ~13 seconds with "no such file or
   directory" for the plugin socket.

Fix by:
- Moving monChan creation and monitor goroutine start before
  restoreState
- Splitting state restore into two phases: restoreState (metadata +
  filesystem setup only) and RestoreMounts (actual FUSE mounts)
- Calling RestoreMounts asynchronously after the server starts
  listening
- Performing mount restoration concurrently across volumes

Fixes #9231
This commit is contained in:
Nick Craig-Wood
2026-07-13 15:30:03 +01:00
parent 4235aa4fe5
commit 560d3928d0
4 changed files with 155 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/serve/docker/docker.go
+4
View File
@@ -73,6 +73,10 @@ var Command = &cobra.Command{
return err
}
srv := NewServer(drv)
// Restore mounts in background after the server starts
// listening so Docker can communicate with the plugin
// even if individual mounts are slow.
go drv.RestoreMounts()
if socketAddr == "" {
// Listen on unix socket at /run/docker/plugins/<pluginName>.sock
return srv.ServeUnix(pluginName, socketGid)
cmd/serve/docker/docker_test.go
+70
View File
@@ -415,6 +415,76 @@ func testMountAPI(t *testing.T, sockAddr string) {
assert.Empty(t, listRes.Volumes)
}
func TestDockerPluginDeferredMounts(t *testing.T) {
ctx := context.Background()
oldCacheDir := config.GetCacheDir()
testDir, testFs := initialise(ctx, t)
err := config.SetCacheDir(testDir)
require.NoError(t, err)
defer func() {
_ = config.SetCacheDir(oldCacheDir)
if !t.Failed() {
fstest.Purge(testFs)
_ = os.RemoveAll(testDir)
}
}()
// Create dummy volume driver with volumes and mounts
drv, err := docker.NewDriver(ctx, testDir, nil, nil, true, true)
require.NoError(t, err)
require.NotNil(t, drv)
defer drv.Exit()
volReq := &docker.CreateRequest{
Name: "vol1",
Options: docker.VolOpts{"remote": testDir},
}
assert.NoError(t, drv.Create(volReq))
volReq.Name = "vol2"
assert.NoError(t, drv.Create(volReq))
// Mount vol2 with two IDs
mountReq := &docker.MountRequest{Name: "vol2", ID: "id1"}
_, err = drv.Mount(mountReq)
assert.NoError(t, err)
mountReq.ID = "id2"
_, err = drv.Mount(mountReq)
assert.NoError(t, err)
// Simulate plugin restart - state is restored but mounts are deferred
// (Don't call drv.Exit() since that clears mounts from state, simulating a crash)
drv2, err := docker.NewDriver(ctx, testDir, nil, nil, true, false)
require.NoError(t, err)
require.NotNil(t, drv2)
defer drv2.Exit()
// Volumes should be listed (metadata restored)
listRes, err := drv2.List()
require.NoError(t, err)
require.Equal(t, 2, len(listRes.Volumes))
// vol2 should have no active mounts yet (deferred)
path2 := filepath.Join(testDir, "vol2")
assertVolumeInfo(t, listRes.Volumes[1], "vol2", path2)
// Now restore mounts
drv2.RestoreMounts()
// After RestoreMounts, vol2 should have its mounts back
listRes, err = drv2.List()
require.NoError(t, err)
require.Equal(t, 2, len(listRes.Volumes))
status := listRes.Volumes[1].Status
require.NotNil(t, status)
mounts, ok := status["Mounts"]
require.True(t, ok)
mountList, ok := mounts.([]string)
require.True(t, ok)
assert.Equal(t, 2, len(mountList))
assert.Contains(t, mountList, "id1")
assert.Contains(t, mountList, "id2")
}
func TestDockerPluginMountTCP(t *testing.T) {
testMountAPI(t, "localhost:53789")
}
cmd/serve/docker/driver.go
+51 -6
View File
@@ -61,6 +61,12 @@ func NewDriver(ctx context.Context, root string, mntOpt *mountlib.Options, vfsOp
}
drv.mntOpt.Daemon = false
// start mount monitoring - must be before restoreState since
// restoring mounts sends on monChan
drv.hupChan = make(chan os.Signal, 1)
drv.monChan = make(chan bool, 1)
go drv.monitor()
// restore from saved state
if !forgetState {
if err = drv.restoreState(ctx); err != nil {
@@ -68,11 +74,6 @@ func NewDriver(ctx context.Context, root string, mntOpt *mountlib.Options, vfsOp
}
}
// start mount monitoring
drv.hupChan = make(chan os.Signal, 1)
drv.monChan = make(chan bool, 1)
go drv.monitor()
// unmount all volumes on exit
atexit.Register(func() {
drv.exitOnce.Do(drv.Exit)
@@ -332,7 +333,11 @@ func (drv *Driver) saveState() error {
return fmt.Errorf("failed to save state: %w", err)
}
// restoreState recreates volumes from saved driver state
// restoreState recreates volumes from saved driver state.
//
// It restores volume metadata and filesystems but defers the actual
// FUSE mounts. Call restoreMounts afterwards (typically after the
// server socket is listening) to perform the mounts.
func (drv *Driver) restoreState(ctx context.Context) error {
fs.Debugf(nil, "Restore state from %s", drv.statePath)
@@ -359,3 +364,43 @@ func (drv *Driver) restoreState(ctx context.Context) error {
}
return nil
}
// RestoreMounts mounts all volumes that were previously mounted.
//
// This should be called after the server socket is listening so that
// Docker can communicate with the plugin even if individual mounts
// are slow or fail. Mounts are performed concurrently.
func (drv *Driver) RestoreMounts() {
// Collect pending mounts under the lock
type pendingMount struct {
vol *Volume
mounts []string
}
drv.mu.Lock()
var pending []pendingMount
for _, vol := range drv.volumes {
mounts := vol.getPendingMounts()
if len(mounts) > 0 {
pending = append(pending, pendingMount{vol: vol, mounts: mounts})
}
}
drv.mu.Unlock()
// Mount concurrently without holding the driver lock
var wg sync.WaitGroup
for _, p := range pending {
wg.Add(1)
go func(vol *Volume, mounts []string) {
defer wg.Done()
for _, id := range mounts {
drv.mu.Lock()
err := vol.mount(id)
drv.mu.Unlock()
if err != nil {
fs.Logf(nil, "Failed to restore mount %q for volume %q: %v", id, vol.Name, err)
}
}
}(p.vol, p.mounts)
}
wg.Wait()
}
cmd/serve/docker/volume.go
+30 -20
View File
@@ -27,20 +27,21 @@ var (
// Volume keeps volume runtime state
// Public members get persisted in saved state
type Volume struct {
Name string `json:"name"`
MountPoint string `json:"mountpoint"`
CreatedAt time.Time `json:"created"`
Fs string `json:"fs"` // remote[,connectString]:path
Type string `json:"type,omitempty"` // same as ":backend:"
Path string `json:"path,omitempty"` // for "remote:path" or ":backend:path"
Options VolOpts `json:"options"` // all options together
Mounts []string `json:"mounts"` // mountReqs as a string list
mountReqs map[string]any
fsString string // result of merging Fs, Type and Options
persist bool
mountType string
drv *Driver
mnt *mountlib.MountPoint
Name string `json:"name"`
MountPoint string `json:"mountpoint"`
CreatedAt time.Time `json:"created"`
Fs string `json:"fs"` // remote[,connectString]:path
Type string `json:"type,omitempty"` // same as ":backend:"
Path string `json:"path,omitempty"` // for "remote:path" or ":backend:path"
Options VolOpts `json:"options"` // all options together
Mounts []string `json:"mounts"` // mountReqs as a string list
mountReqs map[string]any
pendingMounts []string // mount IDs to restore after server starts
fsString string // result of merging Fs, Type and Options
persist bool
mountType string
drv *Driver
mnt *mountlib.MountPoint
}
// VolOpts keeps volume options
@@ -97,12 +98,18 @@ func (vol *Volume) prepareState() {
sort.Strings(vol.Mounts)
}
// restoreState updates volume from saved state
// restoreState updates volume from saved state.
//
// It restores the volume configuration and filesystem but does not
// perform FUSE mounts. The pending mount IDs are saved and can be
// retrieved with getPendingMounts for deferred mounting.
func (vol *Volume) restoreState(ctx context.Context, drv *Driver) error {
vol.drv = drv
vol.mnt = &mountlib.MountPoint{
MountPoint: vol.MountPoint,
}
// Save pending mounts before applyOptions clears them
vol.pendingMounts = vol.Mounts
volOpt := vol.Options
volOpt["fs"] = vol.Fs
volOpt["type"] = vol.Type
@@ -115,14 +122,17 @@ func (vol *Volume) restoreState(ctx context.Context, drv *Driver) error {
if err := vol.setup(ctx); err != nil {
return err
}
for _, id := range vol.Mounts {
if err := vol.mount(id); err != nil {
return err
}
}
return nil
}
// getPendingMounts returns and clears the list of mount IDs that
// were saved from state and need to be re-mounted.
func (vol *Volume) getPendingMounts() []string {
mounts := vol.pendingMounts
vol.pendingMounts = nil
return mounts
}
// validate volume
func (vol *Volume) validate() error {
if vol.Name == "" {