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.

257 lines
9.5 KiB

3 years ago
3 years ago
  1. package command
  2. import (
  3. "context"
  4. "fmt"
  5. "github.com/chrislusf/seaweedfs/weed/filer"
  6. "github.com/chrislusf/seaweedfs/weed/glog"
  7. "github.com/chrislusf/seaweedfs/weed/pb"
  8. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  9. "github.com/chrislusf/seaweedfs/weed/remote_storage"
  10. "github.com/chrislusf/seaweedfs/weed/replication/source"
  11. "github.com/chrislusf/seaweedfs/weed/security"
  12. "github.com/chrislusf/seaweedfs/weed/util"
  13. "github.com/golang/protobuf/proto"
  14. "google.golang.org/grpc"
  15. "time"
  16. )
  17. type RemoteSyncOptions struct {
  18. filerAddress *string
  19. grpcDialOption grpc.DialOption
  20. readChunkFromFiler *bool
  21. debug *bool
  22. timeAgo *time.Duration
  23. dir *string
  24. }
  25. const (
  26. RemoteSyncKeyPrefix = "remote.sync."
  27. )
  28. var _ = filer_pb.FilerClient(&RemoteSyncOptions{})
  29. func (option *RemoteSyncOptions) WithFilerClient(fn func(filer_pb.SeaweedFilerClient) error) error {
  30. return pb.WithFilerClient(*option.filerAddress, option.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  31. return fn(client)
  32. })
  33. }
  34. func (option *RemoteSyncOptions) AdjustedUrl(location *filer_pb.Location) string {
  35. return location.Url
  36. }
  37. var (
  38. remoteSyncOptions RemoteSyncOptions
  39. )
  40. func init() {
  41. cmdFilerRemoteSynchronize.Run = runFilerRemoteSynchronize // break init cycle
  42. remoteSyncOptions.filerAddress = cmdFilerRemoteSynchronize.Flag.String("filer", "localhost:8888", "filer of the SeaweedFS cluster")
  43. remoteSyncOptions.dir = cmdFilerRemoteSynchronize.Flag.String("dir", "/", "a mounted directory on filer")
  44. remoteSyncOptions.readChunkFromFiler = cmdFilerRemoteSynchronize.Flag.Bool("filerProxy", false, "read file chunks from filer instead of volume servers")
  45. remoteSyncOptions.debug = cmdFilerRemoteSynchronize.Flag.Bool("debug", false, "debug mode to print out filer updated remote files")
  46. remoteSyncOptions.timeAgo = cmdFilerRemoteSynchronize.Flag.Duration("timeAgo", 0, "start time before now. \"300ms\", \"1.5h\" or \"2h45m\". Valid time units are \"ns\", \"us\" (or \"µs\"), \"ms\", \"s\", \"m\", \"h\"")
  47. }
  48. var cmdFilerRemoteSynchronize = &Command{
  49. UsageLine: "filer.remote.sync -filer=<filerHost>:<filerPort> -dir=/mount/s3_on_cloud",
  50. Short: "resumable continuously write back updates to remote storage if the directory is mounted to the remote storage",
  51. Long: `resumable continuously write back updates to remote storage if the directory is mounted to the remote storage
  52. filer.remote.sync listens on filer update events.
  53. If any mounted remote file is updated, it will fetch the updated content,
  54. and write to the remote storage.
  55. `,
  56. }
  57. func runFilerRemoteSynchronize(cmd *Command, args []string) bool {
  58. util.LoadConfiguration("security", false)
  59. grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.client")
  60. remoteSyncOptions.grpcDialOption = grpcDialOption
  61. // read filer remote storage mount mappings
  62. mappings, readErr := filer.ReadMountMappings(grpcDialOption, *remoteSyncOptions.filerAddress)
  63. if readErr != nil {
  64. fmt.Printf("read mount mapping: %v", readErr)
  65. return false
  66. }
  67. filerSource := &source.FilerSource{}
  68. filerSource.DoInitialize(
  69. *remoteSyncOptions.filerAddress,
  70. pb.ServerToGrpcAddress(*remoteSyncOptions.filerAddress),
  71. "/", // does not matter
  72. *remoteSyncOptions.readChunkFromFiler,
  73. )
  74. var found bool
  75. for dir, remoteStorageMountLocation := range mappings.Mappings {
  76. if *remoteSyncOptions.dir == dir {
  77. found = true
  78. storageConf, readErr := filer.ReadRemoteStorageConf(grpcDialOption, *remoteSyncOptions.filerAddress, remoteStorageMountLocation.Name)
  79. if readErr != nil {
  80. fmt.Printf("read remote storage configuration for %s: %v", dir, readErr)
  81. continue
  82. }
  83. fmt.Printf("synchronize %s to remote storage...\n", *remoteSyncOptions.dir)
  84. if err := util.Retry("filer.remote.sync "+dir, func() error {
  85. return followUpdatesAndUploadToRemote(&remoteSyncOptions, filerSource, dir, storageConf, remoteStorageMountLocation)
  86. }); err != nil {
  87. fmt.Printf("synchronize %s: %v\n", *remoteSyncOptions.dir, err)
  88. }
  89. break
  90. }
  91. }
  92. if !found {
  93. fmt.Printf("directory %s is not mounted to any remote storage\n", *remoteSyncOptions.dir)
  94. return false
  95. }
  96. return true
  97. }
  98. func followUpdatesAndUploadToRemote(option *RemoteSyncOptions, filerSource *source.FilerSource, mountedDir string, remoteStorage *filer_pb.RemoteConf, remoteStorageMountLocation *filer_pb.RemoteStorageLocation) error {
  99. dirHash := util.HashStringToLong(mountedDir)
  100. // 1. specified by timeAgo
  101. // 2. last offset timestamp for this directory
  102. // 3. directory creation time
  103. var lastOffsetTs time.Time
  104. if *option.timeAgo == 0 {
  105. mountedDirEntry, err := filer_pb.GetEntry(option, util.FullPath(mountedDir))
  106. if err != nil {
  107. return fmt.Errorf("lookup %s: %v", mountedDir, err)
  108. }
  109. lastOffsetTsNs, err := getOffset(option.grpcDialOption, *option.filerAddress, RemoteSyncKeyPrefix, int32(dirHash))
  110. if err == nil && mountedDirEntry.Attributes.Crtime < lastOffsetTsNs/1000000 {
  111. lastOffsetTs = time.Unix(0, lastOffsetTsNs)
  112. glog.V(0).Infof("resume from %v", lastOffsetTs)
  113. } else {
  114. lastOffsetTs = time.Unix(mountedDirEntry.Attributes.Crtime, 0)
  115. }
  116. } else {
  117. lastOffsetTs = time.Now().Add(-*option.timeAgo)
  118. }
  119. client, err := remote_storage.GetRemoteStorage(remoteStorage)
  120. if err != nil {
  121. return err
  122. }
  123. eachEntryFunc := func(resp *filer_pb.SubscribeMetadataResponse) error {
  124. message := resp.EventNotification
  125. if message.OldEntry == nil && message.NewEntry == nil {
  126. return nil
  127. }
  128. if message.OldEntry == nil && message.NewEntry != nil {
  129. if len(message.NewEntry.Chunks) == 0 {
  130. return nil
  131. }
  132. fmt.Printf("create: %+v\n", resp)
  133. if !shouldSendToRemote(message.NewEntry) {
  134. fmt.Printf("skipping creating: %+v\n", resp)
  135. return nil
  136. }
  137. dest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(message.NewParentPath, message.NewEntry.Name), remoteStorageMountLocation)
  138. if message.NewEntry.IsDirectory {
  139. return client.WriteDirectory(dest, message.NewEntry)
  140. }
  141. reader := filer.NewChunkStreamReader(filerSource, message.NewEntry.Chunks)
  142. remoteEntry, writeErr := client.WriteFile(dest, message.NewEntry, reader)
  143. if writeErr != nil {
  144. return writeErr
  145. }
  146. return updateLocalEntry(&remoteSyncOptions, message.NewParentPath, message.NewEntry, remoteEntry)
  147. }
  148. if message.OldEntry != nil && message.NewEntry == nil {
  149. fmt.Printf("delete: %+v\n", resp)
  150. dest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(resp.Directory, message.OldEntry.Name), remoteStorageMountLocation)
  151. return client.DeleteFile(dest)
  152. }
  153. if message.OldEntry != nil && message.NewEntry != nil {
  154. oldDest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(resp.Directory, message.OldEntry.Name), remoteStorageMountLocation)
  155. dest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(message.NewParentPath, message.NewEntry.Name), remoteStorageMountLocation)
  156. if !shouldSendToRemote(message.NewEntry) {
  157. fmt.Printf("skipping updating: %+v\n", resp)
  158. return nil
  159. }
  160. if message.NewEntry.IsDirectory {
  161. return client.WriteDirectory(dest, message.NewEntry)
  162. }
  163. if resp.Directory == message.NewParentPath && message.OldEntry.Name == message.NewEntry.Name {
  164. if isSameChunks(message.OldEntry.Chunks, message.NewEntry.Chunks) {
  165. fmt.Printf("update meta: %+v\n", resp)
  166. return client.UpdateFileMetadata(dest, message.NewEntry)
  167. }
  168. }
  169. fmt.Printf("update: %+v\n", resp)
  170. if err := client.DeleteFile(oldDest); err != nil {
  171. return err
  172. }
  173. reader := filer.NewChunkStreamReader(filerSource, message.NewEntry.Chunks)
  174. remoteEntry, writeErr := client.WriteFile(dest, message.NewEntry, reader)
  175. if writeErr != nil {
  176. return writeErr
  177. }
  178. return updateLocalEntry(&remoteSyncOptions, message.NewParentPath, message.NewEntry, remoteEntry)
  179. }
  180. return nil
  181. }
  182. processEventFnWithOffset := pb.AddOffsetFunc(eachEntryFunc, 3*time.Second, func(counter int64, lastTsNs int64) error {
  183. lastTime := time.Unix(0, lastTsNs)
  184. glog.V(0).Infof("remote sync %s progressed to %v %0.2f/sec", *option.filerAddress, lastTime, float64(counter)/float64(3))
  185. return setOffset(option.grpcDialOption, *option.filerAddress, RemoteSyncKeyPrefix, int32(dirHash), lastTsNs)
  186. })
  187. return pb.FollowMetadata(*option.filerAddress, option.grpcDialOption,
  188. "filer.remote.sync", mountedDir, lastOffsetTs.UnixNano(), 0, processEventFnWithOffset, false)
  189. }
  190. func toRemoteStorageLocation(mountDir, sourcePath util.FullPath, remoteMountLocation *filer_pb.RemoteStorageLocation) *filer_pb.RemoteStorageLocation {
  191. source := string(sourcePath[len(mountDir):])
  192. dest := util.FullPath(remoteMountLocation.Path).Child(source)
  193. return &filer_pb.RemoteStorageLocation{
  194. Name: remoteMountLocation.Name,
  195. Bucket: remoteMountLocation.Bucket,
  196. Path: string(dest),
  197. }
  198. }
  199. func isSameChunks(a, b []*filer_pb.FileChunk) bool {
  200. if len(a) != len(b) {
  201. return false
  202. }
  203. for i := 0; i < len(a); i++ {
  204. x, y := a[i], b[i]
  205. if !proto.Equal(x, y) {
  206. return false
  207. }
  208. }
  209. return true
  210. }
  211. func shouldSendToRemote(entry *filer_pb.Entry) bool {
  212. if entry.RemoteEntry == nil {
  213. return true
  214. }
  215. if entry.RemoteEntry.LocalMtime < entry.Attributes.Mtime {
  216. return true
  217. }
  218. return false
  219. }
  220. func updateLocalEntry(filerClient filer_pb.FilerClient, dir string, entry *filer_pb.Entry, remoteEntry *filer_pb.RemoteEntry) error {
  221. entry.RemoteEntry = remoteEntry
  222. return filerClient.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  223. _, err := client.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{
  224. Directory: dir,
  225. Entry: entry,
  226. })
  227. return err
  228. })
  229. }