command_remote_cache.go 5.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187
  1. package shell
  2. import (
  3. "flag"
  4. "fmt"
  5. "github.com/chrislusf/seaweedfs/weed/filer"
  6. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  7. "github.com/chrislusf/seaweedfs/weed/pb/remote_pb"
  8. "github.com/chrislusf/seaweedfs/weed/util"
  9. "io"
  10. "sync"
  11. )
  12. func init() {
  13. Commands = append(Commands, &commandRemoteCache{})
  14. }
  15. type commandRemoteCache struct {
  16. }
  17. func (c *commandRemoteCache) Name() string {
  18. return "remote.cache"
  19. }
  20. func (c *commandRemoteCache) Help() string {
  21. return `cache the file content for mounted directories or files
  22. # assume a remote storage is configured to name "cloud1"
  23. remote.configure -name=cloud1 -type=s3 -s3.access_key=xxx -s3.secret_key=yyy
  24. # mount and pull one bucket
  25. remote.mount -dir=/xxx -remote=cloud1/bucket
  26. # after mount, run one of these command to cache the content of the files
  27. remote.cache -dir=/xxx
  28. remote.cache -dir=/xxx/some/sub/dir
  29. remote.cache -dir=/xxx/some/sub/dir -include=*.pdf
  30. remote.cache -dir=/xxx/some/sub/dir -exclude=*.txt
  31. remote.cache -maxSize=1024000 # cache files smaller than 100K
  32. remote.cache -maxAge=3600 # cache files less than 1 hour old
  33. This is designed to run regularly. So you can add it to some cronjob.
  34. If a file is already synchronized with the remote copy, the file will be skipped to avoid unnecessary copy.
  35. The actual data copying goes through volume severs in parallel.
  36. `
  37. }
  38. func (c *commandRemoteCache) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
  39. remoteMountCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
  40. dir := remoteMountCommand.String("dir", "", "a mounted directory or one of its sub folders in filer")
  41. concurrency := remoteMountCommand.Int("concurrent", 32, "concurrent file downloading")
  42. fileFiler := newFileFilter(remoteMountCommand)
  43. if err = remoteMountCommand.Parse(args); err != nil {
  44. return nil
  45. }
  46. if *dir != "" {
  47. if err := c.doCacheOneDirectory(commandEnv, writer, *dir, fileFiler, *concurrency); err != nil {
  48. return err
  49. }
  50. return nil
  51. }
  52. mappings, err := filer.ReadMountMappings(commandEnv.option.GrpcDialOption, commandEnv.option.FilerAddress)
  53. if err != nil {
  54. return err
  55. }
  56. for key, _ := range mappings.Mappings {
  57. if err := c.doCacheOneDirectory(commandEnv, writer, key, fileFiler, *concurrency); err != nil {
  58. return err
  59. }
  60. }
  61. return nil
  62. }
  63. func (c *commandRemoteCache) doCacheOneDirectory(commandEnv *CommandEnv, writer io.Writer, dir string, fileFiler *FileFilter, concurrency int) error {
  64. mappings, localMountedDir, remoteStorageMountedLocation, remoteStorageConf, detectErr := detectMountInfo(commandEnv, writer, dir)
  65. if detectErr != nil {
  66. jsonPrintln(writer, mappings)
  67. return detectErr
  68. }
  69. // pull content from remote
  70. if err := c.cacheContentData(commandEnv, writer, util.FullPath(localMountedDir), remoteStorageMountedLocation, util.FullPath(dir), fileFiler, remoteStorageConf, concurrency); err != nil {
  71. return fmt.Errorf("cache content data on %s: %v", localMountedDir, err)
  72. }
  73. return nil
  74. }
  75. func recursivelyTraverseDirectory(filerClient filer_pb.FilerClient, dirPath util.FullPath, visitEntry func(dir util.FullPath, entry *filer_pb.Entry) bool) (err error) {
  76. err = filer_pb.ReadDirAllEntries(filerClient, dirPath, "", func(entry *filer_pb.Entry, isLast bool) error {
  77. if entry.IsDirectory {
  78. if !visitEntry(dirPath, entry) {
  79. return nil
  80. }
  81. subDir := dirPath.Child(entry.Name)
  82. if err := recursivelyTraverseDirectory(filerClient, subDir, visitEntry); err != nil {
  83. return err
  84. }
  85. } else {
  86. if !visitEntry(dirPath, entry) {
  87. return nil
  88. }
  89. }
  90. return nil
  91. })
  92. return
  93. }
  94. func shouldCacheToLocal(entry *filer_pb.Entry) bool {
  95. if entry.IsDirectory {
  96. return false
  97. }
  98. if entry.RemoteEntry == nil {
  99. return false
  100. }
  101. if entry.RemoteEntry.LastLocalSyncTsNs == 0 && entry.RemoteEntry.RemoteSize > 0 {
  102. return true
  103. }
  104. return false
  105. }
  106. func mayHaveCachedToLocal(entry *filer_pb.Entry) bool {
  107. if entry.IsDirectory {
  108. return false
  109. }
  110. if entry.RemoteEntry == nil {
  111. return false // should not uncache an entry that is not in remote
  112. }
  113. if entry.RemoteEntry.LastLocalSyncTsNs > 0 {
  114. return true
  115. }
  116. return false
  117. }
  118. func (c *commandRemoteCache) cacheContentData(commandEnv *CommandEnv, writer io.Writer, localMountedDir util.FullPath, remoteMountedLocation *remote_pb.RemoteStorageLocation, dirToCache util.FullPath, fileFilter *FileFilter, remoteConf *remote_pb.RemoteConf, concurrency int) error {
  119. var wg sync.WaitGroup
  120. limitedConcurrentExecutor := util.NewLimitedConcurrentExecutor(concurrency)
  121. var executionErr error
  122. traverseErr := recursivelyTraverseDirectory(commandEnv, dirToCache, func(dir util.FullPath, entry *filer_pb.Entry) bool {
  123. if !shouldCacheToLocal(entry) {
  124. return true // true means recursive traversal should continue
  125. }
  126. if !fileFilter.matches(entry) {
  127. return true
  128. }
  129. wg.Add(1)
  130. limitedConcurrentExecutor.Execute(func() {
  131. defer wg.Done()
  132. fmt.Fprintf(writer, "Cache %+v ...\n", dir.Child(entry.Name))
  133. remoteLocation := filer.MapFullPathToRemoteStorageLocation(localMountedDir, remoteMountedLocation, dir.Child(entry.Name))
  134. if err := filer.CacheRemoteObjectToLocalCluster(commandEnv, remoteConf, remoteLocation, dir, entry); err != nil {
  135. fmt.Fprintf(writer, "CacheRemoteObjectToLocalCluster %+v: %v\n", remoteLocation, err)
  136. if executionErr == nil {
  137. executionErr = fmt.Errorf("CacheRemoteObjectToLocalCluster %+v: %v\n", remoteLocation, err)
  138. }
  139. return
  140. }
  141. fmt.Fprintf(writer, "Cache %+v Done\n", dir.Child(entry.Name))
  142. })
  143. return true
  144. })
  145. wg.Wait()
  146. if traverseErr != nil {
  147. return traverseErr
  148. }
  149. if executionErr != nil {
  150. return executionErr
  151. }
  152. return nil
  153. }