You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

239 lines
7.8 KiB

4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
3 years ago
4 years ago
4 years ago
  1. package shell
  2. import (
  3. "flag"
  4. "fmt"
  5. "github.com/chrislusf/seaweedfs/weed/glog"
  6. "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
  7. "github.com/chrislusf/seaweedfs/weed/storage/types"
  8. "github.com/chrislusf/seaweedfs/weed/wdclient"
  9. "io"
  10. "path/filepath"
  11. "sync"
  12. "time"
  13. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  14. )
  15. func init() {
  16. Commands = append(Commands, &commandVolumeTierMove{})
  17. }
  18. type commandVolumeTierMove struct {
  19. activeServers map[string]struct{}
  20. activeServersLock sync.Mutex
  21. activeServersCond *sync.Cond
  22. }
  23. func (c *commandVolumeTierMove) Name() string {
  24. return "volume.tier.move"
  25. }
  26. func (c *commandVolumeTierMove) Help() string {
  27. return `change a volume from one disk type to another
  28. volume.tier.move -fromDiskType=hdd -toDiskType=ssd [-collectionPattern=""] [-fullPercent=95] [-quietFor=1h]
  29. Even if the volume is replicated, only one replica will be changed and the rest replicas will be dropped.
  30. So "volume.fix.replication" and "volume.balance" should be followed.
  31. `
  32. }
  33. func (c *commandVolumeTierMove) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
  34. c.activeServers = make(map[string]struct{})
  35. c.activeServersCond = sync.NewCond(new(sync.Mutex))
  36. if err = commandEnv.confirmIsLocked(); err != nil {
  37. return
  38. }
  39. tierCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
  40. collectionPattern := tierCommand.String("collectionPattern", "", "match with wildcard characters '*' and '?'")
  41. fullPercentage := tierCommand.Float64("fullPercent", 95, "the volume reaches the percentage of max volume size")
  42. quietPeriod := tierCommand.Duration("quietFor", 24*time.Hour, "select volumes without no writes for this period")
  43. source := tierCommand.String("fromDiskType", "", "the source disk type")
  44. target := tierCommand.String("toDiskType", "", "the target disk type")
  45. applyChange := tierCommand.Bool("force", false, "actually apply the changes")
  46. if err = tierCommand.Parse(args); err != nil {
  47. return nil
  48. }
  49. fromDiskType := types.ToDiskType(*source)
  50. toDiskType := types.ToDiskType(*target)
  51. if fromDiskType == toDiskType {
  52. return fmt.Errorf("source tier %s is the same as target tier %s", fromDiskType, toDiskType)
  53. }
  54. // collect topology information
  55. topologyInfo, volumeSizeLimitMb, err := collectTopologyInfo(commandEnv)
  56. if err != nil {
  57. return err
  58. }
  59. // collect all volumes that should change
  60. volumeIds, err := collectVolumeIdsForTierChange(commandEnv, topologyInfo, volumeSizeLimitMb, fromDiskType, *collectionPattern, *fullPercentage, *quietPeriod)
  61. if err != nil {
  62. return err
  63. }
  64. fmt.Printf("tier move volumes: %v\n", volumeIds)
  65. _, allLocations := collectVolumeReplicaLocations(topologyInfo)
  66. for _, vid := range volumeIds {
  67. if err = c.doVolumeTierMove(commandEnv, writer, vid, toDiskType, allLocations, *applyChange); err != nil {
  68. fmt.Printf("tier move volume %d: %v\n", vid, err)
  69. }
  70. }
  71. return nil
  72. }
  73. func isOneOf(server string, locations []wdclient.Location) bool {
  74. for _, loc := range locations {
  75. if server == loc.Url {
  76. return true
  77. }
  78. }
  79. return false
  80. }
  81. func (c *commandVolumeTierMove) doVolumeTierMove(commandEnv *CommandEnv, writer io.Writer, vid needle.VolumeId, toDiskType types.DiskType, allLocations []location, applyChanges bool) (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. // find one server with the most empty volume slots with target disk type
  88. hasFoundTarget := false
  89. keepDataNodesSorted(allLocations, toDiskType)
  90. fn := capacityByFreeVolumeCount(toDiskType)
  91. wg := sync.WaitGroup{}
  92. for _, dst := range allLocations {
  93. if fn(dst.dataNode) > 0 && !hasFoundTarget {
  94. // ask the volume server to replicate the volume
  95. if isOneOf(dst.dataNode.Id, locations) {
  96. continue
  97. }
  98. sourceVolumeServer := ""
  99. for _, loc := range locations {
  100. if loc.Url != dst.dataNode.Id {
  101. sourceVolumeServer = loc.Url
  102. }
  103. }
  104. if sourceVolumeServer == "" {
  105. continue
  106. }
  107. fmt.Fprintf(writer, "moving volume %d from %s to %s with disk type %s ...\n", vid, sourceVolumeServer, dst.dataNode.Id, toDiskType.ReadableString())
  108. hasFoundTarget = true
  109. if !applyChanges {
  110. // adjust volume count
  111. dst.dataNode.DiskInfos[string(toDiskType)].VolumeCount++
  112. break
  113. }
  114. c.activeServersCond.L.Lock()
  115. _, isSourceActive := c.activeServers[sourceVolumeServer]
  116. _, isDestActive := c.activeServers[dst.dataNode.Id]
  117. for isSourceActive || isDestActive {
  118. c.activeServersCond.Wait()
  119. _, isSourceActive = c.activeServers[sourceVolumeServer]
  120. _, isDestActive = c.activeServers[dst.dataNode.Id]
  121. }
  122. c.activeServers[sourceVolumeServer] = struct{}{}
  123. c.activeServers[dst.dataNode.Id] = struct{}{}
  124. c.activeServersCond.L.Unlock()
  125. wg.Add(1)
  126. go func(dst location) {
  127. if err := c.doMoveOneVolume(commandEnv, writer, vid, toDiskType, locations, sourceVolumeServer, dst); err != nil {
  128. fmt.Fprintf(writer, "move volume %d %s => %s: %v\n", vid, sourceVolumeServer, dst.dataNode.Id, err)
  129. }
  130. delete(c.activeServers, sourceVolumeServer)
  131. delete(c.activeServers, dst.dataNode.Id)
  132. c.activeServersCond.Signal()
  133. wg.Done()
  134. }(dst)
  135. }
  136. }
  137. wg.Wait()
  138. if !hasFoundTarget {
  139. fmt.Fprintf(writer, "can not find disk type %s for volume %d\n", toDiskType.ReadableString(), vid)
  140. }
  141. return nil
  142. }
  143. func (c *commandVolumeTierMove) doMoveOneVolume(commandEnv *CommandEnv, writer io.Writer, vid needle.VolumeId, toDiskType types.DiskType, locations []wdclient.Location, sourceVolumeServer string, dst location) (err error) {
  144. // mark all replicas as read only
  145. if err = markVolumeReplicasWritable(commandEnv.option.GrpcDialOption, vid, locations, false); err != nil {
  146. return fmt.Errorf("mark volume %d as readonly on %s: %v", vid, locations[0].Url, err)
  147. }
  148. if err = LiveMoveVolume(commandEnv.option.GrpcDialOption, writer, vid, sourceVolumeServer, dst.dataNode.Id, 5*time.Second, toDiskType.ReadableString(), true); err != nil {
  149. // mark all replicas as writable
  150. if err = markVolumeReplicasWritable(commandEnv.option.GrpcDialOption, vid, locations, true); err != nil {
  151. glog.Errorf("mark volume %d as writable on %s: %v", vid, locations[0].Url, err)
  152. }
  153. return fmt.Errorf("move volume %d %s => %s : %v", vid, locations[0].Url, dst.dataNode.Id, err)
  154. }
  155. // adjust volume count
  156. dst.dataNode.DiskInfos[string(toDiskType)].VolumeCount++
  157. // remove the remaining replicas
  158. for _, loc := range locations {
  159. if loc.Url != dst.dataNode.Id && loc.Url != sourceVolumeServer {
  160. if err = deleteVolume(commandEnv.option.GrpcDialOption, vid, loc.Url); err != nil {
  161. fmt.Fprintf(writer, "failed to delete volume %d on %s: %v\n", vid, loc.Url, err)
  162. }
  163. }
  164. }
  165. return nil
  166. }
  167. func collectVolumeIdsForTierChange(commandEnv *CommandEnv, topologyInfo *master_pb.TopologyInfo, volumeSizeLimitMb uint64, sourceTier types.DiskType, collectionPattern string, fullPercentage float64, quietPeriod time.Duration) (vids []needle.VolumeId, err error) {
  168. quietSeconds := int64(quietPeriod / time.Second)
  169. nowUnixSeconds := time.Now().Unix()
  170. fmt.Printf("collect %s volumes quiet for: %d seconds\n", sourceTier, quietSeconds)
  171. vidMap := make(map[uint32]bool)
  172. eachDataNode(topologyInfo, func(dc string, rack RackId, dn *master_pb.DataNodeInfo) {
  173. for _, diskInfo := range dn.DiskInfos {
  174. for _, v := range diskInfo.VolumeInfos {
  175. // check collection name pattern
  176. if collectionPattern != "" {
  177. matched, err := filepath.Match(collectionPattern, v.Collection)
  178. if err != nil {
  179. return
  180. }
  181. if !matched {
  182. continue
  183. }
  184. }
  185. if v.ModifiedAtSecond+quietSeconds < nowUnixSeconds && types.ToDiskType(v.DiskType) == sourceTier {
  186. if float64(v.Size) > fullPercentage/100*float64(volumeSizeLimitMb)*1024*1024 {
  187. vidMap[v.Id] = true
  188. }
  189. }
  190. }
  191. }
  192. })
  193. for vid := range vidMap {
  194. vids = append(vids, needle.VolumeId(vid))
  195. }
  196. return
  197. }