You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

237 lines
9.0 KiB

  1. package command
  2. import (
  3. "context"
  4. "fmt"
  5. "os"
  6. "strings"
  7. "time"
  8. "github.com/chrislusf/seaweedfs/weed/filer"
  9. "github.com/chrislusf/seaweedfs/weed/glog"
  10. "github.com/chrislusf/seaweedfs/weed/pb"
  11. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  12. "github.com/chrislusf/seaweedfs/weed/pb/remote_pb"
  13. "github.com/chrislusf/seaweedfs/weed/remote_storage"
  14. "github.com/chrislusf/seaweedfs/weed/replication/source"
  15. "github.com/chrislusf/seaweedfs/weed/util"
  16. "github.com/golang/protobuf/proto"
  17. "google.golang.org/grpc"
  18. )
  19. func followUpdatesAndUploadToRemote(option *RemoteSyncOptions, filerSource *source.FilerSource, mountedDir string) error {
  20. // read filer remote storage mount mappings
  21. _, _, remoteStorageMountLocation, remoteStorage, detectErr := filer.DetectMountInfo(option.grpcDialOption, pb.ServerAddress(*option.filerAddress), mountedDir)
  22. if detectErr != nil {
  23. return fmt.Errorf("read mount info: %v", detectErr)
  24. }
  25. eachEntryFunc, err := makeEventProcessor(remoteStorage, mountedDir, remoteStorageMountLocation, filerSource)
  26. if err != nil {
  27. return err
  28. }
  29. processEventFnWithOffset := pb.AddOffsetFunc(eachEntryFunc, 3*time.Second, func(counter int64, lastTsNs int64) error {
  30. lastTime := time.Unix(0, lastTsNs)
  31. glog.V(0).Infof("remote sync %s progressed to %v %0.2f/sec", *option.filerAddress, lastTime, float64(counter)/float64(3))
  32. return remote_storage.SetSyncOffset(option.grpcDialOption, pb.ServerAddress(*option.filerAddress), mountedDir, lastTsNs)
  33. })
  34. lastOffsetTs := collectLastSyncOffset(option, option.grpcDialOption, pb.ServerAddress(*option.filerAddress), mountedDir, *option.timeAgo)
  35. return pb.FollowMetadata(pb.ServerAddress(*option.filerAddress), option.grpcDialOption, "filer.remote.sync",
  36. mountedDir, []string{filer.DirectoryEtcRemote}, lastOffsetTs.UnixNano(), 0, processEventFnWithOffset, false)
  37. }
  38. func makeEventProcessor(remoteStorage *remote_pb.RemoteConf, mountedDir string, remoteStorageMountLocation *remote_pb.RemoteStorageLocation, filerSource *source.FilerSource) (pb.ProcessMetadataFunc, error) {
  39. client, err := remote_storage.GetRemoteStorage(remoteStorage)
  40. if err != nil {
  41. return nil, err
  42. }
  43. handleEtcRemoteChanges := func(resp *filer_pb.SubscribeMetadataResponse) error {
  44. message := resp.EventNotification
  45. if message.NewEntry == nil {
  46. return nil
  47. }
  48. if message.NewEntry.Name == filer.REMOTE_STORAGE_MOUNT_FILE {
  49. mappings, readErr := filer.UnmarshalRemoteStorageMappings(message.NewEntry.Content)
  50. if readErr != nil {
  51. return fmt.Errorf("unmarshal mappings: %v", readErr)
  52. }
  53. if remoteLoc, found := mappings.Mappings[mountedDir]; found {
  54. if remoteStorageMountLocation.Bucket != remoteLoc.Bucket || remoteStorageMountLocation.Path != remoteLoc.Path {
  55. glog.Fatalf("Unexpected mount changes %+v => %+v", remoteStorageMountLocation, remoteLoc)
  56. }
  57. } else {
  58. glog.V(0).Infof("unmounted %s exiting ...", mountedDir)
  59. os.Exit(0)
  60. }
  61. }
  62. if message.NewEntry.Name == remoteStorage.Name+filer.REMOTE_STORAGE_CONF_SUFFIX {
  63. conf := &remote_pb.RemoteConf{}
  64. if err := proto.Unmarshal(message.NewEntry.Content, conf); err != nil {
  65. return fmt.Errorf("unmarshal %s/%s: %v", filer.DirectoryEtcRemote, message.NewEntry.Name, err)
  66. }
  67. remoteStorage = conf
  68. if newClient, err := remote_storage.GetRemoteStorage(remoteStorage); err == nil {
  69. client = newClient
  70. } else {
  71. return err
  72. }
  73. }
  74. return nil
  75. }
  76. eachEntryFunc := func(resp *filer_pb.SubscribeMetadataResponse) error {
  77. message := resp.EventNotification
  78. if strings.HasPrefix(resp.Directory, filer.DirectoryEtcRemote) {
  79. return handleEtcRemoteChanges(resp)
  80. }
  81. if message.OldEntry == nil && message.NewEntry == nil {
  82. return nil
  83. }
  84. if message.OldEntry == nil && message.NewEntry != nil {
  85. if !filer.HasData(message.NewEntry) {
  86. return nil
  87. }
  88. glog.V(2).Infof("create: %+v", resp)
  89. if !shouldSendToRemote(message.NewEntry) {
  90. glog.V(2).Infof("skipping creating: %+v", resp)
  91. return nil
  92. }
  93. dest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(message.NewParentPath, message.NewEntry.Name), remoteStorageMountLocation)
  94. if message.NewEntry.IsDirectory {
  95. glog.V(0).Infof("mkdir %s", remote_storage.FormatLocation(dest))
  96. return client.WriteDirectory(dest, message.NewEntry)
  97. }
  98. glog.V(0).Infof("create %s", remote_storage.FormatLocation(dest))
  99. remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewEntry, dest)
  100. if writeErr != nil {
  101. return writeErr
  102. }
  103. return updateLocalEntry(&remoteSyncOptions, message.NewParentPath, message.NewEntry, remoteEntry)
  104. }
  105. if message.OldEntry != nil && message.NewEntry == nil {
  106. glog.V(2).Infof("delete: %+v", resp)
  107. dest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(resp.Directory, message.OldEntry.Name), remoteStorageMountLocation)
  108. if message.OldEntry.IsDirectory {
  109. glog.V(0).Infof("rmdir %s", remote_storage.FormatLocation(dest))
  110. return client.RemoveDirectory(dest)
  111. }
  112. glog.V(0).Infof("delete %s", remote_storage.FormatLocation(dest))
  113. return client.DeleteFile(dest)
  114. }
  115. if message.OldEntry != nil && message.NewEntry != nil {
  116. oldDest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(resp.Directory, message.OldEntry.Name), remoteStorageMountLocation)
  117. dest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(message.NewParentPath, message.NewEntry.Name), remoteStorageMountLocation)
  118. if !shouldSendToRemote(message.NewEntry) {
  119. glog.V(2).Infof("skipping updating: %+v", resp)
  120. return nil
  121. }
  122. if message.NewEntry.IsDirectory {
  123. return client.WriteDirectory(dest, message.NewEntry)
  124. }
  125. if resp.Directory == message.NewParentPath && message.OldEntry.Name == message.NewEntry.Name {
  126. if filer.IsSameData(message.OldEntry, message.NewEntry) {
  127. glog.V(2).Infof("update meta: %+v", resp)
  128. return client.UpdateFileMetadata(dest, message.OldEntry, message.NewEntry)
  129. }
  130. }
  131. glog.V(2).Infof("update: %+v", resp)
  132. glog.V(0).Infof("delete %s", remote_storage.FormatLocation(oldDest))
  133. if err := client.DeleteFile(oldDest); err != nil {
  134. return err
  135. }
  136. remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewEntry, dest)
  137. if writeErr != nil {
  138. return writeErr
  139. }
  140. return updateLocalEntry(&remoteSyncOptions, message.NewParentPath, message.NewEntry, remoteEntry)
  141. }
  142. return nil
  143. }
  144. return eachEntryFunc, nil
  145. }
  146. func retriedWriteFile(client remote_storage.RemoteStorageClient, filerSource *source.FilerSource, newEntry *filer_pb.Entry, dest *remote_pb.RemoteStorageLocation) (remoteEntry *filer_pb.RemoteEntry, err error) {
  147. var writeErr error
  148. err = util.Retry("writeFile", func() error {
  149. reader := filer.NewFileReader(filerSource, newEntry)
  150. glog.V(0).Infof("create %s", remote_storage.FormatLocation(dest))
  151. remoteEntry, writeErr = client.WriteFile(dest, newEntry, reader)
  152. if writeErr != nil {
  153. return writeErr
  154. }
  155. return nil
  156. })
  157. if err != nil {
  158. glog.Errorf("write to %s: %v", dest, err)
  159. }
  160. return
  161. }
  162. func collectLastSyncOffset(filerClient filer_pb.FilerClient, grpcDialOption grpc.DialOption, filerAddress pb.ServerAddress, mountedDir string, timeAgo time.Duration) time.Time {
  163. // 1. specified by timeAgo
  164. // 2. last offset timestamp for this directory
  165. // 3. directory creation time
  166. var lastOffsetTs time.Time
  167. if timeAgo == 0 {
  168. mountedDirEntry, err := filer_pb.GetEntry(filerClient, util.FullPath(mountedDir))
  169. if err != nil {
  170. glog.V(0).Infof("get mounted directory %s: %v", mountedDir, err)
  171. return time.Now()
  172. }
  173. lastOffsetTsNs, err := remote_storage.GetSyncOffset(grpcDialOption, filerAddress, mountedDir)
  174. if mountedDirEntry != nil {
  175. if err == nil && mountedDirEntry.Attributes.Crtime < lastOffsetTsNs/1000000 {
  176. lastOffsetTs = time.Unix(0, lastOffsetTsNs)
  177. glog.V(0).Infof("resume from %v", lastOffsetTs)
  178. } else {
  179. lastOffsetTs = time.Unix(mountedDirEntry.Attributes.Crtime, 0)
  180. }
  181. } else {
  182. lastOffsetTs = time.Now()
  183. }
  184. } else {
  185. lastOffsetTs = time.Now().Add(-timeAgo)
  186. }
  187. return lastOffsetTs
  188. }
  189. func toRemoteStorageLocation(mountDir, sourcePath util.FullPath, remoteMountLocation *remote_pb.RemoteStorageLocation) *remote_pb.RemoteStorageLocation {
  190. source := string(sourcePath[len(mountDir):])
  191. dest := util.FullPath(remoteMountLocation.Path).Child(source)
  192. return &remote_pb.RemoteStorageLocation{
  193. Name: remoteMountLocation.Name,
  194. Bucket: remoteMountLocation.Bucket,
  195. Path: string(dest),
  196. }
  197. }
  198. func shouldSendToRemote(entry *filer_pb.Entry) bool {
  199. if entry.RemoteEntry == nil {
  200. return true
  201. }
  202. if entry.RemoteEntry.RemoteMtime < entry.Attributes.Mtime {
  203. return true
  204. }
  205. return false
  206. }
  207. func updateLocalEntry(filerClient filer_pb.FilerClient, dir string, entry *filer_pb.Entry, remoteEntry *filer_pb.RemoteEntry) error {
  208. remoteEntry.LastLocalSyncTsNs = time.Now().UnixNano()
  209. entry.RemoteEntry = remoteEntry
  210. return filerClient.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  211. _, err := client.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{
  212. Directory: dir,
  213. Entry: entry,
  214. })
  215. return err
  216. })
  217. }