mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2024-01-19 02:48:24 +00:00
refactor data integrity checking code
This commit is contained in:
parent
3018443cd7
commit
dda13def2a
|
@ -107,7 +107,7 @@ func iterateEntries(datFile, idxFile *os.File, visitNeedle func(n *storage.Needl
|
|||
fmt.Println("Recovered in f", r)
|
||||
}
|
||||
}()
|
||||
if err = n.ReadNeedleBody(datFile, version, offset+int64(storage.NeedleHeaderSize), rest, false); err != nil {
|
||||
if _, err = n.ReadNeedleBody(datFile, version, offset+int64(storage.NeedleHeaderSize), rest, false); err != nil {
|
||||
fmt.Printf("cannot read needle body: offset %d body %d %v\n", offset, rest, err)
|
||||
}
|
||||
}()
|
||||
|
|
|
@ -17,6 +17,10 @@ const (
|
|||
NeedleMapBoltDb
|
||||
)
|
||||
|
||||
const (
|
||||
NeedleIndexSize = 16
|
||||
)
|
||||
|
||||
type NeedleMapper interface {
|
||||
Put(key uint64, offset uint32, size uint32) error
|
||||
Get(key uint64) (element *NeedleValue, ok bool)
|
||||
|
|
|
@ -226,37 +226,25 @@ func ReadNeedleHeader(r *os.File, version Version, offset int64) (n *Needle, bod
|
|||
|
||||
//n should be a needle already read the header
|
||||
//the input stream will read until next file entry
|
||||
func (n *Needle) ReadNeedleBody(r *os.File, version Version, offset int64, bodyLength uint32, verifyCheckSum bool) (err error) {
|
||||
func (n *Needle) ReadNeedleBody(r *os.File, version Version, offset int64, bodyLength uint32) (bytes []byte, err error) {
|
||||
if bodyLength <= 0 {
|
||||
return nil
|
||||
return
|
||||
}
|
||||
switch version {
|
||||
case Version1:
|
||||
bytes := make([]byte, bodyLength)
|
||||
bytes = make([]byte, bodyLength)
|
||||
if _, err = r.ReadAt(bytes, offset); err != nil {
|
||||
return
|
||||
}
|
||||
n.Data = bytes[:n.Size]
|
||||
n.Checksum = NewCRC(n.Data)
|
||||
if verifyCheckSum {
|
||||
checksum := util.BytesToUint32(bytes[n.Size : n.Size+NeedleChecksumSize])
|
||||
if n.Checksum.Value() != checksum {
|
||||
err = fmt.Errorf("CRC check failed")
|
||||
}
|
||||
}
|
||||
case Version2:
|
||||
bytes := make([]byte, bodyLength)
|
||||
bytes = make([]byte, bodyLength)
|
||||
if _, err = r.ReadAt(bytes, offset); err != nil {
|
||||
return
|
||||
}
|
||||
n.readNeedleDataVersion2(bytes[0:n.Size])
|
||||
n.Checksum = NewCRC(n.Data)
|
||||
if verifyCheckSum {
|
||||
checksum := util.BytesToUint32(bytes[n.Size : n.Size+NeedleChecksumSize])
|
||||
if n.Checksum.Value() != checksum {
|
||||
err = fmt.Errorf("CRC check failed")
|
||||
}
|
||||
}
|
||||
default:
|
||||
err = fmt.Errorf("Unsupported Version! (%d)", version)
|
||||
}
|
||||
|
|
|
@ -11,6 +11,7 @@ import (
|
|||
"time"
|
||||
|
||||
"github.com/chrislusf/seaweedfs/weed/glog"
|
||||
"github.com/chrislusf/seaweedfs/weed/util"
|
||||
)
|
||||
|
||||
type Volume struct {
|
||||
|
@ -47,55 +48,64 @@ func loadVolumeWithoutIndex(dirname string, collection string, id VolumeId, need
|
|||
return
|
||||
}
|
||||
func verifyIndexFileIntegrity(indexFile *os.File) (indexSize int64, err error) {
|
||||
var fi os.FileInfo
|
||||
if fi, err = indexFile.Stat(); err != nil {
|
||||
return
|
||||
} else if indexSize = fi.Size(); indexSize != 0 && indexSize%16 != 0 {
|
||||
err = fmt.Errorf("index file %s's size is %d bytes, maybe corrupted", indexFile.Name(), fi.Size())
|
||||
return
|
||||
if indexSize, err = util.GetFileSize(indexFile); err == nil {
|
||||
if indexSize%NeedleIndexSize != 0 {
|
||||
err = fmt.Errorf("index file's size is %d bytes, maybe corrupted", indexSize)
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
func readIndexEntryAtOffset(indexFile *os.File, offset int64, v Version) (bytes []byte, err error) {
|
||||
if offset < 0 {
|
||||
err = fmt.Errorf("offset %d for index file %s is invalid", offset, indexFile.Name())
|
||||
err = fmt.Errorf("offset %d for index file is invalid", offset)
|
||||
return
|
||||
}
|
||||
bytes = make([]byte, 16)
|
||||
bytes = make([]byte, NeedleIndexSize)
|
||||
_, err = indexFile.ReadAt(bytes, offset)
|
||||
return
|
||||
}
|
||||
func verifyNeedleIntegrity(datFile *os.File, v Version, offset int64, key uint64) error {
|
||||
if n, bodyLength, err := ReadNeedleHeader(datFile, v, offset); err != nil {
|
||||
return fmt.Errorf("can not read needle header: %s", err.Error())
|
||||
return fmt.Errorf("can not read needle header: %v", err)
|
||||
} else {
|
||||
if n.Id != key {
|
||||
return fmt.Errorf("index key %#x does not match needle's Id %#x", key, n.Id)
|
||||
} else {
|
||||
if err := n.ReadNeedleBody(datFile, v, offset+int64(NeedleHeaderSize), bodyLength, true); err != nil {
|
||||
return fmt.Errorf("dat file %s's body reading failed: %s", datFile.Name(), err.Error())
|
||||
if bytes, err := n.ReadNeedleBody(datFile, v, offset+int64(NeedleHeaderSize), bodyLength); err != nil {
|
||||
return fmt.Errorf("dat file's body reading failed: %v", err)
|
||||
} else {
|
||||
checksum := util.BytesToUint32(bytes[n.Size : n.Size+NeedleChecksumSize])
|
||||
if n.Checksum.Value() != checksum {
|
||||
return fmt.Errorf("CRC check failed")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func volumeDataIntegrityChecking(v *Volume, indexFile *os.File) {
|
||||
func volumeDataIntegrityChecking(v *Volume, indexFile *os.File) error {
|
||||
var indexSize int64
|
||||
var e error
|
||||
if indexSize, e = verifyIndexFileIntegrity(indexFile); e != nil {
|
||||
glog.V(0).Infof("verifyIndexFileIntegrity failed %s", e.Error())
|
||||
v.readOnly = true
|
||||
return fmt.Errorf("verifyIndexFileIntegrity failed: %v", e)
|
||||
}
|
||||
if indexSize != 0 {
|
||||
var lastIdxEntry []byte
|
||||
if lastIdxEntry, e = readIndexEntryAtOffset(indexFile, indexSize-16, v.Version()); e != nil {
|
||||
glog.V(0).Infof("readLastIndexEntry failed %s", e.Error())
|
||||
v.readOnly = true
|
||||
if lastIdxEntry, e = readIndexEntryAtOffset(indexFile, indexSize-NeedleIndexSize, v.Version()); e != nil {
|
||||
return fmt.Errorf("readLastIndexEntry failed: %v", e)
|
||||
}
|
||||
key, offset, _ := idxFileEntry(lastIdxEntry)
|
||||
if e = verifyNeedleIntegrity(v.dataFile, v.Version(), int64(offset)*NeedlePaddingSize, key); e != nil {
|
||||
glog.V(0).Infof("verifyNeedleIntegrity failed %s", e.Error())
|
||||
v.readOnly = true
|
||||
return fmt.Errorf("verifyNeedleIntegrity failed: %v", e)
|
||||
}
|
||||
} else {
|
||||
if datSize, err := util.GetFileSize(v.dataFile); err == nil {
|
||||
if datSize > 0 {
|
||||
return fmt.Errorf("dat file size is %d, not empty while the index file is empty!", datSize)
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func (v *Volume) FileName() (fileName string) {
|
||||
if v.Collection == "" {
|
||||
|
@ -156,7 +166,10 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind
|
|||
return fmt.Errorf("cannot write Volume Index %s.idx: %v", fileName, e)
|
||||
}
|
||||
}
|
||||
volumeDataIntegrityChecking(v, indexFile)
|
||||
if e = volumeDataIntegrityChecking(v, indexFile); e != nil {
|
||||
v.readOnly = true
|
||||
glog.V(0).Infof("volumeDataIntegrityChecking failed %v", e)
|
||||
}
|
||||
switch needleMapKind {
|
||||
case NeedleMapInMemory:
|
||||
glog.V(0).Infoln("loading index file", fileName+".idx", "readonly", v.readOnly)
|
||||
|
@ -384,7 +397,7 @@ func ScanVolumeFile(dirname string, collection string, id VolumeId,
|
|||
}
|
||||
for n != nil {
|
||||
if readNeedleBody {
|
||||
if err = n.ReadNeedleBody(v.dataFile, version, offset+int64(NeedleHeaderSize), rest, false); err != nil {
|
||||
if _, err = n.ReadNeedleBody(v.dataFile, version, offset+int64(NeedleHeaderSize), rest); err != nil {
|
||||
glog.V(0).Infof("cannot read needle body: %v", err)
|
||||
//err = fmt.Errorf("cannot read needle body: %v", err)
|
||||
//return
|
||||
|
|
|
@ -36,3 +36,11 @@ func Readln(r *bufio.Reader) ([]byte, error) {
|
|||
}
|
||||
return ln, err
|
||||
}
|
||||
|
||||
func GetFileSize(file *os.File) (size int64, err error) {
|
||||
var fi os.FileInfo
|
||||
if fi, err = file.Stat(); err == nil {
|
||||
size = fi.Size()
|
||||
}
|
||||
return
|
||||
}
|
||||
|
|
Loading…
Reference in a new issue