command_volume_move.go 9.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222
  1. package shell
  2. import (
  3. "context"
  4. "flag"
  5. "fmt"
  6. "github.com/seaweedfs/seaweedfs/weed/pb"
  7. "github.com/seaweedfs/seaweedfs/weed/wdclient"
  8. "io"
  9. "log"
  10. "time"
  11. "github.com/seaweedfs/seaweedfs/weed/operation"
  12. "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
  13. "github.com/seaweedfs/seaweedfs/weed/storage/needle"
  14. "google.golang.org/grpc"
  15. )
  16. func init() {
  17. Commands = append(Commands, &commandVolumeMove{})
  18. }
  19. type commandVolumeMove struct {
  20. }
  21. func (c *commandVolumeMove) Name() string {
  22. return "volume.move"
  23. }
  24. func (c *commandVolumeMove) Help() string {
  25. return `move a live volume from one volume server to another volume server
  26. volume.move -source <source volume server host:port> -target <target volume server host:port> -volumeId <volume id>
  27. volume.move -source <source volume server host:port> -target <target volume server host:port> -volumeId <volume id> -disk [hdd|ssd|<tag>]
  28. This command move a live volume from one volume server to another volume server. Here are the steps:
  29. 1. This command asks the target volume server to copy the source volume from source volume server, remember the last entry's timestamp.
  30. 2. This command asks the target volume server to mount the new volume
  31. Now the master will mark this volume id as readonly.
  32. 3. This command asks the target volume server to tail the source volume for updates after the timestamp, for 1 minutes to drain the requests.
  33. 4. This command asks the source volume server to unmount the source volume
  34. Now the master will mark this volume id as writable.
  35. 5. This command asks the source volume server to delete the source volume
  36. The option "-disk [hdd|ssd|<tag>]" can be used to change the volume disk type.
  37. `
  38. }
  39. func (c *commandVolumeMove) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
  40. volMoveCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
  41. volumeIdInt := volMoveCommand.Int("volumeId", 0, "the volume id")
  42. sourceNodeStr := volMoveCommand.String("source", "", "the source volume server <host>:<port>")
  43. targetNodeStr := volMoveCommand.String("target", "", "the target volume server <host>:<port>")
  44. diskTypeStr := volMoveCommand.String("disk", "", "[hdd|ssd|<tag>] hard drive or solid state drive or any tag")
  45. ioBytePerSecond := volMoveCommand.Int64("ioBytePerSecond", 0, "limit the speed of move")
  46. if err = volMoveCommand.Parse(args); err != nil {
  47. return nil
  48. }
  49. if err = commandEnv.confirmIsLocked(args); err != nil {
  50. return
  51. }
  52. sourceVolumeServer, targetVolumeServer := pb.ServerAddress(*sourceNodeStr), pb.ServerAddress(*targetNodeStr)
  53. volumeId := needle.VolumeId(*volumeIdInt)
  54. if sourceVolumeServer == targetVolumeServer {
  55. return fmt.Errorf("source and target volume servers are the same!")
  56. }
  57. return LiveMoveVolume(commandEnv.option.GrpcDialOption, writer, volumeId, sourceVolumeServer, targetVolumeServer, 5*time.Second, *diskTypeStr, *ioBytePerSecond, false)
  58. }
  59. // LiveMoveVolume moves one volume from one source volume server to one target volume server, with idleTimeout to drain the incoming requests.
  60. func LiveMoveVolume(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, sourceVolumeServer, targetVolumeServer pb.ServerAddress, idleTimeout time.Duration, diskType string, ioBytePerSecond int64, skipTailError bool) (err error) {
  61. log.Printf("copying volume %d from %s to %s", volumeId, sourceVolumeServer, targetVolumeServer)
  62. lastAppendAtNs, err := copyVolume(grpcDialOption, writer, volumeId, sourceVolumeServer, targetVolumeServer, diskType, ioBytePerSecond)
  63. if err != nil {
  64. return fmt.Errorf("copy volume %d from %s to %s: %v", volumeId, sourceVolumeServer, targetVolumeServer, err)
  65. }
  66. log.Printf("tailing volume %d from %s to %s", volumeId, sourceVolumeServer, targetVolumeServer)
  67. if err = tailVolume(grpcDialOption, volumeId, sourceVolumeServer, targetVolumeServer, lastAppendAtNs, idleTimeout); err != nil {
  68. if skipTailError {
  69. fmt.Fprintf(writer, "tail volume %d from %s to %s: %v\n", volumeId, sourceVolumeServer, targetVolumeServer, err)
  70. } else {
  71. return fmt.Errorf("tail volume %d from %s to %s: %v", volumeId, sourceVolumeServer, targetVolumeServer, err)
  72. }
  73. }
  74. log.Printf("deleting volume %d from %s", volumeId, sourceVolumeServer)
  75. if err = deleteVolume(grpcDialOption, volumeId, sourceVolumeServer); err != nil {
  76. return fmt.Errorf("delete volume %d from %s: %v", volumeId, sourceVolumeServer, err)
  77. }
  78. log.Printf("moved volume %d from %s to %s", volumeId, sourceVolumeServer, targetVolumeServer)
  79. return nil
  80. }
  81. func copyVolume(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, sourceVolumeServer, targetVolumeServer pb.ServerAddress, diskType string, ioBytePerSecond int64) (lastAppendAtNs uint64, err error) {
  82. // check to see if the volume is already read-only and if its not then we need
  83. // to mark it as read-only and then before we return we need to undo what we
  84. // did
  85. var shouldMarkWritable bool
  86. defer func() {
  87. if !shouldMarkWritable {
  88. return
  89. }
  90. clientErr := operation.WithVolumeServerClient(false, sourceVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
  91. _, writableErr := volumeServerClient.VolumeMarkWritable(context.Background(), &volume_server_pb.VolumeMarkWritableRequest{
  92. VolumeId: uint32(volumeId),
  93. })
  94. return writableErr
  95. })
  96. if clientErr != nil {
  97. log.Printf("failed to mark volume %d as writable after copy from %s: %v", volumeId, sourceVolumeServer, clientErr)
  98. }
  99. }()
  100. err = operation.WithVolumeServerClient(false, sourceVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
  101. resp, statusErr := volumeServerClient.VolumeStatus(context.Background(), &volume_server_pb.VolumeStatusRequest{
  102. VolumeId: uint32(volumeId),
  103. })
  104. if statusErr == nil && !resp.IsReadOnly {
  105. shouldMarkWritable = true
  106. _, readonlyErr := volumeServerClient.VolumeMarkReadonly(context.Background(), &volume_server_pb.VolumeMarkReadonlyRequest{
  107. VolumeId: uint32(volumeId),
  108. })
  109. return readonlyErr
  110. }
  111. return statusErr
  112. })
  113. if err != nil {
  114. return
  115. }
  116. err = operation.WithVolumeServerClient(true, targetVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
  117. stream, replicateErr := volumeServerClient.VolumeCopy(context.Background(), &volume_server_pb.VolumeCopyRequest{
  118. VolumeId: uint32(volumeId),
  119. SourceDataNode: string(sourceVolumeServer),
  120. DiskType: diskType,
  121. IoBytePerSecond: ioBytePerSecond,
  122. })
  123. if replicateErr != nil {
  124. return replicateErr
  125. }
  126. for {
  127. resp, recvErr := stream.Recv()
  128. if recvErr != nil {
  129. if recvErr == io.EOF {
  130. break
  131. } else {
  132. return recvErr
  133. }
  134. }
  135. if resp.LastAppendAtNs != 0 {
  136. lastAppendAtNs = resp.LastAppendAtNs
  137. } else {
  138. fmt.Fprintf(writer, "volume %d processed %d bytes\n", volumeId, resp.ProcessedBytes)
  139. }
  140. }
  141. return nil
  142. })
  143. return
  144. }
  145. func tailVolume(grpcDialOption grpc.DialOption, volumeId needle.VolumeId, sourceVolumeServer, targetVolumeServer pb.ServerAddress, lastAppendAtNs uint64, idleTimeout time.Duration) (err error) {
  146. return operation.WithVolumeServerClient(true, targetVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
  147. _, replicateErr := volumeServerClient.VolumeTailReceiver(context.Background(), &volume_server_pb.VolumeTailReceiverRequest{
  148. VolumeId: uint32(volumeId),
  149. SinceNs: lastAppendAtNs,
  150. IdleTimeoutSeconds: uint32(idleTimeout.Seconds()),
  151. SourceVolumeServer: string(sourceVolumeServer),
  152. })
  153. return replicateErr
  154. })
  155. }
  156. func deleteVolume(grpcDialOption grpc.DialOption, volumeId needle.VolumeId, sourceVolumeServer pb.ServerAddress) (err error) {
  157. return operation.WithVolumeServerClient(false, sourceVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
  158. _, deleteErr := volumeServerClient.VolumeDelete(context.Background(), &volume_server_pb.VolumeDeleteRequest{
  159. VolumeId: uint32(volumeId),
  160. })
  161. return deleteErr
  162. })
  163. }
  164. func markVolumeWritable(grpcDialOption grpc.DialOption, volumeId needle.VolumeId, sourceVolumeServer pb.ServerAddress, writable bool) (err error) {
  165. return operation.WithVolumeServerClient(false, sourceVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
  166. if writable {
  167. _, err = volumeServerClient.VolumeMarkWritable(context.Background(), &volume_server_pb.VolumeMarkWritableRequest{
  168. VolumeId: uint32(volumeId),
  169. })
  170. } else {
  171. _, err = volumeServerClient.VolumeMarkReadonly(context.Background(), &volume_server_pb.VolumeMarkReadonlyRequest{
  172. VolumeId: uint32(volumeId),
  173. })
  174. }
  175. return err
  176. })
  177. }
  178. func markVolumeReplicasWritable(grpcDialOption grpc.DialOption, volumeId needle.VolumeId, locations []wdclient.Location, writable bool) error {
  179. for _, location := range locations {
  180. fmt.Printf("markVolumeReadonly %d on %s ...\n", volumeId, location.Url)
  181. if err := markVolumeWritable(grpcDialOption, volumeId, location.ServerAddress(), writable); err != nil {
  182. return err
  183. }
  184. }
  185. return nil
  186. }