diff options
Diffstat (limited to 'weed/command/filer_remote_sync_dir.go')
| -rw-r--r-- | weed/command/filer_remote_sync_dir.go | 21 |
1 files changed, 16 insertions, 5 deletions
diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index 8ff933833..e24c8f9e6 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/weed/command/filer_remote_sync_dir.go @@ -108,8 +108,7 @@ func makeEventProcessor(remoteStorage *remote_pb.RemoteConf, mountedDir string, return client.WriteDirectory(dest, message.NewEntry) } glog.V(0).Infof("create %s", remote_storage.FormatLocation(dest)) - reader := filer.NewFileReader(filerSource, message.NewEntry) - remoteEntry, writeErr := client.WriteFile(dest, message.NewEntry, reader) + remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewEntry, dest) if writeErr != nil { return writeErr } @@ -146,9 +145,7 @@ func makeEventProcessor(remoteStorage *remote_pb.RemoteConf, mountedDir string, if err := client.DeleteFile(oldDest); err != nil { return err } - reader := filer.NewFileReader(filerSource, message.NewEntry) - glog.V(0).Infof("create %s", remote_storage.FormatLocation(dest)) - remoteEntry, writeErr := client.WriteFile(dest, message.NewEntry, reader) + remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewEntry, dest) if writeErr != nil { return writeErr } @@ -160,6 +157,20 @@ func makeEventProcessor(remoteStorage *remote_pb.RemoteConf, mountedDir string, return eachEntryFunc, nil } +func retriedWriteFile(client remote_storage.RemoteStorageClient, filerSource *source.FilerSource, newEntry *filer_pb.Entry, dest *remote_pb.RemoteStorageLocation) (remoteEntry *filer_pb.RemoteEntry, err error) { + var writeErr error + err = util.Retry("writeFile", func() error { + reader := filer.NewFileReader(filerSource, newEntry) + glog.V(0).Infof("create %s", remote_storage.FormatLocation(dest)) + remoteEntry, writeErr = client.WriteFile(dest, newEntry, reader) + if writeErr != nil { + return writeErr + } + return nil + }) + return +} + func collectLastSyncOffset(filerClient filer_pb.FilerClient, grpcDialOption grpc.DialOption, filerAddress pb.ServerAddress, mountedDir string, timeAgo time.Duration) time.Time { // 1. specified by timeAgo // 2. last offset timestamp for this directory |