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

fix: newMultipartUpload should go to same pool (#12106)

Browse source 
avoid potential for duplicates under multi-pool
setup, additionally also make sure CompleteMultipart
is using a more optimal API for uploadID lookup
and never delete the object there is a potential
to create a delete marker during complete multipart.

Signed-off-by: Harshavardhana <harsha@minio.io>
This commit is contained in:
Harshavardhana 2021-04-21 11:30:23 -07:00
parent 07d7dd6321
commit 0fb05489df
1 changed files with 19 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

33
cmd/erasure-server-sets.go
View file

@ -1482,10 +1482,23 @@ func (z *erasureServerSets) NewMultipartUpload(ctx context.Context, bucket, obje
return z.serverSets[0].NewMultipartUpload(ctx, bucket, object, opts)
}
// We don't know the exact size, so we ask for at least 1GiB file.
idx, err := z.getZoneIdx(ctx, bucket, object, opts, 1<<30)
if err != nil {
return "", err
for idx, zone := range z.serverSets {
result, err := zone.ListMultipartUploads(ctx, bucket, object, "", "", "", maxUploadsList)
if err != nil {
return "", err
}
// If there is a multipart upload with the same bucket/object name,
// create the new multipart in the same pool, this will avoid
// creating two multiparts uploads in two different pools
if len(result.Uploads) != 0 {
return z.serverSets[idx].NewMultipartUpload(ctx, bucket, object, opts)
}
}
// We multiply the size by 2 to account for erasure coding.
idx := z.getAvailableZoneIdx(ctx, (1<<30)*2)
if idx < 0 {
return "", toObjectErr(errDiskFull)
}
return z.serverSets[idx].NewMultipartUpload(ctx, bucket, object, opts)
@ -1627,17 +1640,9 @@ func (z *erasureServerSets) CompleteMultipartUpload(ctx context.Context, bucket,
return z.serverSets[0].CompleteMultipartUpload(ctx, bucket, object, uploadID, uploadedParts, opts)
}
// Purge any existing object.
for _, zone := range z.serverSets {
zone.DeleteObject(ctx, bucket, object, opts)
}
for _, zone := range z.serverSets {
result, err := zone.ListMultipartUploads(ctx, bucket, object, "", "", "", maxUploadsList)
if err != nil {
return objInfo, err
}
if result.Lookup(uploadID) {
_, err = zone.GetMultipartInfo(ctx, bucket, object, uploadID, opts)
if err == nil {
return zone.CompleteMultipartUpload(ctx, bucket, object, uploadID, uploadedParts, opts)
}
}