maxmustermann/minio
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

bitrot: Verify file size inside storage interface (#7932)

Browse source
This commit is contained in:
Anis Elleuch 2019-09-11 21:49:53 +01:00 committed by kannappanr
parent 3d65dc8d94
commit 3f258062d8
9 changed files with 39 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
cmd/bitrot.go
View file

@ -155,3 +155,11 @@ func bitrotWriterSum(w io.Writer) []byte {
}
return nil
}
// Returns the size of the file with bitrot protection
func bitrotShardFileSize(size int64, shardSize int64, algo BitrotAlgorithm) int64 {
if algo != HighwayHash256S {
return size
}
return ceilFrac(size, shardSize)*int64(algo.New().Size()) + size
}

4
cmd/naughty-disk_test.go
View file

@ -196,9 +196,9 @@ func (d *naughtyDisk) ReadAll(volume string, path string) (buf []byte, err error
return d.disk.ReadAll(volume, path)
}
func (d *naughtyDisk) VerifyFile(volume, path string, empty bool, algo BitrotAlgorithm, sum []byte, shardSize int64) error {
func (d *naughtyDisk) VerifyFile(volume, path string, size int64, algo BitrotAlgorithm, sum []byte, shardSize int64) error {
if err := d.calcError(); err != nil {
return err
}
return d.disk.VerifyFile(volume, path, empty, algo, sum, shardSize)
return d.disk.VerifyFile(volume, path, size, algo, sum, shardSize)
}

6
cmd/posix.go
View file

@ -1541,7 +1541,7 @@ func (s *posix) RenameFile(srcVolume, srcPath, dstVolume, dstPath string) (err e
return nil
}
func (s *posix) VerifyFile(volume, path string, empty bool, algo BitrotAlgorithm, sum []byte, shardSize int64) (err error) {
func (s *posix) VerifyFile(volume, path string, fileSize int64, algo BitrotAlgorithm, sum []byte, shardSize int64) (err error) {
defer func() {
if err == errFaultyDisk {
atomic.AddInt32(&s.ioErrCount, 1)
@ -1617,7 +1617,9 @@ func (s *posix) VerifyFile(volume, path string, empty bool, algo BitrotAlgorithm
return err
}
if empty && fi.Size() != 0 || !empty && fi.Size() == 0 {
// Calculate the size of the bitrot file and compare
// it with the actual file size.
if fi.Size() != bitrotShardFileSize(fileSize, shardSize, algo) {
return errFileUnexpectedSize
}

18
cmd/posix_test.go
View file

@ -1797,7 +1797,7 @@ func TestPosixVerifyFile(t *testing.T) {
if err := posixStorage.WriteAll(volName, fileName, bytes.NewBuffer(data)); err != nil {
t.Fatal(err)
}
if err := posixStorage.VerifyFile(volName, fileName, false, algo, hashBytes, 0); err != nil {
if err := posixStorage.VerifyFile(volName, fileName, size, algo, hashBytes, 0); err != nil {
t.Fatal(err)
}
@ -1805,7 +1805,14 @@ func TestPosixVerifyFile(t *testing.T) {
if err := posixStorage.AppendFile(volName, fileName, []byte("a")); err != nil {
t.Fatal(err)
}
if err := posixStorage.VerifyFile(volName, fileName, false, algo, hashBytes, 0); err == nil {
// Check if VerifyFile reports the incorrect file length (the correct length is `size+1`)
if err := posixStorage.VerifyFile(volName, fileName, size, algo, hashBytes, 0); err == nil {
t.Fatal("expected to fail bitrot check")
}
// Check if bitrot fails
if err := posixStorage.VerifyFile(volName, fileName, size+1, algo, hashBytes, 0); err == nil {
t.Fatal("expected to fail bitrot check")
}
@ -1833,7 +1840,7 @@ func TestPosixVerifyFile(t *testing.T) {
t.Fatal(err)
}
w.Close()
if err := posixStorage.VerifyFile(volName, fileName, false, algo, nil, shardSize); err != nil {
if err := posixStorage.VerifyFile(volName, fileName, size, algo, nil, shardSize); err != nil {
t.Fatal(err)
}
@ -1847,7 +1854,10 @@ func TestPosixVerifyFile(t *testing.T) {
t.Fatal(err)
}
f.Close()
if err := posixStorage.VerifyFile(volName, fileName, false, algo, nil, shardSize); err == nil {
if err := posixStorage.VerifyFile(volName, fileName, size, algo, nil, shardSize); err == nil {
t.Fatal("expected to fail bitrot check")
}
if err := posixStorage.VerifyFile(volName, fileName, size+1, algo, nil, shardSize); err == nil {
t.Fatal("expected to fail bitrot check")
}
}

2
cmd/storage-interface.go
View file

@ -52,7 +52,7 @@ type StorageAPI interface {
StatFile(volume string, path string) (file FileInfo, err error)
DeleteFile(volume string, path string) (err error)
DeleteFileBulk(volume string, paths []string) (errs []error, err error)
VerifyFile(volume, path string, empty bool, algo BitrotAlgorithm, sum []byte, shardSize int64) error
VerifyFile(volume, path string, size int64, algo BitrotAlgorithm, sum []byte, shardSize int64) error
// Write all data, syncs the data to disk.
WriteAll(volume string, path string, reader io.Reader) (err error)

6
cmd/storage-rest-client.go
View file

@ -439,13 +439,13 @@ func (client *storageRESTClient) getInstanceID() (err error) {
return nil
}
func (client *storageRESTClient) VerifyFile(volume, path string, empty bool, algo BitrotAlgorithm, sum []byte, shardSize int64) error {
func (client *storageRESTClient) VerifyFile(volume, path string, size int64, algo BitrotAlgorithm, sum []byte, shardSize int64) error {
values := make(url.Values)
values.Set(storageRESTVolume, volume)
values.Set(storageRESTFilePath, path)
values.Set(storageRESTBitrotAlgo, algo.String())
values.Set(storageRESTEmpty, strconv.FormatBool(empty))
values.Set(storageRESTLength, strconv.Itoa(int(shardSize)))
values.Set(storageRESTLength, strconv.FormatInt(size, 10))
values.Set(storageRESTShardSize, strconv.Itoa(int(shardSize)))
if len(sum) != 0 {
values.Set(storageRESTBitrotHash, hex.EncodeToString(sum))
}

4
cmd/storage-rest-common.go
View file

@ -16,7 +16,7 @@
package cmd
const storageRESTVersion = "v8"
const storageRESTVersion = "v9"
const storageRESTPath = minioReservedBucketPath + "/storage/" + storageRESTVersion + SlashSeparator
const (
@ -52,7 +52,7 @@ const (
storageRESTDstPath = "destination-path"
storageRESTOffset = "offset"
storageRESTLength = "length"
storageRESTEmpty = "empty"
storageRESTShardSize = "shard-size"
storageRESTCount = "count"
storageRESTMarkerPath = "marker"
storageRESTLeafFile = "leaf-file"

8
cmd/storage-rest-server.go
View file

@ -520,12 +520,12 @@ func (s *storageRESTServer) VerifyFile(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
volume := vars[storageRESTVolume]
filePath := vars[storageRESTFilePath]
empty, err := strconv.ParseBool(vars[storageRESTEmpty])
size, err := strconv.ParseInt(vars[storageRESTLength], 10, 0)
if err != nil {
s.writeErrorResponse(w, err)
return
}
shardSize, err := strconv.Atoi(vars[storageRESTLength])
shardSize, err := strconv.Atoi(vars[storageRESTShardSize])
if err != nil {
s.writeErrorResponse(w, err)
return
@ -547,7 +547,7 @@ func (s *storageRESTServer) VerifyFile(w http.ResponseWriter, r *http.Request) {
algo := BitrotAlgorithmFromString(algoStr)
w.Header().Set(xhttp.ContentType, "text/event-stream")
doneCh := sendWhiteSpaceVerifyFile(w)
err = s.storage.VerifyFile(volume, filePath, empty, algo, hash, int64(shardSize))
err = s.storage.VerifyFile(volume, filePath, size, algo, hash, int64(shardSize))
<-doneCh
gob.NewEncoder(w).Encode(VerifyFileResp{err})
}
@ -600,7 +600,7 @@ func registerStorageRESTHandlers(router *mux.Router, endpoints EndpointList) {
subrouter.Methods(http.MethodPost).Path(SlashSeparator + storageRESTMethodRenameFile).HandlerFunc(httpTraceHdrs(server.RenameFileHandler)).
Queries(restQueries(storageRESTSrcVolume, storageRESTSrcPath, storageRESTDstVolume, storageRESTDstPath)...)
subrouter.Methods(http.MethodPost).Path(SlashSeparator + storageRESTMethodVerifyFile).HandlerFunc(httpTraceHdrs(server.VerifyFile)).
Queries(restQueries(storageRESTVolume, storageRESTFilePath, storageRESTBitrotAlgo, storageRESTLength, storageRESTEmpty)...)
Queries(restQueries(storageRESTVolume, storageRESTFilePath, storageRESTBitrotAlgo, storageRESTLength, storageRESTShardSize)...)
subrouter.Methods(http.MethodPost).Path(SlashSeparator + storageRESTMethodGetInstanceID).HandlerFunc(httpTraceAll(server.GetInstanceID))
}

2
cmd/xl-v1-healing-common.go
View file

@ -183,7 +183,7 @@ func disksWithAllParts(ctx context.Context, onlineDisks []StorageAPI, partsMetad
// it needs healing too.
for _, part := range partsMetadata[i].Parts {
checksumInfo := erasureInfo.GetChecksumInfo(part.Name)
err = onlineDisk.VerifyFile(bucket, pathJoin(object, part.Name), part.Size == 0, checksumInfo.Algorithm, checksumInfo.Hash, erasure.ShardSize())
err = onlineDisk.VerifyFile(bucket, pathJoin(object, part.Name), erasure.ShardFileSize(part.Size), checksumInfo.Algorithm, checksumInfo.Hash, erasure.ShardSize())
if err != nil {
isCorrupt := strings.HasPrefix(err.Error(), "Bitrot verification mismatch - expected ")
if !isCorrupt && err != errFileNotFound && err != errVolumeNotFound && err != errFileUnexpectedSize {