MyFavMirrors
/
minio
mirror of https://github.com/minio/minio.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Optimize listing with leaf check offloaded to posix (#7541) 

Other listing optimizations include

- remove double sorting while filtering object entries
- improve error message when upload-id is not in quorum
- use jsoniter for full unmarshal json, instead of gjson
- remove unused code
This commit is contained in:
Harshavardhana 2019-04-23 14:54:28 -07:00 committed by GitHub
parent 0c75395abe
commit f767a2538a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
27 changed files with 246 additions and 243 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
cmd/disk-cache.go
View File

@ -458,10 +458,6 @@ func (c cacheObjects) listCacheObjects(ctx context.Context, bucket, prefix, mark
eof = true
break
}
// For any walk error return right away.
if walkResult.err != nil {
return result, toObjectErr(walkResult.err, bucket, prefix)
}
entry := walkResult.entry
var objInfo ObjectInfo

4
cmd/fs-v1.go
View File

@ -1010,8 +1010,8 @@ func (fs *FSObjects) listDirFactory(isLeaf IsLeafFunc) ListDirFunc {
logger.LogIf(context.Background(), err)
return
}
entries, delayIsLeaf = filterListEntries(bucket, prefixDir, entries, prefixEntry, isLeaf)
return entries, delayIsLeaf
sort.Strings(entries)
return filterListEntries(bucket, prefixDir, entries, prefixEntry, isLeaf)
}
// Return list factory instance.

4
cmd/naughty-disk_test.go
View File

