mirror of
https://github.com/pgsty/minio.git
synced 2026-11-11 18:15:29 +03:00
Integrate complete R3 encoding and safe-read repairs
Signed-off-by: Feng Ruohang <rh@vonng.com>
This commit is contained in:
@@ -251,8 +251,9 @@ func enforceRetentionBypassForPut(ctx context.Context, r *http.Request, oi Objec
|
|||||||
|
|
||||||
// checkObjectLockedForOverwrite re-checks under the object write lock whether an
|
// checkObjectLockedForOverwrite re-checks under the object write lock whether an
|
||||||
// in-place write to the null version (such as a directory object) would
|
// in-place write to the null version (such as a directory object) would
|
||||||
// overwrite a locked version in an Object Lock bucket.
|
// overwrite a locked version in an Object Lock bucket. Restore context keeps
|
||||||
func checkObjectLockedForOverwrite(ctx context.Context, bucket, object string, getObjectInfoFn GetObjectInfoFn) error {
|
// failed reads from deleting the version being rehydrated.
|
||||||
|
func checkObjectLockedForOverwrite(ctx context.Context, bucket, object string, getObjectInfoFn GetObjectInfoFn, transition TransitionOptions) error {
|
||||||
if isMinioMetaBucketName(bucket) {
|
if isMinioMetaBucketName(bucket) {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
@@ -265,6 +266,7 @@ func checkObjectLockedForOverwrite(ctx context.Context, bucket, object string, g
|
|||||||
}
|
}
|
||||||
objInfo, err := getObjectInfoFn(ctx, bucket, object, ObjectOptions{
|
objInfo, err := getObjectInfoFn(ctx, bucket, object, ObjectOptions{
|
||||||
VersionID: nullVersionID,
|
VersionID: nullVersionID,
|
||||||
|
Transition: transition,
|
||||||
NoLock: true,
|
NoLock: true,
|
||||||
NoAuditLog: true,
|
NoAuditLog: true,
|
||||||
})
|
})
|
||||||
|
|||||||
@@ -1694,7 +1694,7 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
|
|||||||
getObjectInfo = opts.replicaObjectInfo
|
getObjectInfo = opts.replicaObjectInfo
|
||||||
}
|
}
|
||||||
if fi.VersionID == "" || fi.VersionID == nullVersionID {
|
if fi.VersionID == "" || fi.VersionID == nullVersionID {
|
||||||
if err := checkObjectLockedForOverwrite(ctx, bucket, object, getObjectInfo); err != nil {
|
if err := checkObjectLockedForOverwrite(ctx, bucket, object, getObjectInfo, opts.Transition); err != nil {
|
||||||
return oi, toObjectErr(err, bucket, object)
|
return oi, toObjectErr(err, bucket, object)
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
@@ -1702,6 +1702,7 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
|
|||||||
VersionID: fi.VersionID,
|
VersionID: fi.VersionID,
|
||||||
Versioned: opts.Versioned,
|
Versioned: opts.Versioned,
|
||||||
VersionSuspended: opts.VersionSuspended,
|
VersionSuspended: opts.VersionSuspended,
|
||||||
|
Transition: opts.Transition,
|
||||||
NoLock: true,
|
NoLock: true,
|
||||||
})
|
})
|
||||||
switch {
|
switch {
|
||||||
|
|||||||
+26
-6
@@ -156,7 +156,7 @@ func (er erasureObjects) CopyObject(ctx context.Context, srcBucket, srcObject, d
|
|||||||
if dstOpts.replicaObjectInfo != nil {
|
if dstOpts.replicaObjectInfo != nil {
|
||||||
getObjectInfo = dstOpts.replicaObjectInfo
|
getObjectInfo = dstOpts.replicaObjectInfo
|
||||||
}
|
}
|
||||||
if err := checkObjectLockedForOverwrite(ctx, dstBucket, dstObject, getObjectInfo); err != nil {
|
if err := checkObjectLockedForOverwrite(ctx, dstBucket, dstObject, getObjectInfo, dstOpts.Transition); err != nil {
|
||||||
return oi, toObjectErr(err, dstBucket, dstObject)
|
return oi, toObjectErr(err, dstBucket, dstObject)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -1673,7 +1673,7 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st
|
|||||||
if opts.replicaObjectInfo != nil {
|
if opts.replicaObjectInfo != nil {
|
||||||
getObjectInfo = opts.replicaObjectInfo
|
getObjectInfo = opts.replicaObjectInfo
|
||||||
}
|
}
|
||||||
if err := checkObjectLockedForOverwrite(ctx, bucket, object, getObjectInfo); err != nil {
|
if err := checkObjectLockedForOverwrite(ctx, bucket, object, getObjectInfo, opts.Transition); err != nil {
|
||||||
return ObjectInfo{}, toObjectErr(err, bucket, object)
|
return ObjectInfo{}, toObjectErr(err, bucket, object)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -2667,6 +2667,7 @@ func (er erasureObjects) updateRestoreMetadata(ctx context.Context, bucket, obje
|
|||||||
VersionID: oi.VersionID,
|
VersionID: oi.VersionID,
|
||||||
MTime: oi.ModTime,
|
MTime: oi.ModTime,
|
||||||
NoLock: opts.NoLock,
|
NoLock: opts.NoLock,
|
||||||
|
Transition: opts.Transition,
|
||||||
}); err != nil {
|
}); err != nil {
|
||||||
storageLogIf(ctx, fmt.Errorf("Unable to update transition restore metadata for %s/%s(%s): %s", bucket, object, oi.VersionID, err))
|
storageLogIf(ctx, fmt.Errorf("Unable to update transition restore metadata for %s/%s(%s): %s", bucket, object, oi.VersionID, err))
|
||||||
return err
|
return err
|
||||||
@@ -2708,14 +2709,24 @@ func (er erasureObjects) restoreTransitionedObject(ctx context.Context, bucket s
|
|||||||
oi = actualfi.ToObjectInfo(bucket, object, opts.Versioned || opts.VersionSuspended)
|
oi = actualfi.ToObjectInfo(bucket, object, opts.Versioned || opts.VersionSuspended)
|
||||||
ropts := putRestoreOpts(bucket, object, opts.Transition.RestoreRequest, oi)
|
ropts := putRestoreOpts(bucket, object, opts.Transition.RestoreRequest, oi)
|
||||||
ropts.NoLock = opts.NoLock
|
ropts.NoLock = opts.NoLock
|
||||||
if len(oi.Parts) == 1 {
|
ropts.Transition = opts.Transition
|
||||||
|
// Rehydration copies encoded bytes without changing their content identity.
|
||||||
|
ropts.PreserveETag = oi.ETag
|
||||||
|
// Even a one-part multipart upload must retain its original part number:
|
||||||
|
// encrypted multipart data uses that number to derive its decryption key.
|
||||||
|
if len(oi.Parts) == 1 && !oi.isMultipart() {
|
||||||
|
actualSize, err := oi.GetActualSize()
|
||||||
|
if err != nil {
|
||||||
|
return setRestoreHeaderFn(oi, err)
|
||||||
|
}
|
||||||
|
ropts.IndexCB = func() []byte { return oi.Parts[0].Index }
|
||||||
var rs *HTTPRangeSpec
|
var rs *HTTPRangeSpec
|
||||||
gr, err := getTransitionedObjectReader(ctx, bucket, object, rs, http.Header{}, oi, opts)
|
gr, err := getTransitionedObjectReader(ctx, bucket, object, rs, http.Header{}, oi, opts)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return setRestoreHeaderFn(oi, toObjectErr(err, bucket, object))
|
return setRestoreHeaderFn(oi, toObjectErr(err, bucket, object))
|
||||||
}
|
}
|
||||||
defer gr.Close()
|
defer gr.Close()
|
||||||
hashReader, err := hash.NewReader(ctx, gr, gr.ObjInfo.Size, "", "", gr.ObjInfo.Size)
|
hashReader, err := hash.NewReader(ctx, gr, gr.ObjInfo.Size, "", "", actualSize)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return setRestoreHeaderFn(oi, toObjectErr(err, bucket, object))
|
return setRestoreHeaderFn(oi, toObjectErr(err, bucket, object))
|
||||||
}
|
}
|
||||||
@@ -2740,11 +2751,20 @@ func (er erasureObjects) restoreTransitionedObject(ctx context.Context, bucket s
|
|||||||
|
|
||||||
// rehydrate the parts back on disk as per the original xl.meta prior to transition
|
// rehydrate the parts back on disk as per the original xl.meta prior to transition
|
||||||
for _, partInfo := range oi.Parts {
|
for _, partInfo := range oi.Parts {
|
||||||
hr, err := hash.NewReader(ctx, io.LimitReader(gr, partInfo.Size), partInfo.Size, "", "", partInfo.Size)
|
// Old uncompressed parts may not record ActualSize. Encrypted parts
|
||||||
|
// derive it from the written ciphertext in PutObjectPart; compressed
|
||||||
|
// parts need their recorded logical length, including valid zero sizes.
|
||||||
|
actualSize := partInfo.Size
|
||||||
|
if oi.IsCompressed() {
|
||||||
|
actualSize = partInfo.ActualSize
|
||||||
|
}
|
||||||
|
hr, err := hash.NewReader(ctx, io.LimitReader(gr, partInfo.Size), partInfo.Size, "", "", actualSize)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return setRestoreHeaderFn(oi, err)
|
return setRestoreHeaderFn(oi, err)
|
||||||
}
|
}
|
||||||
pInfo, err := er.PutObjectPart(ctx, bucket, object, res.UploadID, partInfo.Number, NewPutObjReader(hr), ObjectOptions{})
|
pInfo, err := er.PutObjectPart(ctx, bucket, object, res.UploadID, partInfo.Number, NewPutObjReader(hr), ObjectOptions{
|
||||||
|
IndexCB: func() []byte { return partInfo.Index },
|
||||||
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return setRestoreHeaderFn(oi, err)
|
return setRestoreHeaderFn(oi, err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1658,8 +1658,12 @@ func (z *erasureServerPools) CopyObject(ctx context.Context, srcBucket, srcObjec
|
|||||||
// Restore status updates address an existing version. Keep ordinary
|
// Restore status updates address an existing version. Keep ordinary
|
||||||
// copy/write allocation unchanged, including its latest-version lookup.
|
// copy/write allocation unchanged, including its latest-version lookup.
|
||||||
poolIdx, err = z.getPoolIdxExistingWithOpts(ctx, dstBucket, dstObject, ObjectOptions{
|
poolIdx, err = z.getPoolIdxExistingWithOpts(ctx, dstBucket, dstObject, ObjectOptions{
|
||||||
VersionID: dstOpts.VersionID, MetadataChg: true, NoLock: true,
|
VersionID: dstOpts.VersionID,
|
||||||
SkipDecommissioned: true, SkipRebalancing: true,
|
MetadataChg: true,
|
||||||
|
NoLock: true,
|
||||||
|
Transition: dstOpts.Transition,
|
||||||
|
SkipDecommissioned: true,
|
||||||
|
SkipRebalancing: true,
|
||||||
})
|
})
|
||||||
} else {
|
} else {
|
||||||
poolIdx, err = z.getPoolIdxNoLock(ctx, dstBucket, dstObject, srcInfo.Size)
|
poolIdx, err = z.getPoolIdxNoLock(ctx, dstBucket, dstObject, srcInfo.Size)
|
||||||
|
|||||||
@@ -3953,6 +3953,8 @@ func (api objectAPIHandlers) PostRestoreObjectHandler(w http.ResponseWriter, r *
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Preserve a failed restore read even before the request body is parsed.
|
||||||
|
opts.Transition.RestoreRequest = &RestoreObjectRequest{}
|
||||||
objInfo, err := getObjectInfo(ctx, bucket, object, opts)
|
objInfo, err := getObjectInfo(ctx, bucket, object, opts)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
|
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
|
||||||
@@ -3978,6 +3980,7 @@ func (api objectAPIHandlers) PostRestoreObjectHandler(w http.ResponseWriter, r *
|
|||||||
writeErrorResponse(ctx, w, apiErr, r.URL)
|
writeErrorResponse(ctx, w, apiErr, r.URL)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
opts.Transition.RestoreRequest = rreq
|
||||||
statusCode := http.StatusOK
|
statusCode := http.StatusOK
|
||||||
alreadyRestored := false
|
alreadyRestored := false
|
||||||
if err == nil {
|
if err == nil {
|
||||||
@@ -4007,8 +4010,10 @@ func (api objectAPIHandlers) PostRestoreObjectHandler(w http.ResponseWriter, r *
|
|||||||
objInfo.UserDefined = metadata
|
objInfo.UserDefined = metadata
|
||||||
if _, err := objectAPI.CopyObject(GlobalContext, bucket, object, bucket, object, objInfo, ObjectOptions{
|
if _, err := objectAPI.CopyObject(GlobalContext, bucket, object, bucket, object, objInfo, ObjectOptions{
|
||||||
VersionID: objInfo.VersionID,
|
VersionID: objInfo.VersionID,
|
||||||
|
Transition: opts.Transition,
|
||||||
}, ObjectOptions{
|
}, ObjectOptions{
|
||||||
VersionID: objInfo.VersionID,
|
VersionID: objInfo.VersionID,
|
||||||
|
Transition: opts.Transition,
|
||||||
MTime: objInfo.ModTime,
|
MTime: objInfo.ModTime,
|
||||||
MetadataChg: true,
|
MetadataChg: true,
|
||||||
}); err != nil {
|
}); err != nil {
|
||||||
|
|||||||
@@ -57,7 +57,7 @@ func TestObjectLockOverwriteConfigUnavailable(t *testing.T) {
|
|||||||
err := checkObjectLockedForOverwrite(t.Context(), "lock-config-unavailable", "object", func(context.Context, string, string, ObjectOptions) (ObjectInfo, error) {
|
err := checkObjectLockedForOverwrite(t.Context(), "lock-config-unavailable", "object", func(context.Context, string, string, ObjectOptions) (ObjectInfo, error) {
|
||||||
t.Fatal("object must not be read without a bucket configuration")
|
t.Fatal("object must not be read without a bucket configuration")
|
||||||
return ObjectInfo{}, nil
|
return ObjectInfo{}, nil
|
||||||
})
|
}, TransitionOptions{})
|
||||||
if !errors.Is(err, errServerNotInitialized) {
|
if !errors.Is(err, errServerNotInitialized) {
|
||||||
t.Fatalf("unavailable lock configuration must fail closed, got %v", err)
|
t.Fatalf("unavailable lock configuration must fail closed, got %v", err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,248 @@
|
|||||||
|
// Copyright (c) 2026 Feng Ruohang
|
||||||
|
//
|
||||||
|
// This file is part of Silo Object Storage stack
|
||||||
|
//
|
||||||
|
// This program is free software: you can redistribute it and/or modify
|
||||||
|
// it under the terms of the GNU Affero General Public License as published by
|
||||||
|
// the Free Software Foundation, either version 3 of the License, or
|
||||||
|
// (at your option) any later version.
|
||||||
|
//
|
||||||
|
// This program is distributed in the hope that it will be useful,
|
||||||
|
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||||
|
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||||
|
// GNU Affero General Public License for more details.
|
||||||
|
//
|
||||||
|
// You should have received a copy of the GNU Affero General Public License
|
||||||
|
// along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||||
|
|
||||||
|
package cmd
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bytes"
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
|
"net/url"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/minio/minio/internal/bucket/versioning"
|
||||||
|
"github.com/minio/minio/internal/event"
|
||||||
|
"github.com/minio/minio/internal/hash"
|
||||||
|
xhttp "github.com/minio/minio/internal/http"
|
||||||
|
"github.com/minio/minio/internal/kms"
|
||||||
|
"github.com/minio/minio/internal/pubsub"
|
||||||
|
"github.com/minio/mux"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Exercise the stored representation produced by real S3 uploads. Native
|
||||||
|
// restore copies encoded bytes, but must retain logical sizes and identity.
|
||||||
|
func TestNativeRestoreEncodedObjects(t *testing.T) {
|
||||||
|
for _, pools := range []int{1, 2} {
|
||||||
|
for _, encoding := range []string{"plain", "compressed", "ssec", "s3-compressed"} {
|
||||||
|
for _, layout := range []string{"single", "multipart", "sparse-one-part"} {
|
||||||
|
t.Run(fmt.Sprintf("pools=%d/%s/%s", pools, encoding, layout), func(t *testing.T) {
|
||||||
|
r := newRestoreSafetyTest(t, pools, versioning.Enabled)
|
||||||
|
initConfigSubsystem(t.Context(), r.z)
|
||||||
|
r.object = "restore-encoded.txt"
|
||||||
|
previousKMS, previousTLS := GlobalKMS, globalIsTLS
|
||||||
|
t.Cleanup(func() { GlobalKMS, globalIsTLS = previousKMS, previousTLS })
|
||||||
|
globalIsTLS = true
|
||||||
|
headers := make(map[string]string)
|
||||||
|
readHeaders := make(map[string]string)
|
||||||
|
if encoding == "compressed" || encoding == "s3-compressed" {
|
||||||
|
enableAttributesPartsCompression(t)
|
||||||
|
}
|
||||||
|
if encoding == "ssec" {
|
||||||
|
headers = ssecTestHeaders(0x53)
|
||||||
|
readHeaders = ssecTestHeaders(0x53)
|
||||||
|
}
|
||||||
|
if encoding == "s3-compressed" {
|
||||||
|
var err error
|
||||||
|
GlobalKMS, err = kms.NewBuiltin("restore-test-key", bytes.Repeat([]byte{0x59}, 32))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
headers[xhttp.AmzServerSideEncryption] = xhttp.AmzEncryptionAES
|
||||||
|
}
|
||||||
|
previousObjectAPI := newObjectLayerFn()
|
||||||
|
t.Cleanup(func() { setObjectLayer(previousObjectAPI) })
|
||||||
|
router := initTestAPIEndPoints(r.z, []string{"NewMultipart", "PutObjectPart", "CompleteMultipart", "PutObject", "GetObject"})
|
||||||
|
bodies := [][]byte{bytes.Repeat([]byte("encoded-restore-"), compMinIndexSize/16+1)}
|
||||||
|
partNumbers := []int{1}
|
||||||
|
switch layout {
|
||||||
|
case "multipart":
|
||||||
|
bodies = append(bodies, []byte("last encoded part"))
|
||||||
|
partNumbers = []int{1, 2}
|
||||||
|
case "sparse-one-part":
|
||||||
|
partNumbers = []int{5}
|
||||||
|
}
|
||||||
|
want := bytes.Join(bodies, nil)
|
||||||
|
if layout == "single" {
|
||||||
|
headers[xhttp.AmzChecksumCRC32] = mustChecksum(t, hash.ChecksumCRC32, want)
|
||||||
|
putCopyChecksumSource(t, router, globalActiveCred, r.bucket, r.object, want, headers)
|
||||||
|
} else {
|
||||||
|
attributesPartsUpload(t, router, globalActiveCred, r.bucket, r.object, headers, bodies, partNumbers)
|
||||||
|
}
|
||||||
|
before, err := r.z.GetObjectInfo(t.Context(), r.bucket, r.object, ObjectOptions{})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if before.IsCompressed() != (encoding == "compressed" || encoding == "s3-compressed") {
|
||||||
|
t.Fatalf("unexpected compression state: %v", before.IsCompressed())
|
||||||
|
}
|
||||||
|
if before.IsCompressed() && len(before.Parts[0].Index) == 0 {
|
||||||
|
t.Fatal("fixture must have a compression index")
|
||||||
|
}
|
||||||
|
if layout == "single" && len(before.Checksum) == 0 {
|
||||||
|
t.Fatal("fixture must have a stored checksum")
|
||||||
|
}
|
||||||
|
objectURL := getGetObjectURL("", r.bucket, r.object) + "?versionId=" + url.QueryEscape(before.VersionID)
|
||||||
|
beforeGet := attributesPartsSignedRequest(t, router, globalActiveCred, http.MethodGet, objectURL, nil, readHeaders)
|
||||||
|
if beforeGet.Code != http.StatusOK || !bytes.Equal(beforeGet.Body.Bytes(), want) {
|
||||||
|
t.Fatalf("fixture GET: %d bytes=%d want=%d", beforeGet.Code, beforeGet.Body.Len(), len(want))
|
||||||
|
}
|
||||||
|
beforeETags := beforeGet.Header()[xhttp.ETag]
|
||||||
|
if len(beforeETags) != 1 || beforeETags[0] == "" {
|
||||||
|
t.Fatal("fixture GET must expose a nonempty ETag")
|
||||||
|
}
|
||||||
|
beforeRaw := nativeRestoreSnapshot(t, r)
|
||||||
|
topts := r.opts(before.VersionID)
|
||||||
|
topts.MTime = before.ModTime
|
||||||
|
topts.Transition = TransitionOptions{Tier: "WARM", ETag: before.ETag}
|
||||||
|
if err := r.z.TransitionObject(t.Context(), r.bucket, r.object, topts); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
observed := nativeRestoreHTTPObserver{r.z, make(chan nativeRestoreHTTPResult, 1)}
|
||||||
|
api := objectAPIHandlers{ObjectAPI: func() ObjectLayer { return observed }}
|
||||||
|
events, done := make(chan event.Event, 1), make(chan struct{})
|
||||||
|
defer close(done)
|
||||||
|
if err := globalHTTPListen.Subscribe(pubsub.MaskFromMaskable(event.ObjectRestoreCompleted), events, done, nil); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
restoreXML := []byte(`<RestoreRequest xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Days>1</Days></RestoreRequest>`)
|
||||||
|
req, err := newTestSignedRequestV4(http.MethodPost, objectURL+"&restore", int64(len(restoreXML)), bytes.NewReader(restoreXML), globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if err := req.ParseForm(); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
req = mux.SetURLVars(req, map[string]string{"bucket": r.bucket, "object": r.object})
|
||||||
|
rec := httptest.NewRecorder()
|
||||||
|
api.PostRestoreObjectHandler(rec, req)
|
||||||
|
if rec.Code != http.StatusOK {
|
||||||
|
t.Fatalf("restore POST: %d %s", rec.Code, rec.Body.String())
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case result := <-observed.done:
|
||||||
|
if result.err != nil {
|
||||||
|
t.Fatalf("background restore: %v", result.err)
|
||||||
|
}
|
||||||
|
case <-time.After(5 * time.Second):
|
||||||
|
t.Fatal("restore did not finish")
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case <-events:
|
||||||
|
case <-time.After(5 * time.Second):
|
||||||
|
t.Fatal("restore completion event missing")
|
||||||
|
}
|
||||||
|
after, err := r.z.GetObjectInfo(t.Context(), r.bucket, r.object, r.opts(before.VersionID))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if after.ETag != before.ETag || after.Size != before.Size || !after.ModTime.Equal(before.ModTime) {
|
||||||
|
t.Error("restore changed content identity")
|
||||||
|
}
|
||||||
|
afterRaw := nativeRestoreSnapshot(t, r)
|
||||||
|
for pool := range beforeRaw {
|
||||||
|
if len(afterRaw[pool]) != len(beforeRaw[pool]) {
|
||||||
|
t.Fatalf("pool %d version count changed", pool)
|
||||||
|
}
|
||||||
|
for i, fi := range beforeRaw[pool] {
|
||||||
|
if !bytes.Equal(fi.Checksum, afterRaw[pool][i].Checksum) {
|
||||||
|
t.Error("restore lost stored checksum")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if len(after.Parts) != len(before.Parts) {
|
||||||
|
t.Fatal("restore changed part count")
|
||||||
|
}
|
||||||
|
for i, part := range after.Parts {
|
||||||
|
old := before.Parts[i]
|
||||||
|
if part.Number != old.Number || part.Size != old.Size || part.ActualSize != old.ActualSize || !bytes.Equal(part.Index, old.Index) {
|
||||||
|
t.Errorf("part %d layout or compression index changed", i)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
tier := globalTierConfigMgr.drivercache["WARM"].(*restoreTestTier)
|
||||||
|
reads := tier.gets.Load()
|
||||||
|
got := attributesPartsSignedRequest(t, router, globalActiveCred, http.MethodGet, objectURL, nil, readHeaders)
|
||||||
|
if got.Code != http.StatusOK || !bytes.Equal(got.Body.Bytes(), want) {
|
||||||
|
t.Errorf("restored GET: %d bytes=%d want=%d", got.Code, got.Body.Len(), len(want))
|
||||||
|
}
|
||||||
|
afterETags := got.Header()[xhttp.ETag]
|
||||||
|
if len(afterETags) != 1 || afterETags[0] == "" || afterETags[0] != beforeETags[0] {
|
||||||
|
t.Error("restore changed client ETag")
|
||||||
|
}
|
||||||
|
for i, number := range partNumbers {
|
||||||
|
part := attributesPartsSignedRequest(t, router, globalActiveCred, http.MethodGet, objectURL+fmt.Sprintf("&partNumber=%d", number), nil, readHeaders)
|
||||||
|
if part.Code != http.StatusPartialContent || !bytes.Equal(part.Body.Bytes(), bodies[i]) {
|
||||||
|
t.Errorf("restored part %d: %d bytes=%d want=%d", number, part.Code, part.Body.Len(), len(bodies[i]))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if tier.gets.Load() != reads {
|
||||||
|
t.Error("restored GET still used remote tier")
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// XL metadata predating compression has no ActualSize. Conversion and healing
|
||||||
|
// preserve those zero values; a plain multipart restore must derive the size.
|
||||||
|
func TestNativeRestoreLegacyPartSizes(t *testing.T) {
|
||||||
|
r := newRestoreSafetyTest(t, 1, "")
|
||||||
|
oi, body := nativeRestoreSeed(t, r, 0, true, true)
|
||||||
|
for _, disk := range r.z.serverPools[0].getHashedSet(r.object).getDisks() {
|
||||||
|
fi, err := disk.ReadVersion(t.Context(), r.bucket, r.bucket, r.object, oi.VersionID, ReadOptions{})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
for i := range fi.Parts {
|
||||||
|
fi.Parts[i].ActualSize = 0
|
||||||
|
}
|
||||||
|
delete(fi.Metadata, ReservedMetadataPrefix+"actual-size")
|
||||||
|
if err := disk.WriteMetadata(t.Context(), "", r.bucket, r.object, fi); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
before, err := r.z.GetObjectInfo(t.Context(), r.bucket, r.object, r.opts(oi.VersionID))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
for _, part := range before.Parts {
|
||||||
|
if part.ActualSize != 0 {
|
||||||
|
t.Fatal("legacy fixture was normalized")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
_, got, err := r.read(oi.VersionID)
|
||||||
|
if err != nil || !bytes.Equal(got, body) {
|
||||||
|
t.Fatalf("legacy fixture read: %v", err)
|
||||||
|
}
|
||||||
|
ctx, cancel := context.WithTimeout(t.Context(), 3*time.Second)
|
||||||
|
defer cancel()
|
||||||
|
if err := r.z.RestoreTransitionedObject(ctx, r.bucket, r.object, nativeRestoreOptions(r, oi.VersionID)); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
tier := globalTierConfigMgr.drivercache["WARM"].(*restoreTestTier)
|
||||||
|
reads := tier.gets.Load()
|
||||||
|
_, got, err = r.read(oi.VersionID)
|
||||||
|
if err != nil || !bytes.Equal(got, body) {
|
||||||
|
t.Fatalf("legacy restored read: %v", err)
|
||||||
|
}
|
||||||
|
if tier.gets.Load() != reads {
|
||||||
|
t.Error("legacy restored read used the remote tier")
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -28,9 +28,11 @@ import (
|
|||||||
"net/http"
|
"net/http"
|
||||||
"net/http/httptest"
|
"net/http/httptest"
|
||||||
"net/url"
|
"net/url"
|
||||||
|
"path/filepath"
|
||||||
"reflect"
|
"reflect"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
|
"sync/atomic"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -742,3 +744,193 @@ func TestMultipartCompletionLockAlias(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type nativeRestoreReadHook struct {
|
||||||
|
io.ReadCloser
|
||||||
|
remaining int
|
||||||
|
mutate func()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *nativeRestoreReadHook) Read(p []byte) (int, error) {
|
||||||
|
n, err := r.ReadCloser.Read(p)
|
||||||
|
r.remaining -= n
|
||||||
|
if r.remaining == 0 && r.mutate != nil {
|
||||||
|
f := r.mutate
|
||||||
|
r.mutate = nil
|
||||||
|
f()
|
||||||
|
}
|
||||||
|
return n, err
|
||||||
|
}
|
||||||
|
|
||||||
|
type nativeRestoreReadHookTier struct {
|
||||||
|
WarmBackend
|
||||||
|
size int
|
||||||
|
mutate func()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (b nativeRestoreReadHookTier) Get(ctx context.Context, object string, version remoteVersionID, opts WarmBackendGetOpts) (io.ReadCloser, error) {
|
||||||
|
rd, err := b.WarmBackend.Get(ctx, object, version, opts)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return &nativeRestoreReadHook{rd, b.size, b.mutate}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Remove only the addressed entry from a majority of disks, keeping shards
|
||||||
|
// and the healthy newer version. Return errors to the synchronous caller.
|
||||||
|
func nativeRestoreLoseMetadata(ctx context.Context, disks []StorageAPI, bucket, object, id string) error {
|
||||||
|
for _, disk := range disks[1:] {
|
||||||
|
raw, err := disk.ReadXL(ctx, bucket, object, true)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
var meta xlMetaV2
|
||||||
|
if err := meta.Load(raw.Buf); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
del := FileInfo{VersionID: id}
|
||||||
|
del.SetSkipTierFreeVersion()
|
||||||
|
if _, err := meta.DeleteVersion(del); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
data, err := meta.AppendTo(nil)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if err := disk.WriteAll(ctx, bucket, filepath.Join(object, xlStorageFormatFile), data); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// A target can lose metadata quorum after the restore snapshot. Nested
|
||||||
|
// completion/retention reads must not turn rehydration into remote GC.
|
||||||
|
func TestNativeRestoreCompletionReadSafety(t *testing.T) {
|
||||||
|
for _, null := range []bool{false, true} {
|
||||||
|
for _, multipart := range []bool{false, true} {
|
||||||
|
t.Run(fmt.Sprintf("null=%v/multipart=%v", null, multipart), func(t *testing.T) {
|
||||||
|
r := newRestoreSafetyTest(t, 1, versioning.Enabled)
|
||||||
|
oi, body := nativeRestoreSeed(t, r, 0, null, multipart)
|
||||||
|
latest := r.put("newer", false, false)
|
||||||
|
enableBucketObjectLock(t, r.bucket)
|
||||||
|
disks := r.z.serverPools[0].getHashedSet(r.object).getDisks()
|
||||||
|
var mutationErr error
|
||||||
|
mutated := false
|
||||||
|
mutate := func() {
|
||||||
|
mutated = true
|
||||||
|
mutationErr = nativeRestoreLoseMetadata(t.Context(), disks, r.bucket, r.object, oi.VersionID)
|
||||||
|
}
|
||||||
|
tier := globalTierConfigMgr.drivercache["WARM"].(*restoreTestTier)
|
||||||
|
globalTierConfigMgr.drivercache["WARM"] = nativeRestoreReadHookTier{tier, len(body), mutate}
|
||||||
|
ctx, cancel := context.WithTimeout(t.Context(), 3*time.Second)
|
||||||
|
defer cancel()
|
||||||
|
err := r.z.RestoreTransitionedObject(ctx, r.bucket, r.object, nativeRestoreOptions(r, oi.VersionID))
|
||||||
|
if ctx.Err() != nil {
|
||||||
|
t.Fatalf("restore timed out: %v", err)
|
||||||
|
}
|
||||||
|
if !mutated || mutationErr != nil {
|
||||||
|
t.Fatalf("invalid injection: mutated=%v err=%v", mutated, mutationErr)
|
||||||
|
}
|
||||||
|
t.Logf("restore after addressed metadata lost: %v", err)
|
||||||
|
for i, disk := range disks {
|
||||||
|
raw, e := disk.ReadXL(t.Context(), r.bucket, r.object, true)
|
||||||
|
if e != nil {
|
||||||
|
t.Fatal(e)
|
||||||
|
}
|
||||||
|
versions, e := getFileInfoVersions(raw.Buf, r.bucket, r.object, false)
|
||||||
|
if e != nil {
|
||||||
|
t.Fatal(e)
|
||||||
|
}
|
||||||
|
if len(versions.FreeVersions) > 0 {
|
||||||
|
t.Errorf("disk %d restore manufactured %d remote GC entries", i, len(versions.FreeVersions))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if err == nil {
|
||||||
|
reads := tier.gets.Load()
|
||||||
|
_, got, readErr := r.read(oi.VersionID)
|
||||||
|
if readErr != nil || !bytes.Equal(got, body) || tier.gets.Load() != reads {
|
||||||
|
t.Errorf("successful restore did not rehydrate its own content: %v", readErr)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
_, got, readErr := r.read(latest.VersionID)
|
||||||
|
if readErr != nil || !bytes.Equal(got, r.bodies["newer"]) {
|
||||||
|
t.Errorf("restore changed the other version: %v", readErr)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
type nativeRestoreStatusHook struct {
|
||||||
|
ObjectLayer
|
||||||
|
beforeCopy func()
|
||||||
|
copyCalled bool
|
||||||
|
backgroundCalled atomic.Bool
|
||||||
|
}
|
||||||
|
|
||||||
|
func (h *nativeRestoreStatusHook) CopyObject(ctx context.Context, srcBucket, srcObject, dstBucket, dstObject string, info ObjectInfo, srcOpts, dstOpts ObjectOptions) (ObjectInfo, error) {
|
||||||
|
h.copyCalled = true
|
||||||
|
h.beforeCopy()
|
||||||
|
return h.ObjectLayer.CopyObject(ctx, srcBucket, srcObject, dstBucket, dstObject, info, srcOpts, dstOpts)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (h *nativeRestoreStatusHook) RestoreTransitionedObject(context.Context, string, string, ObjectOptions) error {
|
||||||
|
h.backgroundCalled.Store(true)
|
||||||
|
return errors.New("unexpected background restore after failed status update")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestNativeRestorePOSTReadSafety(t *testing.T) {
|
||||||
|
for _, pools := range []int{1, 2} {
|
||||||
|
for _, stage := range []string{"initial", "status-copy"} {
|
||||||
|
t.Run(fmt.Sprintf("pools=%d/%s", pools, stage), func(t *testing.T) {
|
||||||
|
r := newRestoreSafetyTest(t, pools, versioning.Enabled)
|
||||||
|
initConfigSubsystem(t.Context(), r.z)
|
||||||
|
oi, _ := nativeRestoreSeed(t, r, 0, false, false)
|
||||||
|
metaOpts := r.opts(oi.VersionID)
|
||||||
|
metaOpts.MTime = oi.ModTime
|
||||||
|
metaOpts.EvalMetadataFn = func(o *ObjectInfo, _ error) (ReplicateDecision, error) {
|
||||||
|
o.UserDefined[xhttp.AmzRestore] = ""
|
||||||
|
return ReplicateDecision{}, nil
|
||||||
|
}
|
||||||
|
if _, err := r.z.PutObjectMetadata(t.Context(), r.bucket, r.object, metaOpts); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
r.put("newer", false, false)
|
||||||
|
disks := r.z.serverPools[0].getHashedSet(r.object).getDisks()
|
||||||
|
var before [][]byte
|
||||||
|
mutate := func() {
|
||||||
|
if err := nativeRestoreLoseMetadata(t.Context(), disks, r.bucket, r.object, oi.VersionID); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
before = restoreBoundarySnapshot(t, r)
|
||||||
|
}
|
||||||
|
hook := &nativeRestoreStatusHook{ObjectLayer: r.z, beforeCopy: mutate}
|
||||||
|
if stage == "initial" {
|
||||||
|
mutate()
|
||||||
|
}
|
||||||
|
api := objectAPIHandlers{ObjectAPI: func() ObjectLayer { return hook }}
|
||||||
|
body := []byte(`<RestoreRequest xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Days>1</Days></RestoreRequest>`)
|
||||||
|
req, err := newTestSignedRequestV4(http.MethodPost, "/"+r.bucket+"/"+r.object+"?restore&versionId="+url.QueryEscape(oi.VersionID), int64(len(body)), bytes.NewReader(body), globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if err := req.ParseForm(); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
req = mux.SetURLVars(req, map[string]string{"bucket": r.bucket, "object": r.object})
|
||||||
|
rec := httptest.NewRecorder()
|
||||||
|
api.PostRestoreObjectHandler(rec, req)
|
||||||
|
if rec.Code < 400 {
|
||||||
|
t.Fatalf("POST acknowledged failed read: %d %s", rec.Code, rec.Body.String())
|
||||||
|
}
|
||||||
|
if hook.copyCalled != (stage == "status-copy") || hook.backgroundCalled.Load() {
|
||||||
|
t.Fatal("request followed the wrong failure path")
|
||||||
|
}
|
||||||
|
if !reflect.DeepEqual(before, restoreBoundarySnapshot(t, r)) {
|
||||||
|
t.Error("failed POST read changed stored metadata or free versions")
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user