command_volume_tier_upload.go 5.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165
  1. package shell
  2. import (
  3. "context"
  4. "flag"
  5. "fmt"
  6. "github.com/seaweedfs/seaweedfs/weed/pb"
  7. "io"
  8. "time"
  9. "google.golang.org/grpc"
  10. "github.com/seaweedfs/seaweedfs/weed/operation"
  11. "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
  12. "github.com/seaweedfs/seaweedfs/weed/storage/needle"
  13. )
  14. func init() {
  15. Commands = append(Commands, &commandVolumeTierUpload{})
  16. }
  17. type commandVolumeTierUpload struct {
  18. }
  19. func (c *commandVolumeTierUpload) Name() string {
  20. return "volume.tier.upload"
  21. }
  22. func (c *commandVolumeTierUpload) Help() string {
  23. return `upload the dat file of a volume to a remote tier
  24. volume.tier.upload [-collection=""] [-fullPercent=95] [-quietFor=1h]
  25. volume.tier.upload [-collection=""] -volumeId=<volume_id> -dest=<storage_backend> [-keepLocalDatFile]
  26. e.g.:
  27. volume.tier.upload -volumeId=7 -dest=s3
  28. volume.tier.upload -volumeId=7 -dest=s3.default
  29. The <storage_backend> is defined in master.toml.
  30. For example, "s3.default" in [storage.backend.s3.default]
  31. This command will move the dat file of a volume to a remote tier.
  32. SeaweedFS enables scalable and fast local access to lots of files,
  33. and the cloud storage is slower by cost efficient. How to combine them together?
  34. Usually the data follows 80/20 rule: only 20% of data is frequently accessed.
  35. We can offload the old volumes to the cloud.
  36. With this, SeaweedFS can be both fast and scalable, and infinite storage space.
  37. Just add more local SeaweedFS volume servers to increase the throughput.
  38. The index file is still local, and the same O(1) disk read is applied to the remote file.
  39. `
  40. }
  41. func (c *commandVolumeTierUpload) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
  42. tierCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
  43. volumeId := tierCommand.Int("volumeId", 0, "the volume id")
  44. collection := tierCommand.String("collection", "", "the collection name")
  45. fullPercentage := tierCommand.Float64("fullPercent", 95, "the volume reaches the percentage of max volume size")
  46. quietPeriod := tierCommand.Duration("quietFor", 24*time.Hour, "select volumes without no writes for this period")
  47. dest := tierCommand.String("dest", "", "the target tier name")
  48. keepLocalDatFile := tierCommand.Bool("keepLocalDatFile", false, "whether keep local dat file")
  49. if err = tierCommand.Parse(args); err != nil {
  50. return nil
  51. }
  52. if err = commandEnv.confirmIsLocked(args); err != nil {
  53. return
  54. }
  55. vid := needle.VolumeId(*volumeId)
  56. // volumeId is provided
  57. if vid != 0 {
  58. return doVolumeTierUpload(commandEnv, writer, *collection, vid, *dest, *keepLocalDatFile)
  59. }
  60. // apply to all volumes in the collection
  61. // reusing collectVolumeIdsForEcEncode for now
  62. volumeIds, err := collectVolumeIdsForEcEncode(commandEnv, *collection, *fullPercentage, *quietPeriod)
  63. if err != nil {
  64. return err
  65. }
  66. fmt.Printf("tier upload volumes: %v\n", volumeIds)
  67. for _, vid := range volumeIds {
  68. if err = doVolumeTierUpload(commandEnv, writer, *collection, vid, *dest, *keepLocalDatFile); err != nil {
  69. return err
  70. }
  71. }
  72. return nil
  73. }
  74. func doVolumeTierUpload(commandEnv *CommandEnv, writer io.Writer, collection string, vid needle.VolumeId, dest string, keepLocalDatFile bool) (err error) {
  75. // find volume location
  76. existingLocations, found := commandEnv.MasterClient.GetLocations(uint32(vid))
  77. if !found {
  78. return fmt.Errorf("volume %d not found", vid)
  79. }
  80. err = markVolumeReplicasWritable(commandEnv.option.GrpcDialOption, vid, existingLocations, false)
  81. if err != nil {
  82. return fmt.Errorf("mark volume %d as readonly on %s: %v", vid, existingLocations[0].Url, err)
  83. }
  84. // copy the .dat file to remote tier
  85. err = uploadDatToRemoteTier(commandEnv.option.GrpcDialOption, writer, vid, collection, existingLocations[0].ServerAddress(), dest, keepLocalDatFile)
  86. if err != nil {
  87. return fmt.Errorf("copy dat file for volume %d on %s to %s: %v", vid, existingLocations[0].Url, dest, err)
  88. }
  89. // now the first replica has the .idx and .vif files.
  90. // ask replicas on other volume server to delete its own local copy
  91. for i, location := range existingLocations {
  92. if i == 0 {
  93. break
  94. }
  95. fmt.Printf("delete volume %d from %s\n", vid, location.Url)
  96. err = deleteVolume(commandEnv.option.GrpcDialOption, vid, location.ServerAddress())
  97. if err != nil {
  98. return fmt.Errorf("deleteVolume %s volume %d: %v", location.Url, vid, err)
  99. }
  100. }
  101. return nil
  102. }
  103. func uploadDatToRemoteTier(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, collection string, sourceVolumeServer pb.ServerAddress, dest string, keepLocalDatFile bool) error {
  104. err := operation.WithVolumeServerClient(true, sourceVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
  105. stream, copyErr := volumeServerClient.VolumeTierMoveDatToRemote(context.Background(), &volume_server_pb.VolumeTierMoveDatToRemoteRequest{
  106. VolumeId: uint32(volumeId),
  107. Collection: collection,
  108. DestinationBackendName: dest,
  109. KeepLocalDatFile: keepLocalDatFile,
  110. })
  111. var lastProcessed int64
  112. for {
  113. resp, recvErr := stream.Recv()
  114. if recvErr != nil {
  115. if recvErr == io.EOF {
  116. break
  117. } else {
  118. return recvErr
  119. }
  120. }
  121. processingSpeed := float64(resp.Processed-lastProcessed) / 1024.0 / 1024.0
  122. fmt.Fprintf(writer, "copied %.2f%%, %d bytes, %.2fMB/s\n", resp.ProcessedPercentage, resp.Processed, processingSpeed)
  123. lastProcessed = resp.Processed
  124. }
  125. return copyErr
  126. })
  127. return err
  128. }