fix_dat.go 4.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144
  1. package main
  2. import (
  3. "flag"
  4. "fmt"
  5. "io"
  6. "os"
  7. "path"
  8. "strconv"
  9. "github.com/chrislusf/seaweedfs/weed/util/log"
  10. "github.com/chrislusf/seaweedfs/weed/storage/backend"
  11. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  12. "github.com/chrislusf/seaweedfs/weed/storage/super_block"
  13. "github.com/chrislusf/seaweedfs/weed/storage/types"
  14. "github.com/chrislusf/seaweedfs/weed/util"
  15. )
  16. var (
  17. fixVolumePath = flag.String("dir", "/tmp", "data directory to store files")
  18. fixVolumeCollection = flag.String("collection", "", "the volume collection name")
  19. fixVolumeId = flag.Int("volumeId", -1, "a volume id. The volume should already exist in the dir. The volume index file should not exist.")
  20. )
  21. /*
  22. This is to resolve an one-time issue that caused inconsistency with .dat and .idx files.
  23. In this case, the .dat file contains all data, but some of deletion caused incorrect offset.
  24. The .idx has all correct offsets.
  25. 1. fix the .dat file, a new .dat_fixed file will be generated.
  26. go run fix_dat.go -volumeId=9 -dir=/Users/chrislu/Downloads
  27. 2. move the original .dat and .idx files to some backup folder, and rename .dat_fixed to .dat file
  28. mv 9.dat_fixed 9.dat
  29. 3. fix the .idx file with the "weed fix"
  30. weed fix -volumeId=9 -dir=/Users/chrislu/Downloads
  31. */
  32. func main() {
  33. flag.Parse()
  34. fileName := strconv.Itoa(*fixVolumeId)
  35. if *fixVolumeCollection != "" {
  36. fileName = *fixVolumeCollection + "_" + fileName
  37. }
  38. indexFile, err := os.OpenFile(path.Join(*fixVolumePath, fileName+".idx"), os.O_RDONLY, 0644)
  39. if err != nil {
  40. log.Fatalf("Read Volume Index %v", err)
  41. }
  42. defer indexFile.Close()
  43. datFileName := path.Join(*fixVolumePath, fileName+".dat")
  44. datFile, err := os.OpenFile(datFileName, os.O_RDONLY, 0644)
  45. if err != nil {
  46. log.Fatalf("Read Volume Data %v", err)
  47. }
  48. datBackend := backend.NewDiskFile(datFile)
  49. defer datBackend.Close()
  50. newDatFile, err := os.Create(path.Join(*fixVolumePath, fileName+".dat_fixed"))
  51. if err != nil {
  52. log.Fatalf("Write New Volume Data %v", err)
  53. }
  54. defer newDatFile.Close()
  55. superBlock, err := super_block.ReadSuperBlock(datBackend)
  56. if err != nil {
  57. log.Fatalf("Read Volume Data superblock %v", err)
  58. }
  59. newDatFile.Write(superBlock.Bytes())
  60. iterateEntries(datBackend, indexFile, func(n *needle.Needle, offset int64) {
  61. fmt.Printf("needle id=%v name=%s size=%d dataSize=%d\n", n.Id, string(n.Name), n.Size, n.DataSize)
  62. _, s, _, e := n.Append(datBackend, superBlock.Version)
  63. fmt.Printf("size %d error %v\n", s, e)
  64. })
  65. }
  66. func iterateEntries(datBackend backend.BackendStorageFile, idxFile *os.File, visitNeedle func(n *needle.Needle, offset int64)) {
  67. // start to read index file
  68. var readerOffset int64
  69. bytes := make([]byte, 16)
  70. count, _ := idxFile.ReadAt(bytes, readerOffset)
  71. readerOffset += int64(count)
  72. // start to read dat file
  73. superBlock, err := super_block.ReadSuperBlock(datBackend)
  74. if err != nil {
  75. fmt.Printf("cannot read dat file super block: %v", err)
  76. return
  77. }
  78. offset := int64(superBlock.BlockSize())
  79. version := superBlock.Version
  80. n, _, rest, err := needle.ReadNeedleHeader(datBackend, version, offset)
  81. if err != nil {
  82. fmt.Printf("cannot read needle header: %v", err)
  83. return
  84. }
  85. fmt.Printf("Needle %+v, rest %d\n", n, rest)
  86. for n != nil && count > 0 {
  87. // parse index file entry
  88. key := util.BytesToUint64(bytes[0:8])
  89. offsetFromIndex := util.BytesToUint32(bytes[8:12])
  90. sizeFromIndex := types.BytesToSize(bytes[12:16])
  91. count, _ = idxFile.ReadAt(bytes, readerOffset)
  92. readerOffset += int64(count)
  93. if offsetFromIndex != 0 && offset != int64(offsetFromIndex)*8 {
  94. //t := offset
  95. offset = int64(offsetFromIndex) * 8
  96. //fmt.Printf("Offset change %d => %d\n", t, offset)
  97. }
  98. fmt.Printf("key: %d offsetFromIndex %d n.Size %d sizeFromIndex:%d\n", key, offsetFromIndex, n.Size, sizeFromIndex)
  99. rest = needle.NeedleBodyLength(sizeFromIndex, version)
  100. func() {
  101. defer func() {
  102. if r := recover(); r != nil {
  103. fmt.Println("Recovered in f", r)
  104. }
  105. }()
  106. if _, err = n.ReadNeedleBody(datBackend, version, offset+int64(types.NeedleHeaderSize), rest); err != nil {
  107. fmt.Printf("cannot read needle body: offset %d body %d %v\n", offset, rest, err)
  108. }
  109. }()
  110. if n.Size <= types.Size(n.DataSize) {
  111. continue
  112. }
  113. visitNeedle(n, offset)
  114. offset += types.NeedleHeaderSize + rest
  115. //fmt.Printf("==> new entry offset %d\n", offset)
  116. if n, _, rest, err = needle.ReadNeedleHeader(datBackend, version, offset); err != nil {
  117. if err == io.EOF {
  118. return
  119. }
  120. fmt.Printf("cannot read needle header: %v\n", err)
  121. return
  122. }
  123. //fmt.Printf("new entry needle size:%d rest:%d\n", n.Size, rest)
  124. }
  125. }