@ -111,11 +111,11 @@ func (d *naughtyDisk) DeleteVol(volume string) (err error) {
return d.disk.DeleteVol(volume)
}
func (d *naughtyDisk) ListDir(volume, path string, count int) (entries []string, err error) {
func (d *naughtyDisk) ListDir(volume, path string, count int, leafFile string) (entries []string, err error) {
if err := d.calcError(); err != nil {
return []string{}, err
}
return d.disk.ListDir(volume, path, count)
return d.disk.ListDir(volume, path, count, leafFile)
}
func (d *naughtyDisk) ReadFile(volume string, path string, offset int64, buf []byte, verifier *BitrotVerifier) (n int64, err error) {

18
cmd/object-api-common.go
View File

@ -116,7 +116,7 @@ func cleanupDir(ctx context.Context, storage StorageAPI, volume, dirPath string)
}
// If it's a directory, list and call delFunc() for each entry.
entries, err := storage.ListDir(volume, entryPath, -1)
entries, err := storage.ListDir(volume, entryPath, -1, "")
// If entryPath prefix never existed, safe to ignore.
if err == errFileNotFound {
return nil
@ -218,14 +218,6 @@ func listObjects(ctx context.Context, obj ObjectLayer, bucket, prefix, marker, d
eof = true
break
}
// For any walk error return right away.
if walkResult.err != nil {
// File not found is a valid case.
if walkResult.err == errFileNotFound {
continue
}
return loi, toObjectErr(walkResult.err, bucket, prefix)
}
var objInfo ObjectInfo
var err error
@ -235,6 +227,14 @@ func listObjects(ctx context.Context, obj ObjectLayer, bucket, prefix, marker, d
if err == nil {
break
}
if err == errFileNotFound {
err = nil
objInfo = ObjectInfo{
Bucket: bucket,
Name: walkResult.entry,
IsDir: true,
}
}
}
} else {
objInfo, err = getObjInfo(ctx, bucket, walkResult.entry)

27
cmd/object-api-errors.go
View File

@ -27,6 +27,22 @@ import (
// underlying storage layer.
func toObjectErr(err error, params ...string) error {
switch err {
case errDiskNotFound:
switch len(params) {
case 1:
err = BucketNotFound{Bucket: params[0]}
case 2:
err = ObjectNotFound{
Bucket: params[0],
Object: params[1],
}
case 3:
err = InvalidUploadID{
Bucket: params[0],
Object: params[1],
UploadID: params[2],
}
}
case errVolumeNotFound:
if len(params) >= 1 {
err = BucketNotFound{Bucket: params[0]}
@ -63,11 +79,18 @@ func toObjectErr(err error, params ...string) error {
}
}
case errFileNotFound:
if len(params) >= 2 {
switch len(params) {
case 2:
err = ObjectNotFound{
Bucket: params[0],
Object: params[1],
}
case 3:
err = InvalidUploadID{
Bucket: params[0],
Object: params[1],
UploadID: params[2],
}
}
case errFileNameTooLong:
if len(params) >= 2 {
@ -321,6 +344,8 @@ func (e MalformedUploadID) Error() string {
// InvalidUploadID invalid upload id.
type InvalidUploadID struct {
Bucket string
Object string
UploadID string
}

26
cmd/object-api-multipart_test.go
View File

@ -228,8 +228,7 @@ func testPutObjectPartDiskNotFound(obj ObjectLayer, instanceType string, disks [
}
// This causes quorum failure verify.
disks = disks[len(disks)-3:]
for _, disk := range disks {
for _, disk := range disks[len(disks)-3:] {
os.RemoveAll(disk)
}
@ -240,9 +239,26 @@ func testPutObjectPartDiskNotFound(obj ObjectLayer, instanceType string, disks [
t.Fatalf("Test %s: expected to fail but passed instead", instanceType)
}
// as majority of xl.json are not available, we expect uploadID to be not available.
expectedErr := InvalidUploadID{UploadID: testCase.uploadID}
if err.Error() != expectedErr.Error() {
t.Fatalf("Test %s: expected error %s, got %s instead.", instanceType, expectedErr, err)
expectedErr1 := InsufficientReadQuorum{}
if err.Error() != expectedErr1.Error() {
t.Fatalf("Test %s: expected error %s, got %s instead.", instanceType, expectedErr1, err)
}
// This causes invalid upload id.
for _, disk := range disks {
os.RemoveAll(disk)
}
// Object part upload should fail with bucket not found.
_, err = obj.PutObjectPart(context.Background(), testCase.bucketName, testCase.objName, testCase.uploadID, testCase.PartID, mustGetPutObjReader(t, bytes.NewBufferString(testCase.inputReaderData), testCase.intputDataSize, testCase.inputMd5, sha256sum), ObjectOptions{})
if err == nil {
t.Fatalf("Test %s: expected to fail but passed instead", instanceType)
}
// As all disks at not available, bucket not found.
expectedErr2 := BucketNotFound{Bucket: testCase.bucketName}
if err.Error() != expectedErr2.Error() {
t.Fatalf("Test %s: expected error %s, got %s instead.", instanceType, expectedErr2, err)
}
}

3
cmd/posix-list-dir_unix.go
View File

@ -20,7 +20,6 @@ package cmd
import (
"os"
"path"
"runtime"
"sync"
"syscall"
@ -79,7 +78,7 @@ func parseDirents(dirPath string, buf []byte) (entries []string, err error) {
// On Linux XFS does not implement d_type for on disk
// format << v5. Fall back to OsStat().
var fi os.FileInfo
fi, err = os.Stat(path.Join(dirPath, name))
fi, err = os.Stat(pathJoin(dirPath, name))
if err != nil {
// If file does not exist, we continue and skip it.
// Could happen if it was deleted in the middle while

3
cmd/posix-list-dir_windows.go
View File

@ -20,7 +20,6 @@ package cmd
import (
"os"
"path"
"strings"
"syscall"
)
@ -82,7 +81,7 @@ func readDirN(dirPath string, count int) (entries []string, err error) {
case data.FileAttributes&syscall.FILE_ATTRIBUTE_REPARSE_POINT != 0:
// If its symbolic link, follow the link using os.Stat()
var fi os.FileInfo
fi, err = os.Stat(path.Join(dirPath, name))
fi, err = os.Stat(pathJoin(dirPath, name))
if err != nil {
// If file does not exist, we continue and skip it.
// Could happen if it was deleted in the middle while

18
cmd/posix.go
View File

@ -651,7 +651,7 @@ func (s *posix) DeleteVol(volume string) (err error) {
// ListDir - return all the entries at the given directory path.
// If an entry is a directory it will be returned with a trailing "/".
func (s *posix) ListDir(volume, dirPath string, count int) (entries []string, err error) {
func (s *posix) ListDir(volume, dirPath string, count int, leafFile string) (entries []string, err error) {
defer func() {
if err == errFaultyDisk {
atomic.AddInt32(&s.ioErrCount, 1)
@ -684,9 +684,21 @@ func (s *posix) ListDir(volume, dirPath string, count int) (entries []string, er
dirPath = pathJoin(volumeDir, dirPath)
if count > 0 {
return readDirN(dirPath, count)
entries, err = readDirN(dirPath, count)
} else {
entries, err = readDir(dirPath)
}
return readDir(dirPath)
// If leaf file is specified, filter out the entries.
if leafFile != "" {
for i, entry := range entries {
if _, serr := os.Stat(pathJoin(dirPath, entry, leafFile)); serr == nil {
entries[i] = strings.TrimSuffix(entry, slashSeparator)
}
}
}
return entries, err
}
// ReadAll reads from r until an error or EOF and returns the data it read.

2
cmd/posix_test.go
View File

@ -839,7 +839,7 @@ func TestPosixPosixListDir(t *testing.T) {
} else {
t.Errorf("Expected the StorageAPI to be of type *posix")
}
dirList, err = posixStorage.ListDir(testCase.srcVol, testCase.srcPath, -1)
dirList, err = posixStorage.ListDir(testCase.srcVol, testCase.srcPath, -1, "")
if err != testCase.expectedErr {
t.Fatalf("TestPosix case %d: Expected: \"%s\", got: \"%s\"", i+1, testCase.expectedErr, err)
}

2
cmd/storage-interface.go
View File

@ -39,7 +39,7 @@ type StorageAPI interface {
DeleteVol(volume string) (err error)
// File operations.
ListDir(volume, dirPath string, count int) ([]string, error)
ListDir(volume, dirPath string, count int, leafFile string) ([]string, error)
ReadFile(volume string, path string, offset int64, buf []byte, verifier *BitrotVerifier) (n int64, err error)
AppendFile(volume string, path string, buf []byte) (err error)
CreateFile(volume, path string, size int64, reader io.Reader) error

3
cmd/storage-rest-client.go
View File

@ -315,11 +315,12 @@ func (client *storageRESTClient) ReadFile(volume, path string, offset int64, buf
}
// ListDir - lists a directory.
func (client *storageRESTClient) ListDir(volume, dirPath string, count int) (entries []string, err error) {
func (client *storageRESTClient) ListDir(volume, dirPath string, count int, leafFile string) (entries []string, err error) {
values := make(url.Values)
values.Set(storageRESTVolume, volume)
values.Set(storageRESTDirPath, dirPath)
values.Set(storageRESTCount, strconv.Itoa(count))
values.Set(storageRESTLeafFile, leafFile)
respBody, err := client.call(storageRESTMethodListDir, values, nil, -1)
if err != nil {
return nil, err

3
cmd/storage-rest-common.go
View File

@ -16,7 +16,7 @@
package cmd
const storageRESTVersion = "v4"
const storageRESTVersion = "v5"
const storageRESTPath = minioReservedBucketPath + "/storage/" + storageRESTVersion + "/"
const (
@ -50,6 +50,7 @@ const (
storageRESTOffset = "offset"
storageRESTLength = "length"
storageRESTCount = "count"
storageRESTLeafFile = "leaf-file"
storageRESTBitrotAlgo = "bitrot-algo"
storageRESTBitrotHash = "bitrot-hash"
storageRESTInstanceID = "instance-id"

5
cmd/storage-rest-server.go
View File

@ -361,12 +361,13 @@ func (s *storageRESTServer) ListDirHandler(w http.ResponseWriter, r *http.Reques
vars := mux.Vars(r)
volume := vars[storageRESTVolume]
dirPath := vars[storageRESTDirPath]
leafFile := vars[storageRESTLeafFile]
count, err := strconv.Atoi(vars[storageRESTCount])
if err != nil {
s.writeErrorResponse(w, err)
return
}
entries, err := s.storage.ListDir(volume, dirPath, count)
entries, err := s.storage.ListDir(volume, dirPath, count, leafFile)
if err != nil {
s.writeErrorResponse(w, err)
return
@ -443,7 +444,7 @@ func registerStorageRESTHandlers(router *mux.Router, endpoints EndpointList) {
subrouter.Methods(http.MethodPost).Path("/" + storageRESTMethodReadFileStream).HandlerFunc(httpTraceHdrs(server.ReadFileStreamHandler)).
Queries(restQueries(storageRESTVolume, storageRESTFilePath, storageRESTOffset, storageRESTLength)...)
subrouter.Methods(http.MethodPost).Path("/" + storageRESTMethodListDir).HandlerFunc(httpTraceHdrs(server.ListDirHandler)).
Queries(restQueries(storageRESTVolume, storageRESTDirPath, storageRESTCount)...)
Queries(restQueries(storageRESTVolume, storageRESTDirPath, storageRESTCount, storageRESTLeafFile)...)
subrouter.Methods(http.MethodPost).Path("/" + storageRESTMethodDeleteFile).HandlerFunc(httpTraceHdrs(server.DeleteFileHandler)).
Queries(restQueries(storageRESTVolume, storageRESTFilePath)...)
subrouter.Methods(http.MethodPost).Path("/" + storageRESTMethodRenameFile).HandlerFunc(httpTraceHdrs(server.RenameFileHandler)).

2
cmd/storage-rest_test.go
View File

@ -260,7 +260,7 @@ func testStorageAPIListDir(t *testing.T, storage StorageAPI) {
}
for i, testCase := range testCases {
result, err := storage.ListDir(testCase.volumeName, testCase.prefix, -1)
result, err := storage.ListDir(testCase.volumeName, testCase.prefix, -1, "")
expectErr := (err != nil)
if expectErr != testCase.expectErr {

18
cmd/tree-walk.go
View File

@ -25,7 +25,6 @@ import (
// TreeWalkResult - Tree walk result carries results of tree walking.
type TreeWalkResult struct {
entry string
err error
end bool
}
@ -99,10 +98,8 @@ type IsLeafFunc func(string, string) bool
// IsLeafDirFunc - A function isLeafDir of type isLeafDirFunc is used to detect if an entry represents an empty directory.
type IsLeafDirFunc func(string, string) bool
// Note: input entries are expected to be sorted.
func filterListEntries(bucket, prefixDir string, entries []string, prefixEntry string, isLeaf IsLeafFunc) ([]string, bool) {
// Listing needs to be sorted.
sort.Strings(entries)
// Filter entries that have the prefix prefixEntry.
entries = filterMatchingPrefix(entries, prefixEntry)
@ -168,9 +165,10 @@ func doTreeWalk(ctx context.Context, bucket, prefixDir, entryPrefixMatch, marker
for i, entry := range entries {
var leaf, leafDir bool
pentry := pathJoin(prefixDir, entry)
// Decision to do isLeaf check was pushed from listDir() to here.
if delayIsLeaf {
leaf = isLeaf(bucket, pathJoin(prefixDir, entry))
leaf = isLeaf(bucket, pentry)
if leaf {
entry = strings.TrimSuffix(entry, slashSeparator)
}
@ -179,7 +177,7 @@ func doTreeWalk(ctx context.Context, bucket, prefixDir, entryPrefixMatch, marker
}
if strings.HasSuffix(entry, slashSeparator) {
leafDir = isLeafDir(bucket, pathJoin(prefixDir, entry))
leafDir = isLeafDir(bucket, pentry)
}
isDir := !leafDir && !leaf
@ -209,17 +207,19 @@ func doTreeWalk(ctx context.Context, bucket, prefixDir, entryPrefixMatch, marker
// markIsEnd is passed to this entry's treeWalk() so that treeWalker.end can be marked
// true at the end of the treeWalk stream.
markIsEnd := i == len(entries)-1 && isEnd
if tErr := doTreeWalk(ctx, bucket, pathJoin(prefixDir, entry), prefixMatch, markerArg, recursive, listDir, isLeaf, isLeafDir, resultCh, endWalkCh, markIsEnd); tErr != nil {
return tErr
if err := doTreeWalk(ctx, bucket, pentry, prefixMatch, markerArg, recursive,
listDir, isLeaf, isLeafDir, resultCh, endWalkCh, markIsEnd); err != nil {
return err
}
continue
}
// EOF is set if we are at last entry and the caller indicated we at the end.
isEOF := ((i == len(entries)-1) && isEnd)
select {
case <-endWalkCh:
return errWalkAbort
case resultCh <- TreeWalkResult{entry: pathJoin(prefixDir, entry), end: isEOF}:
case resultCh <- TreeWalkResult{entry: pentry, end: isEOF}:
}
}

10
cmd/tree-walk_test.go
View File

@ -189,7 +189,7 @@ func TestTreeWalk(t *testing.T) {
}
isLeafDir := func(volume, prefix string) bool {
entries, listErr := disk.ListDir(volume, prefix, 1)
entries, listErr := disk.ListDir(volume, prefix, 1, "")
if listErr != nil {
return false
}
@ -233,7 +233,7 @@ func TestTreeWalkTimeout(t *testing.T) {
}
isLeafDir := func(volume, prefix string) bool {
entries, listErr := disk.ListDir(volume, prefix, 1)
entries, listErr := disk.ListDir(volume, prefix, 1, "")
if listErr != nil {
return false
}
@ -379,7 +379,7 @@ func TestRecursiveTreeWalk(t *testing.T) {
}
isLeafDir := func(volume, prefix string) bool {
entries, listErr := disk1.ListDir(volume, prefix, 1)
entries, listErr := disk1.ListDir(volume, prefix, 1, "")
if listErr != nil {
return false
}
@ -494,7 +494,7 @@ func TestSortedness(t *testing.T) {
}
isLeafDir := func(volume, prefix string) bool {
entries, listErr := disk1.ListDir(volume, prefix, 1)
entries, listErr := disk1.ListDir(volume, prefix, 1, "")
if listErr != nil {
return false
}
@ -577,7 +577,7 @@ func TestTreeWalkIsEnd(t *testing.T) {
}
isLeafDir := func(volume, prefix string) bool {
entries, listErr := disk1.ListDir(volume, prefix, 1)
entries, listErr := disk1.ListDir(volume, prefix, 1, "")
if listErr != nil {
return false
}

19
cmd/xl-sets.go
View File

@ -654,7 +654,7 @@ func listDirSetsFactory(ctx context.Context, isLeaf IsLeafFunc, isLeafDir IsLeaf
wg.Add(1)
go func(index int, disk StorageAPI) {
defer wg.Done()
diskEntries[index], _ = disk.ListDir(bucket, prefixDir, -1)
diskEntries[index], _ = disk.ListDir(bucket, prefixDir, -1, xlMetaJSONFile)
}(index, disk)
}
@ -713,21 +713,10 @@ func listDirSetsFactory(ctx context.Context, isLeaf IsLeafFunc, isLeafDir IsLeaf
// value through the walk channel receives the data properly lexically sorted.
func (s *xlSets) ListObjects(ctx context.Context, bucket, prefix, marker, delimiter string, maxKeys int) (ListObjectsInfo, error) {
isLeaf := func(bucket, entry string) bool {
entry = strings.TrimSuffix(entry, slashSeparator)
// Verify if we are at the leaf, a leaf is where we
// see `xl.json` inside a directory.
return s.getHashedSet(entry).isObject(bucket, entry)
return !hasSuffix(entry, slashSeparator)
}
isLeafDir := func(bucket, entry string) bool {
// Verify prefixes in all sets.
var ok bool
for _, set := range s.sets {
ok = set.isObjectDir(bucket, entry)
if ok {
return true
}
}
return false
}
@ -1271,10 +1260,6 @@ func (s *xlSets) HealObjects(ctx context.Context, bucket, prefix string, healObj
if !ok {
break
}
// For any walk error return right away.
if walkResult.err != nil {
return toObjectErr(walkResult.err, bucket, prefix)
}
if err := healObjectFn(bucket, strings.TrimSuffix(walkResult.entry, slashSeparator+xlMetaJSONFile)); err != nil {
return toObjectErr(err, bucket, walkResult.entry)
}

2
cmd/xl-v1-common.go
View File

@ -98,7 +98,7 @@ func (xl xlObjects) isObjectDir(bucket, prefix string) (ok bool) {
go func(index int, disk StorageAPI) {
defer wg.Done()
// Check if 'prefix' is an object on this 'disk', else continue the check the next disk
entries, err := disk.ListDir(bucket, prefix, 1)
entries, err := disk.ListDir(bucket, prefix, 1, "")
if err != nil {
errs[index] = err
return

113
cmd/xl-v1-healing.go
View File

@ -341,7 +341,7 @@ func (xl xlObjects) healObject(ctx context.Context, bucket string, object string
}
// List and delete the object directory,
files, derr := disk.ListDir(bucket, object, -1)
files, derr := disk.ListDir(bucket, object, -1, "")
if derr == nil {
for _, entry := range files {
_ = disk.DeleteFile(bucket,
@ -478,52 +478,51 @@ func (xl xlObjects) healObjectDir(ctx context.Context, bucket, object string, dr
hr.Before.Drives = make([]madmin.HealDriveInfo, len(storageDisks))
hr.After.Drives = make([]madmin.HealDriveInfo, len(storageDisks))
var wg sync.WaitGroup
errs := statAllDirs(ctx, storageDisks, bucket, object)
if isObjectDirDangling(errs) {
for i, err := range errs {
if err == nil {
storageDisks[i].DeleteFile(bucket, object)
}
}
}
// Prepare object creation in all disks
for i, d := range storageDisks {
wg.Add(1)
go func(idx int, disk StorageAPI) {
defer wg.Done()
if disk == nil {
hr.Before.Drives[idx] = madmin.HealDriveInfo{State: madmin.DriveStateOffline}
hr.After.Drives[idx] = madmin.HealDriveInfo{State: madmin.DriveStateOffline}
return
for i, err := range errs {
var drive string
if storageDisks[i] != nil {
drive = storageDisks[i].String()
}
drive := disk.String()
hr.Before.Drives[idx] = madmin.HealDriveInfo{UUID: "", Endpoint: drive, State: madmin.DriveStateOffline}
hr.After.Drives[idx] = madmin.HealDriveInfo{UUID: "", Endpoint: drive, State: madmin.DriveStateOffline}
_, statErr := disk.StatVol(pathJoin(bucket, object))
switch statErr {
case nil:
hr.Before.Drives[idx].State = madmin.DriveStateOk
hr.After.Drives[idx].State = madmin.DriveStateOk
// Object is fine in this disk, nothing to be done anymore, exiting
return
switch err {
case errDiskNotFound:
hr.Before.Drives[i] = madmin.HealDriveInfo{State: madmin.DriveStateOffline}
hr.After.Drives[i] = madmin.HealDriveInfo{State: madmin.DriveStateOffline}
case errVolumeNotFound:
hr.Before.Drives[idx].State = madmin.DriveStateMissing
hr.After.Drives[idx].State = madmin.DriveStateMissing
hr.Before.Drives[i] = madmin.HealDriveInfo{Endpoint: drive, State: madmin.DriveStateMissing}
hr.After.Drives[i] = madmin.HealDriveInfo{Endpoint: drive, State: madmin.DriveStateMissing}
default:
logger.LogIf(ctx, err)
return
hr.Before.Drives[i] = madmin.HealDriveInfo{Endpoint: drive, State: madmin.DriveStateCorrupt}
hr.After.Drives[i] = madmin.HealDriveInfo{Endpoint: drive, State: madmin.DriveStateCorrupt}
}
}
if dryRun {
return
return hr, nil
}
for i, err := range errs {
switch err {
case errVolumeNotFound:
merr := storageDisks[i].MakeVol(pathJoin(bucket, object))
switch merr {
case nil, errVolumeExists:
hr.After.Drives[i].State = madmin.DriveStateOk
case errDiskNotFound:
hr.After.Drives[i].State = madmin.DriveStateOffline
default:
logger.LogIf(ctx, merr)
hr.After.Drives[i].State = madmin.DriveStateCorrupt
}
if err := disk.MakeVol(pathJoin(bucket, object)); err == nil || err == errVolumeExists {
hr.After.Drives[idx].State = madmin.DriveStateOk
} else {
logger.LogIf(ctx, err)
hr.After.Drives[idx].State = madmin.DriveStateOffline
}
}(i, d)
}
wg.Wait()
return hr, nil
}
@ -587,6 +586,46 @@ func defaultHealResult(latestXLMeta xlMetaV1, storageDisks []StorageAPI, errs []
return result
}
// Stat all directories.
func statAllDirs(ctx context.Context, storageDisks []StorageAPI, bucket, prefix string) []error {
var errs = make([]error, len(storageDisks))
var wg sync.WaitGroup
for index, disk := range storageDisks {
if disk == nil {
continue
}
wg.Add(1)
go func(index int, disk StorageAPI) {
defer wg.Done()
entries, err := disk.ListDir(bucket, prefix, 1, "")
if err != nil {
errs[index] = err
return
}
if len(entries) > 0 {
errs[index] = errVolumeNotEmpty
return
}
}(index, disk)
}
wg.Wait()
return errs
}
// ObjectDir is considered dangling/corrupted if any only
// if total disks - a combination of corrupted and missing
// files is lesser than N/2+1 number of disks.
func isObjectDirDangling(errs []error) (ok bool) {
var notFoundDir int
for _, readErr := range errs {
if readErr == errFileNotFound {
notFoundDir++
}
}
return notFoundDir > len(errs)/2
}
// Object is considered dangling/corrupted if any only
// if total disks - a combination of corrupted and missing
// files is lesser than number of data blocks.

12
cmd/xl-v1-list-objects.go
View File

@ -34,7 +34,7 @@ func listDirFactory(ctx context.Context, isLeaf IsLeafFunc, disks ...StorageAPI)
var entries []string
var newEntries []string
var err error
entries, err = disk.ListDir(bucket, prefixDir, -1)
entries, err = disk.ListDir(bucket, prefixDir, -1, xlMetaJSONFile)
if err != nil {
continue
}
@ -89,10 +89,6 @@ func (xl xlObjects) listObjects(ctx context.Context, bucket, prefix, marker, del
eof = true
break
}
// For any walk error return right away.
if walkResult.err != nil {
return loi, toObjectErr(walkResult.err, bucket, prefix)
}
entry := walkResult.entry
var objInfo ObjectInfo
if hasSuffix(entry, slashSeparator) {
@ -108,8 +104,10 @@ func (xl xlObjects) listObjects(ctx context.Context, bucket, prefix, marker, del
// Ignore errFileNotFound as the object might have got
// deleted in the interim period of listing and getObjectInfo(),
// ignore quorum error as it might be an entry from an outdated disk.
switch err {
case errFileNotFound, errXLReadQuorum:
if IsErrIgnored(err, []error{
errFileNotFound,
errXLReadQuorum,
}...) {
continue
}
return loi, toObjectErr(err, bucket, prefix)

5
cmd/xl-v1-metadata.go
View File

@ -1,5 +1,5 @@
/*
* MinIO Cloud Storage, (C) 2016, 2017, 2017 MinIO, Inc.
* MinIO Cloud Storage, (C) 2016-2019 MinIO, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@ -27,6 +27,7 @@ import (
"sync"
"time"
jsoniter "github.com/json-iterator/go"
"github.com/minio/minio/cmd/logger"
)
@ -74,8 +75,8 @@ func (c ChecksumInfo) MarshalJSON() ([]byte, error) {
// UnmarshalJSON - should never be called, instead xlMetaV1UnmarshalJSON() should be used.
func (c *ChecksumInfo) UnmarshalJSON(data []byte) error {
logger.LogIf(context.Background(), errUnexpected)
var info checksumInfoJSON
var json = jsoniter.ConfigCompatibleWithStandardLibrary
if err := json.Unmarshal(data, &info); err != nil {
return err
}

36
cmd/xl-v1-multipart.go
View File

@ -47,9 +47,10 @@ func (xl xlObjects) getMultipartSHADir(bucket, object string) string {
return getSHA256Hash([]byte(pathJoin(bucket, object)))
}
// isUploadIDExists - verify if a given uploadID exists and is valid.
func (xl xlObjects) isUploadIDExists(ctx context.Context, bucket, object, uploadID string) bool {
return xl.isObject(minioMetaMultipartBucket, xl.getUploadIDDir(bucket, object, uploadID))
// checkUploadIDExists - verify if a given uploadID exists and is valid.
func (xl xlObjects) checkUploadIDExists(ctx context.Context, bucket, object, uploadID string) error {
_, err := xl.getObjectInfo(ctx, minioMetaMultipartBucket, xl.getUploadIDDir(bucket, object, uploadID))
return err
}
// Removes part given by partName belonging to a mulitpart upload from minioMetaBucket
@ -163,7 +164,7 @@ func (xl xlObjects) ListMultipartUploads(ctx context.Context, bucket, object, ke
if disk == nil {
continue
}
uploadIDs, err := disk.ListDir(minioMetaMultipartBucket, xl.getMultipartSHADir(bucket, object), -1)
uploadIDs, err := disk.ListDir(minioMetaMultipartBucket, xl.getMultipartSHADir(bucket, object), -1, "")
if err != nil {
if err == errFileNotFound {
return result, nil
@ -308,9 +309,9 @@ func (xl xlObjects) PutObjectPart(ctx context.Context, bucket, object, uploadID
}
// Validates if upload ID exists.
if !xl.isUploadIDExists(ctx, bucket, object, uploadID) {
if err := xl.checkUploadIDExists(ctx, bucket, object, uploadID); err != nil {
preUploadIDLock.RUnlock()
return pi, InvalidUploadID{UploadID: uploadID}
return pi, toObjectErr(err, bucket, object, uploadID)
}
// Read metadata associated with the object from all disks.
@ -406,9 +407,9 @@ func (xl xlObjects) PutObjectPart(ctx context.Context, bucket, object, uploadID
}
defer postUploadIDLock.Unlock()
// Validate again if upload ID still exists.
if !xl.isUploadIDExists(ctx, bucket, object, uploadID) {
return pi, InvalidUploadID{UploadID: uploadID}
// Validates if upload ID exists.
if err := xl.checkUploadIDExists(ctx, bucket, object, uploadID); err != nil {
return pi, toObjectErr(err, bucket, object, uploadID)
}
// Rename temporary part file to its final location.
@ -499,8 +500,8 @@ func (xl xlObjects) ListObjectParts(ctx context.Context, bucket, object, uploadI
}
defer uploadIDLock.Unlock()
if !xl.isUploadIDExists(ctx, bucket, object, uploadID) {
return result, InvalidUploadID{UploadID: uploadID}
if err := xl.checkUploadIDExists(ctx, bucket, object, uploadID); err != nil {
return result, toObjectErr(err, bucket, object, uploadID)
}
uploadIDPath := xl.getUploadIDDir(bucket, object, uploadID)
@ -617,8 +618,8 @@ func (xl xlObjects) CompleteMultipartUpload(ctx context.Context, bucket string,
}
defer uploadIDLock.Unlock()
if !xl.isUploadIDExists(ctx, bucket, object, uploadID) {
return oi, InvalidUploadID{UploadID: uploadID}
if err := xl.checkUploadIDExists(ctx, bucket, object, uploadID); err != nil {
return oi, toObjectErr(err, bucket, object, uploadID)
}
// Check if an object is present as one of the parent dir.
@ -821,8 +822,9 @@ func (xl xlObjects) AbortMultipartUpload(ctx context.Context, bucket, object, up
}
defer uploadIDLock.Unlock()
if !xl.isUploadIDExists(ctx, bucket, object, uploadID) {
return InvalidUploadID{UploadID: uploadID}
// Validates if upload ID exists.
if err := xl.checkUploadIDExists(ctx, bucket, object, uploadID); err != nil {
return toObjectErr(err, bucket, object, uploadID)
}
// Read metadata associated with the object from all disks.
@ -871,12 +873,12 @@ func (xl xlObjects) cleanupStaleMultipartUploads(ctx context.Context, cleanupInt
// Remove the old multipart uploads on the given disk.
func (xl xlObjects) cleanupStaleMultipartUploadsOnDisk(ctx context.Context, disk StorageAPI, expiry time.Duration) {
now := time.Now()
shaDirs, err := disk.ListDir(minioMetaMultipartBucket, "", -1)
shaDirs, err := disk.ListDir(minioMetaMultipartBucket, "", -1, "")
if err != nil {
return
}
for _, shaDir := range shaDirs {
uploadIDDirs, err := disk.ListDir(minioMetaMultipartBucket, shaDir, -1)
uploadIDDirs, err := disk.ListDir(minioMetaMultipartBucket, shaDir, -1, "")
if err != nil {
continue
}

48
cmd/xl-v1-object.go
View File

@ -173,10 +173,6 @@ func (xl xlObjects) GetObjectNInfo(ctx context.Context, bucket, object string, r
// Handler directory request by returning a reader that
// returns no bytes.
if hasSuffix(object, slashSeparator) {
if !xl.isObjectDir(bucket, object) {
nsUnlocker()
return nil, toObjectErr(errFileNotFound, bucket, object)
}
var objInfo ObjectInfo
if objInfo, err = xl.getObjectInfoDir(ctx, bucket, object); err != nil {
nsUnlocker()
@ -375,19 +371,16 @@ func (xl xlObjects) getObjectInfoDir(ctx context.Context, bucket, object string)
wg.Add(1)
go func(index int, disk StorageAPI) {
defer wg.Done()
if _, err := disk.StatVol(pathJoin(bucket, object)); err != nil {
// Since we are re-purposing StatVol, an object which
// is a directory if it doesn't exist should be
// returned as errFileNotFound instead, convert
// the error right here accordingly.
if err == errVolumeNotFound {
err = errFileNotFound
} else if err == errVolumeAccessDenied {
err = errFileAccessDenied
}
// Save error to reduce it later
// Check if 'prefix' is an object on this 'disk'.
entries, err := disk.ListDir(bucket, object, 1, "")
if err != nil {
errs[index] = err
return
}
if len(entries) > 0 {
// Not a directory if not empty.
errs[index] = errFileNotFound
return
}
}(index, disk)
}
@ -412,13 +405,11 @@ func (xl xlObjects) GetObjectInfo(ctx context.Context, bucket, object string, op
}
if hasSuffix(object, slashSeparator) {
if !xl.isObjectDir(bucket, object) {
return oi, toObjectErr(errFileNotFound, bucket, object)
info, err := xl.getObjectInfoDir(ctx, bucket, object)
if err != nil {
return oi, toObjectErr(err, bucket, object)
}
if oi, e = xl.getObjectInfoDir(ctx, bucket, object); e != nil {
return oi, toObjectErr(e, bucket, object)
}
return oi, nil
return info, nil
}
info, err := xl.getObjectInfo(ctx, bucket, object)
@ -879,8 +870,17 @@ func (xl xlObjects) DeleteObject(ctx context.Context, bucket, object string) (er
var writeQuorum int
var isObjectDir = hasSuffix(object, slashSeparator)
if isObjectDir && !xl.isObjectDir(bucket, object) {
return toObjectErr(errFileNotFound, bucket, object)
if isObjectDir {
_, err = xl.getObjectInfoDir(ctx, bucket, object)
if err == errXLReadQuorum {
if isObjectDirDangling(statAllDirs(ctx, xl.getDisks(), bucket, object)) {
// If object is indeed dangling, purge it.
err = nil
}
}
if err != nil {
return toObjectErr(err, bucket, object)
}
}
if isObjectDir {

84
cmd/xl-v1-utils.go
View File

@ -18,13 +18,13 @@ package cmd
import (
"context"
"encoding/hex"
"errors"
"hash/crc32"
"path"
"sync"
"time"
jsoniter "github.com/json-iterator/go"
"github.com/minio/minio/cmd/logger"
"github.com/tidwall/gjson"
)
@ -139,59 +139,6 @@ func parseXLFormat(xlMetaBuf []byte) string {
return gjson.GetBytes(xlMetaBuf, "format").String()
}
func parseXLRelease(xlMetaBuf []byte) string {
return gjson.GetBytes(xlMetaBuf, "minio.release").String()
}
func parseXLErasureInfo(ctx context.Context, xlMetaBuf []byte) (ErasureInfo, error) {
erasure := ErasureInfo{}
erasureResult := gjson.GetBytes(xlMetaBuf, "erasure")
// parse the xlV1Meta.Erasure.Distribution.
disResult := erasureResult.Get("distribution").Array()
distribution := make([]int, len(disResult))
for i, dis := range disResult {
distribution[i] = int(dis.Int())
}
erasure.Distribution = distribution
erasure.Algorithm = erasureResult.Get("algorithm").String()
erasure.DataBlocks = int(erasureResult.Get("data").Int())
erasure.ParityBlocks = int(erasureResult.Get("parity").Int())
erasure.BlockSize = erasureResult.Get("blockSize").Int()
erasure.Index = int(erasureResult.Get("index").Int())
checkSumsResult := erasureResult.Get("checksum").Array()
// Check for scenario where checksum information missing for some parts.
partsResult := gjson.GetBytes(xlMetaBuf, "parts").Array()
if len(checkSumsResult) != len(partsResult) {
return erasure, errCorruptedFormat
}
// Parse xlMetaV1.Erasure.Checksum array.
checkSums := make([]ChecksumInfo, len(checkSumsResult))
for i, v := range checkSumsResult {
algorithm := BitrotAlgorithmFromString(v.Get("algorithm").String())
if !algorithm.Available() {
logger.LogIf(ctx, errBitrotHashAlgoInvalid)
return erasure, errBitrotHashAlgoInvalid
}
hash, err := hex.DecodeString(v.Get("hash").String())
if err != nil {
logger.LogIf(ctx, err)
return erasure, err
}
name := v.Get("name").String()
if name == "" {
return erasure, errCorruptedFormat
}
checkSums[i] = ChecksumInfo{Name: name, Algorithm: algorithm, Hash: hash}
}
erasure.Checksums = checkSums
return erasure, nil
}
func parseXLParts(xlMetaBuf []byte) []ObjectPartInfo {
// Parse the XL Parts.
partsResult := gjson.GetBytes(xlMetaBuf, "parts").Array()
@ -220,32 +167,9 @@ func parseXLMetaMap(xlMetaBuf []byte) map[string]string {
// Constructs XLMetaV1 using `gjson` lib to retrieve each field.
func xlMetaV1UnmarshalJSON(ctx context.Context, xlMetaBuf []byte) (xlMeta xlMetaV1, e error) {
// obtain version.
xlMeta.Version = parseXLVersion(xlMetaBuf)
// obtain format.
xlMeta.Format = parseXLFormat(xlMetaBuf)
// Parse xlMetaV1.Stat .
stat, err := parseXLStat(xlMetaBuf)
if err != nil {
logger.LogIf(ctx, err)
return xlMeta, err
}
xlMeta.Stat = stat
// parse the xlV1Meta.Erasure fields.
xlMeta.Erasure, err = parseXLErasureInfo(ctx, xlMetaBuf)
if err != nil {
return xlMeta, err
}
// Parse the XL Parts.
xlMeta.Parts = parseXLParts(xlMetaBuf)
// Get the xlMetaV1.Realse field.
xlMeta.Minio.Release = parseXLRelease(xlMetaBuf)
// parse xlMetaV1.
xlMeta.Meta = parseXLMetaMap(xlMetaBuf)
return xlMeta, nil
var json = jsoniter.ConfigCompatibleWithStandardLibrary
e = json.Unmarshal(xlMetaBuf, &xlMeta)
return xlMeta, e
}
// read xl.json from the given disk, parse and return xlV1MetaV1.Parts.

1
go.mod
View File

@ -43,6 +43,7 @@ require (
github.com/howeyc/gopass v0.0.0-20170109162249-bf9dde6d0d2c // indirect
github.com/inconshreveable/go-update v0.0.0-20160112193335-8152e7eb6ccf
github.com/jonboulle/clockwork v0.1.0 // indirect
github.com/json-iterator/go v1.1.6
github.com/klauspost/compress v1.4.1 // indirect
github.com/klauspost/cpuid v1.2.0 // indirect
github.com/klauspost/pgzip v1.2.1

3
go.sum
View File

@ -305,6 +305,7 @@ github.com/jellevandenhooff/dkim v0.0.0-20150330215556-f50fe3d243e1/go.mod h1:E0
github.com/jmespath/go-jmespath v0.0.0-20160202185014-0b12d6b521d8/go.mod h1:Nht3zPeWKUH0NzdCt2Blrr5ys8VGpn0CEB0cQHVjt7k=
github.com/jonboulle/clockwork v0.1.0 h1:VKV+ZcuP6l3yW9doeqz6ziZGgcynBVQO+obU0+0hcPo=
github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo=
github.com/json-iterator/go v1.1.6 h1:MrUvLMLTMxbqFJ9kzlvat/rYZqZnW3u4wkLzWTaFwKs=
github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
github.com/jstemmer/go-junit-report v0.0.0-20190106144839-af01ea7f8024/go.mod h1:6v2b51hI/fHJwM22ozAgKL4VKDeJcHhJFhtBdhmNjmU=
github.com/jtolds/gls v4.2.1+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
@ -407,7 +408,9 @@ github.com/mitchellh/mapstructure v1.1.2 h1:fmNYVwqnSfB9mZU6OS2O6GsXM+wcskZDuKQz
github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
github.com/mitchellh/pointerstructure v0.0.0-20170205204203-f2329fcfa9e2/go.mod h1:KMNPMpc0BU/kZEgyDhBplsDn/mjnJMhyMjq4MWboN20=
github.com/mitchellh/reflectwalk v1.0.0/go.mod h1:mSTlrgnPZtwu0c4WaC2kGObEpuNDbx0jmZXqmk4esnw=
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
github.com/modern-go/reflect2 v1.0.1 h1:9f412s+6RmYXLWZSEzVVgPGK7C2PphHj5RJrvfx9AWI=
github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
github.com/nats-io/gnatsd v1.4.1 h1:RconcfDeWpKCD6QIIwiVFcvForlXpWeJP7i5/lDLy44=