command_ec_decode.go 8.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275
  1. package shell
  2. import (
  3. "context"
  4. "flag"
  5. "fmt"
  6. "github.com/chrislusf/seaweedfs/weed/storage/types"
  7. "io"
  8. "google.golang.org/grpc"
  9. "github.com/chrislusf/seaweedfs/weed/operation"
  10. "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
  11. "github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
  12. "github.com/chrislusf/seaweedfs/weed/storage/erasure_coding"
  13. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  14. )
  15. func init() {
  16. Commands = append(Commands, &commandEcDecode{})
  17. }
  18. type commandEcDecode struct {
  19. }
  20. func (c *commandEcDecode) Name() string {
  21. return "ec.decode"
  22. }
  23. func (c *commandEcDecode) Help() string {
  24. return `decode a erasure coded volume into a normal volume
  25. ec.decode [-collection=""] [-volumeId=<volume_id>]
  26. `
  27. }
  28. func (c *commandEcDecode) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
  29. if err = commandEnv.confirmIsLocked(); err != nil {
  30. return
  31. }
  32. encodeCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
  33. volumeId := encodeCommand.Int("volumeId", 0, "the volume id")
  34. collection := encodeCommand.String("collection", "", "the collection name")
  35. if err = encodeCommand.Parse(args); err != nil {
  36. return nil
  37. }
  38. vid := needle.VolumeId(*volumeId)
  39. // collect topology information
  40. topologyInfo, _, err := collectTopologyInfo(commandEnv)
  41. if err != nil {
  42. return err
  43. }
  44. // volumeId is provided
  45. if vid != 0 {
  46. return doEcDecode(commandEnv, topologyInfo, *collection, vid)
  47. }
  48. // apply to all volumes in the collection
  49. volumeIds := collectEcShardIds(topologyInfo, *collection)
  50. fmt.Printf("ec encode volumes: %v\n", volumeIds)
  51. for _, vid := range volumeIds {
  52. if err = doEcDecode(commandEnv, topologyInfo, *collection, vid); err != nil {
  53. return err
  54. }
  55. }
  56. return nil
  57. }
  58. func doEcDecode(commandEnv *CommandEnv, topoInfo *master_pb.TopologyInfo, collection string, vid needle.VolumeId) (err error) {
  59. // find volume location
  60. nodeToEcIndexBits := collectEcNodeShardBits(topoInfo, vid)
  61. fmt.Printf("ec volume %d shard locations: %+v\n", vid, nodeToEcIndexBits)
  62. // collect ec shards to the server with most space
  63. targetNodeLocation, err := collectEcShards(commandEnv, nodeToEcIndexBits, collection, vid)
  64. if err != nil {
  65. return fmt.Errorf("collectEcShards for volume %d: %v", vid, err)
  66. }
  67. // generate a normal volume
  68. err = generateNormalVolume(commandEnv.option.GrpcDialOption, vid, collection, targetNodeLocation)
  69. if err != nil {
  70. return fmt.Errorf("generate normal volume %d on %s: %v", vid, targetNodeLocation, err)
  71. }
  72. // delete the previous ec shards
  73. err = mountVolumeAndDeleteEcShards(commandEnv.option.GrpcDialOption, collection, targetNodeLocation, nodeToEcIndexBits, vid)
  74. if err != nil {
  75. return fmt.Errorf("delete ec shards for volume %d: %v", vid, err)
  76. }
  77. return nil
  78. }
  79. func mountVolumeAndDeleteEcShards(grpcDialOption grpc.DialOption, collection, targetNodeLocation string, nodeToEcIndexBits map[string]erasure_coding.ShardBits, vid needle.VolumeId) error {
  80. // mount volume
  81. if err := operation.WithVolumeServerClient(targetNodeLocation, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
  82. _, mountErr := volumeServerClient.VolumeMount(context.Background(), &volume_server_pb.VolumeMountRequest{
  83. VolumeId: uint32(vid),
  84. })
  85. return mountErr
  86. }); err != nil {
  87. return fmt.Errorf("mountVolumeAndDeleteEcShards mount volume %d on %s: %v", vid, targetNodeLocation, err)
  88. }
  89. // unmount ec shards
  90. for location, ecIndexBits := range nodeToEcIndexBits {
  91. fmt.Printf("unmount ec volume %d on %s has shards: %+v\n", vid, location, ecIndexBits.ShardIds())
  92. err := unmountEcShards(grpcDialOption, vid, location, ecIndexBits.ToUint32Slice())
  93. if err != nil {
  94. return fmt.Errorf("mountVolumeAndDeleteEcShards unmount ec volume %d on %s: %v", vid, location, err)
  95. }
  96. }
  97. // delete ec shards
  98. for location, ecIndexBits := range nodeToEcIndexBits {
  99. fmt.Printf("delete ec volume %d on %s has shards: %+v\n", vid, location, ecIndexBits.ShardIds())
  100. err := sourceServerDeleteEcShards(grpcDialOption, collection, vid, location, ecIndexBits.ToUint32Slice())
  101. if err != nil {
  102. return fmt.Errorf("mountVolumeAndDeleteEcShards delete ec volume %d on %s: %v", vid, location, err)
  103. }
  104. }
  105. return nil
  106. }
  107. func generateNormalVolume(grpcDialOption grpc.DialOption, vid needle.VolumeId, collection string, sourceVolumeServer string) error {
  108. fmt.Printf("generateNormalVolume from ec volume %d on %s\n", vid, sourceVolumeServer)
  109. err := operation.WithVolumeServerClient(sourceVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
  110. _, genErr := volumeServerClient.VolumeEcShardsToVolume(context.Background(), &volume_server_pb.VolumeEcShardsToVolumeRequest{
  111. VolumeId: uint32(vid),
  112. Collection: collection,
  113. })
  114. return genErr
  115. })
  116. return err
  117. }
  118. func collectEcShards(commandEnv *CommandEnv, nodeToEcIndexBits map[string]erasure_coding.ShardBits, collection string, vid needle.VolumeId) (targetNodeLocation string, err error) {
  119. maxShardCount := 0
  120. var exisitngEcIndexBits erasure_coding.ShardBits
  121. for loc, ecIndexBits := range nodeToEcIndexBits {
  122. toBeCopiedShardCount := ecIndexBits.MinusParityShards().ShardIdCount()
  123. if toBeCopiedShardCount > maxShardCount {
  124. maxShardCount = toBeCopiedShardCount
  125. targetNodeLocation = loc
  126. exisitngEcIndexBits = ecIndexBits
  127. }
  128. }
  129. fmt.Printf("collectEcShards: ec volume %d collect shards to %s from: %+v\n", vid, targetNodeLocation, nodeToEcIndexBits)
  130. var copiedEcIndexBits erasure_coding.ShardBits
  131. for loc, ecIndexBits := range nodeToEcIndexBits {
  132. if loc == targetNodeLocation {
  133. continue
  134. }
  135. needToCopyEcIndexBits := ecIndexBits.Minus(exisitngEcIndexBits).MinusParityShards()
  136. if needToCopyEcIndexBits.ShardIdCount() == 0 {
  137. continue
  138. }
  139. err = operation.WithVolumeServerClient(targetNodeLocation, commandEnv.option.GrpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
  140. fmt.Printf("copy %d.%v %s => %s\n", vid, needToCopyEcIndexBits.ShardIds(), loc, targetNodeLocation)
  141. _, copyErr := volumeServerClient.VolumeEcShardsCopy(context.Background(), &volume_server_pb.VolumeEcShardsCopyRequest{
  142. VolumeId: uint32(vid),
  143. Collection: collection,
  144. ShardIds: needToCopyEcIndexBits.ToUint32Slice(),
  145. CopyEcxFile: false,
  146. CopyEcjFile: true,
  147. CopyVifFile: true,
  148. SourceDataNode: loc,
  149. })
  150. if copyErr != nil {
  151. return fmt.Errorf("copy %d.%v %s => %s : %v\n", vid, needToCopyEcIndexBits.ShardIds(), loc, targetNodeLocation, copyErr)
  152. }
  153. return nil
  154. })
  155. if err != nil {
  156. break
  157. }
  158. copiedEcIndexBits = copiedEcIndexBits.Plus(needToCopyEcIndexBits)
  159. }
  160. nodeToEcIndexBits[targetNodeLocation] = exisitngEcIndexBits.Plus(copiedEcIndexBits)
  161. return targetNodeLocation, err
  162. }
  163. func collectTopologyInfo(commandEnv *CommandEnv) (topoInfo *master_pb.TopologyInfo, volumeSizeLimitMb uint64, err error) {
  164. var resp *master_pb.VolumeListResponse
  165. err = commandEnv.MasterClient.WithClient(func(client master_pb.SeaweedClient) error {
  166. resp, err = client.VolumeList(context.Background(), &master_pb.VolumeListRequest{})
  167. return err
  168. })
  169. if err != nil {
  170. return
  171. }
  172. return resp.TopologyInfo, resp.VolumeSizeLimitMb, nil
  173. }
  174. func collectEcShardInfos(topoInfo *master_pb.TopologyInfo, selectedCollection string, vid needle.VolumeId) (ecShardInfos []*master_pb.VolumeEcShardInformationMessage) {
  175. eachDataNode(topoInfo, func(dc string, rack RackId, dn *master_pb.DataNodeInfo) {
  176. if diskInfo, found := dn.DiskInfos[string(types.HardDriveType)]; found {
  177. for _, v := range diskInfo.EcShardInfos {
  178. if v.Collection == selectedCollection && v.Id == uint32(vid) {
  179. ecShardInfos = append(ecShardInfos, v)
  180. }
  181. }
  182. }
  183. })
  184. return
  185. }
  186. func collectEcShardIds(topoInfo *master_pb.TopologyInfo, selectedCollection string) (vids []needle.VolumeId) {
  187. vidMap := make(map[uint32]bool)
  188. eachDataNode(topoInfo, func(dc string, rack RackId, dn *master_pb.DataNodeInfo) {
  189. if diskInfo, found := dn.DiskInfos[string(types.HardDriveType)]; found {
  190. for _, v := range diskInfo.EcShardInfos {
  191. if v.Collection == selectedCollection {
  192. vidMap[v.Id] = true
  193. }
  194. }
  195. }
  196. })
  197. for vid := range vidMap {
  198. vids = append(vids, needle.VolumeId(vid))
  199. }
  200. return
  201. }
  202. func collectEcNodeShardBits(topoInfo *master_pb.TopologyInfo, vid needle.VolumeId) map[string]erasure_coding.ShardBits {
  203. nodeToEcIndexBits := make(map[string]erasure_coding.ShardBits)
  204. eachDataNode(topoInfo, func(dc string, rack RackId, dn *master_pb.DataNodeInfo) {
  205. if diskInfo, found := dn.DiskInfos[string(types.HardDriveType)]; found {
  206. for _, v := range diskInfo.EcShardInfos {
  207. if v.Id == uint32(vid) {
  208. nodeToEcIndexBits[dn.Id] = erasure_coding.ShardBits(v.EcIndexBits)
  209. }
  210. }
  211. }
  212. })
  213. return nodeToEcIndexBits
  214. }