command_remote_mount.go 5.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196
  1. package shell
  2. import (
  3. "context"
  4. "flag"
  5. "fmt"
  6. "github.com/seaweedfs/seaweedfs/weed/filer"
  7. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  8. "github.com/seaweedfs/seaweedfs/weed/pb/remote_pb"
  9. "github.com/seaweedfs/seaweedfs/weed/remote_storage"
  10. "github.com/seaweedfs/seaweedfs/weed/util"
  11. "google.golang.org/protobuf/proto"
  12. "io"
  13. "os"
  14. "strings"
  15. "time"
  16. )
  17. func init() {
  18. Commands = append(Commands, &commandRemoteMount{})
  19. }
  20. type commandRemoteMount struct {
  21. }
  22. func (c *commandRemoteMount) Name() string {
  23. return "remote.mount"
  24. }
  25. func (c *commandRemoteMount) Help() string {
  26. return `mount remote storage and pull its metadata
  27. # assume a remote storage is configured to name "cloud1"
  28. remote.configure -name=cloud1 -type=s3 -s3.access_key=xxx -s3.secret_key=yyy
  29. # mount and pull one bucket
  30. remote.mount -dir=/xxx -remote=cloud1/bucket
  31. # mount and pull one directory in the bucket
  32. remote.mount -dir=/xxx -remote=cloud1/bucket/dir1
  33. # after mount, start a separate process to write updates to remote storage
  34. weed filer.remote.sync -filer=<filerHost>:<filerPort> -dir=/xxx
  35. `
  36. }
  37. func (c *commandRemoteMount) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
  38. remoteMountCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
  39. dir := remoteMountCommand.String("dir", "", "a directory in filer")
  40. nonEmpty := remoteMountCommand.Bool("nonempty", false, "allows the mounting over a non-empty directory")
  41. remote := remoteMountCommand.String("remote", "", "a directory in remote storage, ex. <storageName>/<bucket>/path/to/dir")
  42. if err = remoteMountCommand.Parse(args); err != nil {
  43. return nil
  44. }
  45. if *dir == "" {
  46. _, err = listExistingRemoteStorageMounts(commandEnv, writer)
  47. return err
  48. }
  49. // find configuration for remote storage
  50. remoteConf, err := filer.ReadRemoteStorageConf(commandEnv.option.GrpcDialOption, commandEnv.option.FilerAddress, remote_storage.ParseLocationName(*remote))
  51. if err != nil {
  52. return fmt.Errorf("find configuration for %s: %v", *remote, err)
  53. }
  54. remoteStorageLocation, err := remote_storage.ParseRemoteLocation(remoteConf.Type, *remote)
  55. if err != nil {
  56. return err
  57. }
  58. // sync metadata from remote
  59. if err = syncMetadata(commandEnv, writer, *dir, *nonEmpty, remoteConf, remoteStorageLocation); err != nil {
  60. return fmt.Errorf("pull metadata: %v", err)
  61. }
  62. // store a mount configuration in filer
  63. if err = filer.InsertMountMapping(commandEnv, *dir, remoteStorageLocation); err != nil {
  64. return fmt.Errorf("save mount mapping: %v", err)
  65. }
  66. return nil
  67. }
  68. func listExistingRemoteStorageMounts(commandEnv *CommandEnv, writer io.Writer) (mappings *remote_pb.RemoteStorageMapping, err error) {
  69. // read current mapping
  70. mappings, err = filer.ReadMountMappings(commandEnv.option.GrpcDialOption, commandEnv.option.FilerAddress)
  71. if err != nil {
  72. return mappings, err
  73. }
  74. jsonPrintln(writer, mappings)
  75. return
  76. }
  77. func jsonPrintln(writer io.Writer, message proto.Message) error {
  78. return filer.ProtoToText(writer, message)
  79. }
  80. func syncMetadata(commandEnv *CommandEnv, writer io.Writer, dir string, nonEmpty bool, remoteConf *remote_pb.RemoteConf, remote *remote_pb.RemoteStorageLocation) error {
  81. // find existing directory, and ensure the directory is empty
  82. err := commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
  83. parent, name := util.FullPath(dir).DirAndName()
  84. _, lookupErr := client.LookupDirectoryEntry(context.Background(), &filer_pb.LookupDirectoryEntryRequest{
  85. Directory: parent,
  86. Name: name,
  87. })
  88. if lookupErr != nil {
  89. if strings.Contains(lookupErr.Error(), filer_pb.ErrNotFound.Error()) {
  90. _, createErr := client.CreateEntry(context.Background(), &filer_pb.CreateEntryRequest{
  91. Directory: parent,
  92. Entry: &filer_pb.Entry{
  93. Name: name,
  94. IsDirectory: true,
  95. Attributes: &filer_pb.FuseAttributes{
  96. Mtime: time.Now().Unix(),
  97. Crtime: time.Now().Unix(),
  98. FileMode: uint32(0644 | os.ModeDir),
  99. },
  100. RemoteEntry: &filer_pb.RemoteEntry{
  101. StorageName: remoteConf.Name,
  102. },
  103. },
  104. })
  105. return createErr
  106. }
  107. }
  108. mountToDirIsEmpty := true
  109. listErr := filer_pb.SeaweedList(client, dir, "", func(entry *filer_pb.Entry, isLast bool) error {
  110. mountToDirIsEmpty = false
  111. return nil
  112. }, "", false, 1)
  113. if listErr != nil {
  114. return fmt.Errorf("list %s: %v", dir, listErr)
  115. }
  116. if !mountToDirIsEmpty {
  117. if !nonEmpty {
  118. return fmt.Errorf("dir %s is not empty", dir)
  119. }
  120. }
  121. return nil
  122. })
  123. if err != nil {
  124. return err
  125. }
  126. // pull metadata from remote
  127. if err = pullMetadata(commandEnv, writer, util.FullPath(dir), remote, util.FullPath(dir), remoteConf); err != nil {
  128. return fmt.Errorf("cache metadata: %v", err)
  129. }
  130. return nil
  131. }
  132. // if an entry has synchronized metadata but has not synchronized content
  133. //
  134. // entry.Attributes.FileSize == entry.RemoteEntry.RemoteSize
  135. // entry.Attributes.Mtime == entry.RemoteEntry.RemoteMtime
  136. // entry.RemoteEntry.LastLocalSyncTsNs == 0
  137. //
  138. // if an entry has synchronized metadata but has synchronized content before
  139. //
  140. // entry.Attributes.FileSize == entry.RemoteEntry.RemoteSize
  141. // entry.Attributes.Mtime == entry.RemoteEntry.RemoteMtime
  142. // entry.RemoteEntry.LastLocalSyncTsNs > 0
  143. //
  144. // if an entry has synchronized metadata but has new updates
  145. //
  146. // entry.Attributes.Mtime * 1,000,000,000 > entry.RemoteEntry.LastLocalSyncTsNs
  147. func doSaveRemoteEntry(client filer_pb.SeaweedFilerClient, localDir string, existingEntry *filer_pb.Entry, remoteEntry *filer_pb.RemoteEntry) error {
  148. existingEntry.RemoteEntry = remoteEntry
  149. existingEntry.Attributes.FileSize = uint64(remoteEntry.RemoteSize)
  150. existingEntry.Attributes.Mtime = remoteEntry.RemoteMtime
  151. existingEntry.Attributes.Md5 = nil
  152. existingEntry.Chunks = nil
  153. existingEntry.Content = nil
  154. _, updateErr := client.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{
  155. Directory: localDir,
  156. Entry: existingEntry,
  157. })
  158. if updateErr != nil {
  159. return updateErr
  160. }
  161. return nil
  162. }