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.

549 lines
17 KiB

3 years ago
6 years ago
4 years ago
6 years ago
3 years ago
4 years ago
3 years ago
3 years ago
4 years ago
4 years ago
4 years ago
3 years ago
4 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
4 years ago
4 years ago
4 years ago
  1. package shell
  2. import (
  3. "context"
  4. "flag"
  5. "fmt"
  6. "github.com/chrislusf/seaweedfs/weed/pb"
  7. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  8. "github.com/chrislusf/seaweedfs/weed/storage/types"
  9. "io"
  10. "path/filepath"
  11. "sort"
  12. "strconv"
  13. "time"
  14. "github.com/chrislusf/seaweedfs/weed/operation"
  15. "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
  16. "github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
  17. "github.com/chrislusf/seaweedfs/weed/storage/super_block"
  18. )
  19. func init() {
  20. Commands = append(Commands, &commandVolumeFixReplication{})
  21. }
  22. type commandVolumeFixReplication struct {
  23. collectionPattern *string
  24. }
  25. func (c *commandVolumeFixReplication) Name() string {
  26. return "volume.fix.replication"
  27. }
  28. func (c *commandVolumeFixReplication) Help() string {
  29. return `add or remove replicas to volumes that are missing replicas or over-replicated
  30. This command finds all over-replicated volumes. If found, it will purge the oldest copies and stop.
  31. This command also finds all under-replicated volumes, and finds volume servers with free slots.
  32. If the free slots satisfy the replication requirement, the volume content is copied over and mounted.
  33. volume.fix.replication -n # do not take action
  34. volume.fix.replication # actually deleting or copying the volume files and mount the volume
  35. volume.fix.replication -collectionPattern=important* # fix any collections with prefix "important"
  36. Note:
  37. * each time this will only add back one replica for each volume id that is under replicated.
  38. If there are multiple replicas are missing, e.g. replica count is > 2, you may need to run this multiple times.
  39. * do not run this too quickly within seconds, since the new volume replica may take a few seconds
  40. to register itself to the master.
  41. `
  42. }
  43. func (c *commandVolumeFixReplication) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
  44. volFixReplicationCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
  45. c.collectionPattern = volFixReplicationCommand.String("collectionPattern", "", "match with wildcard characters '*' and '?'")
  46. skipChange := volFixReplicationCommand.Bool("n", false, "skip the changes")
  47. retryCount := volFixReplicationCommand.Int("retry", 0, "how many times to retry")
  48. volumesPerStep := volFixReplicationCommand.Int("volumesPerStep", 0, "how many volumes to fix in one cycle")
  49. if err = volFixReplicationCommand.Parse(args); err != nil {
  50. return nil
  51. }
  52. if err = commandEnv.confirmIsLocked(args); err != nil {
  53. return
  54. }
  55. takeAction := !*skipChange
  56. underReplicatedVolumeIdsCount := 1
  57. for underReplicatedVolumeIdsCount > 0 {
  58. fixedVolumeReplicas := map[string]int{}
  59. // collect topology information
  60. topologyInfo, _, err := collectTopologyInfo(commandEnv, 15*time.Second)
  61. if err != nil {
  62. return err
  63. }
  64. // find all volumes that needs replication
  65. // collect all data nodes
  66. volumeReplicas, allLocations := collectVolumeReplicaLocations(topologyInfo)
  67. if len(allLocations) == 0 {
  68. return fmt.Errorf("no data nodes at all")
  69. }
  70. // find all under replicated volumes
  71. var underReplicatedVolumeIds, overReplicatedVolumeIds, misplacedVolumeIds []uint32
  72. for vid, replicas := range volumeReplicas {
  73. replica := replicas[0]
  74. replicaPlacement, _ := super_block.NewReplicaPlacementFromByte(byte(replica.info.ReplicaPlacement))
  75. if replicaPlacement.GetCopyCount() > len(replicas) {
  76. underReplicatedVolumeIds = append(underReplicatedVolumeIds, vid)
  77. } else if replicaPlacement.GetCopyCount() < len(replicas) {
  78. overReplicatedVolumeIds = append(overReplicatedVolumeIds, vid)
  79. fmt.Fprintf(writer, "volume %d replication %s, but over replicated %+d\n", replica.info.Id, replicaPlacement, len(replicas))
  80. } else if isMisplaced(replicas, replicaPlacement) {
  81. misplacedVolumeIds = append(misplacedVolumeIds, vid)
  82. fmt.Fprintf(writer, "volume %d replication %s is not well placed %+v\n", replica.info.Id, replicaPlacement, replicas)
  83. }
  84. }
  85. if len(overReplicatedVolumeIds) > 0 {
  86. if err := c.deleteOneVolume(commandEnv, writer, takeAction, overReplicatedVolumeIds, volumeReplicas, allLocations, pickOneReplicaToDelete); err != nil {
  87. return err
  88. }
  89. }
  90. if len(misplacedVolumeIds) > 0 {
  91. if err := c.deleteOneVolume(commandEnv, writer, takeAction, misplacedVolumeIds, volumeReplicas, allLocations, pickOneMisplacedVolume); err != nil {
  92. return err
  93. }
  94. }
  95. underReplicatedVolumeIdsCount = len(underReplicatedVolumeIds)
  96. if underReplicatedVolumeIdsCount > 0 {
  97. // find the most under populated data nodes
  98. fixedVolumeReplicas, err = c.fixUnderReplicatedVolumes(commandEnv, writer, takeAction, underReplicatedVolumeIds, volumeReplicas, allLocations, *retryCount, *volumesPerStep)
  99. if err != nil {
  100. return err
  101. }
  102. }
  103. if *skipChange {
  104. break
  105. }
  106. // check that the topology has been updated
  107. if len(fixedVolumeReplicas) > 0 {
  108. fixedVolumes := make([]string, 0, len(fixedVolumeReplicas))
  109. for k, _ := range fixedVolumeReplicas {
  110. fixedVolumes = append(fixedVolumes, k)
  111. }
  112. volumeIdLocations, err := lookupVolumeIds(commandEnv, fixedVolumes)
  113. if err != nil {
  114. return err
  115. }
  116. for _, volumeIdLocation := range volumeIdLocations {
  117. volumeId := volumeIdLocation.VolumeOrFileId
  118. volumeIdLocationCount := len(volumeIdLocation.Locations)
  119. i := 0
  120. for fixedVolumeReplicas[volumeId] >= volumeIdLocationCount {
  121. fmt.Fprintf(writer, "the number of locations for volume %s has not increased yet, let's wait\n", volumeId)
  122. time.Sleep(time.Duration(i+1) * time.Second * 7)
  123. volumeLocIds, err := lookupVolumeIds(commandEnv, []string{volumeId})
  124. if err != nil {
  125. return err
  126. }
  127. volumeIdLocationCount = len(volumeLocIds[0].Locations)
  128. if *retryCount > i {
  129. return fmt.Errorf("replicas volume %s mismatch in topology", volumeId)
  130. }
  131. i += 1
  132. }
  133. }
  134. }
  135. }
  136. return nil
  137. }
  138. func collectVolumeReplicaLocations(topologyInfo *master_pb.TopologyInfo) (map[uint32][]*VolumeReplica, []location) {
  139. volumeReplicas := make(map[uint32][]*VolumeReplica)
  140. var allLocations []location
  141. eachDataNode(topologyInfo, func(dc string, rack RackId, dn *master_pb.DataNodeInfo) {
  142. loc := newLocation(dc, string(rack), dn)
  143. for _, diskInfo := range dn.DiskInfos {
  144. for _, v := range diskInfo.VolumeInfos {
  145. volumeReplicas[v.Id] = append(volumeReplicas[v.Id], &VolumeReplica{
  146. location: &loc,
  147. info: v,
  148. })
  149. }
  150. }
  151. allLocations = append(allLocations, loc)
  152. })
  153. return volumeReplicas, allLocations
  154. }
  155. type SelectOneVolumeFunc func(replicas []*VolumeReplica, replicaPlacement *super_block.ReplicaPlacement) *VolumeReplica
  156. func (c *commandVolumeFixReplication) deleteOneVolume(commandEnv *CommandEnv, writer io.Writer, takeAction bool, overReplicatedVolumeIds []uint32, volumeReplicas map[uint32][]*VolumeReplica, allLocations []location, selectOneVolumeFn SelectOneVolumeFunc) error {
  157. for _, vid := range overReplicatedVolumeIds {
  158. replicas := volumeReplicas[vid]
  159. replicaPlacement, _ := super_block.NewReplicaPlacementFromByte(byte(replicas[0].info.ReplicaPlacement))
  160. replica := selectOneVolumeFn(replicas, replicaPlacement)
  161. // check collection name pattern
  162. if *c.collectionPattern != "" {
  163. matched, err := filepath.Match(*c.collectionPattern, replica.info.Collection)
  164. if err != nil {
  165. return fmt.Errorf("match pattern %s with collection %s: %v", *c.collectionPattern, replica.info.Collection, err)
  166. }
  167. if !matched {
  168. break
  169. }
  170. }
  171. fmt.Fprintf(writer, "deleting volume %d from %s ...\n", replica.info.Id, replica.location.dataNode.Id)
  172. if !takeAction {
  173. break
  174. }
  175. if err := deleteVolume(commandEnv.option.GrpcDialOption, needle.VolumeId(replica.info.Id), pb.NewServerAddressFromDataNode(replica.location.dataNode)); err != nil {
  176. return fmt.Errorf("deleting volume %d from %s : %v", replica.info.Id, replica.location.dataNode.Id, err)
  177. }
  178. }
  179. return nil
  180. }
  181. func (c *commandVolumeFixReplication) fixUnderReplicatedVolumes(commandEnv *CommandEnv, writer io.Writer, takeAction bool, underReplicatedVolumeIds []uint32, volumeReplicas map[uint32][]*VolumeReplica, allLocations []location, retryCount int, volumesPerStep int) (fixedVolumes map[string]int, err error) {
  182. fixedVolumes = map[string]int{}
  183. if len(underReplicatedVolumeIds) > volumesPerStep && volumesPerStep > 0 {
  184. underReplicatedVolumeIds = underReplicatedVolumeIds[0:volumesPerStep]
  185. }
  186. for _, vid := range underReplicatedVolumeIds {
  187. for i := 0; i < retryCount+1; i++ {
  188. if err = c.fixOneUnderReplicatedVolume(commandEnv, writer, takeAction, volumeReplicas, vid, allLocations); err == nil {
  189. if takeAction {
  190. fixedVolumes[strconv.FormatUint(uint64(vid), 10)] = len(volumeReplicas[vid])
  191. }
  192. break
  193. }
  194. }
  195. }
  196. return fixedVolumes, nil
  197. }
  198. func (c *commandVolumeFixReplication) fixOneUnderReplicatedVolume(commandEnv *CommandEnv, writer io.Writer, takeAction bool, volumeReplicas map[uint32][]*VolumeReplica, vid uint32, allLocations []location) error {
  199. replicas := volumeReplicas[vid]
  200. replica := pickOneReplicaToCopyFrom(replicas)
  201. replicaPlacement, _ := super_block.NewReplicaPlacementFromByte(byte(replica.info.ReplicaPlacement))
  202. foundNewLocation := false
  203. hasSkippedCollection := false
  204. keepDataNodesSorted(allLocations, types.ToDiskType(replica.info.DiskType))
  205. fn := capacityByFreeVolumeCount(types.ToDiskType(replica.info.DiskType))
  206. for _, dst := range allLocations {
  207. // check whether data nodes satisfy the constraints
  208. if fn(dst.dataNode) > 0 && satisfyReplicaPlacement(replicaPlacement, replicas, dst) {
  209. // check collection name pattern
  210. if *c.collectionPattern != "" {
  211. matched, err := filepath.Match(*c.collectionPattern, replica.info.Collection)
  212. if err != nil {
  213. return fmt.Errorf("match pattern %s with collection %s: %v", *c.collectionPattern, replica.info.Collection, err)
  214. }
  215. if !matched {
  216. hasSkippedCollection = true
  217. break
  218. }
  219. }
  220. // ask the volume server to replicate the volume
  221. foundNewLocation = true
  222. fmt.Fprintf(writer, "replicating volume %d %s from %s to dataNode %s ...\n", replica.info.Id, replicaPlacement, replica.location.dataNode.Id, dst.dataNode.Id)
  223. if !takeAction {
  224. // adjust free volume count
  225. dst.dataNode.DiskInfos[replica.info.DiskType].FreeVolumeCount--
  226. break
  227. }
  228. err := operation.WithVolumeServerClient(false, pb.NewServerAddressFromDataNode(dst.dataNode), commandEnv.option.GrpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
  229. stream, replicateErr := volumeServerClient.VolumeCopy(context.Background(), &volume_server_pb.VolumeCopyRequest{
  230. VolumeId: replica.info.Id,
  231. SourceDataNode: string(pb.NewServerAddressFromDataNode(replica.location.dataNode)),
  232. })
  233. if replicateErr != nil {
  234. return fmt.Errorf("copying from %s => %s : %v", replica.location.dataNode.Id, dst.dataNode.Id, replicateErr)
  235. }
  236. for {
  237. resp, recvErr := stream.Recv()
  238. if recvErr != nil {
  239. if recvErr == io.EOF {
  240. break
  241. } else {
  242. return recvErr
  243. }
  244. }
  245. if resp.ProcessedBytes > 0 {
  246. fmt.Fprintf(writer, "volume %d processed %d bytes\n", replica.info.Id, resp.ProcessedBytes)
  247. }
  248. }
  249. return nil
  250. })
  251. if err != nil {
  252. return err
  253. }
  254. // adjust free volume count
  255. dst.dataNode.DiskInfos[replica.info.DiskType].FreeVolumeCount--
  256. break
  257. }
  258. }
  259. if !foundNewLocation && !hasSkippedCollection {
  260. fmt.Fprintf(writer, "failed to place volume %d replica as %s, existing:%+v\n", replica.info.Id, replicaPlacement, len(replicas))
  261. }
  262. return nil
  263. }
  264. func keepDataNodesSorted(dataNodes []location, diskType types.DiskType) {
  265. fn := capacityByFreeVolumeCount(diskType)
  266. sort.Slice(dataNodes, func(i, j int) bool {
  267. return fn(dataNodes[i].dataNode) > fn(dataNodes[j].dataNode)
  268. })
  269. }
  270. /*
  271. if on an existing data node {
  272. return false
  273. }
  274. if different from existing dcs {
  275. if lack on different dcs {
  276. return true
  277. }else{
  278. return false
  279. }
  280. }
  281. if not on primary dc {
  282. return false
  283. }
  284. if different from existing racks {
  285. if lack on different racks {
  286. return true
  287. }else{
  288. return false
  289. }
  290. }
  291. if not on primary rack {
  292. return false
  293. }
  294. if lacks on same rack {
  295. return true
  296. } else {
  297. return false
  298. }
  299. */
  300. func satisfyReplicaPlacement(replicaPlacement *super_block.ReplicaPlacement, replicas []*VolumeReplica, possibleLocation location) bool {
  301. existingDataCenters, _, existingDataNodes := countReplicas(replicas)
  302. if _, found := existingDataNodes[possibleLocation.String()]; found {
  303. // avoid duplicated volume on the same data node
  304. return false
  305. }
  306. primaryDataCenters, _ := findTopKeys(existingDataCenters)
  307. // ensure data center count is within limit
  308. if _, found := existingDataCenters[possibleLocation.DataCenter()]; !found {
  309. // different from existing dcs
  310. if len(existingDataCenters) < replicaPlacement.DiffDataCenterCount+1 {
  311. // lack on different dcs
  312. return true
  313. } else {
  314. // adding this would go over the different dcs limit
  315. return false
  316. }
  317. }
  318. // now this is same as one of the existing data center
  319. if !isAmong(possibleLocation.DataCenter(), primaryDataCenters) {
  320. // not on one of the primary dcs
  321. return false
  322. }
  323. // now this is one of the primary dcs
  324. primaryDcRacks := make(map[string]int)
  325. for _, replica := range replicas {
  326. if replica.location.DataCenter() != possibleLocation.DataCenter() {
  327. continue
  328. }
  329. primaryDcRacks[replica.location.Rack()] += 1
  330. }
  331. primaryRacks, _ := findTopKeys(primaryDcRacks)
  332. sameRackCount := primaryDcRacks[possibleLocation.Rack()]
  333. // ensure rack count is within limit
  334. if _, found := primaryDcRacks[possibleLocation.Rack()]; !found {
  335. // different from existing racks
  336. if len(primaryDcRacks) < replicaPlacement.DiffRackCount+1 {
  337. // lack on different racks
  338. return true
  339. } else {
  340. // adding this would go over the different racks limit
  341. return false
  342. }
  343. }
  344. // now this is same as one of the existing racks
  345. if !isAmong(possibleLocation.Rack(), primaryRacks) {
  346. // not on the primary rack
  347. return false
  348. }
  349. // now this is on the primary rack
  350. // different from existing data nodes
  351. if sameRackCount < replicaPlacement.SameRackCount+1 {
  352. // lack on same rack
  353. return true
  354. } else {
  355. // adding this would go over the same data node limit
  356. return false
  357. }
  358. }
  359. func findTopKeys(m map[string]int) (topKeys []string, max int) {
  360. for k, c := range m {
  361. if max < c {
  362. topKeys = topKeys[:0]
  363. topKeys = append(topKeys, k)
  364. max = c
  365. } else if max == c {
  366. topKeys = append(topKeys, k)
  367. }
  368. }
  369. return
  370. }
  371. func isAmong(key string, keys []string) bool {
  372. for _, k := range keys {
  373. if k == key {
  374. return true
  375. }
  376. }
  377. return false
  378. }
  379. type VolumeReplica struct {
  380. location *location
  381. info *master_pb.VolumeInformationMessage
  382. }
  383. type location struct {
  384. dc string
  385. rack string
  386. dataNode *master_pb.DataNodeInfo
  387. }
  388. func newLocation(dc, rack string, dataNode *master_pb.DataNodeInfo) location {
  389. return location{
  390. dc: dc,
  391. rack: rack,
  392. dataNode: dataNode,
  393. }
  394. }
  395. func (l location) String() string {
  396. return fmt.Sprintf("%s %s %s", l.dc, l.rack, l.dataNode.Id)
  397. }
  398. func (l location) Rack() string {
  399. return fmt.Sprintf("%s %s", l.dc, l.rack)
  400. }
  401. func (l location) DataCenter() string {
  402. return l.dc
  403. }
  404. func pickOneReplicaToCopyFrom(replicas []*VolumeReplica) *VolumeReplica {
  405. mostRecent := replicas[0]
  406. for _, replica := range replicas {
  407. if replica.info.ModifiedAtSecond > mostRecent.info.ModifiedAtSecond {
  408. mostRecent = replica
  409. }
  410. }
  411. return mostRecent
  412. }
  413. func countReplicas(replicas []*VolumeReplica) (diffDc, diffRack, diffNode map[string]int) {
  414. diffDc = make(map[string]int)
  415. diffRack = make(map[string]int)
  416. diffNode = make(map[string]int)
  417. for _, replica := range replicas {
  418. diffDc[replica.location.DataCenter()] += 1
  419. diffRack[replica.location.Rack()] += 1
  420. diffNode[replica.location.String()] += 1
  421. }
  422. return
  423. }
  424. func pickOneReplicaToDelete(replicas []*VolumeReplica, replicaPlacement *super_block.ReplicaPlacement) *VolumeReplica {
  425. sort.Slice(replicas, func(i, j int) bool {
  426. a, b := replicas[i], replicas[j]
  427. if a.info.Size != b.info.Size {
  428. return a.info.Size < b.info.Size
  429. }
  430. if a.info.ModifiedAtSecond != b.info.ModifiedAtSecond {
  431. return a.info.ModifiedAtSecond < b.info.ModifiedAtSecond
  432. }
  433. if a.info.CompactRevision != b.info.CompactRevision {
  434. return a.info.CompactRevision < b.info.CompactRevision
  435. }
  436. return false
  437. })
  438. return replicas[0]
  439. }
  440. // check and fix misplaced volumes
  441. func isMisplaced(replicas []*VolumeReplica, replicaPlacement *super_block.ReplicaPlacement) bool {
  442. for i := 0; i < len(replicas); i++ {
  443. others := otherThan(replicas, i)
  444. if satisfyReplicaPlacement(replicaPlacement, others, *replicas[i].location) {
  445. return false
  446. }
  447. }
  448. return true
  449. }
  450. func otherThan(replicas []*VolumeReplica, index int) (others []*VolumeReplica) {
  451. for i := 0; i < len(replicas); i++ {
  452. if index != i {
  453. others = append(others, replicas[i])
  454. }
  455. }
  456. return
  457. }
  458. func pickOneMisplacedVolume(replicas []*VolumeReplica, replicaPlacement *super_block.ReplicaPlacement) (toDelete *VolumeReplica) {
  459. var deletionCandidates []*VolumeReplica
  460. for i := 0; i < len(replicas); i++ {
  461. others := otherThan(replicas, i)
  462. if !isMisplaced(others, replicaPlacement) {
  463. deletionCandidates = append(deletionCandidates, replicas[i])
  464. }
  465. }
  466. if len(deletionCandidates) > 0 {
  467. return pickOneReplicaToDelete(deletionCandidates, replicaPlacement)
  468. }
  469. return pickOneReplicaToDelete(replicas, replicaPlacement)
  470. }