/
volume.go
326 lines (294 loc) · 7.59 KB
/
volume.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
package docker
import (
"context"
"errors"
"fmt"
"os"
"path/filepath"
"runtime"
"sort"
"time"
"github.com/rclone/rclone/cmd/mountlib"
"github.com/rclone/rclone/fs"
"github.com/rclone/rclone/fs/config"
"github.com/rclone/rclone/fs/rc"
"github.com/rclone/rclone/lib/file"
)
// Errors
var (
ErrVolumeNotFound = errors.New("volume not found")
ErrVolumeExists = errors.New("volume already exists")
ErrMountpointExists = errors.New("non-empty mountpoint already exists")
)
// 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]interface{}
fsString string // result of merging Fs, Type and Options
persist bool
mountType string
drv *Driver
mnt *mountlib.MountPoint
}
// VolOpts keeps volume options
type VolOpts map[string]string
// VolInfo represents a volume for Get and List requests
type VolInfo struct {
Name string
Mountpoint string `json:",omitempty"`
CreatedAt string `json:",omitempty"`
Status map[string]interface{} `json:",omitempty"`
}
func newVolume(ctx context.Context, name string, volOpt VolOpts, drv *Driver) (*Volume, error) {
path := filepath.Join(drv.root, name)
mnt := &mountlib.MountPoint{
MountPoint: path,
}
vol := &Volume{
Name: name,
MountPoint: path,
CreatedAt: time.Now(),
drv: drv,
mnt: mnt,
mountReqs: make(map[string]interface{}),
}
err := vol.applyOptions(volOpt)
if err == nil {
err = vol.setup(ctx)
}
if err != nil {
return nil, err
}
return vol, nil
}
// getInfo returns short digest about volume
func (vol *Volume) getInfo() *VolInfo {
vol.prepareState()
return &VolInfo{
Name: vol.Name,
CreatedAt: vol.CreatedAt.Format(time.RFC3339),
Mountpoint: vol.MountPoint,
Status: rc.Params{"Mounts": vol.Mounts},
}
}
// prepareState prepares volume for saving state
func (vol *Volume) prepareState() {
vol.Mounts = []string{}
for id := range vol.mountReqs {
vol.Mounts = append(vol.Mounts, id)
}
sort.Strings(vol.Mounts)
}
// restoreState updates volume from saved state
func (vol *Volume) restoreState(ctx context.Context, drv *Driver) error {
vol.drv = drv
vol.mnt = &mountlib.MountPoint{
MountPoint: vol.MountPoint,
}
volOpt := vol.Options
volOpt["fs"] = vol.Fs
volOpt["type"] = vol.Type
if err := vol.applyOptions(volOpt); err != nil {
return err
}
if err := vol.validate(); err != nil {
return err
}
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
}
// validate volume
func (vol *Volume) validate() error {
if vol.Name == "" {
return errors.New("volume name is required")
}
if (vol.Type != "" && vol.Fs != "") || (vol.Type == "" && vol.Fs == "") {
return errors.New("volume must have either remote or backend type")
}
if vol.persist && vol.Type == "" {
return errors.New("backend type is required to persist remotes")
}
if vol.persist && !canPersist {
return errors.New("using backend type to persist remotes is prohibited")
}
if vol.MountPoint == "" {
return errors.New("mount point is required")
}
if vol.mountReqs == nil {
vol.mountReqs = make(map[string]interface{})
}
return nil
}
// checkMountpoint verifies that mount point is an existing empty directory
func (vol *Volume) checkMountpoint() error {
path := vol.mnt.MountPoint
if runtime.GOOS == "windows" {
path = filepath.Dir(path)
}
_, err := os.Lstat(path)
if os.IsNotExist(err) {
if err = file.MkdirAll(path, 0700); err != nil {
return fmt.Errorf("failed to create mountpoint: %s: %w", path, err)
}
} else if err != nil {
return err
}
if runtime.GOOS != "windows" {
if err := mountlib.CheckMountEmpty(path); err != nil {
return ErrMountpointExists
}
}
return nil
}
// setup volume filesystem
func (vol *Volume) setup(ctx context.Context) error {
fs.Debugf(nil, "Setup volume %q as %q at path %s", vol.Name, vol.fsString, vol.MountPoint)
if err := vol.checkMountpoint(); err != nil {
return err
}
if vol.drv.dummy {
return nil
}
_, mountFn := mountlib.ResolveMountMethod(vol.mountType)
if mountFn == nil {
if vol.mountType != "" {
return fmt.Errorf("unsupported mount type %q", vol.mountType)
}
return errors.New("mount command unsupported by this build")
}
vol.mnt.MountFn = mountFn
if vol.persist {
// Add remote to config file
params := rc.Params{}
for key, val := range vol.Options {
params[key] = val
}
updateMode := config.UpdateRemoteOpt{}
_, err := config.CreateRemote(ctx, vol.Name, vol.Type, params, updateMode)
if err != nil {
return err
}
}
// Use existing remote
f, err := fs.NewFs(ctx, vol.fsString)
if err == nil {
vol.mnt.Fs = f
}
return err
}
// remove volume filesystem and mounts
func (vol *Volume) remove(ctx context.Context) error {
count := len(vol.mountReqs)
fs.Debugf(nil, "Remove volume %q (count %d)", vol.Name, count)
if count > 0 {
return errors.New("volume is in use")
}
if !vol.drv.dummy {
shutdownFn := vol.mnt.Fs.Features().Shutdown
if shutdownFn != nil {
if err := shutdownFn(ctx); err != nil {
return err
}
}
}
if vol.persist {
// Remote remote from config file
config.DeleteRemote(vol.Name)
}
return nil
}
// clearCache will clear VFS cache for the volume
func (vol *Volume) clearCache() error {
VFS := vol.mnt.VFS
if VFS == nil {
return nil
}
root, err := VFS.Root()
if err != nil {
return fmt.Errorf("error reading root: %v: %w", VFS.Fs(), err)
}
root.ForgetAll()
return nil
}
// mount volume filesystem
func (vol *Volume) mount(id string) error {
drv := vol.drv
count := len(vol.mountReqs)
fs.Debugf(nil, "Mount volume %q for id %q at path %s (count %d)",
vol.Name, id, vol.MountPoint, count)
if _, found := vol.mountReqs[id]; found {
return errors.New("volume is already mounted by this id")
}
if count > 0 { // already mounted
vol.mountReqs[id] = nil
return nil
}
if drv.dummy {
vol.mountReqs[id] = nil
return nil
}
if vol.mnt.Fs == nil {
return errors.New("volume filesystem is not ready")
}
if _, err := vol.mnt.Mount(); err != nil {
return err
}
vol.mountReqs[id] = nil
vol.drv.monChan <- false // ask monitor to refresh channels
return nil
}
// unmount volume
func (vol *Volume) unmount(id string) error {
count := len(vol.mountReqs)
fs.Debugf(nil, "Unmount volume %q from id %q at path %s (count %d)",
vol.Name, id, vol.MountPoint, count)
if count == 0 {
return errors.New("volume is not mounted")
}
if _, found := vol.mountReqs[id]; !found {
return errors.New("volume is not mounted by this id")
}
delete(vol.mountReqs, id)
if len(vol.mountReqs) > 0 {
return nil // more mounts left
}
if vol.drv.dummy {
return nil
}
mnt := vol.mnt
if mnt.UnmountFn != nil {
if err := mnt.UnmountFn(); err != nil {
return err
}
}
mnt.ErrChan = nil
mnt.UnmountFn = nil
mnt.VFS = nil
vol.drv.monChan <- false // ask monitor to refresh channels
return nil
}
func (vol *Volume) unmountAll() error {
var firstErr error
for id := range vol.mountReqs {
err := vol.unmount(id)
if firstErr == nil {
firstErr = err
}
}
return firstErr
}