go imports

`Needle.ReadNeedleBody` add CRC check and warning

go/filer/flat_namespace/flat_namespace_store.go

@@ -1,7 +1,5 @@
 package flat_namespace
 
-import ()
-
 type FlatNamespaceStore interface {
 	Put(fullFileName string, fid string) (err error)
 	Get(fullFileName string) (fid string, err error)

go/storage/needle.go

@@ -14,8 +14,8 @@ import (
 
 	"github.com/chrislusf/seaweedfs/go/glog"
 	"github.com/chrislusf/seaweedfs/go/images"
-	"github.com/chrislusf/seaweedfs/go/util"
 	"github.com/chrislusf/seaweedfs/go/operation"
+	"github.com/chrislusf/seaweedfs/go/util"
 )
 
 const (

go/storage/needle_read_write.go

@@ -238,6 +238,10 @@ func (n *Needle) ReadNeedleBody(r *os.File, version Version, offset int64, bodyL
 		}
 		n.Data = bytes[:n.Size]
 		n.Checksum = NewCRC(n.Data)
+		checksum := util.BytesToUint32(bytes[n.Size : n.Size+NeedleChecksumSize])
+		if n.Checksum.Value() != checksum {
+			glog.V(0).Infof("CRC error! Data On Disk Corrupted, needle id = %x", n.Id)
+		}
 	case Version2:
 		bytes := make([]byte, bodyLength)
 		if _, err = r.ReadAt(bytes, offset); err != nil {
@@ -245,6 +249,10 @@ func (n *Needle) ReadNeedleBody(r *os.File, version Version, offset int64, bodyL
 		}
 		n.readNeedleDataVersion2(bytes[0:n.Size])
 		n.Checksum = NewCRC(n.Data)
+		checksum := util.BytesToUint32(bytes[n.Size : n.Size+NeedleChecksumSize])
+		if n.Checksum.Value() != checksum {
+			glog.V(0).Infof("CRC error! Data On Disk Corrupted, needle id = %x", n.Id)
+		}
 	default:
 		err = fmt.Errorf("Unsupported Version! (%d)", version)
 	}

go/storage/volume_info.go

@@ -2,8 +2,9 @@ package storage
 
 import (
 	"fmt"
-	"github.com/chrislusf/seaweedfs/go/operation"
 	"sort"
+
+	"github.com/chrislusf/seaweedfs/go/operation"
 )
 
 type VolumeInfo struct {

go/storage/volume_sync.go

@@ -202,6 +202,9 @@ func (v *Volume) fetchNeedle(volumeDataContentHandlerUrl string,
 		if err != nil {
 			return fmt.Errorf("Reading from %s error: %v", volumeDataContentHandlerUrl, err)
 		}
+		if needleValue.Size != uint32(len(b)) {
+			return fmt.Errorf("Reading from %s error: size incorrect", volumeDataContentHandlerUrl)
+		}
 		offset, err := v.AppendBlob(b)
 		if err != nil {
 			return fmt.Errorf("Appending volume %d error: %v", v.Id, err)