Fix listDirHealFactory merging of entries across disks (#3959)

For listing of objects needing heal, we list all objects present on all
the disks and return the set union. We were incorrectly dropping objects
that weren't already seen in disks so far.

Sample directory layout of disks in a 4-disk setup:
`/tmp/1`, `/tmp/2`, `/tmp/3`, `/tmp/4` are directories used as disks here.
`test` is the bucket, `obj1` and obj2` are the objects.
```
/tmp/1/test
└── obj2
    ├── part.1
    ├── part.2
    └── xl.json
/tmp/2/test
└── obj1
    ├── part.1
    ├── part.2
    └── xl.json
/tmp/3/test
├── obj1
│   ├── part.1
│   ├── part.2
│   └── xl.json
└── obj2
    ├── part.1
    ├── part.2
    └── xl.json
/tmp/4/test
[This is empty]

```
master
Krishnan Parthasarathi 8 years ago committed by Harshavardhana
parent 80b83a51a3
commit 4e92b2ecb8
  1. 23
      cmd/xl-v1-list-objects-heal.go

@ -35,37 +35,28 @@ func listDirHealFactory(isLeaf isLeafFunc, disks ...StorageAPI) listDirFunc {
if err != nil { if err != nil {
continue continue
} }
// Listing needs to be sorted.
sort.Strings(entries)
// Filter entries that have the prefix prefixEntry. // Filter entries that have the prefix prefixEntry.
entries = filterMatchingPrefix(entries, prefixEntry) entries = filterMatchingPrefix(entries, prefixEntry)
// isLeaf() check has to happen here so that trailing "/" for objects can be removed. // isLeaf() check has to happen here so that
// trailing "/" for objects can be removed.
for i, entry := range entries { for i, entry := range entries {
if isLeaf(bucket, pathJoin(prefixDir, entry)) { if isLeaf(bucket, pathJoin(prefixDir, entry)) {
entries[i] = strings.TrimSuffix(entry, slashSeparator) entries[i] = strings.TrimSuffix(entry, slashSeparator)
} }
} }
// Sort again after removing trailing "/" for objects as the previous sort
// does not hold good anymore. // Find elements in entries which are not in mergedEntries
sort.Strings(entries)
if len(mergedEntries) == 0 {
// For the first successful disk.ListDir()
mergedEntries = entries
sort.Strings(mergedEntries)
continue
}
// find elements in entries which are not in mergedentries
for _, entry := range entries { for _, entry := range entries {
idx := sort.SearchStrings(mergedEntries, entry) idx := sort.SearchStrings(mergedEntries, entry)
// idx different from len(mergedEntries) means entry is not found // if entry is already present in mergedEntries don't add.
// in mergedEntries if idx < len(mergedEntries) && mergedEntries[idx] == entry {
if idx < len(mergedEntries) {
continue continue
} }
newEntries = append(newEntries, entry) newEntries = append(newEntries, entry)
} }
if len(newEntries) > 0 { if len(newEntries) > 0 {
// Merge the entries and sort it. // Merge the entries and sort it.
mergedEntries = append(mergedEntries, newEntries...) mergedEntries = append(mergedEntries, newEntries...)

Loading…
Cancel
Save