command_ec_decode.go 9.4 KB

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