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.

323 lines
10 KiB

2 years ago
2 years ago
4 years ago
4 years ago
2 years ago
4 years ago
4 years ago
4 years ago
4 years ago
9 months ago
2 years ago
2 years ago
2 years ago
2 years ago
9 months ago
4 years ago
4 years ago
  1. package shell
  2. import (
  3. "context"
  4. "errors"
  5. "flag"
  6. "fmt"
  7. "io"
  8. "path/filepath"
  9. "sync"
  10. "time"
  11. "github.com/seaweedfs/seaweedfs/weed/glog"
  12. "github.com/seaweedfs/seaweedfs/weed/pb"
  13. "github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
  14. "github.com/seaweedfs/seaweedfs/weed/storage/types"
  15. "github.com/seaweedfs/seaweedfs/weed/wdclient"
  16. "github.com/seaweedfs/seaweedfs/weed/operation"
  17. "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
  18. "github.com/seaweedfs/seaweedfs/weed/storage/needle"
  19. )
  20. func init() {
  21. Commands = append(Commands, &commandVolumeTierMove{})
  22. }
  23. type volumeTierMoveJob struct {
  24. src pb.ServerAddress
  25. vid needle.VolumeId
  26. }
  27. type commandVolumeTierMove struct {
  28. activeServers sync.Map
  29. queues map[pb.ServerAddress]chan volumeTierMoveJob
  30. //activeServers map[pb.ServerAddress]struct{}
  31. //activeServersLock sync.Mutex
  32. //activeServersCond *sync.Cond
  33. }
  34. func (c *commandVolumeTierMove) Name() string {
  35. return "volume.tier.move"
  36. }
  37. func (c *commandVolumeTierMove) Help() string {
  38. return `change a volume from one disk type to another
  39. volume.tier.move -fromDiskType=hdd -toDiskType=ssd [-collectionPattern=""] [-fullPercent=95] [-quietFor=1h] [-parallelLimit=4] [-toReplication=XYZ]
  40. Even if the volume is replicated, only one replica will be changed and the rest replicas will be dropped.
  41. So "volume.fix.replication" and "volume.balance" should be followed.
  42. `
  43. }
  44. func (c *commandVolumeTierMove) IsResourceHeavy() bool {
  45. return false
  46. }
  47. func (c *commandVolumeTierMove) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
  48. tierCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
  49. collectionPattern := tierCommand.String("collectionPattern", "", "match with wildcard characters '*' and '?'")
  50. fullPercentage := tierCommand.Float64("fullPercent", 95, "the volume reaches the percentage of max volume size")
  51. quietPeriod := tierCommand.Duration("quietFor", 24*time.Hour, "select volumes without no writes for this period")
  52. source := tierCommand.String("fromDiskType", "", "the source disk type")
  53. target := tierCommand.String("toDiskType", "", "the target disk type")
  54. parallelLimit := tierCommand.Int("parallelLimit", 0, "limit the number of parallel copying jobs")
  55. applyChange := tierCommand.Bool("force", false, "actually apply the changes")
  56. ioBytePerSecond := tierCommand.Int64("ioBytePerSecond", 0, "limit the speed of move")
  57. replicationString := tierCommand.String("toReplication", "", "the new target replication setting")
  58. if err = tierCommand.Parse(args); err != nil {
  59. return nil
  60. }
  61. infoAboutSimulationMode(writer, *applyChange, "-force")
  62. if err = commandEnv.confirmIsLocked(args); err != nil {
  63. return
  64. }
  65. fromDiskType := types.ToDiskType(*source)
  66. toDiskType := types.ToDiskType(*target)
  67. if fromDiskType == toDiskType {
  68. return fmt.Errorf("source tier %s is the same as target tier %s", fromDiskType, toDiskType)
  69. }
  70. // collect topology information
  71. topologyInfo, volumeSizeLimitMb, err := collectTopologyInfo(commandEnv, 0)
  72. if err != nil {
  73. return err
  74. }
  75. // collect all volumes that should change
  76. volumeIds, err := collectVolumeIdsForTierChange(topologyInfo, volumeSizeLimitMb, fromDiskType, *collectionPattern, *fullPercentage, *quietPeriod)
  77. if err != nil {
  78. return err
  79. }
  80. fmt.Printf("tier move volumes: %v\n", volumeIds)
  81. _, allLocations := collectVolumeReplicaLocations(topologyInfo)
  82. allLocations = filterLocationsByDiskType(allLocations, toDiskType)
  83. keepDataNodesSorted(allLocations, toDiskType)
  84. if len(allLocations) > 0 && *parallelLimit > 0 && *parallelLimit < len(allLocations) {
  85. allLocations = allLocations[:*parallelLimit]
  86. }
  87. wg := sync.WaitGroup{}
  88. bufferLen := len(allLocations)
  89. c.queues = make(map[pb.ServerAddress]chan volumeTierMoveJob)
  90. for _, dst := range allLocations {
  91. destServerAddress := pb.NewServerAddressFromDataNode(dst.dataNode)
  92. c.queues[destServerAddress] = make(chan volumeTierMoveJob, bufferLen)
  93. wg.Add(1)
  94. go func(dst location, jobs <-chan volumeTierMoveJob, applyChanges bool) {
  95. defer wg.Done()
  96. for job := range jobs {
  97. fmt.Fprintf(writer, "moving volume %d from %s to %s with disk type %s ...\n", job.vid, job.src, dst.dataNode.Id, toDiskType.ReadableString())
  98. locations, found := commandEnv.MasterClient.GetLocationsClone(uint32(job.vid))
  99. if !found {
  100. fmt.Printf("volume %d not found", job.vid)
  101. continue
  102. }
  103. unlock := c.Lock(job.src)
  104. if applyChanges {
  105. if err := c.doMoveOneVolume(commandEnv, writer, job.vid, toDiskType, locations, job.src, dst, *ioBytePerSecond, replicationString); err != nil {
  106. fmt.Fprintf(writer, "move volume %d %s => %s: %v\n", job.vid, job.src, dst.dataNode.Id, err)
  107. }
  108. }
  109. unlock()
  110. }
  111. }(dst, c.queues[destServerAddress], *applyChange)
  112. }
  113. for _, vid := range volumeIds {
  114. if err = c.doVolumeTierMove(commandEnv, writer, vid, toDiskType, allLocations); err != nil {
  115. fmt.Printf("tier move volume %d: %v\n", vid, err)
  116. }
  117. allLocations = rotateDataNodes(allLocations)
  118. }
  119. for key, _ := range c.queues {
  120. close(c.queues[key])
  121. }
  122. wg.Wait()
  123. return nil
  124. }
  125. func (c *commandVolumeTierMove) Lock(key pb.ServerAddress) func() {
  126. value, _ := c.activeServers.LoadOrStore(key, &sync.Mutex{})
  127. mtx := value.(*sync.Mutex)
  128. mtx.Lock()
  129. return func() { mtx.Unlock() }
  130. }
  131. func filterLocationsByDiskType(dataNodes []location, diskType types.DiskType) (ret []location) {
  132. for _, loc := range dataNodes {
  133. _, found := loc.dataNode.DiskInfos[string(diskType)]
  134. if found {
  135. ret = append(ret, loc)
  136. }
  137. }
  138. return
  139. }
  140. func rotateDataNodes(dataNodes []location) []location {
  141. if len(dataNodes) > 0 {
  142. return append(dataNodes[1:], dataNodes[0])
  143. } else {
  144. return dataNodes
  145. }
  146. }
  147. func isOneOf(server string, locations []wdclient.Location) bool {
  148. for _, loc := range locations {
  149. if server == loc.Url {
  150. return true
  151. }
  152. }
  153. return false
  154. }
  155. func (c *commandVolumeTierMove) doVolumeTierMove(commandEnv *CommandEnv, writer io.Writer, vid needle.VolumeId, toDiskType types.DiskType, allLocations []location) (err error) {
  156. // find volume location
  157. locations, found := commandEnv.MasterClient.GetLocationsClone(uint32(vid))
  158. if !found {
  159. return fmt.Errorf("volume %d not found", vid)
  160. }
  161. // find one server with the most empty volume slots with target disk type
  162. hasFoundTarget := false
  163. fn := capacityByFreeVolumeCount(toDiskType)
  164. for _, dst := range allLocations {
  165. if fn(dst.dataNode) > 0 && !hasFoundTarget {
  166. // ask the volume server to replicate the volume
  167. if isOneOf(dst.dataNode.Id, locations) {
  168. continue
  169. }
  170. var sourceVolumeServer pb.ServerAddress
  171. for _, loc := range locations {
  172. if loc.Url != dst.dataNode.Id {
  173. sourceVolumeServer = loc.ServerAddress()
  174. }
  175. }
  176. if sourceVolumeServer == "" {
  177. continue
  178. }
  179. hasFoundTarget = true
  180. // adjust volume count
  181. addVolumeCount(dst.dataNode.DiskInfos[string(toDiskType)], 1)
  182. destServerAddress := pb.NewServerAddressFromDataNode(dst.dataNode)
  183. c.queues[destServerAddress] <- volumeTierMoveJob{sourceVolumeServer, vid}
  184. }
  185. }
  186. if !hasFoundTarget {
  187. fmt.Fprintf(writer, "can not find disk type %s for volume %d\n", toDiskType.ReadableString(), vid)
  188. }
  189. return nil
  190. }
  191. func (c *commandVolumeTierMove) doMoveOneVolume(commandEnv *CommandEnv, writer io.Writer, vid needle.VolumeId, toDiskType types.DiskType, locations []wdclient.Location, sourceVolumeServer pb.ServerAddress, dst location, ioBytePerSecond int64, replicationString *string) (err error) {
  192. if !commandEnv.isLocked() {
  193. return fmt.Errorf("lock is lost")
  194. }
  195. // mark all replicas as read only
  196. if err = markVolumeReplicasWritable(commandEnv.option.GrpcDialOption, vid, locations, false, false); err != nil {
  197. return fmt.Errorf("mark volume %d as readonly on %s: %v", vid, locations[0].Url, err)
  198. }
  199. newAddress := pb.NewServerAddressFromDataNode(dst.dataNode)
  200. if err = LiveMoveVolume(commandEnv.option.GrpcDialOption, writer, vid, sourceVolumeServer, newAddress, 5*time.Second, toDiskType.ReadableString(), ioBytePerSecond, true); err != nil {
  201. // mark all replicas as writable
  202. if err = markVolumeReplicasWritable(commandEnv.option.GrpcDialOption, vid, locations, true, false); err != nil {
  203. glog.Errorf("mark volume %d as writable on %s: %v", vid, locations[0].Url, err)
  204. }
  205. return fmt.Errorf("move volume %d %s => %s : %v", vid, locations[0].Url, dst.dataNode.Id, err)
  206. }
  207. // If move is successful and replication is not empty, alter moved volume's replication setting
  208. if *replicationString != "" {
  209. err = operation.WithVolumeServerClient(false, newAddress, commandEnv.option.GrpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
  210. resp, configureErr := volumeServerClient.VolumeConfigure(context.Background(), &volume_server_pb.VolumeConfigureRequest{
  211. VolumeId: uint32(vid),
  212. Replication: *replicationString,
  213. })
  214. if configureErr != nil {
  215. return configureErr
  216. }
  217. if resp.Error != "" {
  218. return errors.New(resp.Error)
  219. }
  220. return nil
  221. })
  222. if err != nil {
  223. glog.Errorf("update volume %d replication on %s: %v", vid, locations[0].Url, err)
  224. }
  225. }
  226. // remove the remaining replicas
  227. for _, loc := range locations {
  228. if loc.Url != dst.dataNode.Id && loc.ServerAddress() != sourceVolumeServer {
  229. if err = deleteVolume(commandEnv.option.GrpcDialOption, vid, loc.ServerAddress(), false); err != nil {
  230. fmt.Fprintf(writer, "failed to delete volume %d on %s: %v\n", vid, loc.Url, err)
  231. }
  232. // reduce volume count? Not really necessary since they are "more" full and will not be a candidate to move to
  233. }
  234. }
  235. return nil
  236. }
  237. func collectVolumeIdsForTierChange(topologyInfo *master_pb.TopologyInfo, volumeSizeLimitMb uint64, sourceTier types.DiskType, collectionPattern string, fullPercentage float64, quietPeriod time.Duration) (vids []needle.VolumeId, err error) {
  238. quietSeconds := int64(quietPeriod / time.Second)
  239. nowUnixSeconds := time.Now().Unix()
  240. fmt.Printf("collect %s volumes quiet for: %d seconds\n", sourceTier, quietSeconds)
  241. vidMap := make(map[uint32]bool)
  242. eachDataNode(topologyInfo, func(dc string, rack RackId, dn *master_pb.DataNodeInfo) {
  243. for _, diskInfo := range dn.DiskInfos {
  244. for _, v := range diskInfo.VolumeInfos {
  245. // check collection name pattern
  246. if collectionPattern != "" {
  247. matched, err := filepath.Match(collectionPattern, v.Collection)
  248. if err != nil {
  249. return
  250. }
  251. if !matched {
  252. continue
  253. }
  254. }
  255. if v.ModifiedAtSecond+quietSeconds < nowUnixSeconds && types.ToDiskType(v.DiskType) == sourceTier {
  256. if float64(v.Size) > fullPercentage/100*float64(volumeSizeLimitMb)*1024*1024 {
  257. vidMap[v.Id] = true
  258. }
  259. }
  260. }
  261. }
  262. })
  263. for vid := range vidMap {
  264. vids = append(vids, needle.VolumeId(vid))
  265. }
  266. return
  267. }