diff --git a/weed/command/filer_remote_sync.go b/weed/command/filer_remote_sync.go index 7ab94ad40d..43a5da3d2e 100644 --- a/weed/command/filer_remote_sync.go +++ b/weed/command/filer_remote_sync.go @@ -7,6 +7,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/remote_pb" "github.com/seaweedfs/seaweedfs/weed/replication/source" "github.com/seaweedfs/seaweedfs/weed/security" "github.com/seaweedfs/seaweedfs/weed/util" @@ -20,9 +21,15 @@ type RemoteSyncOptions struct { readChunkFromFiler *bool timeAgo *time.Duration dir *string + allMounts *bool + mountsCheck *time.Duration clientId int32 clientEpoch int32 resumeTsNs int64 + // what resumeTsNs was earned under: the directory's remote location, and + // the offset stored when it was kept (resumeFor) + resumeLoc *remote_pb.RemoteStorageLocation + resumeStoredNs int64 } var _ = filer_pb.FilerClient(&RemoteSyncOptions{}) @@ -48,6 +55,8 @@ func init() { cmdFilerRemoteSynchronize.Run = runFilerRemoteSynchronize // break init cycle remoteSyncOptions.filerAddress = cmdFilerRemoteSynchronize.Flag.String("filer", "localhost:8888", "filer of the SeaweedFS cluster") remoteSyncOptions.dir = cmdFilerRemoteSynchronize.Flag.String("dir", "", "a mounted directory on filer") + remoteSyncOptions.allMounts = cmdFilerRemoteSynchronize.Flag.Bool("allMounts", false, "write back every mounted directory from this one process, following mounts and unmounts (instead of -dir)") + remoteSyncOptions.mountsCheck = cmdFilerRemoteSynchronize.Flag.Duration("mountsCheckInterval", 5*time.Second, "with -allMounts, how often the mount table is read for new and removed mounts") remoteSyncOptions.storageClass = cmdFilerRemoteSynchronize.Flag.String("storageClass", "", "override amz storage class, empty to delete") remoteSyncOptions.readChunkFromFiler = cmdFilerRemoteSynchronize.Flag.Bool("filerProxy", false, "read file chunks from filer instead of volume servers") remoteSyncOptions.timeAgo = cmdFilerRemoteSynchronize.Flag.Duration("timeAgo", 0, "start time before now, skipping previous metadata changes. \"300ms\", \"1.5h\" or \"2h45m\". Valid time units are \"ns\", \"us\" (or \"µs\"), \"ms\", \"s\", \"m\", \"h\"") @@ -70,6 +79,12 @@ var cmdFilerRemoteSynchronize = &Command{ marker over a version with a legal hold or retention keeps the remote object until a later change to that version releases the lock. + With -allMounts, one process writes back every mounted directory, each + from its own position as a separate -dir process would, and starts or + stops a directory's write-back as it is mounted or unmounted: + + weed filer.remote.sync -allMounts + The metadata sync starting time is determined with the following priority order: 1. specified by timeAgo 2. last sync timestamp for this directory @@ -95,6 +110,15 @@ func runFilerRemoteSynchronize(cmd *Command, args []string) bool { *remoteSyncOptions.readChunkFromFiler, ) + if *remoteSyncOptions.allMounts { + if dir != "" { + glog.Fatalf("-dir and -allMounts are exclusive") + } + fmt.Printf("synchronize every mounted directory to remote storage...\n") + newMountSupervisor(&remoteSyncOptions, filerSource).run(nil) + return true + } + if dir != "" { fmt.Printf("synchronize %s to remote storage...\n", dir) util.RetryUntil("filer.remote.sync "+dir, func() error { diff --git a/weed/command/filer_remote_sync_all.go b/weed/command/filer_remote_sync_all.go new file mode 100644 index 0000000000..05b918f390 --- /dev/null +++ b/weed/command/filer_remote_sync_all.go @@ -0,0 +1,366 @@ +package command + +import ( + "errors" + "sort" + "sync" + "sync/atomic" + "time" + + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/remote_pb" + "github.com/seaweedfs/seaweedfs/weed/remote_storage" + "github.com/seaweedfs/seaweedfs/weed/replication/source" +) + +// The write-backs of -allMounts start this many at a time, this far apart: +// each one replays the filer's log from its own offset, and a process that +// starts a hundred of them at once makes the filer replay a hundred logs +// together. +const ( + allMountsStartBatch = 10 + allMountsStartPause = 500 * time.Millisecond + allMountsRetryMin = time.Second + allMountsRetryMax = 30 * time.Second + // A directory mounted while this process runs, with no offset saved yet, + // waits this long for one: remote.mount saves it right after the + // mapping, and other mount tooling may save it later. + allMountsOffsetWait = 2 * time.Minute + // A directory whose write-back keeps ending at an unmount this many times + // in a row, while the live table still mounts it at the same location, is + // warned about: its replayed log holds an unmount it stops at every time. + allMountsGoneWarnAfter = 5 +) + +// mountSupervisor runs one write-back per mounted directory in this process. +// Each is the -dir write-back: its own subscription, offset, processor and +// retry loop, under its own client id (the filer keeps one subscription per +// client id, so a shared id would make each new follow replace the last). +// +// The supervisor only starts write-backs. A write-back ends itself when the +// mount table it receives, in log order, no longer maps its directory to its +// location, exactly where the -dir process exits: everything before that +// point is written back, nothing after it (remote.unmount's purge) is. The +// supervisor never stops a running write-back because the live table changed, +// which would cut off a write-back still behind in the log. A directory whose +// write-back ended is started again, from its saved offset, while the live +// table still mounts it (with a growing pause if it keeps ending at once). +type mountSupervisor struct { + option *RemoteSyncOptions + filerSource *source.FilerSource + + // readMappings, follow, readOffset, now and warn are the seams the tests + // replace. + readMappings func() (*remote_pb.RemoteStorageMapping, error) + follow func(option *RemoteSyncOptions, dir string, watch *mountWatch) error + readOffset func(dir string) (int64, error) + now func() time.Time + warn func(format string, args ...interface{}) + + mu sync.Mutex + followers map[string]*mountFollower + // notBefore holds a directory whose write-back ended until its next start; + // ends counts its consecutive quick ends, for the pause. + notBefore map[string]time.Time + ends map[string]int + // firstSeen is when a mounted directory without a saved offset was first + // read in the table (allMountsOffsetWait). + firstSeen map[string]time.Time + // gone counts, per directory, the write-backs in a row that ended at an + // unmount while each was started at the same location and stored offset + // (goneStreak). + gone map[string]*goneStreak + wake chan struct{} + // clientIds hands each follow its own non-zero client id. + clientIds atomic.Int32 +} + +// goneStreak is a directory's run of write-backs that ended at an unmount, +// each started at loc from the stored offset; warned is set once the run has +// been reported. +type goneStreak struct { + loc *remote_pb.RemoteStorageLocation + offset int64 + ends int + warned bool +} + +type mountFollower struct { + stop chan struct{} + done chan struct{} +} + +func newMountSupervisor(option *RemoteSyncOptions, filerSource *source.FilerSource) *mountSupervisor { + m := &mountSupervisor{ + option: option, + filerSource: filerSource, + followers: map[string]*mountFollower{}, + notBefore: map[string]time.Time{}, + ends: map[string]int{}, + firstSeen: map[string]time.Time{}, + gone: map[string]*goneStreak{}, + wake: make(chan struct{}, 1), + now: time.Now, + warn: glog.Warningf, + } + m.clientIds.Store(int32(time.Now().UnixNano())) + m.readMappings = func() (*remote_pb.RemoteStorageMapping, error) { return readMountMappings(option) } + m.follow = func(o *RemoteSyncOptions, dir string, watch *mountWatch) error { + return followMountedDir(o, filerSource, dir, watch) + } + m.readOffset = func(dir string) (int64, error) { + return remote_storage.GetSyncOffset(option.grpcDialOption, pb.ServerAddress(*option.filerAddress), dir) + } + return m +} + +// run starts write-backs until quit closes (never, for the command), then +// stops every write-back and waits for them. +func (m *mountSupervisor) run(quit <-chan struct{}) { + interval := 5 * time.Second + if m.option.mountsCheck != nil && *m.option.mountsCheck > 0 { + interval = *m.option.mountsCheck + } + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + if err := m.reconcile(); err != nil { + glog.Errorf("filer.remote.sync -allMounts: read mount table: %v", err) + } + select { + case <-quit: + m.stopAll() + return + case <-ticker.C: + case <-m.wake: + } + } +} + +// reconcile starts a write-back for every mounted directory that has none +// running, is past its pause, and has a saved offset (or waited for one). +func (m *mountSupervisor) reconcile() error { + mappings, err := m.readMappings() + if err != nil { + return err + } + now := m.now() + var starting []string + m.mu.Lock() + for dir := range m.firstSeen { + if mappings == nil || mappings.Mappings[dir] == nil { + delete(m.firstSeen, dir) + } + } + for dir := range m.gone { + if mappings == nil || mappings.Mappings[dir] == nil { + delete(m.gone, dir) + } + } + for dir := range m.notBefore { + if _, running := m.followers[dir]; !running && (mappings == nil || mappings.Mappings[dir] == nil) { + delete(m.notBefore, dir) + delete(m.ends, dir) + } + } + if mappings != nil { + for dir := range mappings.Mappings { + if _, running := m.followers[dir]; running { + continue + } + if t, paused := m.notBefore[dir]; paused && now.Before(t) { + continue + } + starting = append(starting, dir) + } + } + m.mu.Unlock() + + sort.Strings(starting) + ready := starting[:0] + offsets := map[string]int64{} + for _, dir := range starting { + if settled, offset := m.offsetSettled(dir, now); settled { + ready = append(ready, dir) + offsets[dir] = offset + } + } + for i, dir := range ready { + if i > 0 && i%allMountsStartBatch == 0 { + time.Sleep(allMountsStartPause) + } + m.noteStart(dir, mappings.Mappings[dir], offsets[dir]) + m.start(dir) + } + return nil +} + +// noteStart keeps a directory's run of ends at an unmount going while it is +// started again at the same location from the same stored offset (an end +// saves none; a write-back that saved progress, or remote.unmount, moved +// it), and warns once when the run reaches +// allMountsGoneWarnAfter: a write-back that keeps replaying an unmount from +// its saved offset ends there each time and never writes the directory back. +// Nothing changes but the log; the operator decides (setting the directory's +// sync offset to now skips the replay). +func (m *mountSupervisor) noteStart(dir string, loc *remote_pb.RemoteStorageLocation, offset int64) { + m.mu.Lock() + defer m.mu.Unlock() + streak := m.gone[dir] + if streak == nil || !sameMountLocation(streak.loc, loc) || streak.offset != offset { + m.gone[dir] = &goneStreak{loc: loc, offset: offset} + return + } + if streak.ends >= allMountsGoneWarnAfter && !streak.warned { + streak.warned = true + m.warn("filer.remote.sync -allMounts: the write-back of %s has ended %d times in a row at an unmount in its replayed log while the mount table still mounts it at %+v; it is not writing the directory back. Set its sync offset to now to skip the replay", dir, streak.ends, loc) + } +} + +// offsetSettled reports whether a directory may start, and its stored +// offset: it has a saved offset, or has waited allMountsOffsetWait for one. +// An unreadable offset waits too (the follow could not start from it +// anyway). +func (m *mountSupervisor) offsetSettled(dir string, now time.Time) (bool, int64) { + offset, err := m.readOffset(dir) + m.mu.Lock() + defer m.mu.Unlock() + if err == nil && offset > 0 { + delete(m.firstSeen, dir) + return true, offset + } + seen, ok := m.firstSeen[dir] + if !ok { + m.firstSeen[dir] = now + seen = now + } + if err != nil { + glog.Errorf("filer.remote.sync -allMounts: read sync offset of %s: %v", dir, err) + return false, 0 + } + if now.Sub(seen) >= allMountsOffsetWait { + glog.V(0).Infof("filer.remote.sync -allMounts: %s has no saved offset after %v; starting from its creation", dir, allMountsOffsetWait) + delete(m.firstSeen, dir) + return true, 0 + } + return false, 0 +} + +func (m *mountSupervisor) poke() { + select { + case m.wake <- struct{}{}: + default: + } +} + +func (m *mountSupervisor) start(dir string) { + f := &mountFollower{stop: make(chan struct{}), done: make(chan struct{})} + m.mu.Lock() + m.followers[dir] = f + m.mu.Unlock() + + option := *m.option + option.clientId = m.clientIds.Add(1) + if option.clientId == 0 { // the filer does not register client id 0 + option.clientId = m.clientIds.Add(1) + } + option.clientEpoch = 0 + watch := &mountWatch{stop: f.stop} + + go func() { + defer close(f.done) + glog.V(0).Infof("filer.remote.sync -allMounts: synchronize %s to remote storage", dir) + began := m.now() + wait := allMountsRetryMin + for { + attempt := time.Now() + err := m.follow(&option, dir, watch) + if errors.Is(err, errFollowStopped) || isClosed(f.stop) { + m.forget(dir, f) + return + } + if errors.Is(err, errMountGone) { + m.mu.Lock() + if streak := m.gone[dir]; streak != nil { + streak.ends++ + } + m.mu.Unlock() + // ended where the -dir process exits; the next reconcile + // starts it again if the live table still mounts it + m.ended(dir, f, m.now().Sub(began)) + return + } + if err != nil { + glog.Errorf("synchronize %s: %v", dir, err) + } + // no stream is running here: a directory the live table no longer + // mounts is retired, and started again if it is mounted again + if !m.stillMounted(dir) { + m.forget(dir, f) + return + } + if time.Since(attempt) > allMountsRetryMax { + wait = allMountsRetryMin + } + select { + case <-f.stop: + m.forget(dir, f) + return + case <-time.After(wait): + } + if wait *= 2; wait > allMountsRetryMax { + wait = allMountsRetryMax + } + } + }() +} + +// ended records a write-back that ended itself: its directory waits before +// its next start, longer each time it ends again within a minute. +func (m *mountSupervisor) ended(dir string, f *mountFollower, ran time.Duration) { + m.mu.Lock() + if m.followers[dir] == f { + delete(m.followers, dir) + } + if ran > time.Minute { + m.ends[dir] = 0 + } + pause := allMountsRetryMin << m.ends[dir] + if pause > allMountsRetryMax || pause <= 0 { + pause = allMountsRetryMax + } else { + m.ends[dir]++ + } + m.notBefore[dir] = m.now().Add(pause) + m.mu.Unlock() + m.poke() +} + +// stillMounted reports whether the live table maps dir; an unreadable table +// counts as mapped. +func (m *mountSupervisor) stillMounted(dir string) bool { + mappings, err := m.readMappings() + return err != nil || (mappings != nil && mappings.Mappings[dir] != nil) +} + +func (m *mountSupervisor) forget(dir string, f *mountFollower) { + m.mu.Lock() + if m.followers[dir] == f { + delete(m.followers, dir) + } + m.mu.Unlock() +} + +func (m *mountSupervisor) stopAll() { + m.mu.Lock() + var all []*mountFollower + for _, f := range m.followers { + close(f.stop) + all = append(all, f) + } + m.mu.Unlock() + for _, f := range all { + <-f.done + } +} diff --git a/weed/command/filer_remote_sync_all_test.go b/weed/command/filer_remote_sync_all_test.go new file mode 100644 index 0000000000..5f6a771704 --- /dev/null +++ b/weed/command/filer_remote_sync_all_test.go @@ -0,0 +1,629 @@ +package command + +import ( + "errors" + "slices" + "sync" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/remote_pb" +) + +// fakeMounts is a mount table and a record of the follows run against it. +type fakeMounts struct { + mu sync.Mutex + table map[string]*remote_pb.RemoteStorageLocation + active map[string]int // follows running per directory + overlap bool // two follows ever ran at once for one directory, or a client id 0 + started map[string]int + clients map[int32]string + gone map[string]chan struct{} // closing one makes that directory's follow reach an unmount + noOffset map[string]bool // directories without a saved offset + failing map[string]int // follows of these directories return an error this many times +} + +func newFakeMounts() *fakeMounts { + return &fakeMounts{ + table: map[string]*remote_pb.RemoteStorageLocation{}, + active: map[string]int{}, + started: map[string]int{}, + clients: map[int32]string{}, + gone: map[string]chan struct{}{}, + noOffset: map[string]bool{}, + failing: map[string]int{}, + } +} + +func (f *fakeMounts) mount(dir string) { + f.mu.Lock() + defer f.mu.Unlock() + f.table[dir] = &remote_pb.RemoteStorageLocation{Name: "r", Bucket: "zone", Path: dir} +} + +func (f *fakeMounts) unmount(dir string) { + f.mu.Lock() + defer f.mu.Unlock() + delete(f.table, dir) +} + +// reachUnmount makes the running follow of dir receive a table without it. +func (f *fakeMounts) reachUnmount(dir string) { + f.mu.Lock() + defer f.mu.Unlock() + close(f.gone[dir]) +} + +func (f *fakeMounts) supervisor() *mountSupervisor { + interval := 10 * time.Millisecond + m := newMountSupervisor(&RemoteSyncOptions{mountsCheck: &interval}, nil) + m.readMappings = func() (*remote_pb.RemoteStorageMapping, error) { + f.mu.Lock() + defer f.mu.Unlock() + out := &remote_pb.RemoteStorageMapping{Mappings: map[string]*remote_pb.RemoteStorageLocation{}} + for k, v := range f.table { + out.Mappings[k] = v + } + return out, nil + } + m.readOffset = func(dir string) (int64, error) { + f.mu.Lock() + defer f.mu.Unlock() + if f.noOffset[dir] { + return 0, nil + } + return 100, nil + } + m.follow = func(o *RemoteSyncOptions, dir string, watch *mountWatch) error { + f.mu.Lock() + f.active[dir]++ + if f.active[dir] > 1 || o.clientId == 0 { + f.overlap = true + } + f.started[dir]++ + f.clients[o.clientId] = dir + gone := make(chan struct{}) + f.gone[dir] = gone + if f.table[dir] == nil { // DetectMountInfo + f.active[dir]-- + f.mu.Unlock() + return errors.New(dir + " is not mounted") + } + if f.failing[dir] > 0 { + f.failing[dir]-- + f.active[dir]-- + f.mu.Unlock() + return errors.New("stream reset") + } + f.mu.Unlock() + var err error + select { + case <-watch.stop: + err = errFollowStopped + case <-gone: + err = errMountGone + } + f.mu.Lock() + f.active[dir]-- + f.mu.Unlock() + return err + } + return m +} + +func (f *fakeMounts) snapshot() (active map[string]int, started map[string]int, clients int, overlap bool) { + f.mu.Lock() + defer f.mu.Unlock() + active, started = map[string]int{}, map[string]int{} + for k, v := range f.active { + active[k] = v + } + for k, v := range f.started { + started[k] = v + } + return active, started, len(f.clients), f.overlap +} + +func waitFor(t *testing.T, what string, cond func() bool) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for !cond() { + if time.Now().After(deadline) { + t.Fatalf("timed out waiting for %s", what) + } + time.Sleep(5 * time.Millisecond) + } +} + +func runSupervisor(m *mountSupervisor) (stop func()) { + quit := make(chan struct{}) + stopped := make(chan struct{}) + go func() { m.run(quit); close(stopped) }() + return func() { close(quit); <-stopped } +} + +func TestAllMountsFollowsEachMountOnceUnderItsOwnClient(t *testing.T) { + f := newFakeMounts() + for _, d := range []string{"/buckets/a", "/buckets/b", "/buckets/c"} { + f.mount(d) + } + m := f.supervisor() + stop := runSupervisor(m) + + waitFor(t, "three follows", func() bool { + a, _, _, _ := f.snapshot() + return a["/buckets/a"] == 1 && a["/buckets/b"] == 1 && a["/buckets/c"] == 1 + }) + time.Sleep(50 * time.Millisecond) // more reconciles change nothing + _, started, clients, _ := f.snapshot() + for d, n := range started { + if n != 1 { + t.Errorf("%s started %d times, want 1", d, n) + } + } + // the filer keeps one subscription per client id: a shared id would make + // each follow replace the others + if clients != 3 { + t.Errorf("%d distinct client ids for 3 follows", clients) + } + + f.mount("/buckets/d") + waitFor(t, "a new mount followed", func() bool { a, _, _, _ := f.snapshot(); return a["/buckets/d"] == 1 }) + + stop() + active, _, _, overlap := f.snapshot() + for d, n := range active { + if n != 0 { + t.Errorf("%s still has %d follows after the supervisor stopped", d, n) + } + } + if overlap { + t.Error("two follows at once for one directory, or a client id 0") + } +} + +// An unmount in the live table does not stop a write-back still behind in +// the log: the changes made before the unmount would never reach the remote +// (remote.unmount then sets the offset past them). It ends where it reaches +// the unmount, and is not started again while the table does not mount it. +func TestAllMountsWriteBackEndsAtTheUnmountNotBefore(t *testing.T) { + f := newFakeMounts() + f.mount("/buckets/a") + m := f.supervisor() + stop := runSupervisor(m) + defer stop() + waitFor(t, "a followed", func() bool { a, _, _, _ := f.snapshot(); return a["/buckets/a"] == 1 }) + + f.unmount("/buckets/a") + time.Sleep(80 * time.Millisecond) // several reconciles + if a, _, _, _ := f.snapshot(); a["/buckets/a"] != 1 { + t.Fatal("the live unmount stopped a write-back that had not reached it") + } + f.reachUnmount("/buckets/a") + waitFor(t, "a ended", func() bool { a, _, _, _ := f.snapshot(); return a["/buckets/a"] == 0 }) + time.Sleep(80 * time.Millisecond) + if _, s, _, _ := f.snapshot(); s["/buckets/a"] != 1 { + t.Errorf("an unmounted directory was started again (%d starts)", s["/buckets/a"]) + } + // its restart pause goes with the mount, or the maps grow with every + // directory ever mounted + m.mu.Lock() + _, paused := m.notBefore["/buckets/a"] + _, counted := m.ends["/buckets/a"] + m.mu.Unlock() + if paused || counted { + t.Errorf("an unmounted directory keeps its retry state: notBefore=%v ends=%v", paused, counted) + } +} + +// A write-back that ended at an unmount while the directory is mounted again +// (or ends at once, replaying an old table) is started again after a pause, +// never beside the one that ended. +func TestAllMountsRestartsAnEndedWriteBackWhileMounted(t *testing.T) { + f := newFakeMounts() + f.mount("/buckets/a") + m := f.supervisor() + clock := time.Now() + var clockMu sync.Mutex + m.now = func() time.Time { clockMu.Lock(); defer clockMu.Unlock(); return clock } + stop := runSupervisor(m) + defer stop() + waitFor(t, "a followed", func() bool { a, _, _, _ := f.snapshot(); return a["/buckets/a"] == 1 }) + + f.reachUnmount("/buckets/a") + // its pause runs from when the supervisor recorded the end on the frozen clock + waitFor(t, "a ended", func() bool { + m.mu.Lock() + defer m.mu.Unlock() + _, paused := m.notBefore["/buckets/a"] + return paused + }) + time.Sleep(60 * time.Millisecond) + if _, s, _, _ := f.snapshot(); s["/buckets/a"] != 1 { + t.Fatalf("restarted before its pause: %d starts", s["/buckets/a"]) + } + clockMu.Lock() + clock = clock.Add(allMountsRetryMin) + clockMu.Unlock() + waitFor(t, "a restarted", func() bool { a, s, _, _ := f.snapshot(); return a["/buckets/a"] == 1 && s["/buckets/a"] == 2 }) + if _, _, _, overlap := f.snapshot(); overlap { + t.Error("two follows at once for one directory") + } +} + +func TestAllMountsGuards(t *testing.T) { + f := newFakeMounts() + f.mount("/buckets/a") + f.failing["/buckets/a"] = 2 // two stream errors, then it runs + m := f.supervisor() + clock := time.Now() + var clockMu sync.Mutex + m.now = func() time.Time { clockMu.Lock(); defer clockMu.Unlock(); return clock } + stop := runSupervisor(m) + defer stop() + + // a follow that ends on a stream error is started again, never twice at once + waitFor(t, "a running after its errors", func() bool { + a, s, _, _ := f.snapshot() + return a["/buckets/a"] == 1 && s["/buckets/a"] == 3 + }) + + // a new mount without a saved offset waits for one, then starts + f.mu.Lock() + f.noOffset["/buckets/c"] = true + f.mu.Unlock() + f.mount("/buckets/c") + // the wait runs from when a reconcile first saw it on the frozen clock + waitFor(t, "c seen without an offset", func() bool { + m.mu.Lock() + defer m.mu.Unlock() + _, seen := m.firstSeen["/buckets/c"] + return seen + }) + time.Sleep(60 * time.Millisecond) + if _, s, _, _ := f.snapshot(); s["/buckets/c"] != 0 { + t.Fatalf("started before its offset was written") + } + clockMu.Lock() + clock = clock.Add(allMountsOffsetWait) + clockMu.Unlock() + waitFor(t, "c started after the wait", func() bool { a, _, _, _ := f.snapshot(); return a["/buckets/c"] == 1 }) + + if _, _, _, overlap := f.snapshot(); overlap { + t.Error("two follows at once for one directory, or a client id 0") + } +} + +// A follow that fails while no stream runs, for a directory the live table no +// longer mounts, is retired instead of retried forever, and the directory is +// followed again once it is mounted again. +func TestAllMountsRetiresAFailedFollowWhoseMountIsGone(t *testing.T) { + f := newFakeMounts() + f.mount("/buckets/a") + f.failing["/buckets/a"] = 1 + m := f.supervisor() + stop := runSupervisor(m) + defer stop() + waitFor(t, "the first follow failed", func() bool { _, s, _, _ := f.snapshot(); return s["/buckets/a"] == 1 }) + + f.unmount("/buckets/a") + waitFor(t, "the follower retired", func() bool { + m.mu.Lock() + defer m.mu.Unlock() + _, running := m.followers["/buckets/a"] + return !running + }) + _, started, _, _ := f.snapshot() + time.Sleep(80 * time.Millisecond) + if _, s, _, _ := f.snapshot(); s["/buckets/a"] != started["/buckets/a"] { + t.Fatalf("a retired follow was tried again: %d starts, then %d", started["/buckets/a"], s["/buckets/a"]) + } + + f.mount("/buckets/a") + waitFor(t, "a followed again", func() bool { a, _, _, _ := f.snapshot(); return a["/buckets/a"] == 1 }) + if _, _, _, overlap := f.snapshot(); overlap { + t.Error("two follows at once for one directory") + } +} + +func fileEvent(tsNs int64, dir, name string, deleted bool) *filer_pb.SubscribeMetadataResponse { + e := &filer_pb.Entry{Name: name, Attributes: &filer_pb.FuseAttributes{}} + n := &filer_pb.EventNotification{NewEntry: e, NewParentPath: dir} + if deleted { + n = &filer_pb.EventNotification{OldEntry: e} + } + return &filer_pb.SubscribeMetadataResponse{Directory: dir, TsNs: tsNs, EventNotification: n} +} + +// runFence feeds events, in arrival order, through a -allMounts follow's +// stream callback and returns the positions whose jobs ran, and whether the +// write-back ended and where to. +func runFence(events []*filer_pb.SubscribeMetadataResponse) (admitted []int64, ended bool, moved *remote_pb.RemoteStorageLocation) { + loc := &remote_pb.RemoteStorageLocation{Name: "r", Bucket: "zone", Path: "/a"} + var mu sync.Mutex + processor := NewMetadataProcessor(func(resp *filer_pb.SubscribeMetadataResponse) error { + mu.Lock() + defer mu.Unlock() + admitted = append(admitted, resp.TsNs) + return nil + }, 128, 0) + admit := admitUntilMountEnds(processor, "/buckets/a", "/buckets/a", loc, func(to *remote_pb.RemoteStorageLocation) { + ended, moved = true, to + }) + for _, e := range events { + _ = admit(e) + } + processor.Drain() + slices.Sort(admitted) + return admitted, ended, moved +} + +// remote.unmount deletes the mapping, then purges the directory: nothing +// arriving after the table without the directory may be admitted, or the +// purge deletes the remote objects. Not when the directory is mounted again +// at the same place right after (a -allMounts follow keeps receiving events +// until its stream ends, unlike the -dir process, which exits), nor for an +// event stamped earlier that arrives later. +func TestNothingIsAdmittedAfterAnUnmount(t *testing.T) { + loc := &remote_pb.RemoteStorageLocation{Name: "r", Bucket: "zone", Path: "/a"} + mapped := map[string]*remote_pb.RemoteStorageLocation{"/buckets/a": loc} + none := map[string]*remote_pb.RemoteStorageLocation{} + admitted, ended, moved := runFence([]*filer_pb.SubscribeMetadataResponse{ + mountTableEvent(t, 1, none, mapped), + fileEvent(2, "/buckets/a/ns/t/data", "f1.parquet", false), + mountTableEvent(t, 3, mapped, none), // unmounted + fileEvent(2, "/buckets/a/ns/t/data", "late.parquet", true), // stamped before, arrives after + fileEvent(4, "/buckets/a/ns/t/data", "f1.parquet", true), // the purge + mountTableEvent(t, 5, none, mapped), + fileEvent(6, "/buckets/a/ns/t/data", "f2.parquet", false), + }) + if want := []int64{1, 2}; !slices.Equal(admitted, want) { + t.Errorf("jobs run %v, want %v", admitted, want) + } + if !ended || moved != nil { + t.Errorf("ended=%v moved=%v, want an unmount", ended, moved) + } + + // the mount table file removed outright is an unmount too + admitted, ended, moved = runFence([]*filer_pb.SubscribeMetadataResponse{ + mountTableEvent(t, 2, mapped, nil), + fileEvent(3, "/buckets/a/x", "f.parquet", true), + }) + if len(admitted) != 0 || !ended || moved != nil { + t.Errorf("mount table removed: jobs run %v, ended=%v moved=%v", admitted, ended, moved) + } +} + +func TestNothingIsAdmittedAfterAMove(t *testing.T) { + loc := &remote_pb.RemoteStorageLocation{Name: "r", Bucket: "zone", Path: "/a"} + other := &remote_pb.RemoteStorageLocation{Name: "r", Bucket: "zone2", Path: "/a"} + admitted, ended, moved := runFence([]*filer_pb.SubscribeMetadataResponse{ + fileEvent(1, "/buckets/a/ns", "f.parquet", false), + mountTableEvent(t, 2, map[string]*remote_pb.RemoteStorageLocation{"/buckets/a": loc}, map[string]*remote_pb.RemoteStorageLocation{"/buckets/a": other}), + fileEvent(3, "/buckets/a/ns", "g.parquet", false), + }) + if !slices.Equal(admitted, []int64{1}) || !ended || moved.GetBucket() != "zone2" { + t.Errorf("moved: jobs run %v, ended=%v moved=%v", admitted, ended, moved) + } +} + +// A panic in one directory's job is that job's error, not the end of the +// process (and of every other directory's write-back). +func TestAPanickingJobIsAnError(t *testing.T) { + job := recoverJob("/buckets/a", func(*filer_pb.SubscribeMetadataResponse) error { + var m map[string]int + m["x"] = 1 // nil map write + return nil + }) + if err := job(fileEvent(7, "/buckets/a", "f.parquet", false)); err == nil { + t.Fatal("a panicking job returned no error") + } +} + +// A write-back behind in the log must not save its older position over the +// offset remote.unmount (or a mount) wrote: its successor would replay the +// unmount, end there, and never write the directory back. +func TestAFollowNeverMovesAnOffsetSomeoneElseWrote(t *testing.T) { + stored := int64(100) + g := &offsetGuard{dir: "/buckets/a", last: 100, + read: func() (int64, error) { return stored, nil }, + write: func(ts int64) error { stored = ts; return nil }, + } + if err := g.save(150); err != nil || stored != 150 { + t.Fatalf("own offset not saved: stored=%d err=%v", stored, err) + } + stored = 900 // remote.unmount, after its purge + if err := g.save(160); err != nil || stored != 900 || !g.moved { + t.Fatalf("saved over someone else's offset: stored=%d moved=%v err=%v", stored, g.moved, err) + } + if err := g.save(170); err != nil || stored != 900 { + t.Fatalf("saved later over someone else's offset: stored=%d err=%v", stored, err) + } +} + +// A resume position kept for the next attempt in this process is only good +// for the mount and the offset it was earned under: a mount that now maps +// the directory elsewhere drops it (one that cannot be read leaves it to the +// next attempt), and so does an offset someone else moved since (a +// same-location unmount and remount): +// the attempt then starts from the stored offset. Its own exit save, landed +// although the call failed, is not someone else's. +func TestAResumePositionStaysWithItsMount(t *testing.T) { + locA := &remote_pb.RemoteStorageLocation{Name: "r", Bucket: "zone", Path: "/a"} + locB := &remote_pb.RemoteStorageLocation{Name: "r", Bucket: "zone", Path: "/b"} + kept := func() *RemoteSyncOptions { + return &RemoteSyncOptions{resumeTsNs: 50, resumeLoc: locA, resumeStoredNs: 100} + } + + o := kept() + if err := o.resumeFor("/buckets/a", "/buckets/a", locA, nil, 100); err != nil || o.resumeTsNs != 50 { + t.Fatalf("same mount, same offset: resume=%d err=%v", o.resumeTsNs, err) + } + o = kept() + if err := o.resumeFor("/buckets/a", "/buckets/a", locA, nil, 50); err != nil || o.resumeTsNs != 50 { + t.Fatalf("its own exit save read back: resume=%d err=%v", o.resumeTsNs, err) + } + o = kept() + if err := o.resumeFor("/buckets/a", "/buckets/a", locB, nil, 100); err != nil || o.resumeTsNs != 0 { + t.Fatalf("remounted elsewhere: resume=%d err=%v", o.resumeTsNs, err) + } + // an unreadable mount keeps it: with -timeAgo and no checkpoint yet, the + // retry would otherwise start at a later now minus timeAgo, past the + // event the follow failed on + o = kept() + if err := o.resumeFor("/buckets/a", "", nil, errors.New("no table"), 100); err == nil || o.resumeTsNs != 50 { + t.Fatalf("unreadable mount: resume=%d err=%v", o.resumeTsNs, err) + } + if err := o.resumeFor("/buckets/a", "/buckets/a", locB, nil, 100); err != nil || o.resumeTsNs != 0 { + t.Fatalf("remounted elsewhere while the table was unreadable: resume=%d err=%v", o.resumeTsNs, err) + } + // a directory below a mount is judged by its own remote location: the + // same remote mounted one level up maps it to another key + below := &RemoteSyncOptions{resumeTsNs: 50, resumeLoc: dirRemoteLocation("/buckets/a", "/buckets/a/x", locA), resumeStoredNs: 100} + if err := below.resumeFor("/buckets/a/x", "/buckets/a", locA, nil, 100); err != nil || below.resumeTsNs != 50 { + t.Fatalf("below the same mount: resume=%d err=%v", below.resumeTsNs, err) + } + if err := below.resumeFor("/buckets/a/x", "/buckets", locA, nil, 100); err != nil || below.resumeTsNs != 0 { + t.Fatalf("the same remote mounted above the old mount: resume=%d err=%v", below.resumeTsNs, err) + } + o = kept() + if err := o.resumeFor("/buckets/a", "/buckets/a", locA, nil, 900); err != nil || o.resumeTsNs != 0 { + t.Fatalf("offset moved since, same mount: resume=%d err=%v", o.resumeTsNs, err) + } + // with the position dropped, the attempt starts at the stored offset + mounted := func() (*filer_pb.Entry, error) { + return &filer_pb.Entry{Attributes: &filer_pb.FuseAttributes{Crtime: 1}}, nil + } + if start, err := startPosition("/buckets/a", 0, o.resumeTsNs, 900, mounted); err != nil || start.UnixNano() != 900 { + t.Fatalf("start = %v (%v), want the stored offset 900", start.UnixNano(), err) + } + // positive control: a kept position is where the attempt starts + if start, err := startPosition("/buckets/a", 0, 50, 900, mounted); err != nil || start.UnixNano() != 50 { + t.Fatalf("start with a kept position = %v (%v), want 50", start.UnixNano(), err) + } +} + +// A save whose write failed may have landed: read back, that value is the +// guard's own, and it keeps saving. Anything else is someone else's move. +func TestAGuardsOwnWriteThatFailedIsNotAMove(t *testing.T) { + stored := int64(100) + failNext := true + g := &offsetGuard{dir: "/buckets/a", last: 100, + read: func() (int64, error) { return stored, nil }, + write: func(ts int64) error { + stored = ts + if failNext { + failNext = false + return errors.New("deadline exceeded") + } + return nil + }, + } + if err := g.save(150); err == nil { + t.Fatal("the failed write reported no error") + } + if err := g.save(160); err != nil || stored != 160 || g.moved { + t.Fatalf("own landed write taken as a move: stored=%d moved=%v err=%v", stored, g.moved, err) + } + stored = 900 // remote.unmount + if err := g.save(170); err != nil || stored != 900 || !g.moved { + t.Fatalf("someone else's offset: stored=%d moved=%v err=%v", stored, g.moved, err) + } +} + +// A directory whose write-back keeps ending at an unmount while the live +// table still mounts it at the same location, and its stored offset stays +// put, is warned about once per run of ends; nothing else changes (it is +// started again each time). +func TestAllMountsWarnsAboutAWriteBackThatKeepsEndingAtAnUnmount(t *testing.T) { + loc := &remote_pb.RemoteStorageLocation{Name: "r", Bucket: "zone", Path: "/a"} + table := &remote_pb.RemoteStorageMapping{Mappings: map[string]*remote_pb.RemoteStorageLocation{"/buckets/a": loc}} + var mu sync.Mutex + clock := time.Unix(1000, 0) + follows, warns := 0, 0 + interval := time.Hour + m := newMountSupervisor(&RemoteSyncOptions{mountsCheck: &interval}, nil) + m.readMappings = func() (*remote_pb.RemoteStorageMapping, error) { mu.Lock(); defer mu.Unlock(); return table, nil } + offset := int64(100) + m.readOffset = func(string) (int64, error) { mu.Lock(); defer mu.Unlock(); return offset, nil } + m.now = func() time.Time { mu.Lock(); defer mu.Unlock(); return clock } + m.warn = func(string, ...interface{}) { mu.Lock(); warns++; mu.Unlock() } + m.follow = func(*RemoteSyncOptions, string, *mountWatch) error { + mu.Lock() + follows++ + mu.Unlock() + return errMountGone + } + round := func() { + t.Helper() + if err := m.reconcile(); err != nil { + t.Fatal(err) + } + deadline := time.Now().Add(5 * time.Second) + for { + m.mu.Lock() + _, running := m.followers["/buckets/a"] + m.mu.Unlock() + if !running { + break + } + if time.Now().After(deadline) { + t.Fatal("the follow did not end") + } + time.Sleep(time.Millisecond) + } + mu.Lock() + clock = clock.Add(time.Minute) // past any pause + mu.Unlock() + } + for i := 0; i < allMountsGoneWarnAfter; i++ { + round() + } + mu.Lock() + if follows != allMountsGoneWarnAfter || warns != 0 { + t.Fatalf("after %d ends: follows=%d warns=%d", allMountsGoneWarnAfter, follows, warns) + } + mu.Unlock() + round() + round() + mu.Lock() + if warns != 1 { + t.Fatalf("warns = %d after the run reached %d, want exactly 1", warns, allMountsGoneWarnAfter) + } + // mounted elsewhere: a new run, warned about again only when it too + // reaches the count + table = &remote_pb.RemoteStorageMapping{Mappings: map[string]*remote_pb.RemoteStorageLocation{ + "/buckets/a": {Name: "r", Bucket: "zone", Path: "/b"}}} + mu.Unlock() + for i := 0; i < allMountsGoneWarnAfter; i++ { + round() + } + mu.Lock() + if warns != 1 { + t.Fatalf("warns = %d before the new run reached the count", warns) + } + mu.Unlock() + round() + mu.Lock() + if warns != 2 { + t.Fatalf("warns = %d after the new run reached the count, want 2", warns) + } + + // ends whose stored offset moves in between are not one replay: each + // follow saved progress, or a remote.unmount and remount ran between + table = &remote_pb.RemoteStorageMapping{Mappings: map[string]*remote_pb.RemoteStorageLocation{ + "/buckets/a": {Name: "r", Bucket: "zone", Path: "/c"}}} + mu.Unlock() + for i := 0; i < 2*allMountsGoneWarnAfter; i++ { + mu.Lock() + offset += 10 + mu.Unlock() + round() + } + mu.Lock() + defer mu.Unlock() + if warns != 2 { + t.Fatalf("warns = %d after ends from moving offsets, want still 2", warns) + } +} diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index 6e10eb210b..fca603a84f 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/weed/command/filer_remote_sync_dir.go @@ -8,6 +8,8 @@ import ( "io" "math" "os" + "reflect" + "runtime/debug" "strings" "time" @@ -50,20 +52,55 @@ import ( // and must change together: a guard added to one and not the other is how // the zero-byte re-upload on remote.uncache happened. +// mountWatch lets a -allMounts supervisor end a follow: when stop closes, the +// follow ends after its admitted jobs drain. A nil watch is the -dir process, +// which exits (unmounted) or dies (moved) where the mount ends, as it always +// has; under a watch the follow returns errMountGone there instead, and the +// supervisor starts a successor, from the saved offset, while the directory +// is still mounted. +type mountWatch struct { + stop <-chan struct{} +} + func followUpdatesAndUploadToRemote(option *RemoteSyncOptions, filerSource *source.FilerSource, mountedDir string) error { + return followMountedDir(option, filerSource, mountedDir, nil) +} + +// followMountedDir writes one mounted directory back. +func followMountedDir(option *RemoteSyncOptions, filerSource *source.FilerSource, mountedDir string, watch *mountWatch) error { + + // The offset as stored now, read once and before the mount table: it is + // both where this follow starts (without a resume position) and the + // value its offsetGuard may replace. Read after the table, a + // remote.unmount completing in between (its mapping write, its purge, + // then its offset) would start a follow that saw the directory mounted + // past the unmount's table, into its purge. + storedOffsetNs, err := remote_storage.GetSyncOffset(option.grpcDialOption, pb.ServerAddress(*option.filerAddress), mountedDir) + if err != nil { + return fmt.Errorf("read sync offset of %s: %w", mountedDir, err) + } // read filer remote storage mount mappings _, mountKey, remoteStorageMountLocation, remoteStorage, detectErr := filer.DetectMountInfo(option.grpcDialOption, pb.ServerAddress(*option.filerAddress), mountedDir) - if detectErr != nil { - return fmt.Errorf("read mount info: %w", detectErr) + if err := option.resumeFor(mountedDir, mountKey, remoteStorageMountLocation, detectErr, storedOffsetNs); err != nil { + return err + } + dirLoc := dirRemoteLocation(mountKey, mountedDir, remoteStorageMountLocation) + guard := &offsetGuard{dir: mountedDir, last: storedOffsetNs, + read: func() (int64, error) { + return remote_storage.GetSyncOffset(option.grpcDialOption, pb.ServerAddress(*option.filerAddress), mountedDir) + }, + write: func(ts int64) error { + return remote_storage.SetSyncOffset(option.grpcDialOption, pb.ServerAddress(*option.filerAddress), mountedDir, ts) + }, } - eachEntryFunc, err := option.makeEventProcessor(remoteStorage, mountedDir, dirRemoteLocation(mountKey, mountedDir, remoteStorageMountLocation), filerSource) + eachEntryFunc, err := option.makeEventProcessor(remoteStorage, mountedDir, dirLoc, filerSource) if err != nil { return err } - lastOffsetTs, err := collectLastSyncOffset(option, option.grpcDialOption, pb.ServerAddress(*option.filerAddress), mountedDir, *option.timeAgo, option.resumeTsNs) + lastOffsetTs, err := collectLastSyncOffsetFrom(option, mountedDir, *option.timeAgo, option.resumeTsNs, storedOffsetNs) if err != nil { return err } @@ -75,17 +112,34 @@ func followUpdatesAndUploadToRemote(option *RemoteSyncOptions, filerSource *sour } processor := NewMetadataProcessor(eachEntryFunc, 128, lastOffsetTs.UnixNano()) + // Only the stream goroutine reads or writes mountEnded. + var mountEnded bool + mountGone := make(chan struct{}) + onMountEnd := func(*remote_pb.RemoteStorageLocation) { + mountEnded = true + close(mountGone) + } + if watch == nil { + onMountEnd = exitProcessAtMountEnd(processor, mountedDir, remoteStorageMountLocation) + } + followDone := make(chan struct{}) + defer close(followDone) + var lastLogTsNs = time.Now().UnixNano() - processEventFnWithOffset := pb.AddOffsetFunc(admitUntilMountEnds(processor, mountedDir, mountKey, remoteStorageMountLocation), 3*time.Second, func(counter int64, lastTsNs int64) error { + processEventFnWithOffset := pb.AddOffsetFunc(admitUntilMountEnds(processor, mountedDir, mountKey, remoteStorageMountLocation, onMountEnd), 3*time.Second, func(counter int64, lastTsNs int64) error { + if mountEnded { + // the offset remote.unmount sets after its purge must stay + return nil + } offsetTsNs := processor.processedTsWatermark.Load() if offsetTsNs == 0 { return nil } // use processor.processedTsWatermark instead of the lastTsNs from the most recent job now := time.Now().UnixNano() - glog.V(0).Infof("remote sync %s progressed to %v %0.2f/sec", *option.filerAddress, time.Unix(0, offsetTsNs), float64(counter)/(float64(now-lastLogTsNs)/1e9)) + glog.V(0).Infof("remote sync %s progressed to %v %0.2f/sec", mountedDir, time.Unix(0, offsetTsNs), float64(counter)/(float64(now-lastLogTsNs)/1e9)) lastLogTsNs = now - return remote_storage.SetSyncOffset(option.grpcDialOption, pb.ServerAddress(*option.filerAddress), mountedDir, resumeTsNsFor(processor)) + return guard.save(resumeTsNsFor(processor)) }) option.clientEpoch++ @@ -95,6 +149,10 @@ func followUpdatesAndUploadToRemote(option *RemoteSyncOptions, filerSource *sour prefix = prefix + "/" } + var stop <-chan struct{} + if watch != nil { + stop = watch.stop + } metadataFollowOption := &pb.MetadataFollowOption{ ClientName: "filer.remote.sync", ClientId: option.clientId, @@ -109,12 +167,34 @@ func followUpdatesAndUploadToRemote(option *RemoteSyncOptions, filerSource *sour GetResumeTsNs: func() int64 { return resumeTsNsFor(processor) }, - Resubscribe: processor.ResubscribeCh(), + Resubscribe: anyClosed(followDone, processor.ResubscribeCh(), stop, mountGone), } - err = pb.FollowMetadata(pb.ServerAddress(*option.filerAddress), option.grpcDialOption, metadataFollowOption, processEventFnWithOffset) - option.resumeTsNs = saveResumeOffset(option.grpcDialOption, pb.ServerAddress(*option.filerAddress), mountedDir, processor) - return err + followErr := pb.FollowMetadata(pb.ServerAddress(*option.filerAddress), option.grpcDialOption, metadataFollowOption, processEventFnWithOffset) + // however the follow ended, its admitted jobs finish before the caller + // starts another for the directory: a replayed event must not race a + // job still running for the same path + drainLogged(processor, mountedDir) + if mountEnded { + // No offset is saved past the end, and no resume position is kept: + // remote.unmount sets the offset after its purge, and a remount sets + // its own; a position saved here, behind them, would make the next + // write-back replay the unmount and stop there again. + option.clearResume() + return errMountGone + } + resumeTsNs := saveResumeOffsetWith(guard.save, mountedDir, processor) + if guard.moved { + // someone else set the offset (remote.unmount after its purge, a + // mount after its mapping write): the next attempt starts from theirs + option.clearResume() + } else { + option.resumeTsNs, option.resumeLoc, option.resumeStoredNs = resumeTsNs, dirLoc, guard.last + } + if isClosed(stop) { + return errFollowStopped + } + return followErr } // dirRemoteLocation is where dir, at or below the mount at mountKey, lives @@ -154,11 +234,19 @@ func startAfterMountOffset(start time.Time, mountKey, dir string, timeAgo time.D // would otherwise start the retry after the event it failed on. The offset is // returned for the retry in this process, which must not depend on the write. func saveResumeOffset(grpcDialOption grpc.DialOption, filerAddress pb.ServerAddress, dir string, processor *MetadataProcessor) int64 { + return saveResumeOffsetWith(func(ts int64) error { + return remote_storage.SetSyncOffset(grpcDialOption, filerAddress, dir, ts) + }, dir, processor) +} + +// saveResumeOffsetWith is saveResumeOffset through save: a mounted +// directory's write-back saves through its offsetGuard. +func saveResumeOffsetWith(save func(ts int64) error, dir string, processor *MetadataProcessor) int64 { offsetTsNs := resumeTsNsFor(processor) if offsetTsNs <= 0 { return 0 } - if err := remote_storage.SetSyncOffset(grpcDialOption, filerAddress, dir, offsetTsNs); err != nil { + if err := save(offsetTsNs); err != nil { glog.Errorf("save the sync offset of %s: %v", dir, err) } return offsetTsNs @@ -167,25 +255,43 @@ func saveResumeOffset(grpcDialOption grpc.DialOption, filerAddress pb.ServerAddr // admitUntilMountEnds judges each mount-table version before the next event // is admitted. remote.unmount purges the directory right after writing the // table, and a purge delete admitted under a directory no longer mounted -// deletes the remote objects. At the end nothing more is admitted and no -// offset is saved, keeping the one remote.unmount writes after its purge; -// the admitted jobs drain before the exit. -func admitUntilMountEnds(processor *MetadataProcessor, mountedDir, mountKey string, mountLocation *remote_pb.RemoteStorageLocation) pb.ProcessMetadataFunc { +// deletes the remote objects. At the end nothing more is admitted and onEnd +// is called with the new location (nil for an unmount); the caller saves no +// offset after it, keeping the one remote.unmount writes after its purge. +func admitUntilMountEnds(processor *MetadataProcessor, mountedDir, mountKey string, mountLocation *remote_pb.RemoteStorageLocation, onEnd func(moved *remote_pb.RemoteStorageLocation)) pb.ProcessMetadataFunc { + ended := false return func(resp *filer_pb.SubscribeMetadataResponse) error { - ended, moved, err := mountTableEnds(mountedDir, mountKey, mountLocation, resp) + if ended { + return nil + } + end, moved, err := mountTableEnds(mountedDir, mountKey, mountLocation, resp) if err != nil { return err } - if !ended { + if !end { processor.AddSyncJob(resp) return nil } - processor.Drain() - exitAtMountEnd(mountedDir, mountLocation, moved) + ended = true + if moved != nil { + glog.V(0).Infof("%s moved %+v => %+v at %d: its write-back stops", mountedDir, mountLocation, moved, resp.TsNs) + } else { + glog.V(0).Infof("%s unmounted at %d: its write-back stops", mountedDir, resp.TsNs) + } + onEnd(moved) return nil } } +// exitProcessAtMountEnd is the -dir process's end: the admitted jobs drain, +// then it exits (unmounted) or dies (moved). +func exitProcessAtMountEnd(processor *MetadataProcessor, mountedDir string, mountLocation *remote_pb.RemoteStorageLocation) func(*remote_pb.RemoteStorageLocation) { + return func(moved *remote_pb.RemoteStorageLocation) { + drainLogged(processor, mountedDir) + exitAtMountEnd(mountedDir, mountLocation, moved) + } +} + var exitAtMountEnd = func(mountedDir string, from, to *remote_pb.RemoteStorageLocation) { if to != nil { glog.Fatalf("Unexpected mount changes %+v => %+v", from, to) @@ -194,6 +300,125 @@ var exitAtMountEnd = func(mountedDir string, from, to *remote_pb.RemoteStorageLo os.Exit(0) } +// resumeFor checks the resume position a previous attempt in this process +// kept against the mount and the stored offset as this attempt finds them. +// A mount that cannot be read keeps it for the next attempt to check. The +// position is dropped when the mount at mountKey maps the directory to +// another remote location than the one it was earned under (a remount +// elsewhere, or by a mount above or below the old one, never resumes from +// the old mount's position), and when the +// stored offset is no longer the one that attempt left, nor the position +// itself (its own save at exit, which can land although the call failed): +// someone else set it, remote.unmount after its purge or a mount after its +// mapping write, and this attempt starts from theirs, as a follow whose +// guard saw the offset move drops the position. Replaying the old position +// across an unmount and remount would upload pre-unmount content after the +// remount's pull. +func (option *RemoteSyncOptions) resumeFor(dir, mountKey string, mountLoc *remote_pb.RemoteStorageLocation, detectErr error, storedNs int64) error { + if detectErr != nil { + return fmt.Errorf("read mount info: %w", detectErr) + } + if option.resumeTsNs <= 0 { + return nil + } + loc := dirRemoteLocation(mountKey, dir, mountLoc) + switch { + case !sameMountLocation(option.resumeLoc, loc): + glog.V(0).Infof("%s: mounted at %+v now, not %+v where it stopped: not resuming from %d", dir, loc, option.resumeLoc, option.resumeTsNs) + option.clearResume() + case storedNs != option.resumeStoredNs && storedNs != option.resumeTsNs: + glog.V(0).Infof("%s: its sync offset was moved to %d by someone else since it stopped: starting there, not at %d", dir, storedNs, option.resumeTsNs) + option.clearResume() + } + return nil +} + +func (option *RemoteSyncOptions) clearResume() { + option.resumeTsNs, option.resumeLoc, option.resumeStoredNs = 0, nil, 0 +} + +// offsetGuard saves a follow's offset only over the value it last saved (or +// found at its start). Someone else moved it otherwise: remote.unmount sets it +// after its purge, a mount sets it after its own mapping write, and a +// write-back still behind in the log that saved its older position over +// those would make its successor replay the unmount, end there, and never +// write the directory back again. Read then write, not atomic: it narrows the +// window to one round trip. Once moved, it saves nothing more. +// +// pending is a value this guard wrote whose write reported an error: it may +// have landed all the same (a timeout after the filer applied it), so read +// back later it is this guard's own, not someone else's move. +type offsetGuard struct { + dir string + last int64 + pending int64 + moved bool + read func() (int64, error) + write func(ts int64) error +} + +func (g *offsetGuard) save(ts int64) error { + if g.moved { + return nil + } + cur, err := g.read() + if err != nil { + return err + } + if cur != g.last && g.pending != 0 && cur == g.pending { + g.last = g.pending + } + g.pending = 0 + if cur != g.last { + g.moved = true + glog.V(0).Infof("%s: its sync offset was moved to %d by someone else; this write-back no longer saves it", g.dir, cur) + return nil + } + if err := g.write(ts); err != nil { + g.pending = ts + return err + } + g.last = ts + return nil +} + +// errFollowStopped ends a follow its mountWatch stopped; errMountGone one +// whose directory the mount table no longer maps to its location. +var ( + errFollowStopped = errors.New("follow stopped") + errMountGone = errors.New("unmounted or moved") +) + +// anyClosed closes when any of chans does (nil ones never do), and gives up +// when done closes, so its goroutine never outlives the follow. +func anyClosed(done <-chan struct{}, chans ...<-chan struct{}) <-chan struct{} { + merged := make(chan struct{}) + cases := []reflect.SelectCase{{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(done)}} + for _, ch := range chans { + if ch != nil { + cases = append(cases, reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(ch)}) + } + } + go func() { + if i, _, _ := reflect.Select(cases); i > 0 { + close(merged) + } + }() + return merged +} + +func isClosed(ch <-chan struct{}) bool { + if ch == nil { + return false + } + select { + case <-ch: + return true + default: + return false + } +} + // mountTableEnds reports whether a mount-table event ends the write-back of // mountedDir, which mountKey mounts at loc: a version in which another mount // or none covers the directory. moved is the location of the covering mount, @@ -323,9 +548,49 @@ func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.Rem return nil } - return eachEntryFunc, nil + return recoverJob(mountedDir, eachEntryFunc), nil } +// recoverJob turns a panic in one directory's job into that job's error: the +// processor pins the event and resubscribes this directory alone, instead of +// the panic ending the process and, with -allMounts, every directory's +// write-back with it. +func recoverJob(dir string, fn pb.ProcessMetadataFunc) pb.ProcessMetadataFunc { + return func(resp *filer_pb.SubscribeMetadataResponse) (err error) { + defer func() { + if r := recover(); r != nil { + glog.Errorf("%s: job at %d panicked: %v\n%s", dir, resp.GetTsNs(), r, debug.Stack()) + err = fmt.Errorf("%s: job at %d panicked: %v", dir, resp.GetTsNs(), r) + } + }() + return fn(resp) + } +} + +// drainLogged is processor.Drain, logged every drainWarnEvery while it waits: +// a job hung on a remote call keeps its directory from resubscribing or +// ending, and that must not be silent. +func drainLogged(processor *MetadataProcessor, dir string) { + done := make(chan struct{}) + go func() { + ticker := time.NewTicker(drainWarnEvery) + defer ticker.Stop() + began := time.Now() + for { + select { + case <-done: + return + case <-ticker.C: + glog.Warningf("%s: still waiting for its admitted jobs after %v; its write-back is held until they finish", dir, time.Since(began).Round(time.Second)) + } + } + }() + processor.Drain() + close(done) +} + +const drainWarnEvery = 5 * time.Minute + // reshapeCrossMountRename maps an update event onto the side of it the mount // covers. The subscription delivers a rename when either of its paths lies // under the mounted directory, and only a path under it maps to a remote @@ -1242,6 +1507,27 @@ func collectLastSyncOffset(filerClient filer_pb.FilerClient, grpcDialOption grpc func() (int64, error) { return remote_storage.GetSyncOffset(grpcDialOption, filerAddress, mountedDir) }) } +// collectLastSyncOffsetFrom is collectLastSyncOffset with the saved offset +// already read (storedNs): the caller uses the same value for its guard. +func collectLastSyncOffsetFrom(filerClient filer_pb.FilerClient, mountedDir string, timeAgo time.Duration, resumeTsNs, storedNs int64) (time.Time, error) { + return startPosition(mountedDir, timeAgo, resumeTsNs, storedNs, func() (*filer_pb.Entry, error) { + ctx, cancel := lookupContext() + defer cancel() + entry, _, _, err := filer_pb.GetEntry(ctx, filerClient, util.FullPath(mountedDir)) + return entry, err + }) +} + +// startPosition is where a follow subscribes from: the resume position when +// one is kept, otherwise resolveLastSyncOffset over the stored offset. +func startPosition(mountedDir string, timeAgo time.Duration, resumeTsNs, storedNs int64, lookupMount func() (*filer_pb.Entry, error)) (time.Time, error) { + if resumeTsNs > 0 { + glog.V(0).Infof("resuming previous attempt at %v", time.Unix(0, resumeTsNs)) + return time.Unix(0, resumeTsNs), nil + } + return resolveLastSyncOffset(mountedDir, timeAgo, lookupMount, func() (int64, error) { return storedNs, nil }) +} + // resolveLastSyncOffset picks the position to subscribe from: // 1. now minus timeAgo, or the saved offset when it is older // 2. the saved offset for this directory diff --git a/weed/command/filer_remote_sync_dir_test.go b/weed/command/filer_remote_sync_dir_test.go index 9cfcd1c772..4dba231b62 100644 --- a/weed/command/filer_remote_sync_dir_test.go +++ b/weed/command/filer_remote_sync_dir_test.go @@ -3385,7 +3385,7 @@ func TestAdmitUntilMountEndsDrainsBeforeTheExit(t *testing.T) { }(exitAtMountEnd) exitAtMountEnd = func(_ string, _, to *remote_pb.RemoteStorageLocation) { exits <- exitCall{to, len(ran)} } - admit := admitUntilMountEnds(processor, mountedDir, mountedDir, loc) + admit := admitUntilMountEnds(processor, mountedDir, mountedDir, loc, exitProcessAtMountEnd(processor, mountedDir, loc)) if err := admit(&filer_pb.SubscribeMetadataResponse{Directory: mountedDir, TsNs: 1, EventNotification: &filer_pb.EventNotification{ NewParentPath: mountedDir, NewEntry: &filer_pb.Entry{Name: "a.txt"}}}); err != nil { t.Fatal(err) diff --git a/weed/shell/command_remote_mount.go b/weed/shell/command_remote_mount.go index 2dcc661098..04404cfb75 100644 --- a/weed/shell/command_remote_mount.go +++ b/weed/shell/command_remote_mount.go @@ -127,6 +127,51 @@ func (c *commandRemoteMount) Do(args []string, commandEnv *CommandEnv, writer io return fmt.Errorf("save mount mapping: %w", err) } + saveMountSyncOffset(commandEnv, writer, *dir) + + return nil +} + +// saveMountSyncOffset saves the directory's creation time as its sync +// offset, where filer.remote.sync starts a mount without one, so -allMounts +// need not wait for it. An offset already saved, by remote.unmount for a +// remount, is kept. A failure only warns: the mount is saved, and +// filer.remote.sync starts it from the same creation time after its wait. +func saveMountSyncOffset(commandEnv *CommandEnv, writer io.Writer, dir string) { + err := saveMountSyncOffsetWith(func() (int64, error) { + return remote_storage.GetSyncOffset(commandEnv.option.GrpcDialOption, commandEnv.FilerAddress(), dir) + }, func() (int64, error) { + entry, _, _, err := filer_pb.GetEntry(context.Background(), commandEnv, util.FullPath(dir)) + if err != nil { + return 0, err + } + return entry.GetAttributes().GetCrtime(), nil + }, func(ts int64) error { + return remote_storage.SetSyncOffset(commandEnv.option.GrpcDialOption, commandEnv.FilerAddress(), dir, ts) + }, dir) + if err != nil { + fmt.Fprintf(writer, "warning: mounted %s, but %v\n", dir, err) + } +} + +func saveMountSyncOffsetWith(read func() (int64, error), crtime func() (int64, error), write func(ts int64) error, dir string) error { + saved, err := read() + if err != nil { + return fmt.Errorf("read remote.sync offset for %s: %v", dir, err) + } + if saved > 0 { + return nil + } + created, err := crtime() + if err != nil { + return fmt.Errorf("look up %s: %v", dir, err) + } + if created <= 0 { + return nil + } + if err = write(time.Unix(created, 0).UnixNano()); err != nil { + return fmt.Errorf("save remote.sync offset for %s: %v", dir, err) + } return nil } diff --git a/weed/shell/command_remote_mount_buckets.go b/weed/shell/command_remote_mount_buckets.go index c011a15e57..208d042f23 100644 --- a/weed/shell/command_remote_mount_buckets.go +++ b/weed/shell/command_remote_mount_buckets.go @@ -120,6 +120,7 @@ func (c *commandRemoteMountBuckets) Do(args []string, commandEnv *CommandEnv, wr if err = filer.InsertMountMapping(commandEnv, string(dir), remoteStorageLocation); err != nil { return fmt.Errorf("save mount mapping %s to %+v: %v", dir, remoteStorageLocation, err) } + saveMountSyncOffset(commandEnv, writer, string(dir)) } } diff --git a/weed/shell/command_remote_mount_test.go b/weed/shell/command_remote_mount_test.go new file mode 100644 index 0000000000..0c47c4f4e5 --- /dev/null +++ b/weed/shell/command_remote_mount_test.go @@ -0,0 +1,48 @@ +package shell + +import ( + "errors" + "testing" + "time" +) + +func TestSaveMountSyncOffset(t *testing.T) { + const crtime = int64(1700000000) + noOffset := func() (int64, error) { return 0, nil } + created := func() (int64, error) { return crtime, nil } + failed := errors.New("kv down") + + var written []int64 + write := func(ts int64) error { written = append(written, ts); return nil } + + // a new mount starts where a write-back without an offset would: at the + // directory's creation, so local files already there are written back + if err := saveMountSyncOffsetWith(noOffset, created, write, "/buckets/b"); err != nil { + t.Fatal(err) + } + if want := time.Unix(crtime, 0).UnixNano(); len(written) != 1 || written[0] != want { + t.Fatalf("a new mount saved %v, want [%d]", written, want) + } + + written = nil + if err := saveMountSyncOffsetWith(func() (int64, error) { return 42, nil }, created, write, "/buckets/b"); err != nil { + t.Fatal(err) + } + if len(written) != 0 { + t.Fatalf("a remount overwrote the offset remote.unmount saved: %v", written) + } + + if err := saveMountSyncOffsetWith(noOffset, func() (int64, error) { return 0, nil }, write, "/buckets/b"); err != nil || len(written) != 0 { + t.Fatalf("no creation time must save nothing: err=%v written=%v", err, written) + } + + if err := saveMountSyncOffsetWith(func() (int64, error) { return 0, failed }, created, write, "/buckets/b"); err == nil || len(written) != 0 { + t.Fatalf("an unread offset must fail without a write: err=%v written=%v", err, written) + } + if err := saveMountSyncOffsetWith(noOffset, func() (int64, error) { return 0, failed }, write, "/buckets/b"); err == nil || len(written) != 0 { + t.Fatalf("a failed lookup must fail without a write: err=%v written=%v", err, written) + } + if err := saveMountSyncOffsetWith(noOffset, created, func(int64) error { return failed }, "/buckets/b"); err == nil { + t.Fatal("a failed write must be reported") + } +}