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.

270 lines
8.5 KiB

3 years ago
4 years ago
3 years ago
4 years ago
3 years ago
  1. package shell
  2. import (
  3. "bytes"
  4. "context"
  5. "flag"
  6. "fmt"
  7. "github.com/chrislusf/seaweedfs/weed/operation"
  8. "github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
  9. "github.com/chrislusf/seaweedfs/weed/storage/needle_map"
  10. "io"
  11. "math"
  12. "sort"
  13. )
  14. func init() {
  15. Commands = append(Commands, &commandVolumeCheckDisk{})
  16. }
  17. type commandVolumeCheckDisk struct {
  18. env *CommandEnv
  19. }
  20. func (c *commandVolumeCheckDisk) Name() string {
  21. return "volume.check.disk"
  22. }
  23. func (c *commandVolumeCheckDisk) Help() string {
  24. return `check all replicated volumes to find and fix inconsistencies
  25. How it works:
  26. find all volumes that are replicated
  27. for each volume id, if there are more than 2 replicas, find one pair with the largest 2 in file count.
  28. for the pair volume A and B
  29. append entries in A and not in B to B
  30. append entries in B and not in A to A
  31. `
  32. }
  33. func (c *commandVolumeCheckDisk) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
  34. if err = commandEnv.confirmIsLocked(); err != nil {
  35. return
  36. }
  37. fsckCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
  38. slowMode := fsckCommand.Bool("slow", false, "slow mode checks all replicas even file counts are the same")
  39. verbose := fsckCommand.Bool("v", false, "verbose mode")
  40. applyChanges := fsckCommand.Bool("force", false, "apply the fix")
  41. nonRepairThreshold := fsckCommand.Float64("nonRepairThreshold", 0.3, "repair when missing keys is not more than this limit")
  42. if err = fsckCommand.Parse(args); err != nil {
  43. return nil
  44. }
  45. c.env = commandEnv
  46. // collect topology information
  47. topologyInfo, _, err := collectTopologyInfo(commandEnv)
  48. if err != nil {
  49. return err
  50. }
  51. volumeReplicas, _ := collectVolumeReplicaLocations(topologyInfo)
  52. // pick 1 pairs of volume replica
  53. fileCount := func(replica *VolumeReplica) uint64 {
  54. return replica.info.FileCount - replica.info.DeleteCount
  55. }
  56. aDB, bDB := needle_map.NewMemDb(), needle_map.NewMemDb()
  57. defer aDB.Close()
  58. defer bDB.Close()
  59. for _, replicas := range volumeReplicas {
  60. sort.Slice(replicas, func(i, j int) bool {
  61. return fileCount(replicas[i]) > fileCount(replicas[j])
  62. })
  63. for len(replicas) >= 2 {
  64. a, b := replicas[0], replicas[1]
  65. if !*slowMode {
  66. if fileCount(a) == fileCount(b) {
  67. replicas = replicas[1:]
  68. continue
  69. }
  70. }
  71. if a.info.ReadOnly || b.info.ReadOnly {
  72. fmt.Fprintf(writer, "skipping readonly volume %d on %s and %s\n", a.info.Id, a.location.dataNode.Id, b.location.dataNode.Id)
  73. replicas = replicas[1:]
  74. continue
  75. }
  76. if err := c.syncTwoReplicas(aDB, bDB, a, verbose, writer, b, err, applyChanges, nonRepairThreshold); err != nil {
  77. fmt.Fprintf(writer, "sync volume %d on %s and %s: %v\n", a.info.Id, a.location.dataNode.Id, b.location.dataNode.Id, err)
  78. }
  79. replicas = replicas[1:]
  80. }
  81. }
  82. return nil
  83. }
  84. func (c *commandVolumeCheckDisk) syncTwoReplicas(aDB *needle_map.MemDb, bDB *needle_map.MemDb, a *VolumeReplica, verbose *bool, writer io.Writer, b *VolumeReplica, err error, applyChanges *bool, nonRepairThreshold *float64) error {
  85. aHasChanges, bHasChanges := true, true
  86. for aHasChanges || bHasChanges {
  87. // reset index db
  88. aDB.Close()
  89. bDB.Close()
  90. aDB, bDB = needle_map.NewMemDb(), needle_map.NewMemDb()
  91. // read index db
  92. if err := c.readIndexDatabase(aDB, a.info.Collection, a.info.Id, a.location.dataNode.Id, *verbose, writer); err != nil {
  93. return err
  94. }
  95. if err := c.readIndexDatabase(bDB, b.info.Collection, b.info.Id, b.location.dataNode.Id, *verbose, writer); err != nil {
  96. return err
  97. }
  98. // find and make up the differences
  99. if aHasChanges, err = c.doVolumeCheckDisk(aDB, bDB, a, b, *verbose, writer, *applyChanges, *nonRepairThreshold); err != nil {
  100. return err
  101. }
  102. if bHasChanges, err = c.doVolumeCheckDisk(bDB, aDB, b, a, *verbose, writer, *applyChanges, *nonRepairThreshold); err != nil {
  103. return err
  104. }
  105. }
  106. return nil
  107. }
  108. func (c *commandVolumeCheckDisk) doVolumeCheckDisk(subtrahend, minuend *needle_map.MemDb, source, target *VolumeReplica, verbose bool, writer io.Writer, applyChanges bool, nonRepairThreshold float64) (hasChanges bool, err error) {
  109. // find missing keys
  110. // hash join, can be more efficient
  111. var missingNeedles []needle_map.NeedleValue
  112. var counter int
  113. subtrahend.AscendingVisit(func(value needle_map.NeedleValue) error {
  114. counter++
  115. if _, found := minuend.Get(value.Key); !found {
  116. missingNeedles = append(missingNeedles, value)
  117. }
  118. return nil
  119. })
  120. fmt.Fprintf(writer, "volume %d %s has %d entries, %s missed %d entries\n", source.info.Id, source.location.dataNode.Id, counter, target.location.dataNode.Id, len(missingNeedles))
  121. if counter == 0 || len(missingNeedles) == 0 {
  122. return false, nil
  123. }
  124. missingNeedlesFraction := float64(len(missingNeedles)) / float64(counter)
  125. if missingNeedlesFraction > nonRepairThreshold {
  126. return false, fmt.Errorf(
  127. "failed to start repair volume %d, percentage of missing keys is greater than the threshold: %.2f > %.2f",
  128. source.info.Id, missingNeedlesFraction, nonRepairThreshold)
  129. }
  130. for _, needleValue := range missingNeedles {
  131. needleBlob, err := c.readSourceNeedleBlob(source.location.dataNode.Id, source.info.Id, needleValue)
  132. if err != nil {
  133. return hasChanges, err
  134. }
  135. if !applyChanges {
  136. continue
  137. }
  138. if verbose {
  139. fmt.Fprintf(writer, "read %d,%x %s => %s \n", source.info.Id, needleValue.Key, source.location.dataNode.Id, target.location.dataNode.Id)
  140. }
  141. hasChanges = true
  142. if err = c.writeNeedleBlobToTarget(target.location.dataNode.Id, source.info.Id, needleValue, needleBlob); err != nil {
  143. return hasChanges, err
  144. }
  145. }
  146. return
  147. }
  148. func (c *commandVolumeCheckDisk) readSourceNeedleBlob(sourceVolumeServer string, volumeId uint32, needleValue needle_map.NeedleValue) (needleBlob []byte, err error) {
  149. err = operation.WithVolumeServerClient(sourceVolumeServer, c.env.option.GrpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
  150. resp, err := client.ReadNeedleBlob(context.Background(), &volume_server_pb.ReadNeedleBlobRequest{
  151. VolumeId: volumeId,
  152. NeedleId: uint64(needleValue.Key),
  153. Offset: needleValue.Offset.ToActualOffset(),
  154. Size: int32(needleValue.Size),
  155. })
  156. if err != nil {
  157. return err
  158. }
  159. needleBlob = resp.NeedleBlob
  160. return nil
  161. })
  162. return
  163. }
  164. func (c *commandVolumeCheckDisk) writeNeedleBlobToTarget(targetVolumeServer string, volumeId uint32, needleValue needle_map.NeedleValue, needleBlob []byte) error {
  165. return operation.WithVolumeServerClient(targetVolumeServer, c.env.option.GrpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
  166. _, err := client.WriteNeedleBlob(context.Background(), &volume_server_pb.WriteNeedleBlobRequest{
  167. VolumeId: volumeId,
  168. NeedleId: uint64(needleValue.Key),
  169. Size: int32(needleValue.Size),
  170. NeedleBlob: needleBlob,
  171. })
  172. return err
  173. })
  174. }
  175. func (c *commandVolumeCheckDisk) readIndexDatabase(db *needle_map.MemDb, collection string, volumeId uint32, volumeServer string, verbose bool, writer io.Writer) error {
  176. var buf bytes.Buffer
  177. if err := c.copyVolumeIndexFile(collection, volumeId, volumeServer, &buf, verbose, writer); err != nil {
  178. return err
  179. }
  180. if verbose {
  181. fmt.Fprintf(writer, "load collection %s volume %d index size %d from %s ...\n", collection, volumeId, buf.Len(), volumeServer)
  182. }
  183. return db.LoadFromReaderAt(bytes.NewReader(buf.Bytes()))
  184. }
  185. func (c *commandVolumeCheckDisk) copyVolumeIndexFile(collection string, volumeId uint32, volumeServer string, buf *bytes.Buffer, verbose bool, writer io.Writer) error {
  186. return operation.WithVolumeServerClient(volumeServer, c.env.option.GrpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
  187. ext := ".idx"
  188. copyFileClient, err := volumeServerClient.CopyFile(context.Background(), &volume_server_pb.CopyFileRequest{
  189. VolumeId: volumeId,
  190. Ext: ".idx",
  191. CompactionRevision: math.MaxUint32,
  192. StopOffset: math.MaxInt64,
  193. Collection: collection,
  194. IsEcVolume: false,
  195. IgnoreSourceFileNotFound: false,
  196. })
  197. if err != nil {
  198. return fmt.Errorf("failed to start copying volume %d%s: %v", volumeId, ext, err)
  199. }
  200. err = writeToBuffer(copyFileClient, buf)
  201. if err != nil {
  202. return fmt.Errorf("failed to copy %d%s from %s: %v", volumeId, ext, volumeServer, err)
  203. }
  204. return nil
  205. })
  206. }
  207. func writeToBuffer(client volume_server_pb.VolumeServer_CopyFileClient, buf *bytes.Buffer) error {
  208. for {
  209. resp, receiveErr := client.Recv()
  210. if receiveErr == io.EOF {
  211. break
  212. }
  213. if receiveErr != nil {
  214. return fmt.Errorf("receiving: %v", receiveErr)
  215. }
  216. buf.Write(resp.FileContent)
  217. }
  218. return nil
  219. }