diff --git a/weed/command/filer_remote_gateway_buckets.go b/weed/command/filer_remote_gateway_buckets.go index bd3e76859..d70e96904 100644 --- a/weed/command/filer_remote_gateway_buckets.go +++ b/weed/command/filer_remote_gateway_buckets.go @@ -199,8 +199,7 @@ func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *sour 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 } @@ -264,9 +263,7 @@ func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *sour return client.UpdateFileMetadata(oldDest, message.OldEntry, message.NewEntry) } else { newDest := toRemoteStorageLocation(newBucket, util.NewFullPath(message.NewParentPath, message.NewEntry.Name), newRemoteStorageMountLocation) - reader := filer.NewFileReader(filerSource, message.NewEntry) - glog.V(0).Infof("create %s", remote_storage.FormatLocation(newDest)) - remoteEntry, writeErr := client.WriteFile(newDest, message.NewEntry, reader) + remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewEntry, newDest) if writeErr != nil { return writeErr } @@ -303,9 +300,7 @@ func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *sour if message.NewEntry.IsDirectory { return client.WriteDirectory(newDest, message.NewEntry) } - reader := filer.NewFileReader(filerSource, message.NewEntry) - glog.V(0).Infof("create %s", remote_storage.FormatLocation(newDest)) - remoteEntry, writeErr := client.WriteFile(newDest, message.NewEntry, reader) + remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewEntry, newDest) if writeErr != nil { return writeErr } 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