astelm/minio
1
0
Fork 0
mirror of https://github.com/pgsty/minio.git synced 2026-07-20 20:50:22 +03:00
Code Issues Packages Projects Releases Wiki Activity

re-implement pickValidInfo dataDir, move to quorum calculation (#13681)

Browse Source 
dataDir loosely based on maxima is incorrect and does not
work in all situations such as disks in the following order

- xl.json migration to xl.meta there may be partial xl.json's
  leftover if some disks are not yet connected when the disk
  is yet to come up, since xl.json mtime and xl.meta is
  same the dataDir maxima doesn't work properly leading to
  quorum issues.

- its also possible that XLV1 might be true among the disks
  available, make sure to keep FileInfo based on common quorum
  and skip unexpected disks with the older data format.

Also, this PR tests upgrade from older to a newer release if the 
data is readable and matches the checksum.

NOTE: this is just initial work we can build on top of this to do further tests.
This commit is contained in:
Harshavardhana
2021-11-21 10:41:30 -08:00
committed by GitHub
parent 36b5426f6e
commit c791de0e1e
20 changed files with 378 additions and 155 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/erasure-object.go
+26 -19
View File
@@ -88,10 +88,10 @@ func (er erasureObjects) CopyObject(ctx context.Context, srcBucket, srcObject, d
}
// List all online disks.
onlineDisks, modTime, dataDir := listOnlineDisks(storageDisks, metaArr, errs)
onlineDisks, modTime := listOnlineDisks(storageDisks, metaArr, errs)
// Pick latest valid metadata.
fi, err := pickValidFileInfo(ctx, metaArr, modTime, dataDir, readQuorum)
fi, err := pickValidFileInfo(ctx, metaArr, modTime, readQuorum)
if err != nil {
return oi, toObjectErr(err, srcBucket, srcObject)
}
@@ -102,6 +102,8 @@ func (er erasureObjects) CopyObject(ctx context.Context, srcBucket, srcObject, d
return fi.ToObjectInfo(srcBucket, srcObject), toObjectErr(errMethodNotAllowed, srcBucket, srcObject)
}
filterOnlineDisksInplace(fi, metaArr, onlineDisks)
versionID := srcInfo.VersionID
if srcInfo.versionOnly {
versionID = dstOpts.VersionID
@@ -404,14 +406,16 @@ func (er erasureObjects) getObjectFileInfo(ctx context.Context, bucket, object s
}
// List all online disks.
onlineDisks, modTime, dataDir := listOnlineDisks(disks, metaArr, errs)
onlineDisks, modTime := listOnlineDisks(disks, metaArr, errs)
// Pick latest valid metadata.
fi, err = pickValidFileInfo(ctx, metaArr, modTime, dataDir, readQuorum)
fi, err = pickValidFileInfo(ctx, metaArr, modTime, readQuorum)
if err != nil {
return fi, nil, nil, err
}
filterOnlineDisksInplace(fi, metaArr, onlineDisks)
// if one of the disk is offline, return right here no need
// to attempt a heal on the object.
if countErrs(errs, errDiskNotFound) > 0 {
@@ -424,7 +428,7 @@ func (er erasureObjects) getObjectFileInfo(ctx context.Context, bucket, object s
missingBlocks++
continue
}
if metaArr[i].IsValid() && metaArr[i].ModTime.Equal(fi.ModTime) && metaArr[i].DataDir == fi.DataDir {
if metaArr[i].IsValid() && metaArr[i].ModTime.Equal(fi.ModTime) {
continue
}
missingBlocks++
@@ -1455,17 +1459,20 @@ func (er erasureObjects) PutObjectMetadata(ctx context.Context, bucket, object s
}
// List all online disks.
_, modTime, dataDir := listOnlineDisks(disks, metaArr, errs)
onlineDisks, modTime := listOnlineDisks(disks, metaArr, errs)
// Pick latest valid metadata.
fi, err := pickValidFileInfo(ctx, metaArr, modTime, dataDir, readQuorum)
fi, err := pickValidFileInfo(ctx, metaArr, modTime, readQuorum)
if err != nil {
return ObjectInfo{}, toObjectErr(err, bucket, object)
}
if fi.Deleted {
return ObjectInfo{}, toObjectErr(errMethodNotAllowed, bucket, object)
}
filterOnlineDisksInplace(fi, metaArr, onlineDisks)
// if version-id is not specified retention is supposed to be set on the latest object.
if opts.VersionID == "" {
opts.VersionID = fi.VersionID
@@ -1483,7 +1490,7 @@ func (er erasureObjects) PutObjectMetadata(ctx context.Context, bucket, object s
fi.ModTime = opts.MTime
fi.VersionID = opts.VersionID
if err = er.updateObjectMeta(ctx, bucket, object, fi); err != nil {
if err = er.updateObjectMeta(ctx, bucket, object, fi, onlineDisks); err != nil {
return ObjectInfo{}, toObjectErr(err, bucket, object)
}
@@ -1512,10 +1519,10 @@ func (er erasureObjects) PutObjectTags(ctx context.Context, bucket, object strin
}
// List all online disks.
_, modTime, dataDir := listOnlineDisks(disks, metaArr, errs)
onlineDisks, modTime := listOnlineDisks(disks, metaArr, errs)
// Pick latest valid metadata.
fi, err := pickValidFileInfo(ctx, metaArr, modTime, dataDir, readQuorum)
fi, err := pickValidFileInfo(ctx, metaArr, modTime, readQuorum)
if err != nil {
return ObjectInfo{}, toObjectErr(err, bucket, object)
}
@@ -1526,12 +1533,14 @@ func (er erasureObjects) PutObjectTags(ctx context.Context, bucket, object strin
return ObjectInfo{}, toObjectErr(errMethodNotAllowed, bucket, object)
}
filterOnlineDisksInplace(fi, metaArr, onlineDisks)
fi.Metadata[xhttp.AmzObjectTagging] = tags
for k, v := range opts.UserDefined {
fi.Metadata[k] = v
}
if err = er.updateObjectMeta(ctx, bucket, object, fi); err != nil {
if err = er.updateObjectMeta(ctx, bucket, object, fi, onlineDisks); err != nil {
return ObjectInfo{}, toObjectErr(err, bucket, object)
}
@@ -1539,30 +1548,28 @@ func (er erasureObjects) PutObjectTags(ctx context.Context, bucket, object strin
}
// updateObjectMeta will update the metadata of a file.
func (er erasureObjects) updateObjectMeta(ctx context.Context, bucket, object string, fi FileInfo) error {
func (er erasureObjects) updateObjectMeta(ctx context.Context, bucket, object string, fi FileInfo, onlineDisks []StorageAPI) error {
if len(fi.Metadata) == 0 {
return nil
}
disks := er.getDisks()
g := errgroup.WithNErrs(len(disks))
g := errgroup.WithNErrs(len(onlineDisks))
// Start writing `xl.meta` to all disks in parallel.
for index := range disks {
for index := range onlineDisks {
index := index
g.Go(func() error {
if disks[index] == nil {
if onlineDisks[index] == nil {
return errDiskNotFound
}
return disks[index].UpdateMetadata(ctx, bucket, object, fi)
return onlineDisks[index].UpdateMetadata(ctx, bucket, object, fi)
}, index)
}
// Wait for all the routines.
mErrs := g.Wait()
return reduceWriteQuorumErrs(ctx, mErrs, objectOpIgnoredErrs, getWriteQuorum(len(disks)))
return reduceWriteQuorumErrs(ctx, mErrs, objectOpIgnoredErrs, getWriteQuorum(len(onlineDisks)))
}
// DeleteObjectTags - delete object tags from an existing object