command_ec_encode.go 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321
  1. package shell
  2. import (
  3. "context"
  4. "flag"
  5. "fmt"
  6. "github.com/seaweedfs/seaweedfs/weed/glog"
  7. "github.com/seaweedfs/seaweedfs/weed/pb"
  8. "io"
  9. "math/rand"
  10. "sync"
  11. "time"
  12. "google.golang.org/grpc"
  13. "github.com/seaweedfs/seaweedfs/weed/operation"
  14. "github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
  15. "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
  16. "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
  17. "github.com/seaweedfs/seaweedfs/weed/storage/needle"
  18. "github.com/seaweedfs/seaweedfs/weed/wdclient"
  19. )
  20. func init() {
  21. Commands = append(Commands, &commandEcEncode{})
  22. }
  23. type commandEcEncode struct {
  24. }
  25. func (c *commandEcEncode) Name() string {
  26. return "ec.encode"
  27. }
  28. func (c *commandEcEncode) Help() string {
  29. return `apply erasure coding to a volume
  30. ec.encode [-collection=""] [-fullPercent=95 -quietFor=1h]
  31. ec.encode [-collection=""] [-volumeId=<volume_id>]
  32. This command will:
  33. 1. freeze one volume
  34. 2. apply erasure coding to the volume
  35. 3. move the encoded shards to multiple volume servers
  36. The erasure coding is 10.4. So ideally you have more than 14 volume servers, and you can afford
  37. to lose 4 volume servers.
  38. If the number of volumes are not high, the worst case is that you only have 4 volume servers,
  39. and the shards are spread as 4,4,3,3, respectively. You can afford to lose one volume server.
  40. If you only have less than 4 volume servers, with erasure coding, at least you can afford to
  41. have 4 corrupted shard files.
  42. `
  43. }
  44. func (c *commandEcEncode) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
  45. encodeCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
  46. volumeId := encodeCommand.Int("volumeId", 0, "the volume id")
  47. collection := encodeCommand.String("collection", "", "the collection name")
  48. fullPercentage := encodeCommand.Float64("fullPercent", 95, "the volume reaches the percentage of max volume size")
  49. quietPeriod := encodeCommand.Duration("quietFor", time.Hour, "select volumes without no writes for this period")
  50. parallelCopy := encodeCommand.Bool("parallelCopy", true, "copy shards in parallel")
  51. forceChanges := encodeCommand.Bool("force", false, "force the encoding even if the cluster has less than recommended 4 nodes")
  52. if err = encodeCommand.Parse(args); err != nil {
  53. return nil
  54. }
  55. if err = commandEnv.confirmIsLocked(args); err != nil {
  56. return
  57. }
  58. // collect topology information
  59. topologyInfo, _, err := collectTopologyInfo(commandEnv, 0)
  60. if err != nil {
  61. return err
  62. }
  63. if !*forceChanges {
  64. var nodeCount int
  65. eachDataNode(topologyInfo, func(dc string, rack RackId, dn *master_pb.DataNodeInfo) {
  66. nodeCount++
  67. })
  68. if nodeCount < erasure_coding.ParityShardsCount {
  69. glog.V(0).Infof("skip erasure coding with %d nodes, less than recommended %d nodes", nodeCount, erasure_coding.ParityShardsCount)
  70. return nil
  71. }
  72. }
  73. vid := needle.VolumeId(*volumeId)
  74. // volumeId is provided
  75. if vid != 0 {
  76. return doEcEncode(commandEnv, *collection, vid, *parallelCopy)
  77. }
  78. // apply to all volumes in the collection
  79. volumeIds, err := collectVolumeIdsForEcEncode(commandEnv, *collection, *fullPercentage, *quietPeriod)
  80. if err != nil {
  81. return err
  82. }
  83. fmt.Printf("ec encode volumes: %v\n", volumeIds)
  84. for _, vid := range volumeIds {
  85. if err = doEcEncode(commandEnv, *collection, vid, *parallelCopy); err != nil {
  86. return err
  87. }
  88. }
  89. return nil
  90. }
  91. func doEcEncode(commandEnv *CommandEnv, collection string, vid needle.VolumeId, parallelCopy bool) (err error) {
  92. if !commandEnv.isLocked() {
  93. return fmt.Errorf("lock is lost")
  94. }
  95. // find volume location
  96. locations, found := commandEnv.MasterClient.GetLocationsClone(uint32(vid))
  97. if !found {
  98. return fmt.Errorf("volume %d not found", vid)
  99. }
  100. // fmt.Printf("found ec %d shards on %v\n", vid, locations)
  101. // mark the volume as readonly
  102. err = markVolumeReplicasWritable(commandEnv.option.GrpcDialOption, vid, locations, false)
  103. if err != nil {
  104. return fmt.Errorf("mark volume %d as readonly on %s: %v", vid, locations[0].Url, err)
  105. }
  106. // generate ec shards
  107. err = generateEcShards(commandEnv.option.GrpcDialOption, vid, collection, locations[0].ServerAddress())
  108. if err != nil {
  109. return fmt.Errorf("generate ec shards for volume %d on %s: %v", vid, locations[0].Url, err)
  110. }
  111. // balance the ec shards to current cluster
  112. err = spreadEcShards(commandEnv, vid, collection, locations, parallelCopy)
  113. if err != nil {
  114. return fmt.Errorf("spread ec shards for volume %d from %s: %v", vid, locations[0].Url, err)
  115. }
  116. return nil
  117. }
  118. func generateEcShards(grpcDialOption grpc.DialOption, volumeId needle.VolumeId, collection string, sourceVolumeServer pb.ServerAddress) error {
  119. fmt.Printf("generateEcShards %s %d on %s ...\n", collection, volumeId, sourceVolumeServer)
  120. err := operation.WithVolumeServerClient(false, sourceVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
  121. _, genErr := volumeServerClient.VolumeEcShardsGenerate(context.Background(), &volume_server_pb.VolumeEcShardsGenerateRequest{
  122. VolumeId: uint32(volumeId),
  123. Collection: collection,
  124. })
  125. return genErr
  126. })
  127. return err
  128. }
  129. func spreadEcShards(commandEnv *CommandEnv, volumeId needle.VolumeId, collection string, existingLocations []wdclient.Location, parallelCopy bool) (err error) {
  130. allEcNodes, totalFreeEcSlots, err := collectEcNodes(commandEnv, "")
  131. if err != nil {
  132. return err
  133. }
  134. if totalFreeEcSlots < erasure_coding.TotalShardsCount {
  135. return fmt.Errorf("not enough free ec shard slots. only %d left", totalFreeEcSlots)
  136. }
  137. allocatedDataNodes := allEcNodes
  138. if len(allocatedDataNodes) > erasure_coding.TotalShardsCount {
  139. allocatedDataNodes = allocatedDataNodes[:erasure_coding.TotalShardsCount]
  140. }
  141. // calculate how many shards to allocate for these servers
  142. allocatedEcIds := balancedEcDistribution(allocatedDataNodes)
  143. // ask the data nodes to copy from the source volume server
  144. copiedShardIds, err := parallelCopyEcShardsFromSource(commandEnv.option.GrpcDialOption, allocatedDataNodes, allocatedEcIds, volumeId, collection, existingLocations[0], parallelCopy)
  145. if err != nil {
  146. return err
  147. }
  148. // unmount the to be deleted shards
  149. err = unmountEcShards(commandEnv.option.GrpcDialOption, volumeId, existingLocations[0].ServerAddress(), copiedShardIds)
  150. if err != nil {
  151. return err
  152. }
  153. // ask the source volume server to clean up copied ec shards
  154. err = sourceServerDeleteEcShards(commandEnv.option.GrpcDialOption, collection, volumeId, existingLocations[0].ServerAddress(), copiedShardIds)
  155. if err != nil {
  156. return fmt.Errorf("source delete copied ecShards %s %d.%v: %v", existingLocations[0].Url, volumeId, copiedShardIds, err)
  157. }
  158. // ask the source volume server to delete the original volume
  159. for _, location := range existingLocations {
  160. fmt.Printf("delete volume %d from %s\n", volumeId, location.Url)
  161. err = deleteVolume(commandEnv.option.GrpcDialOption, volumeId, location.ServerAddress(), false)
  162. if err != nil {
  163. return fmt.Errorf("deleteVolume %s volume %d: %v", location.Url, volumeId, err)
  164. }
  165. }
  166. return err
  167. }
  168. func parallelCopyEcShardsFromSource(grpcDialOption grpc.DialOption, targetServers []*EcNode, allocatedEcIds [][]uint32, volumeId needle.VolumeId, collection string, existingLocation wdclient.Location, parallelCopy bool) (actuallyCopied []uint32, err error) {
  169. fmt.Printf("parallelCopyEcShardsFromSource %d %s\n", volumeId, existingLocation.Url)
  170. var wg sync.WaitGroup
  171. shardIdChan := make(chan []uint32, len(targetServers))
  172. copyFunc := func(server *EcNode, allocatedEcShardIds []uint32) {
  173. defer wg.Done()
  174. copiedShardIds, copyErr := oneServerCopyAndMountEcShardsFromSource(grpcDialOption, server,
  175. allocatedEcShardIds, volumeId, collection, existingLocation.ServerAddress())
  176. if copyErr != nil {
  177. err = copyErr
  178. } else {
  179. shardIdChan <- copiedShardIds
  180. server.addEcVolumeShards(volumeId, collection, copiedShardIds)
  181. }
  182. }
  183. cleanupFunc := func(server *EcNode, allocatedEcShardIds []uint32) {
  184. if err := unmountEcShards(grpcDialOption, volumeId, pb.NewServerAddressFromDataNode(server.info), allocatedEcShardIds); err != nil {
  185. fmt.Printf("unmount aborted shards %d.%v on %s: %v\n", volumeId, allocatedEcShardIds, server.info.Id, err)
  186. }
  187. if err := sourceServerDeleteEcShards(grpcDialOption, collection, volumeId, pb.NewServerAddressFromDataNode(server.info), allocatedEcShardIds); err != nil {
  188. fmt.Printf("remove aborted shards %d.%v on %s: %v\n", volumeId, allocatedEcShardIds, server.info.Id, err)
  189. }
  190. }
  191. // maybe parallelize
  192. for i, server := range targetServers {
  193. if len(allocatedEcIds[i]) <= 0 {
  194. continue
  195. }
  196. wg.Add(1)
  197. if parallelCopy {
  198. go copyFunc(server, allocatedEcIds[i])
  199. } else {
  200. copyFunc(server, allocatedEcIds[i])
  201. }
  202. }
  203. wg.Wait()
  204. close(shardIdChan)
  205. if err != nil {
  206. for i, server := range targetServers {
  207. if len(allocatedEcIds[i]) <= 0 {
  208. continue
  209. }
  210. cleanupFunc(server, allocatedEcIds[i])
  211. }
  212. return nil, err
  213. }
  214. for shardIds := range shardIdChan {
  215. actuallyCopied = append(actuallyCopied, shardIds...)
  216. }
  217. return
  218. }
  219. func balancedEcDistribution(servers []*EcNode) (allocated [][]uint32) {
  220. allocated = make([][]uint32, len(servers))
  221. allocatedShardIdIndex := uint32(0)
  222. serverIndex := rand.Intn(len(servers))
  223. for allocatedShardIdIndex < erasure_coding.TotalShardsCount {
  224. if servers[serverIndex].freeEcSlot > 0 {
  225. allocated[serverIndex] = append(allocated[serverIndex], allocatedShardIdIndex)
  226. allocatedShardIdIndex++
  227. }
  228. serverIndex++
  229. if serverIndex >= len(servers) {
  230. serverIndex = 0
  231. }
  232. }
  233. return allocated
  234. }
  235. func collectVolumeIdsForEcEncode(commandEnv *CommandEnv, selectedCollection string, fullPercentage float64, quietPeriod time.Duration) (vids []needle.VolumeId, err error) {
  236. // collect topology information
  237. topologyInfo, volumeSizeLimitMb, err := collectTopologyInfo(commandEnv, 0)
  238. if err != nil {
  239. return
  240. }
  241. quietSeconds := int64(quietPeriod / time.Second)
  242. nowUnixSeconds := time.Now().Unix()
  243. fmt.Printf("collect volumes quiet for: %d seconds and %.1f%% full\n", quietSeconds, fullPercentage)
  244. vidMap := make(map[uint32]bool)
  245. eachDataNode(topologyInfo, func(dc string, rack RackId, dn *master_pb.DataNodeInfo) {
  246. for _, diskInfo := range dn.DiskInfos {
  247. for _, v := range diskInfo.VolumeInfos {
  248. if v.Collection == selectedCollection && v.ModifiedAtSecond+quietSeconds < nowUnixSeconds {
  249. if float64(v.Size) > fullPercentage/100*float64(volumeSizeLimitMb)*1024*1024 {
  250. vidMap[v.Id] = true
  251. }
  252. }
  253. }
  254. }
  255. })
  256. for vid := range vidMap {
  257. vids = append(vids, needle.VolumeId(vid))
  258. }
  259. return
  260. }