diff --git a/cmd/serve/docker/docker.go b/cmd/serve/docker/docker.go index 8221a1ac0..d3cde5b84 100644 --- a/cmd/serve/docker/docker.go +++ b/cmd/serve/docker/docker.go @@ -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/.sock return srv.ServeUnix(pluginName, socketGid) diff --git a/cmd/serve/docker/docker_test.go b/cmd/serve/docker/docker_test.go index 4c0940a7c..e5df97676 100644 --- a/cmd/serve/docker/docker_test.go +++ b/cmd/serve/docker/docker_test.go @@ -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") } diff --git a/cmd/serve/docker/driver.go b/cmd/serve/docker/driver.go index c60f77aa0..4cb8f31f2 100644 --- a/cmd/serve/docker/driver.go +++ b/cmd/serve/docker/driver.go @@ -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() +} diff --git a/cmd/serve/docker/volume.go b/cmd/serve/docker/volume.go index 70414d076..fc4034850 100644 --- a/cmd/serve/docker/volume.go +++ b/cmd/serve/docker/volume.go @@ -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 == "" {