command_volume_tier_download.go 4.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172
  1. package shell
  2. import (
  3. "context"
  4. "flag"
  5. "fmt"
  6. "io"
  7. "google.golang.org/grpc"
  8. "github.com/chrislusf/seaweedfs/weed/operation"
  9. "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
  10. "github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
  11. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  12. )
  13. func init() {
  14. Commands = append(Commands, &commandVolumeTierDownload{})
  15. }
  16. type commandVolumeTierDownload struct {
  17. }
  18. func (c *commandVolumeTierDownload) Name() string {
  19. return "volume.tier.download"
  20. }
  21. func (c *commandVolumeTierDownload) Help() string {
  22. return `download the dat file of a volume from a remote tier
  23. volume.tier.download [-collection=""]
  24. volume.tier.download [-collection=""] -volumeId=<volume_id>
  25. e.g.:
  26. volume.tier.download -volumeId=7
  27. volume.tier.download -volumeId=7
  28. This command will download the dat file of a volume from a remote tier to a volume server in local cluster.
  29. `
  30. }
  31. func (c *commandVolumeTierDownload) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
  32. if err = commandEnv.confirmIsLocked(); err != nil {
  33. return
  34. }
  35. tierCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
  36. volumeId := tierCommand.Int("volumeId", 0, "the volume id")
  37. collection := tierCommand.String("collection", "", "the collection name")
  38. if err = tierCommand.Parse(args); err != nil {
  39. return nil
  40. }
  41. vid := needle.VolumeId(*volumeId)
  42. // collect topology information
  43. topologyInfo, _, err := collectTopologyInfo(commandEnv)
  44. if err != nil {
  45. return err
  46. }
  47. // volumeId is provided
  48. if vid != 0 {
  49. return doVolumeTierDownload(commandEnv, writer, *collection, vid)
  50. }
  51. // apply to all volumes in the collection
  52. // reusing collectVolumeIdsForEcEncode for now
  53. volumeIds := collectRemoteVolumes(topologyInfo, *collection)
  54. if err != nil {
  55. return err
  56. }
  57. fmt.Printf("tier download volumes: %v\n", volumeIds)
  58. for _, vid := range volumeIds {
  59. if err = doVolumeTierDownload(commandEnv, writer, *collection, vid); err != nil {
  60. return err
  61. }
  62. }
  63. return nil
  64. }
  65. func collectRemoteVolumes(topoInfo *master_pb.TopologyInfo, selectedCollection string) (vids []needle.VolumeId) {
  66. vidMap := make(map[uint32]bool)
  67. eachDataNode(topoInfo, func(dc string, rack RackId, dn *master_pb.DataNodeInfo) {
  68. for _, diskInfo := range dn.DiskInfos {
  69. for _, v := range diskInfo.VolumeInfos {
  70. if v.Collection == selectedCollection && v.RemoteStorageKey != "" && v.RemoteStorageName != "" {
  71. vidMap[v.Id] = true
  72. }
  73. }
  74. }
  75. })
  76. for vid := range vidMap {
  77. vids = append(vids, needle.VolumeId(vid))
  78. }
  79. return
  80. }
  81. func doVolumeTierDownload(commandEnv *CommandEnv, writer io.Writer, collection string, vid needle.VolumeId) (err error) {
  82. // find volume location
  83. locations, found := commandEnv.MasterClient.GetLocations(uint32(vid))
  84. if !found {
  85. return fmt.Errorf("volume %d not found", vid)
  86. }
  87. // TODO parallelize this
  88. for _, loc := range locations {
  89. // copy the .dat file from remote tier to local
  90. err = downloadDatFromRemoteTier(commandEnv.option.GrpcDialOption, writer, needle.VolumeId(vid), collection, loc.Url)
  91. if err != nil {
  92. return fmt.Errorf("download dat file for volume %d to %s: %v", vid, loc.Url, err)
  93. }
  94. }
  95. return nil
  96. }
  97. func downloadDatFromRemoteTier(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, collection string, targetVolumeServer string) error {
  98. err := operation.WithVolumeServerClient(targetVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
  99. stream, downloadErr := volumeServerClient.VolumeTierMoveDatFromRemote(context.Background(), &volume_server_pb.VolumeTierMoveDatFromRemoteRequest{
  100. VolumeId: uint32(volumeId),
  101. Collection: collection,
  102. })
  103. var lastProcessed int64
  104. for {
  105. resp, recvErr := stream.Recv()
  106. if recvErr != nil {
  107. if recvErr == io.EOF {
  108. break
  109. } else {
  110. return recvErr
  111. }
  112. }
  113. processingSpeed := float64(resp.Processed-lastProcessed) / 1024.0 / 1024.0
  114. fmt.Fprintf(writer, "downloaded %.2f%%, %d bytes, %.2fMB/s\n", resp.ProcessedPercentage, resp.Processed, processingSpeed)
  115. lastProcessed = resp.Processed
  116. }
  117. if downloadErr != nil {
  118. return downloadErr
  119. }
  120. _, unmountErr := volumeServerClient.VolumeUnmount(context.Background(), &volume_server_pb.VolumeUnmountRequest{
  121. VolumeId: uint32(volumeId),
  122. })
  123. if unmountErr != nil {
  124. return unmountErr
  125. }
  126. _, mountErr := volumeServerClient.VolumeMount(context.Background(), &volume_server_pb.VolumeMountRequest{
  127. VolumeId: uint32(volumeId),
  128. })
  129. if mountErr != nil {
  130. return mountErr
  131. }
  132. return nil
  133. })
  134. return err
  135. }