volume_checking.go 2.2 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667
  1. package storage
  2. import (
  3. "fmt"
  4. "os"
  5. "github.com/chrislusf/seaweedfs/weed/storage/backend"
  6. "github.com/chrislusf/seaweedfs/weed/storage/idx"
  7. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  8. . "github.com/chrislusf/seaweedfs/weed/storage/types"
  9. "github.com/chrislusf/seaweedfs/weed/util"
  10. )
  11. func CheckVolumeDataIntegrity(v *Volume, indexFile *os.File) (lastAppendAtNs uint64, e error) {
  12. var indexSize int64
  13. if indexSize, e = verifyIndexFileIntegrity(indexFile); e != nil {
  14. return 0, fmt.Errorf("verifyIndexFileIntegrity %s failed: %v", indexFile.Name(), e)
  15. }
  16. if indexSize == 0 {
  17. return 0, nil
  18. }
  19. var lastIdxEntry []byte
  20. if lastIdxEntry, e = readIndexEntryAtOffset(indexFile, indexSize-NeedleMapEntrySize); e != nil {
  21. return 0, fmt.Errorf("readLastIndexEntry %s failed: %v", indexFile.Name(), e)
  22. }
  23. key, offset, size := idx.IdxFileEntry(lastIdxEntry)
  24. if offset.IsZero() {
  25. return 0, nil
  26. }
  27. if size == TombstoneFileSize {
  28. size = 0
  29. }
  30. if lastAppendAtNs, e = verifyNeedleIntegrity(v.DataBackend, v.Version(), offset.ToAcutalOffset(), key, size); e != nil {
  31. return lastAppendAtNs, fmt.Errorf("verifyNeedleIntegrity %s failed: %v", indexFile.Name(), e)
  32. }
  33. return
  34. }
  35. func verifyIndexFileIntegrity(indexFile *os.File) (indexSize int64, err error) {
  36. if indexSize, err = util.GetFileSize(indexFile); err == nil {
  37. if indexSize%NeedleMapEntrySize != 0 {
  38. err = fmt.Errorf("index file's size is %d bytes, maybe corrupted", indexSize)
  39. }
  40. }
  41. return
  42. }
  43. func readIndexEntryAtOffset(indexFile *os.File, offset int64) (bytes []byte, err error) {
  44. if offset < 0 {
  45. err = fmt.Errorf("offset %d for index file is invalid", offset)
  46. return
  47. }
  48. bytes = make([]byte, NeedleMapEntrySize)
  49. _, err = indexFile.ReadAt(bytes, offset)
  50. return
  51. }
  52. func verifyNeedleIntegrity(datFile backend.BackendStorageFile, v needle.Version, offset int64, key NeedleId, size uint32) (lastAppendAtNs uint64, err error) {
  53. n := new(needle.Needle)
  54. if err = n.ReadData(datFile, offset, size, v); err != nil {
  55. return n.AppendAtNs, fmt.Errorf("read data [%d,%d) : %v", offset, offset+int64(size), err)
  56. }
  57. if n.Id != key {
  58. return n.AppendAtNs, fmt.Errorf("index key %#x does not match needle's Id %#x", key, n.Id)
  59. }
  60. return n.AppendAtNs, err
  61. }