heal legacy objects when versioning is enabled after upgrade (#13671)

legacy objects in 'xl.json' after upgrade, should have
following sequence of events - bucket should have versioning
enabled and the object should have been overwritten with
another version of an object.

this situation was not handled, which would lead to older
objects to stay perpetually with "legacy" dataDir, however
these objects were readable by all means - there weren't
converted to newer format.

This PR fixes this situation properly.
This commit is contained in:
Harshavardhana 2021-11-17 15:49:12 -08:00 committed by GitHub
parent 9c5d9ae376
commit 886262e58a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 19 additions and 3 deletions

View File

@ -18,9 +18,11 @@
package cmd package cmd
import ( import (
"fmt"
"sort" "sort"
jsoniter "github.com/json-iterator/go" jsoniter "github.com/json-iterator/go"
"github.com/minio/minio/internal/logger"
) )
// versionsSorter sorts FileInfo slices by version. // versionsSorter sorts FileInfo slices by version.
@ -121,15 +123,17 @@ func getFileInfo(xlMetaBuf []byte, volume, path, versionID string, data bool) (F
xlMeta := &xlMetaV1Object{} xlMeta := &xlMetaV1Object{}
var json = jsoniter.ConfigCompatibleWithStandardLibrary var json = jsoniter.ConfigCompatibleWithStandardLibrary
if err := json.Unmarshal(xlMetaBuf, xlMeta); err != nil { if err := json.Unmarshal(xlMetaBuf, xlMeta); err != nil {
logger.LogIf(GlobalContext, fmt.Errorf("unable to unmarshal json object: %v", err))
return FileInfo{}, errFileCorrupt return FileInfo{}, errFileCorrupt
} }
fi, err := xlMeta.ToFileInfo(volume, path) fi, err := xlMeta.ToFileInfo(volume, path)
if err == errFileNotFound && versionID != "" { if err != nil {
return fi, errFileVersionNotFound return FileInfo{}, err
} }
fi.IsLatest = true // No versions so current version is latest.
fi.XLV1 = true // indicates older version fi.XLV1 = true // indicates older version
fi.IsLatest = true // No versions so current version is latest.
return fi, err return fi, err
} }

View File

@ -199,7 +199,9 @@ func (m *xlMetaV1Object) ToFileInfo(volume, path string) (FileInfo, error) {
Erasure: m.Erasure, Erasure: m.Erasure,
VersionID: m.VersionID, VersionID: m.VersionID,
DataDir: m.DataDir, DataDir: m.DataDir,
XLV1: true,
} }
return fi, nil return fi, nil
} }

View File

@ -2133,6 +2133,10 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f
xlMeta.AddFreeVersion(fi) xlMeta.AddFreeVersion(fi)
} }
// indicates if RenameData() is called by healing.
// healing doesn't preserve the dataDir as 'legacy'
healing := fi.XLV1 && fi.DataDir != legacyDataDir
if err = xlMeta.AddVersion(fi); err != nil { if err = xlMeta.AddVersion(fi); err != nil {
if legacyPreserved { if legacyPreserved {
// Any failed rename calls un-roll previous transaction. // Any failed rename calls un-roll previous transaction.
@ -2164,6 +2168,12 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f
// renameAll only for objects that have xl.meta not saved inline. // renameAll only for objects that have xl.meta not saved inline.
if len(fi.Data) == 0 && fi.Size > 0 { if len(fi.Data) == 0 && fi.Size > 0 {
s.moveToTrash(dstDataPath, true) s.moveToTrash(dstDataPath, true)
if healing {
// If we are healing we should purge any legacyDataPath content,
// that was previously preserved during PutObject() call
// on a versioned bucket.
s.moveToTrash(legacyDataPath, true)
}
if err = renameAll(srcDataPath, dstDataPath); err != nil { if err = renameAll(srcDataPath, dstDataPath); err != nil {
if legacyPreserved { if legacyPreserved {
// Any failed rename calls un-roll previous transaction. // Any failed rename calls un-roll previous transaction.